Merge pull request #19 from seqsense/add-con-lost-cb

Add connection lost handler

.codecov.yml

@@ -0,0 +1,2 @@
+ignore:
+  - "sample"

connection.go

@@ -84,6 +84,9 @@ func connectionHandler(c *DeviceClient) {
 					c.reconnectPeriod = c.opt.MaximumReconnectTime
 				}
 				log.Printf("Trying to reconnect (%d ms)\n", c.reconnectPeriod/time.Millisecond)
+				if c.opt.OnConnectionLost != nil {
+					c.opt.OnConnectionLost(c.opt)
+				}
 
 				go func() {
 					time.Sleep(c.reconnectPeriod)

connection_test.go

@@ -164,6 +164,49 @@ func TestDeviceClient(t *testing.T) {
 	}
 }
 
+func TestConnectionLostHandler(t *testing.T) {
+	newClient = func(opt *mqtt.ClientOptions) mqtt.Client {
+		return &MockClient{}
+	}
+	connectionLostHandled := false
+	o := &Options{
+		BaseReconnectTime:        time.Millisecond * 10,
+		MaximumReconnectTime:     time.Millisecond * 50,
+		MinimumConnectionTime:    time.Millisecond * 50,
+		Keepalive:                time.Second * 2,
+		Url:                      "mock://",
+		OfflineQueueing:          true,
+		OfflineQueueMaxSize:      100,
+		OfflineQueueDropBehavior: "oldest",
+		AutoResubscribe:          true,
+		OnConnectionLost: func(opt *Options) {
+			connectionLostHandled = true
+		},
+	}
+	cli := New(o)
+	cli.Connect()
+
+	// Establish connection
+	cli.mqttOpt.OnConnect(cli.cli)
+	time.Sleep(time.Millisecond * 100)
+
+	if connectionLostHandled != false {
+		t.Fatal("ConnectionLostHandler is called before connection lost")
+	}
+
+	// Disconnect
+	cli.mqttOpt.OnConnectionLost(cli.cli, errors.New("Disconnect test"))
+	time.Sleep(time.Millisecond * 100)
+
+	// Establish connection
+	cli.mqttOpt.OnConnect(cli.cli)
+	time.Sleep(time.Millisecond * 100)
+
+	if connectionLostHandled != true {
+		t.Fatal("ConnectionLostHandler was not called on connection lost")
+	}
+}
+
 type MockMessage struct {
 	topic   string
 	payload []byte

options.go

@@ -23,6 +23,8 @@ type TopicPayload struct {
 	Payload string
 }
 
+type ConnectionLostHandler func(*Options)
+
 type Options struct {
 	KeyPath                  string
 	CertPath                 string
@@ -44,4 +46,8 @@ type Options struct {
 	OfflineQueueMaxSize      int
 	OfflineQueueDropBehavior string
 	AutoResubscribe          bool
+
+	// OnConnectionLost is called if the MQTT connection is lost.
+	// Pointer to the Options passed as the argument can be modified for the next reconnect.
+	OnConnectionLost ConnectionLostHandler
 }

sample/main.go

@@ -70,6 +70,9 @@ func main() {
 		OfflineQueueMaxSize:      100,
 		OfflineQueueDropBehavior: "oldest",
 		AutoResubscribe:          true,
+		OnConnectionLost: func(opt *awsiot.Options) {
+			fmt.Printf("Connection lost handler function called\n")
+		},
 	}
 	cli := awsiot.New(o)
 	cli.Connect()