Vendor new queue-worker version

Introduces 0.4.6 of queue-worker - see upstream repo for changes.

Signed-off-by: Alex Ellis (VMware) <alexellis2@gmail.com>

gateway/vendor/github.com/nats-io/go-nats-streaming/stan.go

@@ -1,4 +1,15 @@
-// Copyright 2016 Apcera Inc. All rights reserved.
+// Copyright 2016-2018 The NATS Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
 
 // Package stan is a Go client for the NATS Streaming messaging system (https://nats.io).
 package stan
@@ -16,7 +27,7 @@ import (
 )
 
 // Version is the NATS Streaming Go Client version
-const Version = "0.3.4"
+const Version = "0.4.0"
 
 const (
 	// DefaultNatsURL is the default URL the client connects to
@@ -30,6 +41,10 @@ const (
 	// DefaultMaxPubAcksInflight is the default maximum number of published messages
 	// without outstanding ACKs from the server
 	DefaultMaxPubAcksInflight = 16384
+	// DefaultPingInterval is the default interval (in seconds) at which a connection sends a PING to the server
+	DefaultPingInterval = 5
+	// DefaultPingMaxOut is the number of PINGs without a response before the connection is considered lost.
+	DefaultPingMaxOut = 3
 )
 
 // Conn represents a connection to the NATS Streaming subsystem. It can Publish and
@@ -54,6 +69,13 @@ type Conn interface {
 	NatsConn() *nats.Conn
 }
 
+const (
+	// Client send connID in ConnectRequest and PubMsg, and server
+	// listens and responds to client PINGs. The validity of the
+	// connection (based on connID) is checked on incoming PINGs.
+	protocolOne = int32(1)
+)
+
 // Errors
 var (
 	ErrConnectReqTimeout = errors.New("stan: connect request timeout")
@@ -68,13 +90,20 @@ var (
 	ErrManualAck         = errors.New("stan: cannot manually ack in auto-ack mode")
 	ErrNilMsg            = errors.New("stan: nil message")
 	ErrNoServerSupport   = errors.New("stan: not supported by server")
+	ErrMaxPings          = errors.New("stan: connection lost due to PING failure")
 )
 
+var testAllowMillisecInPings = false
+
 // AckHandler is used for Async Publishing to provide status of the ack.
-// The func will be passed teh GUID and any error state. No error means the
+// The func will be passed the GUID and any error state. No error means the
 // message was successfully received by NATS Streaming.
 type AckHandler func(string, error)
 
+// ConnectionLostHandler is used to be notified if the Streaming connection
+// is closed due to unexpected errors.
+type ConnectionLostHandler func(Conn, error)
+
 // Options can be used to a create a customized connection.
 type Options struct {
 	NatsURL            string
@@ -83,6 +112,9 @@ type Options struct {
 	AckTimeout         time.Duration
 	DiscoverPrefix     string
 	MaxPubAcksInflight int
+	PingIterval        int // In seconds
+	PingMaxOut         int
+	ConnectionLostCB   ConnectionLostHandler
 }
 
 // DefaultOptions are the NATS Streaming client's default options
@@ -92,6 +124,8 @@ var DefaultOptions = Options{
 	AckTimeout:         DefaultAckWait,
 	DiscoverPrefix:     DefaultDiscoverPrefix,
 	MaxPubAcksInflight: DefaultMaxPubAcksInflight,
+	PingIterval:        DefaultPingInterval,
+	PingMaxOut:         DefaultPingMaxOut,
 }
 
 // Option is a function on the options for a connection.
@@ -140,11 +174,45 @@ func NatsConn(nc *nats.Conn) Option {
 	}
 }
 
+// Pings is an Option to set the ping interval and max out values.
+// The interval needs to be at least 1 and represents the number
+// of seconds.
+// The maxOut needs to be at least 2, since the count of sent PINGs
+// increase whenever a PING is sent and reset to 0 when a response
+// is received. Setting to 1 would cause the library to close the
+// connection right away.
+func Pings(interval, maxOut int) Option {
+	return func(o *Options) error {
+		// For tests, we may pass negative value that will be interpreted
+		// by the library as milliseconds. If this test boolean is set,
+		// do not check values.
+		if !testAllowMillisecInPings {
+			if interval < 1 || maxOut <= 2 {
+				return fmt.Errorf("Invalid ping values: interval=%v (min>0) maxOut=%v (min=2)", interval, maxOut)
+			}
+		}
+		o.PingIterval = interval
+		o.PingMaxOut = maxOut
+		return nil
+	}
+}
+
+// SetConnectionLostHandler is an Option to set the connection lost handler.
+// This callback will be invoked should the client permanently lose
+// contact with the server (or another client replaces it while being
+// disconnected). The callback will not be invoked on normal Conn.Close().
+func SetConnectionLostHandler(handler ConnectionLostHandler) Option {
+	return func(o *Options) error {
+		o.ConnectionLostCB = handler
+		return nil
+	}
+}
+
 // A conn represents a bare connection to a stan cluster.
 type conn struct {
 	sync.RWMutex
 	clientID         string
-	serverID         string
+	connID           []byte // This is a NUID that uniquely identify connections.
 	pubPrefix        string // Publish prefix set by stan, append our subject.
 	subRequests      string // Subject to send subscription requests.
 	unsubRequests    string // Subject to send unsubscribe requests.
@@ -158,7 +226,19 @@ type conn struct {
 	pubAckChan       chan (struct{})
 	opts             Options
 	nc               *nats.Conn
-	ncOwned          bool // NATS Streaming created the connection, so needs to close it.
+	ncOwned          bool       // NATS Streaming created the connection, so needs to close it.
+	pubNUID          *nuid.NUID // NUID generator for published messages.
+	connLostCB       ConnectionLostHandler
+
+	pingMu       sync.Mutex
+	pingSub      *nats.Subscription
+	pingTimer    *time.Timer
+	pingBytes    []byte
+	pingRequests string
+	pingInbox    string
+	pingInterval time.Duration
+	pingMaxOut   int
+	pingOut      int
 }
 
 // Closure for ack contexts.
@@ -169,9 +249,10 @@ type ack struct {
 }
 
 // Connect will form a connection to the NATS Streaming subsystem.
+// Note that clientID can contain only alphanumeric and `-` or `_` characters.
 func Connect(stanClusterID, clientID string, options ...Option) (Conn, error) {
 	// Process Options
-	c := conn{clientID: clientID, opts: DefaultOptions}
+	c := conn{clientID: clientID, opts: DefaultOptions, connID: []byte(nuid.Next()), pubNUID: nuid.New()}
 	for _, opt := range options {
 		if err := opt(&c.opts); err != nil {
 			return nil, err
@@ -181,7 +262,15 @@ func Connect(stanClusterID, clientID string, options ...Option) (Conn, error) {
 	c.nc = c.opts.NatsConn
 	// Create a NATS connection if it doesn't exist.
 	if c.nc == nil {
-		nc, err := nats.Connect(c.opts.NatsURL)
+		// We will set the max reconnect attempts to -1 (infinite)
+		// and the reconnect buffer to -1 to prevent any buffering
+		// (which may cause a published message to be flushed on
+		// reconnect while the API may have returned an error due
+		// to PubAck timeout.
+		nc, err := nats.Connect(c.opts.NatsURL,
+			nats.Name(clientID),
+			nats.MaxReconnects(-1),
+			nats.ReconnectBufSize(-1))
 		if err != nil {
 			return nil, err
 		}
@@ -200,9 +289,25 @@ func Connect(stanClusterID, clientID string, options ...Option) (Conn, error) {
 		return nil, err
 	}
 
+	// Prepare a subscription on ping responses, even if we are not
+	// going to need it, so that if that fails, it fails before initiating
+	// a connection.
+	pingSub, err := c.nc.Subscribe(nats.NewInbox(), c.processPingResponse)
+	if err != nil {
+		c.Close()
+		return nil, err
+	}
+
 	// Send Request to discover the cluster
 	discoverSubject := c.opts.DiscoverPrefix + "." + stanClusterID
-	req := &pb.ConnectRequest{ClientID: clientID, HeartbeatInbox: hbInbox}
+	req := &pb.ConnectRequest{
+		ClientID:       clientID,
+		HeartbeatInbox: hbInbox,
+		ConnID:         c.connID,
+		Protocol:       protocolOne,
+		PingInterval:   int32(c.opts.PingIterval),
+		PingMaxOut:     int32(c.opts.PingMaxOut),
+	}
 	b, _ := req.Marshal()
 	reply, err := c.nc.Request(discoverSubject, b, c.opts.ConnectTimeout)
 	if err != nil {
@@ -245,18 +350,172 @@ func Connect(stanClusterID, clientID string, options ...Option) (Conn, error) {
 
 	c.pubAckChan = make(chan struct{}, c.opts.MaxPubAcksInflight)
 
+	// Capture the connection error cb
+	c.connLostCB = c.opts.ConnectionLostCB
+
+	unsubPingSub := true
+	// Do this with servers which are at least at protcolOne.
+	if cr.Protocol >= protocolOne {
+		// Note that in the future server may override client ping
+		// interval value sent in ConnectRequest, so use the
+		// value in ConnectResponse to decide if we send PINGs
+		// and at what interval.
+		// In tests, the interval could be negative to indicate
+		// milliseconds.
+		if cr.PingInterval != 0 {
+			unsubPingSub = false
+
+			// These will be immutable.
+			c.pingRequests = cr.PingRequests
+			c.pingInbox = pingSub.Subject
+			// In test, it is possible that we get a negative value
+			// to represent milliseconds.
+			if testAllowMillisecInPings && cr.PingInterval < 0 {
+				c.pingInterval = time.Duration(cr.PingInterval*-1) * time.Millisecond
+			} else {
+				// PingInterval is otherwise assumed to be in seconds.
+				c.pingInterval = time.Duration(cr.PingInterval) * time.Second
+			}
+			c.pingMaxOut = int(cr.PingMaxOut)
+			c.pingBytes, _ = (&pb.Ping{ConnID: c.connID}).Marshal()
+			c.pingSub = pingSub
+			// Set the timer now that we are set. Use lock to create
+			// synchronization point.
+			c.pingMu.Lock()
+			c.pingTimer = time.AfterFunc(c.pingInterval, c.pingServer)
+			c.pingMu.Unlock()
+		}
+	}
+	if unsubPingSub {
+		pingSub.Unsubscribe()
+	}
+
 	// Attach a finalizer
 	runtime.SetFinalizer(&c, func(sc *conn) { sc.Close() })
 
 	return &c, nil
 }
 
+// Sends a PING (containing the connection's ID) to the server at intervals
+// specified by PingInterval option when connection is created.
+// Everytime a PING is sent, the number of outstanding PINGs is increased.
+// If the total number is > than the PingMaxOut option, then the connection
+// is closed, and connection error callback invoked if one was specified.
+func (sc *conn) pingServer() {
+	sc.pingMu.Lock()
+	// In case the timer fired while we were stopping it.
+	if sc.pingTimer == nil {
+		sc.pingMu.Unlock()
+		return
+	}
+	sc.pingOut++
+	if sc.pingOut > sc.pingMaxOut {
+		sc.pingMu.Unlock()
+		sc.closeDueToPing(ErrMaxPings)
+		return
+	}
+	sc.pingTimer.Reset(sc.pingInterval)
+	nc := sc.nc
+	sc.pingMu.Unlock()
+	// Send the PING now. If the NATS connection is reported closed,
+	// we are done.
+	if err := nc.PublishRequest(sc.pingRequests, sc.pingInbox, sc.pingBytes); err == nats.ErrConnectionClosed {
+		sc.closeDueToPing(err)
+	}
+}
+
+// Receives PING responses from the server.
+// If the response contains an error message, the connection is closed
+// and the connection error callback is invoked (if one is specified).
+// If no error, the number of ping out is reset to 0. There is no
+// decrement by one since for a given PING, the client may received
+// many responses when servers are running in channel partitioning mode.
+// Regardless, any positive response from the server ought to signal
+// that the connection is ok.
+func (sc *conn) processPingResponse(m *nats.Msg) {
+	// No data means OK (we don't have to call Unmarshal)
+	if len(m.Data) > 0 {
+		pingResp := &pb.PingResponse{}
+		if err := pingResp.Unmarshal(m.Data); err != nil {
+			return
+		}
+		if pingResp.Error != "" {
+			sc.closeDueToPing(errors.New(pingResp.Error))
+			return
+		}
+	}
+	// Do not attempt to decrement, simply reset to 0.
+	sc.pingMu.Lock()
+	sc.pingOut = 0
+	sc.pingMu.Unlock()
+}
+
+// Closes a connection and invoke the connection error callback if one
+// was registered when the connection was created.
+func (sc *conn) closeDueToPing(err error) {
+	sc.Lock()
+	if sc.nc == nil {
+		sc.Unlock()
+		return
+	}
+	// Stop timer, unsubscribe, fail the pubs, etc..
+	sc.cleanupOnClose(err)
+	// No need to send Close prototol, so simply close the underlying
+	// NATS connection (if we own it, and if not already closed)
+	if sc.ncOwned && !sc.nc.IsClosed() {
+		sc.nc.Close()
+	}
+	// Mark this streaming connection as closed. Do this under pingMu lock.
+	sc.pingMu.Lock()
+	sc.nc = nil
+	sc.pingMu.Unlock()
+	// Capture callback (even though this is immutable).
+	cb := sc.connLostCB
+	sc.Unlock()
+	if cb != nil {
+		// Execute in separate go routine.
+		go cb(sc, err)
+	}
+}
+
+// Do some cleanup when connection is lost or closed.
+// Connection lock is held on entry, and sc.nc is guaranteed not to be nil.
+func (sc *conn) cleanupOnClose(err error) {
+	sc.pingMu.Lock()
+	if sc.pingTimer != nil {
+		sc.pingTimer.Stop()
+		sc.pingTimer = nil
+	}
+	sc.pingMu.Unlock()
+
+	// Unsubscribe only if the NATS connection is not already closed...
+	if !sc.nc.IsClosed() {
+		if sc.ackSubscription != nil {
+			sc.ackSubscription.Unsubscribe()
+		}
+		if sc.pingSub != nil {
+			sc.pingSub.Unsubscribe()
+		}
+	}
+	// Fail all pending pubs
+	for guid, pubAck := range sc.pubAckMap {
+		if pubAck.t != nil {
+			pubAck.t.Stop()
+		}
+		if pubAck.ah != nil {
+			pubAck.ah(guid, err)
+		} else if pubAck.ch != nil {
+			pubAck.ch <- err
+		}
+		delete(sc.pubAckMap, guid)
+		if len(sc.pubAckChan) > 0 {
+			<-sc.pubAckChan
+		}
+	}
+}
+
 // Close a connection to the stan system.
 func (sc *conn) Close() error {
-	if sc == nil {
-		return ErrBadConnection
-	}
-
 	sc.Lock()
 	defer sc.Unlock()
 
@@ -271,13 +530,15 @@ func (sc *conn) Close() error {
 		defer nc.Close()
 	}
 
-	// Signals we are closed.
-	sc.nc = nil
-
 	// Now close ourselves.
-	if sc.ackSubscription != nil {
-		sc.ackSubscription.Unsubscribe()
-	}
+	sc.cleanupOnClose(ErrConnectionClosed)
+
+	// Signals we are closed.
+	// Do this also under pingMu lock so that we don't need
+	// to grab sc's lock in pingServer.
+	sc.pingMu.Lock()
+	sc.nc = nil
+	sc.pingMu.Unlock()
 
 	req := &pb.CloseRequest{ClientID: sc.clientID}
 	b, _ := req.Marshal()
@@ -303,7 +564,10 @@ func (sc *conn) Close() error {
 // closing the wrapped NATS conn will put the NATS Streaming Conn in an invalid
 // state.
 func (sc *conn) NatsConn() *nats.Conn {
-	return sc.nc
+	sc.RLock()
+	nc := sc.nc
+	sc.RUnlock()
+	return nc
 }
 
 // Process a heartbeat from the NATS Streaming cluster
@@ -322,9 +586,7 @@ func (sc *conn) processAck(m *nats.Msg) {
 	pa := &pb.PubAck{}
 	err := pa.Unmarshal(m.Data)
 	if err != nil {
-		// FIXME, make closure to have context?
-		fmt.Printf("Error processing unmarshal\n")
-		return
+		panic(fmt.Errorf("Error during ack unmarshal: %v", err))
 	}
 
 	// Remove
@@ -346,7 +608,10 @@ func (sc *conn) processAck(m *nats.Msg) {
 
 // Publish will publish to the cluster and wait for an ACK.
 func (sc *conn) Publish(subject string, data []byte) error {
-	ch := make(chan error)
+	// Need to make this a buffered channel of 1 in case
+	// a publish call is blocked in pubAckChan but cleanupOnClose()
+	// is trying to push the error to this channel.
+	ch := make(chan error, 1)
 	_, err := sc.publishAsync(subject, data, nil, ch)
 	if err == nil {
 		err = <-ch
@@ -370,9 +635,11 @@ func (sc *conn) publishAsync(subject string, data []byte, ah AckHandler, ch chan
 
 	subj := sc.pubPrefix + "." + subject
 	// This is only what we need from PubMsg in the timer below,
-	// so do this so that pe doesn't escape (and we same on new object)
-	peGUID := nuid.Next()
-	pe := &pb.PubMsg{ClientID: sc.clientID, Guid: peGUID, Subject: subject, Data: data}
+	// so do this so that pe doesn't escape.
+	peGUID := sc.pubNUID.Next()
+	// We send connID regardless of server we connect to. Older server
+	// will simply not decode it.
+	pe := &pb.PubMsg{ClientID: sc.clientID, Guid: peGUID, Subject: subject, Data: data, ConnID: sc.connID}
 	b, _ := pe.Marshal()
 
 	// Map ack to guid.
@@ -381,25 +648,44 @@ func (sc *conn) publishAsync(subject string, data []byte, ah AckHandler, ch chan
 	ackSubject := sc.ackSubject
 	ackTimeout := sc.opts.AckTimeout
 	pac := sc.pubAckChan
+	nc := sc.nc
 	sc.Unlock()
 
 	// Use the buffered channel to control the number of outstanding acks.
 	pac <- struct{}{}
 
-	err := sc.nc.PublishRequest(subj, ackSubject, b)
-	if err != nil {
-		sc.removeAck(peGUID)
-		return "", err
-	}
+	err := nc.PublishRequest(subj, ackSubject, b)
 
 	// Setup the timer for expiration.
 	sc.Lock()
+	if err != nil || sc.nc == nil {
+		sc.Unlock()
+		// If we got and error on publish or the connection has been closed,
+		// we need to return an error only if:
+		// - we can remove the pubAck from the map
+		// - we can't, but this is an async pub with no provided AckHandler
+		removed := sc.removeAck(peGUID) != nil
+		if removed || (ch == nil && ah == nil) {
+			if err == nil {
+				err = ErrConnectionClosed
+			}
+			return "", err
+		}
+		// pubAck was removed from cleanupOnClose() and error will be sent
+		// to appropriate go channel (ah or ch).
+		return peGUID, nil
+	}
 	a.t = time.AfterFunc(ackTimeout, func() {
-		sc.removeAck(peGUID)
-		if a.ah != nil {
-			ah(peGUID, ErrTimeout)
+		pubAck := sc.removeAck(peGUID)
+		// processAck could get here before and handle the ack.
+		// If that's the case, we would get nil here and simply return.
+		if pubAck == nil {
+			return
+		}
+		if pubAck.ah != nil {
+			pubAck.ah(peGUID, ErrTimeout)
 		} else if a.ch != nil {
-			a.ch <- ErrTimeout
+			pubAck.ch <- ErrTimeout
 		}
 	})
 	sc.Unlock()
@@ -436,7 +722,7 @@ func (sc *conn) processMsg(raw *nats.Msg) {
 	msg := &Msg{}
 	err := msg.Unmarshal(raw.Data)
 	if err != nil {
-		panic("Error processing unmarshal for msg")
+		panic(fmt.Errorf("Error processing unmarshal for msg: %v", err))
 	}
 	// Lookup the subscription
 	sc.RLock()
@@ -465,12 +751,11 @@ func (sc *conn) processMsg(raw *nats.Msg) {
 		cb(msg)
 	}
 
-	// Proces auto-ack
+	// Process auto-ack
 	if !isManualAck && nc != nil {
 		ack := &pb.Ack{Subject: msg.Subject, Sequence: msg.Sequence}
 		b, _ := ack.Marshal()
-		if err := nc.Publish(ackSubject, b); err != nil {
-			// FIXME(dlc) - Async error handler? Retry?
-		}
+		// FIXME(dlc) - Async error handler? Retry?
+		nc.Publish(ackSubject, b)
 	}
 }