Bugfix: c.stop closed twice in some startup conditions.

Sometimes net.connect sees an issue and pushes an error on c.errors before returning an rc of CONN_FAILURE. It is also possible
to exit net.connect with a non-successful rc but NOT get an error on c.errors. So I updated net.connect to not push errors on
c.errors, but instead just return CONN_FAILURE and allow client.Start to put the erorr on the channel. In addition, I made c.errors
a buffer size 1 and updated all pushes to that channel to be a select with an empty default. This allows us to use c.errors as a flag channel so that
the first error can always get through, but subsequent attempts to push errors will not block forever.

Also, the begin chan ConnRC in MqttClient was not longer being initialized so I removed it. I also removed the call to close it in net.connect, since
closing a nil channel was causing a crash.

Change-Id: I0988ad51e7740b03fff832084ea65dcbce173884
Signed-off-by: Christie Deiters <cdeiters@lutron.com>

Author: cdeiters

client.go

@@ -48,7 +48,6 @@ type MqttClient struct {
 	ibound          chan *Message
 	obound          chan sendable
 	oboundP         chan *Message
-	begin           chan ConnRC
 	errors          chan error
 	stop            chan struct{}
 	receipts        *receiptMap
@@ -116,7 +115,7 @@ func (c *MqttClient) Start() ([]Receipt, error) {
 	c.obound = make(chan sendable)
 	c.ibound = make(chan *Message)
 	c.oboundP = make(chan *Message)
-	c.errors = make(chan error)
+	c.errors = make(chan error, 1) // all we need is one error to trigger alllogic to clean up
 	c.stop = make(chan struct{})
 
 	go outgoing(c)
@@ -127,11 +126,14 @@ func (c *MqttClient) Start() ([]Receipt, error) {
 	c.oboundP <- cm
 
 	rc := connect(c)
-	if rc != CONN_ACCEPTED {
+	if chkrc(rc) != nil {
 		CRITICAL.Println(CLI, "CONNACK was not CONN_ACCEPTED, but rather", rc2str(rc))
-		// Stop all go routines except outgoing
-		close(c.stop)
-		c.conn.Close()
+		err := errors.New("CONNACK was not CONN_ACCEPTED, but rather " + rc2str(rc))
+		select {
+		case c.errors <- err:
+		default:
+			// c.errors is a buffer of one, so there must already be an error closing this connection.
+		}
 		return nil, chkrc(rc)
 	}
 
@@ -157,12 +159,7 @@ func (c *MqttClient) Start() ([]Receipt, error) {
 	go incoming(c)
 
 	DEBUG.Println(CLI, "exit startMqttClient")
-	if chkrc(rc) != nil {
-		// Cleanup before returning.
-		close(c.stop)
-		c.conn.Close()
-	}
-	return leftovers, chkrc(rc)
+	return leftovers, nil
 }
 
 // Disconnect will end the connection with the server, but not before waiting

net.go

@@ -56,19 +56,22 @@ func connect(c *MqttClient) (rc ConnRC) {
 	_, err := io.ReadFull(c.bufferedConn, ca)
 	if err != nil {
 		ERROR.Println(NET, "connect got error")
-		c.errors <- err
+		select {
+		case c.errors <- err:
+		default:
+			// c.errors is a buffer of one, so there must already be an error closing this connection.
+		}
 		return
 	}
 	msg := decode(ca)
 
 	if msg == nil || msg.msgType() != CONNACK {
-		close(c.begin)
 		ERROR.Println(NET, "received msg that was nil or not CONNACK")
-	} else {
-		DEBUG.Println(NET, "received connack")
-		rc = msg.connRC()
+		return
 	}
-	return
+
+	DEBUG.Println(NET, "received connack")
+	return msg.connRC()
 }
 
 // actually read incoming messages off the wire
@@ -121,7 +124,11 @@ func incoming(c *MqttClient) {
 		// Not trying to disconnect, send the error to the errors channel
 	default:
 		ERROR.Println(NET, "incoming stopped with error")
-		c.errors <- err
+		select {
+		case c.errors <- err:
+		default:
+			// c.errors is a buffer of one, so there must already be an error closing this connection.
+		}
 		return
 	}
 }
@@ -154,7 +161,11 @@ func outgoing(c *MqttClient) {
 
 			if _, err := c.conn.Write(msg.Bytes()); err != nil {
 				ERROR.Println(NET, "outgoing stopped with error")
-				c.errors <- err
+				select {
+				case c.errors <- err:
+				default:
+					// c.errors is a buffer of one, so there must already be an error closing this connection.
+				}
 				return
 			}
 
@@ -177,7 +188,11 @@ func outgoing(c *MqttClient) {
 			_, err := c.conn.Write(msg.Bytes())
 			if err != nil {
 				ERROR.Println(NET, "outgoing stopped with error")
-				c.errors <- err
+				select {
+				case c.errors <- err:
+				default:
+					// c.errors is a buffer of one, so there must already be an error closing this connection.
+				}
 				return
 			}
 			c.lastContact.update()
@@ -249,7 +264,11 @@ func alllogic(c *MqttClient) {
 						// select can handle it appropriately.
 						if ok {
 							go func(errVal error, errChan chan error) {
-								errChan <- errVal
+								select {
+								case errChan <- errVal:
+								default:
+									// c.errors is a buffer of one, so there must already be an error closing this connection.
+								}
 							}(err, c.errors)
 						}
 					}