ConnectionNotificationHandler - generic callback for all types of connection events.

Enables users to receive updates on connection related events (e.g. connecting, connection failed, broker connection failed etc).

Author: MattBrittan

client.go

@@ -258,12 +258,15 @@ func (c *client) Connect() Token {
 			return
 		}
 
+		var attemptCount int
+
 	RETRYCONN:
 		var conn net.Conn
 		var rc byte
 		var err error
-		conn, rc, t.sessionPresent, err = c.attemptConnection()
+		conn, rc, t.sessionPresent, err = c.attemptConnection(false, attemptCount)
 		if err != nil {
+			attemptCount++
 			if c.options.ConnectRetry {
 				DEBUG.Println(CLI, "Connect failed, sleeping for", int(c.options.ConnectRetryInterval.Seconds()), "seconds and will then retry, error:", err.Error())
 				time.Sleep(c.options.ConnectRetryInterval)
@@ -315,15 +318,17 @@ func (c *client) reconnect(connectionUp connCompletedFn) {
 		DEBUG.Println(CLI, "Detect continual connection lost after reconnect, slept for", int(slp.Seconds()), "seconds")
 	}
 
+	var attemptCount int
 	for {
 		if nil != c.options.OnReconnecting {
 			c.options.OnReconnecting(c, &c.options)
 		}
 		var err error
-		conn, _, _, err = c.attemptConnection()
+		conn, _, _, err = c.attemptConnection(true, attemptCount)
 		if err == nil {
 			break
 		}
+		attemptCount++
 		sleep, _ := c.backoff.sleepWithBackoff("attemptReconnection", initSleep, c.options.MaxReconnectInterval, c.options.ConnectTimeout, false)
 		DEBUG.Println(CLI, "Reconnect failed, slept for", int(sleep.Seconds()), "seconds:", err)
 
@@ -351,7 +356,7 @@ func (c *client) reconnect(connectionUp connCompletedFn) {
 // byte - Return code (packets.Accepted indicates a successful connection).
 // bool - SessionPresent flag from the connect ack (only valid if packets.Accepted)
 // err - Error (err != nil guarantees that conn has been set to active connection).
-func (c *client) attemptConnection() (net.Conn, byte, bool, error) {
+func (c *client) attemptConnection(isReconnect bool, attempt int) (net.Conn, byte, bool, error) {
 	protocolVersion := c.options.ProtocolVersion
 	var (
 		sessionPresent bool
@@ -360,6 +365,10 @@ func (c *client) attemptConnection() (net.Conn, byte, bool, error) {
 		rc             byte
 	)
 
+	if c.options.OnConnectionNotification != nil {
+		c.options.OnConnectionNotification(c, ConnectionNotificationConnecting{isReconnect, attempt})
+	}
+
 	c.optionsMu.Lock() // Protect c.options.Servers so that servers can be added in test cases
 	brokers := c.options.Servers
 	c.optionsMu.Unlock()
@@ -372,6 +381,9 @@ func (c *client) attemptConnection() (net.Conn, byte, bool, error) {
 			DEBUG.Println(CLI, "using custom onConnectAttempt handler...")
 			tlsCfg = c.options.OnConnectAttempt(broker, c.options.TLSConfig)
 		}
+		if c.options.OnConnectionNotification != nil {
+			c.options.OnConnectionNotification(c, ConnectionNotificationBroker{broker})
+		}
 		connDeadline := time.Now().Add(c.options.ConnectTimeout) // Time by which connection must be established
 		dialer := c.options.Dialer
 		if dialer == nil { //
@@ -388,6 +400,9 @@ func (c *client) attemptConnection() (net.Conn, byte, bool, error) {
 			ERROR.Println(CLI, err.Error())
 			WARN.Println(CLI, "failed to connect to broker, trying next")
 			rc = packets.ErrNetworkError
+			if c.options.OnConnectionNotification != nil {
+				c.options.OnConnectionNotification(c, ConnectionNotificationBrokerFailed{broker, err})
+			}
 			continue
 		}
 		DEBUG.Println(CLI, "socket connected to broker")
@@ -430,6 +445,9 @@ func (c *client) attemptConnection() (net.Conn, byte, bool, error) {
 			err = fmt.Errorf("%w : %w", packets.ConnErrors[rc], err)
 		}
 	}
+	if err != nil && c.options.OnConnectionNotification != nil {
+		c.options.OnConnectionNotification(c, ConnectionNotificationFailed{err})
+	}
 	return conn, rc, sessionPresent, err
 }
 
@@ -564,6 +582,9 @@ func (c *client) internalConnLost(whyConnLost error) {
 		if c.options.OnConnectionLost != nil {
 			go c.options.OnConnectionLost(c, whyConnLost)
 		}
+		if c.options.OnConnectionNotification != nil {
+			go c.options.OnConnectionNotification(c, ConnectionNotificationLost{whyConnLost})
+		}
 		DEBUG.Println(CLI, "internalConnLost complete")
 	}()
 }
@@ -613,6 +634,9 @@ func (c *client) startCommsWorkers(conn net.Conn, connectionUp connCompletedFn,
 	if c.options.OnConnect != nil {
 		go c.options.OnConnect(c)
 	}
+	if c.options.OnConnectionNotification != nil {
+		go c.options.OnConnectionNotification(c, ConnectionNotificationConnected{})
+	}
 
 	// c.oboundP and c.obound need to stay active for the life of the client because, depending upon the options,
 	// messages may be published while the client is disconnected (they will block unless in a goroutine). However

cmd/simple/main.go

@@ -39,6 +39,22 @@ func main() {
 	opts.SetKeepAlive(2 * time.Second)
 	opts.SetDefaultPublishHandler(f)
 	opts.SetPingTimeout(1 * time.Second)
+	opts.SetConnectionNotificationHandler(func(client mqtt.Client, notification mqtt.ConnectionNotification) {
+		switch n := notification.(type) {
+		case mqtt.ConnectionNotificationConnected:
+			fmt.Printf("[NOTIFICATION] connected\n")
+		case mqtt.ConnectionNotificationConnecting:
+			fmt.Printf("[NOTIFICATION] connecting (isReconnect=%t) [%d]\n", n.IsReconnect, n.Attempt)
+		case mqtt.ConnectionNotificationFailed:
+			fmt.Printf("[NOTIFICATION] connection failed: %v\n", n.Reason)
+		case mqtt.ConnectionNotificationLost:
+			fmt.Printf("[NOTIFICATION] connection lost: %v\n", n.Reason)
+		case mqtt.ConnectionNotificationBroker:
+			fmt.Printf("[NOTIFICATION] broker connection: %s\n", n.Broker.String())
+		case mqtt.ConnectionNotificationBrokerFailed:
+			fmt.Printf("[NOTIFICATION] broker connection failed: %v [%s]\n", n.Reason, n.Broker.String())
+		}
+	})
 
 	c := mqtt.NewClient(opts)
 	if token := c.Connect(); token.Wait() && token.Error() != nil {

connnotf.go

@@ -0,0 +1,79 @@
+package mqtt
+
+import "net/url"
+
+type ConnectionNotificationType int64
+
+const (
+	ConnectionNotificationTypeConnected ConnectionNotificationType = iota
+	ConnectionNotificationTypeConnecting
+	ConnectionNotificationTypeFailed
+	ConnectionNotificationTypeLost
+	ConnectionNotificationTypeBroker
+	ConnectionNotificationTypeBrokerFailed
+)
+
+type ConnectionNotification interface {
+	Type() ConnectionNotificationType
+}
+
+// Connected
+
+type ConnectionNotificationConnected struct {
+}
+
+func (n ConnectionNotificationConnected) Type() ConnectionNotificationType {
+	return ConnectionNotificationTypeConnected
+}
+
+// Connecting
+
+type ConnectionNotificationConnecting struct {
+	IsReconnect bool
+	Attempt     int
+}
+
+func (n ConnectionNotificationConnecting) Type() ConnectionNotificationType {
+	return ConnectionNotificationTypeConnecting
+}
+
+// Connection Failed
+
+type ConnectionNotificationFailed struct {
+	Reason error
+}
+
+func (n ConnectionNotificationFailed) Type() ConnectionNotificationType {
+	return ConnectionNotificationTypeFailed
+}
+
+// Connection Lost
+
+type ConnectionNotificationLost struct {
+	Reason error // may be nil
+}
+
+func (n ConnectionNotificationLost) Type() ConnectionNotificationType {
+	return ConnectionNotificationTypeLost
+}
+
+// Broker Connection
+
+type ConnectionNotificationBroker struct {
+	Broker *url.URL
+}
+
+func (n ConnectionNotificationBroker) Type() ConnectionNotificationType {
+	return ConnectionNotificationTypeBroker
+}
+
+// Broker Connection Failed
+
+type ConnectionNotificationBrokerFailed struct {
+	Broker *url.URL
+	Reason error
+}
+
+func (n ConnectionNotificationBrokerFailed) Type() ConnectionNotificationType {
+	return ConnectionNotificationTypeBrokerFailed
+}