API updates in preparation for 1.0 release

Author: Al S-M

README.md

@@ -43,8 +43,7 @@ Samples are available in the `/samples` directory for reference.
 Runtime tracing
 ---------------
 
-Tracing is enabled by using the `SetTraceLevel` option when creating a ClientOptions struct. See the ClientOptions
-documentation for more details.
+Tracing is enabled by assigning logs (from the Go log package) to the logging endpoints, ERROR, CRITICAL, WARN and DEBUG
 
 
 Reporting bugs

client.go

@@ -72,7 +72,7 @@ type Client struct {
 	workers         sync.WaitGroup
 }
 
-// NewClient will create an MQTT v3.1 client with all of the options specified
+// NewClient will create an MQTT v3.1.1 client with all of the options specified
 // in the provided ClientOptions. The client must have the Start method called
 // on it before it may be used. This is to make sure resources (such as a net
 // connection) are created before the application is actually ready.
@@ -112,6 +112,10 @@ func (c *Client) setConnected(status bool) {
 	c.connected = status
 }
 
+//ErrNotConnected is the error returned from function calls that are
+//made when the client is not connected to a broker
+var ErrNotConnected = errors.New("Not Connected")
+
 // Connect will create a connection to the message broker
 // If clean session is false, then a slice will
 // be returned containing Receipts for all messages
