Reset timer on token updates

eventhub/client.go

@@ -11,14 +11,9 @@ import (
 	"time"
 
 	"github.com/goautomotive/iothub/common"
-
 	"pack.ag/amqp"
 )
 
-var (
-	DurationOfTokenGenerate = 1 * time.Hour
-)
-
 // Dial connects to the named amqp broker and returns an eventhub client.
 func Dial(addr string, tlsConfig *tls.Config) (*Client, error) {
 	conn, err := amqp.Dial(addr,
@@ -106,6 +101,14 @@ func SubscribePartitions(ctx context.Context, sess *amqp.Session, name, group st
 	}
 }
 
+const (
+	tokenUpdateInterval = time.Hour
+
+	// we need to update tokens before they expire to prevent disconnects
+	// from azure, without interrupting the message flow
+	tokenUpdateSpan = 10 * time.Minute
+)
+
 // PutTokenContinuously writes token first time in blocking mode and returns
 // maintaining token updates in the background until stopCh is closed.
 func (c *Client) PutTokenContinuously(
@@ -114,12 +117,7 @@ func (c *Client) PutTokenContinuously(
 	cred *common.Credentials,
 	stopCh chan struct{},
 ) error {
-	// validate
-	if int64(DurationOfTokenGenerate-10*time.Minute) <= 0 {
-		return errors.New("duration is minus")
-	}
-
-	token, err := cred.SAS(cred.HostName, DurationOfTokenGenerate)
+	token, err := cred.SAS(cred.HostName, tokenUpdateInterval)
 	if err != nil {
 		return err
 	}
@@ -128,21 +126,22 @@ func (c *Client) PutTokenContinuously(
 	}
 
 	go func() {
-		ticker := time.NewTimer(DurationOfTokenGenerate - 10*time.Minute) // 10min is a safe buffer
+		ticker := time.NewTimer(tokenUpdateInterval - tokenUpdateSpan)
 		defer ticker.Stop()
 
 		for {
 			select {
 			case <-ticker.C:
-				token, err := cred.SAS(cred.HostName, DurationOfTokenGenerate)
+				token, err := cred.SAS(cred.HostName, tokenUpdateInterval)
 				if err != nil {
-					log.Printf("create SAS token error: %s", err)
+					log.Printf("genegate SAS token error: %s", err)
 					return
 				}
 				if err := c.PutToken(context.Background(), audience, token); err != nil {
 					log.Printf("put token error: %s", err)
 					return
 				}
+				ticker.Reset(tokenUpdateInterval - tokenUpdateSpan)
 			case <-stopCh:
 				return
 			}

go.sum

@@ -0,0 +1,5 @@
+github.com/eclipse/paho.mqtt.golang v1.1.1/go.mod h1:H9keYFcgq3Qr5OUJm/JZI/i6U7joQ8SYLhZwfeOo6Ts=
+github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
+github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+golang.org/x/net v0.0.0-20180218175443-cbe0f9307d01/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+pack.ag/amqp v0.5.0/go.mod h1:4/cbmt4EJXSKlG6LCfWHoqmN0uFdy5i/+YFz+fTfhV4=

iotdevice/client.go

@@ -35,6 +35,9 @@ func WithLogger(l *log.Logger) ClientOption {
 
 // WithTransport changes default transport.
 func WithTransport(tr transport.Transport) ClientOption {
+	if tr == nil {
+		panic("transport is nil")
+	}
 	return func(c *Client) error {
 		c.tr = tr
 		return nil
@@ -156,6 +159,7 @@ func (c *Client) Connect(ctx context.Context) error {
 		close(c.ready)
 	}
 	c.mu.Unlock()
+	// TODO: c.err = err
 	return err
 }
 

iotservice/client.go

@@ -156,9 +156,7 @@ func (c *Client) connectToEventHub(ctx context.Context) (*amqp.Client, string, e
 	if err != nil {
 		return nil, "", err
 	}
-	defer func(conn *amqp.Client) {
-		conn.Close()
-	}(conn)
+	defer conn.Close()
 
 	sess, err := conn.NewSession()
 	if err != nil {
@@ -173,7 +171,6 @@ func (c *Client) connectToEventHub(ctx context.Context) (*amqp.Client, string, e
 	}
 	defer recv.Close(context.Background())
 	_, err = recv.Receive(ctx)
-
 	if err == nil {
 		return nil, "", errors.New("expected redirect error")
 	}