func (c *client) internalConnLost(err error) {
	DEBUG.Println(CLI, "internalConnLost called")
	stopDone := c.stopCommsWorkers()
	if stopDone != nil {  
		go func() {
			DEBUG.Println(CLI, "internalConnLost waiting on workers")
			<-stopDone
			DEBUG.Println(CLI, "internalConnLost workers stopped")
			reconnect := c.options.AutoReconnect && c.connectionStatus() > connecting
			if c.options.CleanSession && !reconnect {
				c.messageIds.cleanUp()
			}
			if reconnect {
				c.setConnected(reconnecting)
				go c.reconnect()
			} else {
				c.setConnected(disconnected)
			}
			if c.options.OnConnectionLost != nil {
				go c.options.OnConnectionLost(c, err)
			}
			DEBUG.Println(CLI, "internalConnLost complete")
		}()
	}
}
internalConnLost cleanup when connection is lost or an error occurs
Note: This function will not block
package mqtt
import (
	"bytes"
	"errors"
	"fmt"
	"net"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"github.com/eclipse/paho.mqtt.golang/packets"
)
const (
	disconnected uint32 = iota
	connecting
	reconnecting
	connected
)
type Client interface {
	IsConnected() bool
	IsConnectionOpen() bool
	Connect() Token
	Disconnect(quiesce uint)
	Publish(topic string, qos byte, retained bool, payload interface{}) Token
	Subscribe(topic string, qos byte, callback MessageHandler) Token
	SubscribeMultiple(filters map[string]byte, callback MessageHandler) Token
	Unsubscribe(topics ...string) Token
	AddRoute(topic string, callback MessageHandler)
	OptionsReader() ClientOptionsReader
}
type client struct {
	lastSent        atomic.Value  
	lastReceived    atomic.Value  
	pingOutstanding int32         
	status       uint32  
	sync.RWMutex         
	messageIds  
	obound    chan *PacketAndToken  
	oboundP   chan *PacketAndToken  
	msgRouter *router               
	persist   Store
	options   ClientOptions
	optionsMu sync.Mutex  
	conn   net.Conn    
	connMu sync.Mutex  
	stop         chan struct{}   
	workers      sync.WaitGroup  
	commsStopped chan struct{}   
}
func NewClient(o *ClientOptions) Client {
	c := &client{}
	c.options = *o
	if c.options.Store == nil {
		c.options.Store = NewMemoryStore()
	}
	switch c.options.ProtocolVersion {
	case 3, 4:
		c.options.protocolVersionExplicit = true
	case 0x83, 0x84:
		c.options.protocolVersionExplicit = true
	default:
		c.options.ProtocolVersion = 4
		c.options.protocolVersionExplicit = false
	}
	c.persist = c.options.Store
	c.status = disconnected
	c.messageIds = messageIds{index: make(map[uint16]tokenCompletor)}
	c.msgRouter = newRouter()
	c.msgRouter.setDefaultHandler(c.options.DefaultPublishHandler)
	c.obound = make(chan *PacketAndToken)
	c.oboundP = make(chan *PacketAndToken)
	return c
}
func (c *client) AddRoute(topic string, callback MessageHandler) {
	if callback != nil {
		c.msgRouter.addRoute(topic, callback)
	}
}
func (c *client) IsConnected() bool {
	c.RLock()
	defer c.RUnlock()
	status := atomic.LoadUint32(&c.status)
	switch {
	case status == connected:
		return true
	case c.options.AutoReconnect && status > connecting:
		return true
	case c.options.ConnectRetry && status == connecting:
		return true
	default:
		return false
	}
}
func (c *client) IsConnectionOpen() bool {
	c.RLock()
	defer c.RUnlock()
	status := atomic.LoadUint32(&c.status)
	switch {
	case status == connected:
		return true
	default:
		return false
	}
}
func (c *client) connectionStatus() uint32 {
	c.RLock()
	defer c.RUnlock()
	status := atomic.LoadUint32(&c.status)
	return status
}
func (c *client) setConnected(status uint32) {
	c.Lock()
	defer c.Unlock()
	atomic.StoreUint32(&c.status, status)
}
var ErrNotConnected = errors.New("not Connected")
func (c *client) Connect() Token {
	t := newToken(packets.Connect).(*ConnectToken)
	DEBUG.Println(CLI, "Connect()")
	if c.options.ConnectRetry && atomic.LoadUint32(&c.status) != disconnected {
		WARN.Println(CLI, "Connect() called but not disconnected")
		t.returnCode = packets.Accepted
		t.flowComplete()
		return t
	}
	c.persist.Open()
	if c.options.ConnectRetry {
		c.reserveStoredPublishIDs()  
	}
	c.setConnected(connecting)
	go func() {
		if len(c.options.Servers) == 0 {
			t.setError(fmt.Errorf("no servers defined to connect to"))
			return
		}
	RETRYCONN:
		var conn net.Conn
		var rc byte
		var err error
		conn, rc, t.sessionPresent, err = c.attemptConnection()
		if err != nil {
			if c.options.ConnectRetry {
				DEBUG.Println(CLI, "Connect failed, sleeping for", int(c.options.ConnectRetryInterval.Seconds()), "seconds and will then retry")
				time.Sleep(c.options.ConnectRetryInterval)
				if atomic.LoadUint32(&c.status) == connecting {
					goto RETRYCONN
				}
			}
			ERROR.Println(CLI, "Failed to connect to a broker")
			c.setConnected(disconnected)
			c.persist.Close()
			t.returnCode = rc
			t.setError(err)
			return
		}
		inboundFromStore := make(chan packets.ControlPacket)  
		if c.startCommsWorkers(conn, inboundFromStore) {
			if !c.options.CleanSession {
				c.resume(c.options.ResumeSubs, inboundFromStore)
			} else {
				c.persist.Reset()
			}
		} else {
			WARN.Println(CLI, "Connect() called but connection established in another goroutine")
		}
		close(inboundFromStore)
		t.flowComplete()
		DEBUG.Println(CLI, "exit startClient")
	}()
	return t
}
func (c *client) reconnect() {
	DEBUG.Println(CLI, "enter reconnect")
	var (
		sleep = 1 * time.Second
		conn  net.Conn
	)
	for {
		if nil != c.options.OnReconnecting {
			c.options.OnReconnecting(c, &c.options)
		}
		var err error
		conn, _, _, err = c.attemptConnection()
		if err == nil {
			break
		}
		DEBUG.Println(CLI, "Reconnect failed, sleeping for", int(sleep.Seconds()), "seconds:", err)
		time.Sleep(sleep)
		if sleep < c.options.MaxReconnectInterval {
			sleep *= 2
		}
		if sleep > c.options.MaxReconnectInterval {
			sleep = c.options.MaxReconnectInterval
		}
		if atomic.LoadUint32(&c.status) == disconnected {
			break
		}
	}
	if c.connectionStatus() == disconnected {
		if conn != nil {
			conn.Close()
		}
		DEBUG.Println(CLI, "Client moved to disconnected state while reconnecting, abandoning reconnect")
		return
	}
	inboundFromStore := make(chan packets.ControlPacket)  
	if c.startCommsWorkers(conn, inboundFromStore) {
		c.resume(c.options.ResumeSubs, inboundFromStore)
	}
	close(inboundFromStore)
}
func (c *client) attemptConnection() (net.Conn, byte, bool, error) {
	protocolVersion := c.options.ProtocolVersion
	var (
		sessionPresent bool
		conn           net.Conn
		err            error
		rc             byte
	)
	c.optionsMu.Lock()  
	brokers := c.options.Servers
	c.optionsMu.Unlock()
	for _, broker := range brokers {
		cm := newConnectMsgFromOptions(&c.options, broker)
		DEBUG.Println(CLI, "about to write new connect msg")
	CONN:
		conn, err = openConnection(broker, c.options.TLSConfig, c.options.ConnectTimeout, c.options.HTTPHeaders, c.options.WebsocketOptions)
		if err != nil {
			ERROR.Println(CLI, err.Error())
			WARN.Println(CLI, "failed to connect to broker, trying next")
			rc = packets.ErrNetworkError
			continue
		}
		DEBUG.Println(CLI, "socket connected to broker")
		rc, sessionPresent, err = connectMQTT(conn, cm, protocolVersion)
		if rc == packets.Accepted {
			break  
		}
		if conn != nil {
			conn.Close()
		}
		if !c.options.protocolVersionExplicit && protocolVersion == 4 {  
			DEBUG.Println(CLI, "Trying reconnect using MQTT 3.1 protocol")
			protocolVersion = 3
			goto CONN
		}
		if c.options.protocolVersionExplicit {  
			ERROR.Println(CLI, "Connecting to", broker, "CONNACK was not CONN_ACCEPTED, but rather", packets.ConnackReturnCodes[rc])
		}
	}
	if rc == packets.Accepted {
		c.options.ProtocolVersion = protocolVersion
		c.options.protocolVersionExplicit = true
	} else {
		if rc != packets.ErrNetworkError {  
			err = packets.ConnErrors[rc]
		} else {  
			err = fmt.Errorf("%s : %s", packets.ConnErrors[rc], err)
		}
	}
	return conn, rc, sessionPresent, err
}
func (c *client) Disconnect(quiesce uint) {
	status := atomic.LoadUint32(&c.status)
	if status == connected {
		DEBUG.Println(CLI, "disconnecting")
		c.setConnected(disconnected)
		dm := packets.NewControlPacket(packets.Disconnect).(*packets.DisconnectPacket)
		dt := newToken(packets.Disconnect)
		c.oboundP <- &PacketAndToken{p: dm, t: dt}
		DEBUG.Println(CLI, "calling WaitTimeout")
		dt.WaitTimeout(time.Duration(quiesce) * time.Millisecond)
		DEBUG.Println(CLI, "WaitTimeout done")
	} else {
		WARN.Println(CLI, "Disconnect() called but not connected (disconnected/reconnecting)")
		c.setConnected(disconnected)
	}
	c.disconnect()
}
func (c *client) forceDisconnect() {
	if !c.IsConnected() {
		WARN.Println(CLI, "already disconnected")
		return
	}
	c.setConnected(disconnected)
	DEBUG.Println(CLI, "forcefully disconnecting")
	c.disconnect()
}
func (c *client) disconnect() {
	done := c.stopCommsWorkers()
	if done != nil {
		<-done  
		DEBUG.Println(CLI, "forcefully disconnecting")
		c.messageIds.cleanUp()
		DEBUG.Println(CLI, "disconnected")
		c.persist.Close()
	}
}