@@ -165,17 +169,17 @@ func (c *Client) Connect() Token {
 			} else {
 				ERROR.Println(CLI, err.Error())
 				WARN.Println(CLI, "failed to connect to broker, trying next")
-				rc = packets.NetworkError
+				rc = packets.ErrNetworkError
 			}
 		}
 
 		if c.conn == nil {
 			ERROR.Println(CLI, "Failed to connect to a broker")
 			t.returnCode = rc
-			if rc != packets.NetworkError {
-				t.err = connErrors[rc]
+			if rc != packets.ErrNetworkError {
+				t.err = packets.ConnErrors[rc]
 			} else {
-				t.err = errors.New(connErrors[rc].Error() + " : " + err.Error())
+				t.err = errors.New(packets.ConnErrors[rc].Error() + " : " + err.Error())
 			}
 			t.flowComplete()
 			return
@@ -274,7 +278,7 @@ func (c *Client) reconnect() {
 			} else {
 				ERROR.Println(CLI, err.Error())
 				WARN.Println(CLI, "failed to connect to broker, trying next")
-				rc = packets.NetworkError
+				rc = packets.ErrNetworkError
 			}
 		}
 		if rc != 0 {
@@ -318,7 +322,7 @@ func (c *Client) connect() byte {
 	if err != nil {
 		ERROR.Println(NET, "connect got error", err)
 		//c.errors <- err
-		return packets.NetworkError
+		return packets.ErrNetworkError
 	}
 	msg := ca.(*packets.ConnackPacket)
 
@@ -474,3 +478,9 @@ func (c *Client) Unsubscribe(topics ...string) Token {
 	DEBUG.Println(CLI, "exit Unsubscribe")
 	return token
 }
+
+//DefaultConnectionLostHandler is a definition of a function that simply
+//reports to the DEBUG log the reason for the client losing a connection.
+func DefaultConnectionLostHandler(client *Client, reason error) {
+	DEBUG.Println("Connection lost:", reason.Error())
+}

fvt/README.md

@@ -65,10 +65,10 @@ Launch mosquitto from the fvt directory, specifiying mosquitto.cfg as config fil
 
 ``ex: /usr/bin/mosquitto -c ./mosquitto.cfg``
 
-Note: Mosquitto requires SSL 1.1 or better, while Go 1.1.2 (current) supports
-only SSL v1.0. However, Go 1.2.* (near release) supports SSL v1.1 and SSL v1.2.
+Note: Mosquitto requires SSL 1.1 or better, while Go 1.1.2 supports
+only SSL v1.0. However, Go 1.2+ supports SSL v1.1 and SSL v1.2.
 
 
 Other Notes
 -----------
-Go 1.1.2 does not support intermediate certificates, however Go 1.2 does.
+Go 1.1.2 does not support intermediate certificates, however Go 1.2+ does.

fvt/setup_IMA.sh

@@ -36,7 +36,6 @@ const (
 func (mids *messageIds) freeID(id uint16) {
 	mids.Lock()
 	defer mids.Unlock()
-	//trace_v(MID, "freeing message id: %v", id)
 	delete(mids.index, id)
 }
 

messageids.go

@@ -110,7 +110,7 @@ func outgoing(c *Client) {
 			msg := pub.p.(*packets.PublishPacket)
 			if msg.Qos != 0 && msg.MessageID == 0 {
 				msg.MessageID = c.getID(pub.t)
-				pub.t.(*PublishToken).messageId = msg.MessageID
+				pub.t.(*PublishToken).messageID = msg.MessageID
 			}
 			//persist_obound(c.persist, msg)
 
@@ -273,7 +273,9 @@ func alllogic(c *Client) {
 			// Call onConnectionLost or default error handler
 			if c.IsConnected() {
 				go c.options.OnConnectionLost(c, err)
-				go c.reconnect()
+				if c.options.AutoReconnect {
+					go c.reconnect()
+				}
 			}
 			return
 		}

net.go

@@ -14,45 +14,6 @@
 
 package mqtt
 
-import (
-	"errors"
-
-	"git.eclipse.org/gitroot/paho/org.eclipse.paho.mqtt.golang.git/packets"
-)
-
-/*
- * Connect Errors
- */
-var connErrors = map[byte]error{
-	packets.Accepted:                     nil,
-	packets.RefusedBadProtocolVersion:    errors.New("Unnacceptable protocol version"),
-	packets.RefusedIDRejected:            errors.New("Identifier rejected"),
-	packets.RefusedServerUnavailable:     errors.New("Server Unavailable"),
-	packets.RefusedBadUsernameOrPassword: errors.New("Bad user name or password"),
-	packets.RefusedNotAuthorised:         errors.New("Not Authorized"),
-	packets.NetworkError:                 errors.New("Network Error"),
-	packets.ProtocolViolation:            errors.New("Protocol Violation"),
-}
-
-var ErrNotConnected = errors.New("Not Connected")
-
-/*
- * Topic Errors
- */
-var ErrInvalidTopicNameEmptyString = errors.New("Invalid TopicName - may not be empty string")
-var ErrInvalidTopicNameWildcard = errors.New("Invalid TopicName - may not contain wild card")
-var ErrInvalidTopicFilterEmptyString = errors.New("Invalid TopicFilter - may not be empty string")
-var ErrInvalidTopicFilterMultilevel = errors.New("Invalid TopicFilter - multi-level wildcard must be last level")
-
-/*
- * QoS Errors
- */
-var ErrInvalidQoS = errors.New("Invalid QoS")
-
-func DefaultConnectionLostHandler(client *Client, reason error) {
-	DEBUG.Println("Connection lost:", reason.Error())
-}
-
 func chkerr(e error) {
 	if e != nil {
 		panic(e)

oops.go

@@ -54,6 +54,7 @@ type ClientOptions struct {
 	TLSConfig               tls.Config
 	KeepAlive               time.Duration
 	MaxReconnectInterval    time.Duration
+	AutoReconnect           bool
 	Store                   Store
 	DefaultPublishHander    MessageHandler
 	OnConnect               OnConnectHandler
@@ -85,6 +86,7 @@ func NewClientOptions() *ClientOptions {
 		TLSConfig:               tls.Config{},
 		KeepAlive:               30 * time.Second,
 		MaxReconnectInterval:    10 * time.Minute,
+		AutoReconnect:           true,
 		Store:                   nil,
 		OnConnect:               nil,
 		OnConnectionLost:        DefaultConnectionLostHandler,
@@ -234,12 +236,22 @@ func (o *ClientOptions) SetConnectionLostHandler(onLost ConnectionLostHandler) *
 
 // SetWriteTimeout puts a limit on how long a mqtt publish should block until it unblocks with a
 // timeout error. A duration of 0 never times out.
-func (o *ClientOptions) SetWriteTimeout(t time.Duration) {
+func (o *ClientOptions) SetWriteTimeout(t time.Duration) *ClientOptions {
 	o.WriteTimeout = t
+	return o
 }
 
 // SetMaxReconnectInterval sets the maximum time that will be waited between reconnection attempts
 // when connection is lost
-func (o *ClientOptions) SetMaxReconnectInterval(t time.Duration) {
+func (o *ClientOptions) SetMaxReconnectInterval(t time.Duration) *ClientOptions {
 	o.MaxReconnectInterval = t
+	return o
+}
+
+// SetAutoReconnect sets whether the automatic reconnection logic should be used
+// when the connection is lost, even if disabled the ConnectionLostHandler is still
+// called
+func (o *ClientOptions) SetAutoReconnect(a bool) *ClientOptions {
+	o.AutoReconnect = a
+	return o
 }

options.go

@@ -7,8 +7,8 @@ import (
 	"io"
 )
 
-//CONNACK packet
-
+//ConnackPacket is an internal representation of the fields of the
+//Connack MQTT packet
 type ConnackPacket struct {
 	FixedHeader
 	TopicNameCompression byte
@@ -36,15 +36,22 @@ func (ca *ConnackPacket) Write(w io.Writer) error {
 	return err
 }
 
+//Unpack decodes the details of a ControlPacket after the fixed
+//header has been read
 func (ca *ConnackPacket) Unpack(b io.Reader) {
 	ca.TopicNameCompression = decodeByte(b)
 	ca.ReturnCode = decodeByte(b)
 }
 
+//Details returns a Details struct containing the Qos and
+//MessageID of this ControlPacket
 func (ca *ConnackPacket) Details() Details {
 	return Details{Qos: 0, MessageID: 0}
 }
 
+//UUID returns the unique ID assigned to the ControlPacket when
+//it was originally received. Note: this is not related to the
+//MessageID field for MQTT packets
 func (ca *ConnackPacket) UUID() uuid.UUID {
 	return ca.uuid
 }

packets/connack.go

@@ -7,8 +7,8 @@ import (
 	"io"
 )
 
-//CONNECT packet
-
+//ConnectPacket is an internal representation of the fields of the
+//Connect MQTT packet
 type ConnectPacket struct {
 	FixedHeader
 	ProtocolName    string
@@ -63,6 +63,8 @@ func (c *ConnectPacket) Write(w io.Writer) error {
 	return err
 }
 
+//Unpack decodes the details of a ControlPacket after the fixed
+//header has been read
 func (c *ConnectPacket) Unpack(b io.Reader) {
 	c.ProtocolName = decodeString(b)
 	c.ProtocolVersion = decodeByte(b)
@@ -88,32 +90,39 @@ func (c *ConnectPacket) Unpack(b io.Reader) {
 	}
 }
 
+//Validate performs validation of the fields of a Connect packet
 func (c *ConnectPacket) Validate() byte {
 	if c.PasswordFlag && !c.UsernameFlag {
-		return RefusedBadUsernameOrPassword
+		return ErrRefusedBadUsernameOrPassword
 	}
 	if c.ReservedBit != 0 {
-		fmt.Println("Bad reserved bit")
-		return ProtocolViolation
+		//Bad reserved bit
+		return ErrProtocolViolation
 	}
 	if (c.ProtocolName == "MQIsdp" && c.ProtocolVersion != 3) || (c.ProtocolName == "MQTT" && c.ProtocolVersion != 4) {
-		return RefusedBadProtocolVersion
+		//Mismatched or unsupported protocol version
+		return ErrRefusedBadProtocolVersion
 	}
 	if c.ProtocolName != "MQIsdp" && c.ProtocolName != "MQTT" {
-		fmt.Println("Bad protocol name")
-		return ProtocolViolation
+		//Bad protocol name
+		return ErrProtocolViolation
 	}
 	if len(c.ClientIdentifier) > 65535 || len(c.Username) > 65535 || len(c.Password) > 65535 {
-		fmt.Println("Bad size field")
-		return ProtocolViolation
+		//Bad size field
+		return ErrProtocolViolation
 	}
 	return Accepted
 }
 
+//Details returns a Details struct containing the Qos and
+//MessageID of this ControlPacket
 func (c *ConnectPacket) Details() Details {
 	return Details{Qos: 0, MessageID: 0}
 }
 
+//UUID returns the unique ID assigned to the ControlPacket when
+//it was originally received. Note: this is not related to the
+//MessageID field for MQTT packets
 func (c *ConnectPacket) UUID() uuid.UUID {
 	return c.uuid
 }

packets/connect.go

@@ -6,8 +6,8 @@ import (
 	"io"
 )
 
-//DISCONNECT packet
-
+//DisconnectPacket is an internal representation of the fields of the
+//Disconnect MQTT packet
 type DisconnectPacket struct {
 	FixedHeader
 	uuid uuid.UUID
@@ -25,13 +25,20 @@ func (d *DisconnectPacket) Write(w io.Writer) error {
 	return err
 }
 
+//Unpack decodes the details of a ControlPacket after the fixed
+//header has been read
 func (d *DisconnectPacket) Unpack(b io.Reader) {
 }
 
+//Details returns a Details struct containing the Qos and
+//MessageID of this ControlPacket
 func (d *DisconnectPacket) Details() Details {
 	return Details{Qos: 0, MessageID: 0}
 }
 
+//UUID returns the unique ID assigned to the ControlPacket when
+//it was originally received. Note: this is not related to the
+//MessageID field for MQTT packets
 func (d *DisconnectPacket) UUID() uuid.UUID {
 	return d.uuid
 }