Repository: nats-io/nats-server
Pull request #1658 (2020-10-20 23:43:09)

[JetStream] Upgrade for wait queues for pull based consumers
Upgraded the wait queue to be more robust and process formal requests with options for batch size, no waiting on messages and expiration.

/cc @nats-io/core



Patch:
diff --git a/README.md b/README.md
index 04f79f84..3aac13d3 100644
--- a/README.md
+++ b/README.md
@@ -1,6 +1,6 @@
 ## <img src="logos/nats-server.png" width="300">
 
-[NATS](https://nats.io) is a simple, secure and performant communications system for digital systems, services and devices. NATS is part of the Cloud Native Computing Foundation ([CNCF](https://cncf.io)). NATS has over [40 client language implementations](https://nats.io/download/), and its server can run on-premise, in the cloud, at the edge, and even on a Raspberry Pi. NATS can secure and simplify design and operation of modern distributed systems.
+[NATS](https://nats.io) is a simple, secure and performant communications system for digital systems, services and devices. NATS is part of the Cloud Native Computing Foundation ([CNCF](https://cncf.io)). NATS has over [30 client language implementations](https://nats.io/download/), and its server can run on-premise, in the cloud, at the edge, and even on a Raspberry Pi. NATS can secure and simplify design and operation of modern distributed systems.
 
 [![License][License-Image]][License-Url] [![FOSSA Status][Fossa-Image]][Fossa-Url] [![ReportCard][ReportCard-Image]][ReportCard-Url] [![Build][Build-Status-Image]][Build-Status-Url] [![Release][Release-Image]][Release-Url] [![Coverage][Coverage-Image]][Coverage-Url] [![CII Best Practices](https://bestpractices.coreinfrastructure.org/projects/1895/badge)](https://bestpractices.coreinfrastructure.org/projects/1895)
 
diff --git a/go.mod b/go.mod
index 3744c01f..85acbd73 100644
--- a/go.mod
+++ b/go.mod
@@ -3,10 +3,10 @@ module github.com/nats-io/nats-server/v2
 require (
 	github.com/minio/highwayhash v1.0.0
 	github.com/nats-io/jwt/v2 v2.0.0-20201015190852-e11ce317263c
-	github.com/nats-io/nats.go v1.10.1-0.20200606002146-fc6fed82929a
+	github.com/nats-io/nats.go v1.10.1-0.20201021145452-94be476ad6e0
 	github.com/nats-io/nkeys v0.2.0
 	github.com/nats-io/nuid v1.0.1
-	golang.org/x/crypto v0.0.0-20200323165209-0ec3e9974c59
+	golang.org/x/crypto v0.0.0-20201016220609-9e8e0b390897
 	golang.org/x/sys v0.0.0-20191022100944-742c48ecaeb7
 	golang.org/x/time v0.0.0-20200416051211-89c76fbcd5d1
 )
diff --git a/go.sum b/go.sum
index a41b8040..a20b5346 100644
--- a/go.sum
+++ b/go.sum
@@ -14,14 +14,20 @@ github.com/minio/highwayhash v1.0.0/go.mod h1:xQboMTeM9nY9v/LlAOxFctujiv5+Aq2hR5
 github.com/nats-io/jwt v0.3.2/go.mod h1:/euKqTS1ZD+zzjYrY7pseZrTtWQSjujC7xjPc8wL6eU=
 github.com/nats-io/jwt v0.3.3-0.20200519195258-f2bf5ce574c7 h1:RnGotxlghqR5D2KDAu4TyuLqyjuylOsJiAFhXvMvQIc=
 github.com/nats-io/jwt v0.3.3-0.20200519195258-f2bf5ce574c7/go.mod h1:n3cvmLfBfnpV4JJRN7lRYCyZnw48ksGsbThGXEk4w9M=
+github.com/nats-io/jwt/v2 v2.0.0-20200916203241-1f8ce17dff02/go.mod h1:vs+ZEjP+XKy8szkBmQwCB7RjYdIlMaPsFPs4VdS4bTQ=
 github.com/nats-io/jwt/v2 v2.0.0-20201015190852-e11ce317263c h1:Hc1D9ChlsCMVwCxJ6QT5xqfk2zJ4XNea+LtdfaYhd20=
 github.com/nats-io/jwt/v2 v2.0.0-20201015190852-e11ce317263c/go.mod h1:vs+ZEjP+XKy8szkBmQwCB7RjYdIlMaPsFPs4VdS4bTQ=
 github.com/nats-io/nats-server/v2 v2.1.8-0.20200524125952-51ebd92a9093/go.mod h1:rQnBf2Rv4P9adtAs/Ti6LfFmVtFG6HLhl/H7cVshcJU=
 github.com/nats-io/nats-server/v2 v2.1.8-0.20200601203034-f8d6dd992b71/go.mod h1:Nan/1L5Sa1JRW+Thm4HNYcIDcVRFc5zK9OpSZeI2kk4=
+github.com/nats-io/nats-server/v2 v2.1.8-0.20200929001935-7f44d075f7ad/go.mod h1:TkHpUIDETmTI7mrHN40D1pzxfzHZuGmtMbtb83TGVQw=
 github.com/nats-io/nats.go v1.10.0/go.mod h1:AjGArbfyR50+afOUotNX2Xs5SYHf+CoOa5HH1eEl2HE=
 github.com/nats-io/nats.go v1.10.1-0.20200531124210-96f2130e4d55/go.mod h1:ARiFsjW9DVxk48WJbO3OSZ2DG8fjkMi7ecLmXoY/n9I=
 github.com/nats-io/nats.go v1.10.1-0.20200606002146-fc6fed82929a h1:gzSKZOBlu/DpbuPbt34paXCOvA6+E+lVfU2BmomQ9HA=
 github.com/nats-io/nats.go v1.10.1-0.20200606002146-fc6fed82929a/go.mod h1:8eAIv96Mo9QW6Or40jUHejS7e4VwZ3VRYD6Sf0BTDp4=
+github.com/nats-io/nats.go v1.10.1-0.20201013114232-5a33ce07522f h1:ip3GUJTsw7ybH4qIRtv/GjNzMyHmsJnxXzRmsbqYh8M=
+github.com/nats-io/nats.go v1.10.1-0.20201013114232-5a33ce07522f/go.mod h1:VU2zERjp8xmF+Lw2NH4u2t5qWZxwc7jB3+7HVMWQXPI=
+github.com/nats-io/nats.go v1.10.1-0.20201021145452-94be476ad6e0 h1:s//xxvMgNRVJNTRXTRNvLlES0zvJoDLXfcmbK0YgN0E=
+github.com/nats-io/nats.go v1.10.1-0.20201021145452-94be476ad6e0/go.mod h1:VU2zERjp8xmF+Lw2NH4u2t5qWZxwc7jB3+7HVMWQXPI=
 github.com/nats-io/nkeys v0.1.3/go.mod h1:xpnFELMwJABBLVhffcfd1MZx6VsNRFpEugbxziKVo7w=
 github.com/nats-io/nkeys v0.1.4/go.mod h1:XdZpAbhgyyODYqjTawOnIOI7VlbKSarI9Gfy1tqEu/s=
 github.com/nats-io/nkeys v0.2.0 h1:WXKF7diOaPU9cJdLD7nuzwasQy9vT1tBqzXZZf3AMJM=
@@ -32,6 +38,8 @@ golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACk
 golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20200323165209-0ec3e9974c59 h1:3zb4D3T4G8jdExgVU/95+vQXfpEPiMdCaZgmGVxjNHM=
 golang.org/x/crypto v0.0.0-20200323165209-0ec3e9974c59/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
+golang.org/x/crypto v0.0.0-20201016220609-9e8e0b390897 h1:pLI5jrR7OSLijeIDcmRxNmw2api+jEfxLoykJVice/E=
+golang.org/x/crypto v0.0.0-20201016220609-9e8e0b390897/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/sys v0.0.0-20190130150945-aca44879d564/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
diff --git a/server/accounts.go b/server/accounts.go
index 11c5d6a0..10186371 100644
--- a/server/accounts.go
+++ b/server/accounts.go
@@ -32,7 +32,6 @@ import (
 	"time"
 
 	"github.com/nats-io/jwt/v2"
-	"github.com/nats-io/nkeys"
 	"github.com/nats-io/nuid"
 )
 
@@ -3302,49 +3301,14 @@ func handleListRequest(store *DirJWTStore, s *Server, reply string) {
 }
 
 func handleDeleteRequest(store *DirJWTStore, s *Server, msg []byte, reply string) {
-	var accIds []interface{}
-	var subj, sysAccName string
-	if sysAcc := s.SystemAccount(); sysAcc != nil {
-		sysAccName = sysAcc.GetName()
-	}
-	// TODO Can allow keys (issuer) to delete accounts they issued and operator key to delete all accounts.
-	//      For now only operator is allowed to delete
-	gk, err := jwt.DecodeGeneric(string(msg))
-	if err == nil {
-		subj = gk.Subject
-		if store.deleteType == NoDelete {
-			err = fmt.Errorf("delete must be enabled in server config")
-		} else if subj != gk.Issuer {
-			err = fmt.Errorf("not self signed")
-		} else if !s.isTrustedIssuer(gk.Issuer) {
-			err = fmt.Errorf("not trusted")
-		} else if store.operator != gk.Issuer {
-			err = fmt.Errorf("needs to be the operator operator")
-		} else if list, ok := gk.Data["accounts"]; !ok {
-			err = fmt.Errorf("malformed request")
-		} else if accIds, ok = list.([]interface{}); !ok {
-			err = fmt.Errorf("malformed request")
-		} else {
-			for _, entry := range accIds {
-				if acc, ok := entry.(string); !ok ||
-					acc == "" || !nkeys.IsValidPublicAccountKey(acc) {
-					err = fmt.Errorf("malformed request")
-					break
-				} else if acc == sysAccName {
-					err = fmt.Errorf("not allowed to delete system account")
-					break
-				}
-			}
-		}
-	}
-	if err != nil {
-		respondToUpdate(s, reply, "", fmt.Sprintf("delete accounts request by %s failed", subj), err)
-		return
-	}
+	accIds := strings.Split(strings.Replace(string(msg), "\r\n", "\n", -1), "\n")
 	errs := []string{}
 	passCnt := 0
 	for _, acc := range accIds {
-		if err := store.delete(acc.(string)); err != nil {
+		if acc == "" {
+			continue
+		}
+		if err := store.delete(acc); err != nil {
 			errs = append(errs, err.Error())
 		} else {
 			passCnt++
@@ -3358,26 +3322,10 @@ func handleDeleteRequest(store *DirJWTStore, s *Server, msg []byte, reply string
 	}
 }
 
-func getOperator(s *Server) (string, error) {
-	var op string
-	if opts := s.getOpts(); opts != nil && len(opts.TrustedOperators) > 0 {
-		op = opts.TrustedOperators[0].Subject
-	}
-	if op == "" {
-		return "", fmt.Errorf("no operator found")
-	}
-	return op, nil
-}
-
 func (dr *DirAccResolver) Start(s *Server) error {
-	op, err := getOperator(s)
-	if err != nil {
-		return err
-	}
 	dr.Lock()
 	defer dr.Unlock()
 	dr.Server = s
-	dr.operator = op
 	dr.DirJWTStore.changed = func(pubKey string) {
 		if v, ok := s.accounts.Load(pubKey); !ok {
 		} else if jwt, err := dr.LoadAcc(pubKey); err != nil {
@@ -3528,18 +3476,14 @@ func (dr *DirAccResolver) Store(name, jwt string) error {
 	return dr.saveIfNewer(name, jwt)
 }
 
-func NewDirAccResolver(path string, limit int64, syncInterval time.Duration, delete bool) (*DirAccResolver, error) {
+func NewDirAccResolver(path string, limit int64, syncInterval time.Duration) (*DirAccResolver, error) {
 	if limit == 0 {
 		limit = math.MaxInt64
 	}
 	if syncInterval <= 0 {
 		syncInterval = time.Minute
 	}
-	deleteType := NoDelete
-	if delete {
-		deleteType = RenameDeleted
-	}
-	store, err := NewExpiringDirJWTStore(path, false, true, deleteType, 0, limit, false, 0, nil)
+	store, err := NewExpiringDirJWTStore(path, false, true, 0, limit, false, 0, nil)
 	if err != nil {
 		return nil, err
 	}
@@ -3600,11 +3544,11 @@ func (s *Server) fetch(res AccountResolver, name string) (string, error) {
 	return theJWT, err
 }
 
-func NewCacheDirAccResolver(path string, limit int64, ttl time.Duration, _ ...dirJWTStoreOption) (*CacheDirAccResolver, error) {
+func NewCacheDirAccResolver(path string, limit int64, ttl time.Duration) (*CacheDirAccResolver, error) {
 	if limit <= 0 {
 		limit = 1_000
 	}
-	store, err := NewExpiringDirJWTStore(path, false, true, HardDelete, 0, limit, true, ttl, nil)
+	store, err := NewExpiringDirJWTStore(path, false, true, 0, limit, true, ttl, nil)
 	if err != nil {
 		return nil, err
 	}
@@ -3612,14 +3556,9 @@ func NewCacheDirAccResolver(path string, limit int64, ttl time.Duration, _ ...di
 }
 
 func (dr *CacheDirAccResolver) Start(s *Server) error {
-	op, err := getOperator(s)
-	if err != nil {
-		return err
-	}
 	dr.Lock()
 	defer dr.Unlock()
 	dr.Server = s
-	dr.operator = op
 	dr.DirJWTStore.changed = func(pubKey string) {
 		if v, ok := s.accounts.Load(pubKey); !ok {
 		} else if jwt, err := dr.LoadAcc(pubKey); err != nil {
diff --git a/server/const.go b/server/const.go
index 1928963c..ba96f83c 100644
--- a/server/const.go
+++ b/server/const.go
@@ -40,7 +40,7 @@ var (
 
 const (
 	// VERSION is the current version for the server.
-	VERSION = "2.2.0-beta.26"
+	VERSION = "2.2.0-beta.25"
 
 	// PROTO is the currently supported protocol.
 	// 0 was the original
diff --git a/server/consumer.go b/server/consumer.go
index 2dde289d..ad2183f7 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -18,6 +18,7 @@ import (
 	"crypto/rand"
 	"crypto/sha256"
 	"encoding/json"
+	"errors"
 	"fmt"
 	mrand "math/rand"
 	"reflect"
@@ -40,6 +41,7 @@ type ConsumerInfo struct {
 	AckFloor       SequencePair   `json:"ack_floor"`
 	NumPending     int            `json:"num_pending"`
 	NumRedelivered int            `json:"num_redelivered"`
+	NumWaiting     int            `json:"num_waiting"`
 }
 
 type ConsumerConfig struct {
@@ -55,6 +57,7 @@ type ConsumerConfig struct {
 	ReplayPolicy    ReplayPolicy  `json:"replay_policy"`
 	RateLimit       uint64        `json:"rate_limit_bps,omitempty"` // Bits per sec
 	SampleFrequency string        `json:"sample_freq,omitempty"`
+	MaxWaiting      int           `json:"max_waiting,omitempty"`
 }
 
 type CreateConsumerRequest struct {
@@ -178,7 +181,7 @@ type Consumer struct {
 	rdq               []uint64
 	rdc               map[uint64]uint64
 	maxdc             uint64
-	waiting           []string
+	waiting           *waitQueue
 	config            ConsumerConfig
 	store             ConsumerStore
 	active            bool
@@ -218,6 +221,9 @@ func (mset *Stream) AddConsumer(config *ConsumerConfig) (*Consumer, error) {
 		if mset.deliveryFormsCycle(config.DeliverSubject) {
 			return nil, fmt.Errorf("consumer deliver subject forms a cycle")
 		}
+		if config.MaxWaiting != 0 {
+			return nil, fmt.Errorf("consumer in push mode can not set max waiting")
+		}
 	} else {
 		// Pull mode / work queue mode require explicit ack.
 		if config.AckPolicy != AckExplicit {
@@ -231,6 +237,13 @@ func (mset *Stream) AddConsumer(config *ConsumerConfig) (*Consumer, error) {
 		if config.RateLimit > 0 {
 			return nil, fmt.Errorf("consumer in pull mode can not have rate limit set")
 		}
+		if config.MaxWaiting < 0 {
+			return nil, fmt.Errorf("consumer max waiting needs to be positive")
+		}
+		// Set to default if not specified.
+		if config.MaxWaiting == 0 {
+			config.MaxWaiting = JSWaitQueueDefaultMax
+		}
 	}
 
 	// Setup proper default for ack wait if we are in explicit ack mode.
@@ -380,6 +393,9 @@ func (mset *Stream) AddConsumer(config *ConsumerConfig) (*Consumer, error) {
 			return nil, fmt.Errorf("consumer name is too long, maximum allowed is %d", JSMaxNameLen)
 		}
 		o.name = config.Durable
+		if o.isPullMode() {
+			o.waiting = newWaitQueue(config.MaxWaiting)
+		}
 	} else {
 		for {
 			o.name = createConsumerName()
@@ -699,7 +715,7 @@ func (o *Consumer) processAck(_ *subscription, _ *client, subject, reply string,
 		o.ackMsg(sseq, dseq, dcount)
 	case bytes.Equal(msg, AckNext):
 		o.ackMsg(sseq, dseq, dcount)
-		o.processNextMsgReq(nil, nil, subject, reply, nil)
+		o.processNextMsgReq(nil, nil, subject, reply, msg)
 		skipAckReply = true
 	case bytes.Equal(msg, AckNak):
 		o.processNak(sseq, dseq)
@@ -885,6 +901,10 @@ func (o *Consumer) Info() *ConsumerInfo {
 		NumPending:     len(o.pending),
 		NumRedelivered: len(o.rdc),
 	}
+	// If we are a pull mode consumer, report on number of waiting requests.
+	if o.isPullMode() {
+		info.NumWaiting = o.waiting.len()
+	}
 	o.mu.Unlock()
 	return info
 }
@@ -1031,47 +1051,175 @@ func (o *Consumer) needAck(sseq uint64) bool {
 	return needAck
 }
 
-// Default is 1 if msg is nil.
-func batchSizeFromMsg(msg []byte) int {
+// Helper for the next message requests.
+func nextReqFromMsg(msg []byte) (time.Time, int, bool, error) {
+	req := strings.TrimSpace(string(msg))
+	if len(req) == 0 {
+		return time.Time{}, 1, false, nil
+	}
+	if req[0] == '{' {
+		var cr JSApiConsumerGetNextRequest
+		if err := json.Unmarshal(msg, &cr); err != nil {
+			return time.Time{}, -1, false, err
+		}
+		return cr.Expires, cr.Batch, cr.NoWait, nil
+	}
+	// Naked batch size here for backward compatibility.
 	bs := 1
-	if len(msg) > 0 {
-		if n, err := strconv.Atoi(string(msg)); err == nil {
-			bs = n
+	if n, err := strconv.Atoi(req); err == nil {
+		bs = n
+	}
+	return time.Time{}, bs, false, nil
+}
+
+// Represents a request that is on the internal waiting queue
+type waitingRequest struct {
+	client  *client
+	reply   string
+	n       int // For batching
+	expires time.Time
+	noWait  bool
+}
+
+// waiting queue for requests that are waiting for new messages to arrive.
+type waitQueue struct {
+	rp, wp int
+	reqs   []*waitingRequest
+}
+
+// Create a new ring buffer with at most max items.
+func newWaitQueue(max int) *waitQueue {
+	return &waitQueue{rp: -1, reqs: make([]*waitingRequest, max)}
+}
+
+var (
+	errWaitQueueFull = errors.New("wait queue is full")
+	errWaitQueueNil  = errors.New("wait queue is nil")
+)
+
+// Adds in a new request.
+func (wq *waitQueue) add(req *waitingRequest) error {
+	if wq == nil {
+		return errWaitQueueNil
+	}
+	if wq.isFull() {
+		return errWaitQueueFull
+	}
+	wq.reqs[wq.wp] = req
+	// TODO(dlc) - Could make pow2 and get rid of mod.
+	wq.wp = (wq.wp + 1) % cap(wq.reqs)
+
+	// Adjust read pointer if we were empty.
+	if wq.rp < 0 {
+		wq.rp = 0
+	}
+
+	return nil
+}
+
+func (wq *waitQueue) isFull() bool {
+	return wq.rp == wq.wp
+}
+
+func (wq *waitQueue) len() int {
+	if wq == nil || wq.rp < 0 {
+		return 0
+	}
+	if wq.rp < wq.wp {
+		return wq.wp - wq.rp
+	}
+	return cap(wq.reqs) - wq.rp + wq.wp
+}
+
+// Peek will return the next request waiting or nil if empty.
+func (wq *waitQueue) peek() *waitingRequest {
+	if wq == nil {
+		return nil
+	}
+	var wr *waitingRequest
+	if wq.rp >= 0 {
+		wr = wq.reqs[wq.rp]
+	}
+	return wr
+}
+
+// pop will return the next request and move the read cursor.
+func (wq *waitQueue) pop() *waitingRequest {
+	wr := wq.peek()
+	if wr != nil {
+		wr.n--
+		if wr.n <= 0 {
+			wq.reqs[wq.rp] = nil
+			wq.rp = (wq.rp + 1) % cap(wq.reqs)
+			// Check if we are empty.
+			if wq.rp == wq.wp {
+				wq.rp, wq.wp = -1, 0
+			}
 		}
 	}
-	return bs
+	return wr
 }
 
 // processNextMsgReq will process a request for the next message available. A nil message payload means deliver
 // a single message. If the payload is a number parseable with Atoi(), then we will send a batch of messages without
 // requiring another request to this endpoint, or an ACK.
-func (o *Consumer) processNextMsgReq(_ *subscription, _ *client, _, reply string, msg []byte) {
-	// Check payload here to see if they sent in batch size.
-	batchSize := batchSizeFromMsg(msg)
-
+func (o *Consumer) processNextMsgReq(_ *subscription, c *client, _, reply string, msg []byte) {
 	o.mu.Lock()
 	mset := o.mset
 	if mset == nil || o.isPushMode() {
 		o.mu.Unlock()
 		return
 	}
-	shouldSignal := false
+
+	sendErr := func(status int, description string) {
+		sendq := mset.sendq
+		o.mu.Unlock()
+		hdr := []byte(fmt.Sprintf("NATS/1.0 %d %s\r\n\r\n", status, description))
+		pmsg := &jsPubMsg{reply, reply, _EMPTY_, hdr, nil, nil, 0}
+		sendq <- pmsg // Send message.
+	}
+
+	if o.waiting.isFull() {
+		// Try to expire some of the requests.
+		if expired := o.expireWaiting(); expired == 0 {
+			// Force expiration if needed.
+			o.forceExpireFirstWaiting()
+		}
+	}
+
+	// Check payload here to see if they sent in batch size or a formal request.
+	expires, batchSize, noWait, err := nextReqFromMsg(msg)
+	if err != nil {
+		sendErr(400, fmt.Sprintf("Bad Request - %v", err))
+		return
+	}
+
+	// In case we have to queue up this request. This is all on stack pre-allocated.
+	wr := waitingRequest{client: c, reply: reply, n: batchSize, noWait: noWait, expires: expires}
+
+	// If we are in replay mode, defer to processReplay for delivery.
+	if o.replay {
+		o.waiting.add(&wr)
+		o.mu.Unlock()
+		mset.signalConsumers()
+		return
+	}
 
 	for i := 0; i < batchSize; i++ {
-		// If we are in replay mode, defer to processReplay for delivery.
-		if o.replay {
-			o.waiting = append(o.waiting, reply)
-			shouldSignal = true
-		} else if subj, hdr, msg, seq, dc, ts, err := o.getNextMsg(); err == nil {
+		if subj, hdr, msg, seq, dc, ts, err := o.getNextMsg(); err == nil {
 			o.deliverMsg(reply, subj, hdr, msg, seq, dc, ts)
+			// Need to discount this from the total n for the request.
+			wr.n--
 		} else {
-			o.waiting = append(o.waiting, reply)
+			if wr.noWait {
+				sendErr(404, "No Messages")
+				return
+			}
+			o.waiting.add(&wr)
+			break
 		}
 	}
 	o.mu.Unlock()
-	if shouldSignal {
-		mset.signalConsumers()
-	}
 }
 
 // Increase the delivery count for this message.
@@ -1161,17 +1309,52 @@ func (o *Consumer) getNextMsg() (subj string, hdr, msg []byte, seq uint64, dcoun
 	}
 }
 
-// Will check to make sure those waiting still have registered interest.
-func (o *Consumer) checkWaitingForInterest() bool {
-	for len(o.waiting) > 0 {
-		rr := o.acc.sl.Match(o.waiting[0])
+// forceExpireFirstWaiting will force expire the first waiting.
+// Lock should be held.
+func (o *Consumer) forceExpireFirstWaiting() *waitingRequest {
+	// FIXME(dlc) - Should we do advisory here as well?
+	wr := o.waiting.pop()
+	if wr == nil {
+		return wr
+	}
+	// If we are expiring this and we think there is still interest, alert.
+	if rr := o.acc.sl.Match(wr.reply); len(rr.psubs)+len(rr.qsubs) > 0 && o.mset != nil {
+		// We still appear to have interest, so send alert as courtesy.
+		sendq := o.mset.sendq
+		o.mu.Unlock()
+		hdr := []byte("NATS/1.0 408 Request Timeout\r\n\r\n")
+		pmsg := &jsPubMsg{wr.reply, wr.reply, _EMPTY_, hdr, nil, nil, 0}
+		sendq <- pmsg // Send message.
+		o.mu.Lock()
+	}
+	return wr
+}
+
+// Will check for expiration and lack of interest on waiting requests.
+func (o *Consumer) expireWaiting() int {
+	var expired int
+	now := time.Now()
+	for wr := o.waiting.peek(); wr != nil; wr = o.waiting.peek() {
+		if !wr.expires.IsZero() && now.After(wr.expires) {
+			o.forceExpireFirstWaiting()
+			expired++
+			continue
+		}
+		rr := o.acc.sl.Match(wr.reply)
 		if len(rr.psubs)+len(rr.qsubs) > 0 {
 			break
 		}
 		// No more interest so go ahead and remove this one from our list.
-		o.waiting = append(o.waiting[:0], o.waiting[1:]...)
+		o.forceExpireFirstWaiting()
+		expired++
 	}
-	return len(o.waiting) > 0
+	return expired
+}
+
+// Will check to make sure those waiting still have registered interest.
+func (o *Consumer) checkWaitingForInterest() bool {
+	o.expireWaiting()
+	return o.waiting.len() > 0
 }
 
 func (o *Consumer) loopAndDeliverMsgs(s *Server, a *Account) {
@@ -1235,9 +1418,8 @@ func (o *Consumer) loopAndDeliverMsgs(s *Server, a *Account) {
 			}
 		}
 
-		if len(o.waiting) > 0 {
-			dsubj = o.waiting[0]
-			o.waiting = append(o.waiting[:0], o.waiting[1:]...)
+		if wr := o.waiting.pop(); wr != nil {
+			dsubj = wr.reply
 		} else {
 			dsubj = o.dsubj
 		}
@@ -1326,10 +1508,10 @@ func (o *Consumer) deliverCurrentMsg(subj string, hdr, msg []byte, seq uint64, t
 		o.mu.Unlock()
 		return true
 	}
+
 	var dsubj string
-	if len(o.waiting) > 0 {
-		dsubj = o.waiting[0]
-		o.waiting = append(o.waiting[:0], o.waiting[1:]...)
+	if wr := o.waiting.pop(); wr != nil {
+		dsubj = wr.reply
 	} else {
 		dsubj = o.dsubj
 	}
@@ -1725,6 +1907,7 @@ func (o *Consumer) stop(dflag, doSignal, advisory bool) error {
 	stopAndClearTimer(&o.ptmr)
 	stopAndClearTimer(&o.dtmr)
 	delivery := o.config.DeliverSubject
+	o.waiting = nil
 	o.mu.Unlock()
 
 	if delivery != "" {
diff --git a/server/dirstore.go b/server/dirstore.go
index 5a6af63a..76ddf295 100644
--- a/server/dirstore.go
+++ b/server/dirstore.go
@@ -81,8 +81,6 @@ type DirJWTStore struct {
 	directory  string
 	shard      bool
 	readonly   bool
-	deleteType deleteType
-	operator   string
 	expiration *expirationTracker
 	changed    JWTChanged
 }
@@ -103,13 +101,10 @@ func newDir(dirPath string, create bool) (string, error) {
 	return fullPath, nil
 }
 
-// future proofing in case new options will be added
-type dirJWTStoreOption interface{}
-
 // Creates a directory based jwt store.
 // Reads files only, does NOT watch directories and files.
-func NewImmutableDirJWTStore(dirPath string, shard bool, _ ...dirJWTStoreOption) (*DirJWTStore, error) {
-	theStore, err := NewDirJWTStore(dirPath, shard, false, nil)
+func NewImmutableDirJWTStore(dirPath string, shard bool) (*DirJWTStore, error) {
+	theStore, err := NewDirJWTStore(dirPath, shard, false)
 	if err != nil {
 		return nil, err
 	}
@@ -119,7 +114,7 @@ func NewImmutableDirJWTStore(dirPath string, shard bool, _ ...dirJWTStoreOption)
 
 // Creates a directory based jwt store.
 // Operates on files only, does NOT watch directories and files.
-func NewDirJWTStore(dirPath string, shard bool, create bool, _ ...dirJWTStoreOption) (*DirJWTStore, error) {
+func NewDirJWTStore(dirPath string, shard bool, create bool) (*DirJWTStore, error) {
 	fullPath, err := newDir(dirPath, create)
 	if err != nil {
 		return nil, err
@@ -131,14 +126,6 @@ func NewDirJWTStore(dirPath string, shard bool, create bool, _ ...dirJWTStoreOpt
 	return theStore, nil
 }
 
-type deleteType int
-
-const (
-	NoDelete deleteType = iota
-	RenameDeleted
-	HardDelete
-)
-
 // Creates a directory based jwt store.
 //
 // When ttl is set deletion of file is based on it and not on the jwt expiration
@@ -148,17 +135,16 @@ const (
 // evictOnLimit determines the behavior once limit is reached.
 //     true - Evict based on lru strategy
 //     false - return an error
-func NewExpiringDirJWTStore(dirPath string, shard bool, create bool, delete deleteType, expireCheck time.Duration, limit int64,
-	evictOnLimit bool, ttl time.Duration, changeNotification JWTChanged, _ ...dirJWTStoreOption) (*DirJWTStore, error) {
+func NewExpiringDirJWTStore(dirPath string, shard bool, create bool, expireCheck time.Duration, limit int64,
+	evictOnLimit bool, ttl time.Duration, changeNotification JWTChanged) (*DirJWTStore, error) {
 	fullPath, err := newDir(dirPath, create)
 	if err != nil {
 		return nil, err
 	}
 	theStore := &DirJWTStore{
-		directory:  fullPath,
-		shard:      shard,
-		deleteType: delete,
-		changed:    changeNotification,
+		directory: fullPath,
+		shard:     shard,
+		changed:   changeNotification,
 	}
 	if expireCheck <= 0 {
 		if ttl != 0 {
@@ -437,27 +423,17 @@ func (store *DirJWTStore) write(path string, publicKey string, theJWT string) (b
 func (store *DirJWTStore) delete(publicKey string) error {
 	if store.readonly {
 		return fmt.Errorf("store is read-only")
-	} else if store.deleteType == NoDelete {
-		return fmt.Errorf("store is not set up to for delete")
 	}
 	store.Lock()
 	defer store.Unlock()
-	name := store.pathForKey(publicKey)
-	if store.deleteType == RenameDeleted {
-		if err := os.Rename(name, name+".deleted"); err != nil {
-			if os.IsNotExist(err) {
-				return nil
-			}
-			return err
-		}
-	} else if err := os.Remove(name); err != nil {
-		if os.IsNotExist(err) {
+	if err := os.Remove(store.pathForKey(publicKey)); err != nil {
+		if _, ok := err.(*os.PathError); ok || err == os.ErrNotExist {
 			return nil
 		}
 		return err
 	}
 	store.expiration.unTrack(publicKey)
-	// TODO do cb so server can evict the account and associated clients
+	// TODO do cb
 	return nil
 }
 
diff --git a/server/events.go b/server/events.go
index 4d5b7330..21b38ca7 100644
--- a/server/events.go
+++ b/server/events.go
@@ -134,7 +134,6 @@ const DisconnectEventMsgType = "io.nats.server.advisory.v1.client_disconnect"
 // a given account when the number of connections changes. It will also HB
 // updates in the absence of any changes.
 type AccountNumConns struct {
-	TypedEvent
 	Server     ServerInfo `json:"server"`
 	Account    string     `json:"acc"`
 	Conns      int        `json:"conns"`
@@ -142,8 +141,6 @@ type AccountNumConns struct {
 	TotalConns int        `json:"total_conns"`
 }
 
-const AccountNumConnsMsgType = "io.nats.server.advisory.v1.account_connections"
-
 // accNumConnsReq is sent when we are starting to track an account for the first
 // time. We will request others send info to us about their local state.
 type accNumConnsReq struct {
@@ -1167,16 +1164,10 @@ func (s *Server) sendAccConnsUpdate(a *Account, subj ...string) {
 		return
 	}
 	// Build event with account name and number of local clients and leafnodes.
-	eid := s.nextEventID()
 	a.mu.Lock()
 	s.mu.Unlock()
 	localConns := a.numLocalConnections()
 	m := &AccountNumConns{
-		TypedEvent: TypedEvent{
-			Type: AccountNumConnsMsgType,
-			ID:   eid,
-			Time: time.Now().UTC(),
-		},
 		Account:    a.Name,
 		Conns:      localConns,
 		LeafNodes:  a.numLocalLeafNodes(),
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index c15ec57e..47ff5e25 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -366,6 +366,9 @@ type JSApiMsgGetResponse struct {
 
 const JSApiMsgGetResponseType = "io.nats.jetstream.api.v1.stream_msg_get_response"
 
+// JSWaitQueueDefaultMax is the default max number of outstanding requests for pull consumers.
+const JSWaitQueueDefaultMax = 512
+
 // JSApiConsumerCreateResponse.
 type JSApiConsumerCreateResponse struct {
 	ApiResponse
@@ -413,6 +416,13 @@ type JSApiConsumerListResponse struct {
 
 const JSApiConsumerListResponseType = "io.nats.jetstream.api.v1.consumer_list_response"
 
+// JSApiConsumerGetNextRequest is for getting next messages for pull based consumers.
+type JSApiConsumerGetNextRequest struct {
+	Expires time.Time `json:"expires,omitempty"`
+	Batch   int       `json:"batch,omitempty"`
+	NoWait  bool      `json:"no_wait,omitempty"`
+}
+
 // JSApiStreamTemplateCreateResponse for creating templates.
 type JSApiStreamTemplateCreateResponse struct {
 	ApiResponse
diff --git a/server/leafnode.go b/server/leafnode.go
index 9af10cfb..f7d7cd84 100644
--- a/server/leafnode.go
+++ b/server/leafnode.go
@@ -617,17 +617,6 @@ func (s *Server) createLeafNode(conn net.Conn, remote *leafNodeCfg) *client {
 	// Grab lock
 	c.mu.Lock()
 
-	// If client has been closed, this function will unlock and make sure
-	// the the connection is closed for proper clean-up.
-	isClosedUnlockAndReturn := func() bool {
-		if c.isClosed() {
-			c.mu.Unlock()
-			c.closeConnection(WriteError)
-			return true
-		}
-		return false
-	}
-
 	if solicited {
 		// We need to wait here for the info, but not for too long.
 		c.nc.SetReadDeadline(time.Now().Add(DEFAULT_LEAFNODE_INFO_WAIT))
@@ -660,12 +649,6 @@ func (s *Server) createLeafNode(conn net.Conn, remote *leafNodeCfg) *client {
 			return nil
 		}
 
-		// Not sure that can happen, but in case the connection was marked
-		// as closed during the call to parse...
-		if isClosedUnlockAndReturn() {
-			return nil
-		}
-
 		// Do TLS here as needed.
 		tlsRequired := remote.TLS || remote.TLSConfig != nil
 		if tlsRequired {
@@ -729,11 +712,6 @@ func (s *Server) createLeafNode(conn net.Conn, remote *leafNodeCfg) *client {
 
 			// Re-Grab lock
 			c.mu.Lock()
-
-			// Timeout may have closed the connection while the lock was released.
-			if isClosedUnlockAndReturn() {
-				return nil
-			}
 		}
 
 		if err := c.sendLeafConnect(clusterName, tlsRequired); err != nil {
@@ -758,12 +736,6 @@ func (s *Server) createLeafNode(conn net.Conn, remote *leafNodeCfg) *client {
 		// this before it can initiate the TLS handshake..
 		c.sendProtoNow(bytes.Join(pcs, []byte(" ")))
 
-		// The above call could have marked the connection as closed (due to
-		// TCP error), so if that is the case, bail out here.
-		if isClosedUnlockAndReturn() {
-			return nil
-		}
-
 		// Check to see if we need to spin up TLS.
 		if info.TLSRequired {
 			c.Debugf("Starting TLS leafnode server handshake")
@@ -790,17 +762,14 @@ func (s *Server) createLeafNode(conn net.Conn, remote *leafNodeCfg) *client {
 
 			// Indicate that handshake is complete (used in monitoring)
 			c.flags.set(handshakeComplete)
-
-			// Timeout may have closed the connection while the lock was released.
-			if isClosedUnlockAndReturn() {
-				return nil
-			}
 		}
 
 		// Leaf nodes will always require a CONNECT to let us know
 		// when we are properly bound to an account.
 		// The connection may have been closed
-		c.setAuthTimer(secondsToDuration(opts.LeafNode.AuthTimeout))
+		if !c.isClosed() {
+			c.setAuthTimer(secondsToDuration(opts.LeafNode.AuthTimeout))
+		}
 	}
 
 	// Keep track in case server is shutdown before we can successfully register.
diff --git a/server/monitor.go b/server/monitor.go
index 3451c62e..85598133 100644
--- a/server/monitor.go
+++ b/server/monitor.go
@@ -1934,12 +1934,6 @@ type ExtExport struct {
 	ApprovedAccounts []string `json:"approved_accounts,omitempty"`
 }
 
-type ExtVrIssues struct {
-	Description string `json:"description"`
-	Blocking    bool   `json:"blocking"`
-	Time        bool   `json:"time_check"`
-}
-
 type ExtMap map[string][]*MapDest
 
 type AccountInfo struct {
@@ -1956,7 +1950,6 @@ type AccountInfo struct {
 	Imports     []ExtImport        `json:"imports,omitempty"`
 	Jwt         string             `json:"jwt,omitempty"`
 	Claim       *jwt.AccountClaims `json:"decoded_jwt,omitempty"`
-	Vr          []ExtVrIssues      `json:"validation_result_jwt,omitempty"`
 }
 
 type Accountz struct {
@@ -2012,105 +2005,68 @@ func (s *Server) accountInfo(accName string) (*AccountInfo, error) {
 	}
 	a.mu.RLock()
 	defer a.mu.RUnlock()
-	var vrIssues []ExtVrIssues
-	claim, _ := jwt.DecodeAccountClaims(a.claimJWT) //ignore error
-	if claim != nil {
-		vr := jwt.ValidationResults{}
-		claim.Validate(&vr)
-		vrIssues = make([]ExtVrIssues, len(vr.Issues))
-		for i, v := range vr.Issues {
-			vrIssues[i] = ExtVrIssues{v.Description, v.Blocking, v.TimeCheck}
-		}
-	}
+	claim, _ := jwt.DecodeAccountClaims(a.claimJWT) // ignore error
 	exports := []ExtExport{}
 	for k, v := range a.exports.services {
 		e := ExtExport{
 			Export: jwt.Export{
-				Subject: jwt.Subject(k),
-				Type:    jwt.Service,
+				Subject:      jwt.Subject(k),
+				Type:         jwt.Service,
+				TokenReq:     v.tokenReq,
+				ResponseType: jwt.ResponseType(v.respType.String()),
 			},
 			ApprovedAccounts: []string{},
 		}
-		if v != nil {
-			e.TokenReq = v.tokenReq
-			e.ResponseType = jwt.ResponseType(v.respType.String())
-			for name := range v.approved {
-				e.ApprovedAccounts = append(e.ApprovedAccounts, name)
-			}
+		for name := range v.approved {
+			e.ApprovedAccounts = append(e.ApprovedAccounts, name)
 		}
 		exports = append(exports, e)
 	}
 	for k, v := range a.exports.streams {
 		e := ExtExport{
 			Export: jwt.Export{
-				Subject: jwt.Subject(k),
-				Type:    jwt.Stream,
+				Subject:  jwt.Subject(k),
+				Type:     jwt.Stream,
+				TokenReq: v.tokenReq,
 			},
 			ApprovedAccounts: []string{},
 		}
-		if v != nil {
-			e.TokenReq = v.tokenReq
-			for name := range v.approved {
-				e.ApprovedAccounts = append(e.ApprovedAccounts, name)
-			}
+		for name := range v.approved {
+			e.ApprovedAccounts = append(e.ApprovedAccounts, name)
 		}
 		exports = append(exports, e)
 	}
 	imports := []ExtImport{}
 	for _, v := range a.imports.streams {
-		imp := ExtImport{
-			Invalid: true,
-			Import:  jwt.Import{Type: jwt.Stream},
-		}
-		if v != nil {
-			imp.Invalid = v.invalid
-			imp.Import = jwt.Import{
+		imports = append(imports, ExtImport{
+			Import: jwt.Import{
 				Subject: jwt.Subject(v.from),
 				Account: v.acc.Name,
 				Type:    jwt.Stream,
 				To:      jwt.Subject(v.to),
-			}
-		}
-		imports = append(imports, imp)
+			},
+			Invalid: v.invalid,
+		})
 	}
 	for _, v := range a.imports.services {
-		imp := ExtImport{
-			Invalid: true,
-			Import:  jwt.Import{Type: jwt.Service},
-		}
-		if v != nil {
-			imp.Invalid = v.invalid
-			imp.Import = jwt.Import{
+		imports = append(imports, ExtImport{
+			Import: jwt.Import{
 				Subject: jwt.Subject(v.from),
 				Account: v.acc.Name,
 				Type:    jwt.Service,
 				To:      jwt.Subject(v.to),
-			}
-		}
-		imports = append(imports, imp)
+			},
+			Invalid: v.invalid,
+		})
 	}
+
 	mappings := ExtMap{}
 	for _, m := range a.mappings {
 		var dests []*MapDest
-		src := ""
-		if m == nil {
-			src = "nil"
-			if _, ok := mappings[src]; ok { // only set if not present (keep orig in case nil is used)
-				continue
-			}
-			dests = append(dests, &MapDest{})
-		} else {
-			src = m.src
-			for _, d := range m.dests {
-				dests = append(dests, &MapDest{d.tr.dest, d.weight, ""})
-			}
-			for c, cd := range m.cdests {
-				for _, d := range cd {
-					dests = append(dests, &MapDest{d.tr.dest, d.weight, c})
-				}
-			}
+		for _, d := range m.dests {
+			dests = append(dests, &MapDest{d.tr.dest, d.weight, ""})
 		}
-		mappings[src] = dests
+		mappings[m.src] = dests
 	}
 
 	return &AccountInfo{
@@ -2127,6 +2083,5 @@ func (s *Server) accountInfo(accName string) (*AccountInfo, error) {
 		imports,
 		a.claimJWT,
 		claim,
-		vrIssues,
 	}, nil
 }
diff --git a/server/opts.go b/server/opts.go
index 1dfbd054..71b905fd 100644
--- a/server/opts.go
+++ b/server/opts.go
@@ -849,7 +849,6 @@ func (o *Options) processConfigFileLine(k string, v interface{}, errors *[]error
 				}
 			}
 		case map[string]interface{}:
-			del := false
 			dir := ""
 			dirType := ""
 			limit := int64(0)
@@ -864,10 +863,6 @@ func (o *Options) processConfigFileLine(k string, v interface{}, errors *[]error
 				_, v := unwrapValue(v, &lt)
 				dirType = v.(string)
 			}
-			if v, ok := v["allow_delete"]; ok {
-				_, v := unwrapValue(v, &lt)
-				del = v.(bool)
-			}
 			if v, ok := v["limit"]; ok {
 				_, v := unwrapValue(v, &lt)
 				limit = v.(int64)
@@ -898,15 +893,12 @@ func (o *Options) processConfigFileLine(k string, v interface{}, errors *[]error
 				if sync != 0 {
 					*errors = append(*errors, &configErr{tk, "CACHE does not accept sync"})
 				}
-				if del {
-					*errors = append(*errors, &configErr{tk, "CACHE does not accept allow_delete"})
-				}
 				res, err = NewCacheDirAccResolver(dir, limit, ttl)
 			case "FULL":
 				if ttl != 0 {
 					*errors = append(*errors, &configErr{tk, "FULL does not accept ttl"})
 				}
-				res, err = NewDirAccResolver(dir, limit, sync, del)
+				res, err = NewDirAccResolver(dir, limit, sync)
 			}
 			if err != nil {
 				*errors = append(*errors, &configErr{tk, err.Error()})
diff --git a/server/server.go b/server/server.go
index ef3a0eb0..f46692b3 100644
--- a/server/server.go
+++ b/server/server.go
@@ -1221,7 +1221,7 @@ func (s *Server) lookupAccount(name string) (*Account, error) {
 	}
 	// If we have a resolver see if it can fetch the account.
 	if s.AccountResolver() == nil {
-		return nil, ErrNoAccountResolver
+		return nil, ErrMissingAccount
 	}
 	return s.fetchAccount(name)
 }
@@ -1262,8 +1262,7 @@ func (s *Server) updateAccountWithClaimJWT(acc *Account, claimJWT string) error
 		acc.mu.Lock()
 		if acc.Issuer == "" {
 			acc.Issuer = accClaims.Issuer
-		}
-		if acc.Name != accClaims.Subject {
+		} else if acc.Issuer != accClaims.Issuer {
 			acc.mu.Unlock()
 			return ErrAccountValidation
 		}
diff --git a/server/stream.go b/server/stream.go
index 0361ee57..a2b2d288 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -945,6 +945,7 @@ func (mset *Stream) internalSendLoop() {
 				c.pa.hdb = nil
 				msg = append(pm.msg, _CRLF_...)
 			}
+
 			didDeliver := c.processInboundClientMsg(msg)
 			c.pa.szb = nil
 			c.flushClients(0)
diff --git a/vendor/github.com/nats-io/nats.go/.travis.yml b/vendor/github.com/nats-io/nats.go/.travis.yml
index 3b00ae19..dcbac5be 100644
--- a/vendor/github.com/nats-io/nats.go/.travis.yml
+++ b/vendor/github.com/nats-io/nats.go/.travis.yml
@@ -1,7 +1,7 @@
 language: go
 go:
+- 1.15.x
 - 1.14.x
-- 1.13.x
 env:
 - GO111MODULE=off
 go_import_path: github.com/nats-io/nats.go
@@ -20,4 +20,4 @@ before_script:
 script:
 - go test -i -race ./...
 - go test -v -run=TestNoRace -p=1 ./...
-- if [[ "$TRAVIS_GO_VERSION" =~ 1.14 ]]; then ./scripts/cov.sh TRAVIS; else go test -race -v -p=1 ./... --failfast; fi
+- if [[ "$TRAVIS_GO_VERSION" =~ 1.15 ]]; then ./scripts/cov.sh TRAVIS; else go test -race -v -p=1 ./... --failfast; fi
diff --git a/vendor/github.com/nats-io/nats.go/MAINTAINERS.md b/vendor/github.com/nats-io/nats.go/MAINTAINERS.md
index 323faa8e..23214655 100644
--- a/vendor/github.com/nats-io/nats.go/MAINTAINERS.md
+++ b/vendor/github.com/nats-io/nats.go/MAINTAINERS.md
@@ -2,9 +2,7 @@
 
 Maintainership is on a per project basis.
 
-### Core-maintainers
+### Maintainers
   - Derek Collison <derek@nats.io> [@derekcollison](https://github.com/derekcollison)
   - Ivan Kozlovic <ivan@nats.io> [@kozlovic](https://github.com/kozlovic)
-  
-### Maintainers
-  - Waldemar Quevedo <wally@nats.io> [@wallyqs](https://github.com/wallyqs)
\ No newline at end of file
+  - Waldemar Quevedo <wally@nats.io> [@wallyqs](https://github.com/wallyqs)
diff --git a/vendor/github.com/nats-io/nats.go/README.md b/vendor/github.com/nats-io/nats.go/README.md
index 5eb7c968..8b6e6f08 100644
--- a/vendor/github.com/nats-io/nats.go/README.md
+++ b/vendor/github.com/nats-io/nats.go/README.md
@@ -3,7 +3,8 @@ A [Go](http://golang.org) client for the [NATS messaging system](https://nats.io
 
 [![License Apache 2](https://img.shields.io/badge/License-Apache2-blue.svg)](https://www.apache.org/licenses/LICENSE-2.0)
 [![FOSSA Status](https://app.fossa.io/api/projects/git%2Bgithub.com%2Fnats-io%2Fgo-nats.svg?type=shield)](https://app.fossa.io/projects/git%2Bgithub.com%2Fnats-io%2Fgo-nats?ref=badge_shield)
-[![Go Report Card](https://goreportcard.com/badge/github.com/nats-io/nats.go)](https://goreportcard.com/report/github.com/nats-io/nats.go) [![Build Status](https://travis-ci.org/nats-io/nats.go.svg?branch=master)](http://travis-ci.org/nats-io/nats.go) [![GoDoc](https://godoc.org/github.com/nats-io/nats.go?status.svg)](http://godoc.org/github.com/nats-io/nats.go) [![Coverage Status](https://coveralls.io/repos/nats-io/nats.go/badge.svg?branch=master)](https://coveralls.io/r/nats-io/nats.go?branch=master)
+[![Go Report Card](https://goreportcard.com/badge/github.com/nats-io/nats.go)](https://goreportcard.com/report/github.com/nats-io/nats.go) [![Build Status](https://travis-ci.org/nats-io/nats.go.svg?branch=master)](http://travis-ci.org/nats-io/nats.go) [![GoDoc](https://img.shields.io/badge/GoDoc-reference-007d9c)](https://pkg.go.dev/github.com/nats-io/nats.go)
+ [![Coverage Status](https://coveralls.io/repos/nats-io/nats.go/badge.svg?branch=master)](https://coveralls.io/r/nats-io/nats.go?branch=master)
 
 ## Installation
 
@@ -284,6 +285,21 @@ nc.QueueSubscribe("foo", "job_workers", func(_ *Msg) {
 
 ```go
 
+// Normally, the library will return an error when trying to connect and
+// there is no server running. The RetryOnFailedConnect option will set
+// the connection in reconnecting state if it failed to connect right away.
+nc, err := nats.Connect(nats.DefaultURL,
+    nats.RetryOnFailedConnect(true),
+    nats.MaxReconnects(10),
+    nats.ReconnectWait(time.Second),
+    nats.ReconnectHandler(func(_ *nats.Conn) {
+        // Note that this will be invoked for the first asynchronous connect.
+    }))
+if err != nil {
+    // Should not return an error even if it can't connect, but you still
+    // need to check in case there are some configuration errors.
+}
+
 // Flush connection to server, returns when all messages have been processed.
 nc.Flush()
 fmt.Println("All clear!")
diff --git a/vendor/github.com/nats-io/nats.go/context.go b/vendor/github.com/nats-io/nats.go/context.go
index 769f88a0..4aa3bf0e 100644
--- a/vendor/github.com/nats-io/nats.go/context.go
+++ b/vendor/github.com/nats-io/nats.go/context.go
@@ -58,34 +58,37 @@ func (nc *Conn) requestWithContext(ctx context.Context, subj string, hdr, data [
 		return nil, ctx.Err()
 	}
 
-	nc.mu.Lock()
-	// If user wants the old style.
-	if nc.Opts.UseOldRequestStyle {
-		nc.mu.Unlock()
-		return nc.oldRequestWithContext(ctx, subj, hdr, data)
-	}
-
-	mch, token, err := nc.createNewRequestAndSend(subj, hdr, data)
-	if err != nil {
-		return nil, err
-	}
+	var m *Msg
+	var err error
 
-	var ok bool
-	var msg *Msg
+	// If user wants the old style.
+	if nc.useOldRequestStyle() {
+		m, err = nc.oldRequestWithContext(ctx, subj, hdr, data)
+	} else {
+		mch, token, err := nc.createNewRequestAndSend(subj, hdr, data)
+		if err != nil {
+			return nil, err
+		}
 
-	select {
-	case msg, ok = <-mch:
-		if !ok {
-			return nil, ErrConnectionClosed
+		var ok bool
+
+		select {
+		case m, ok = <-mch:
+			if !ok {
+				return nil, ErrConnectionClosed
+			}
+		case <-ctx.Done():
+			nc.mu.Lock()
+			delete(nc.respMap, token)
+			nc.mu.Unlock()
+			return nil, ctx.Err()
 		}
-	case <-ctx.Done():
-		nc.mu.Lock()
-		delete(nc.respMap, token)
-		nc.mu.Unlock()
-		return nil, ctx.Err()
 	}
-
-	return msg, nil
+	// Check for no responder status.
+	if err == nil && len(m.Data) == 0 && m.Header.Get(statusHdr) == noResponders {
+		m, err = nil, ErrNoResponders
+	}
+	return m, err
 }
 
 // oldRequestWithContext utilizes inbox and subscription per request.
diff --git a/vendor/github.com/nats-io/nats.go/enc.go b/vendor/github.com/nats-io/nats.go/enc.go
index ee832f69..38149c98 100644
--- a/vendor/github.com/nats-io/nats.go/enc.go
+++ b/vendor/github.com/nats-io/nats.go/enc.go
@@ -130,7 +130,7 @@ func (c *EncodedConn) Request(subject string, v interface{}, vPtr interface{}, t
 
 // Handler is a specific callback used for Subscribe. It is generalized to
 // an interface{}, but we will discover its format and arguments at runtime
-// and perform the correct callback, including de-marshaling JSON strings
+// and perform the correct callback, including de-marshaling encoded data
 // back into the appropriate struct based on the signature of the Handler.
 //
 // Handlers are expected to have one of four signatures.
diff --git a/vendor/github.com/nats-io/nats.go/go.mod b/vendor/github.com/nats-io/nats.go/go.mod
index 59ee43d3..383f2230 100644
--- a/vendor/github.com/nats-io/nats.go/go.mod
+++ b/vendor/github.com/nats-io/nats.go/go.mod
@@ -4,7 +4,7 @@ go 1.14
 
 require (
 	github.com/golang/protobuf v1.4.2
-	github.com/nats-io/nats-server/v2 v2.1.8-0.20200601203034-f8d6dd992b71
+	github.com/nats-io/nats-server/v2 v2.1.8-0.20200929001935-7f44d075f7ad
 	github.com/nats-io/nkeys v0.2.0
 	github.com/nats-io/nuid v1.0.1
 	google.golang.org/protobuf v1.23.0
diff --git a/vendor/github.com/nats-io/nats.go/go.sum b/vendor/github.com/nats-io/nats.go/go.sum
index f31d0af3..1419245b 100644
--- a/vendor/github.com/nats-io/nats.go/go.sum
+++ b/vendor/github.com/nats-io/nats.go/go.sum
@@ -7,17 +7,23 @@ github.com/golang/protobuf v1.4.2 h1:+Z5KGCizgyZCbGh1KZqA0fcLLkwbsjIzS4aV2v7wJX0
 github.com/golang/protobuf v1.4.2/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
 github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
 github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
+github.com/google/go-cmp v0.4.0 h1:xsAVV57WRhGj6kEIi8ReJzQlHHqcBYCElAvkovg3B/4=
 github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/minio/highwayhash v1.0.0 h1:iMSDhgUILCr0TNm8LWlSjF8N0ZIj2qbO8WHp6Q/J2BA=
 github.com/minio/highwayhash v1.0.0/go.mod h1:xQboMTeM9nY9v/LlAOxFctujiv5+Aq2hR5dxBpaMbdc=
 github.com/nats-io/jwt v0.3.2/go.mod h1:/euKqTS1ZD+zzjYrY7pseZrTtWQSjujC7xjPc8wL6eU=
 github.com/nats-io/jwt v0.3.3-0.20200519195258-f2bf5ce574c7 h1:RnGotxlghqR5D2KDAu4TyuLqyjuylOsJiAFhXvMvQIc=
 github.com/nats-io/jwt v0.3.3-0.20200519195258-f2bf5ce574c7/go.mod h1:n3cvmLfBfnpV4JJRN7lRYCyZnw48ksGsbThGXEk4w9M=
+github.com/nats-io/jwt/v2 v2.0.0-20200916203241-1f8ce17dff02 h1:WloZv3SCb55D/rOHYy1rWBXLrj3BYc9zw8VIq6X54lI=
+github.com/nats-io/jwt/v2 v2.0.0-20200916203241-1f8ce17dff02/go.mod h1:vs+ZEjP+XKy8szkBmQwCB7RjYdIlMaPsFPs4VdS4bTQ=
 github.com/nats-io/nats-server/v2 v2.1.8-0.20200524125952-51ebd92a9093/go.mod h1:rQnBf2Rv4P9adtAs/Ti6LfFmVtFG6HLhl/H7cVshcJU=
 github.com/nats-io/nats-server/v2 v2.1.8-0.20200601203034-f8d6dd992b71 h1:nexMtKbOeM+w3vGQMNF0BEt+2xZDmVCtYXql2Ym+RWg=
 github.com/nats-io/nats-server/v2 v2.1.8-0.20200601203034-f8d6dd992b71/go.mod h1:Nan/1L5Sa1JRW+Thm4HNYcIDcVRFc5zK9OpSZeI2kk4=
+github.com/nats-io/nats-server/v2 v2.1.8-0.20200929001935-7f44d075f7ad h1:oRb9MIi1Y4N5cTZWciqH68aVNt1e+o4N2uRnjVzv/UE=
+github.com/nats-io/nats-server/v2 v2.1.8-0.20200929001935-7f44d075f7ad/go.mod h1:TkHpUIDETmTI7mrHN40D1pzxfzHZuGmtMbtb83TGVQw=
 github.com/nats-io/nats.go v1.10.0/go.mod h1:AjGArbfyR50+afOUotNX2Xs5SYHf+CoOa5HH1eEl2HE=
 github.com/nats-io/nats.go v1.10.1-0.20200531124210-96f2130e4d55/go.mod h1:ARiFsjW9DVxk48WJbO3OSZ2DG8fjkMi7ecLmXoY/n9I=
+github.com/nats-io/nats.go v1.10.1-0.20200606002146-fc6fed82929a/go.mod h1:8eAIv96Mo9QW6Or40jUHejS7e4VwZ3VRYD6Sf0BTDp4=
 github.com/nats-io/nkeys v0.1.3/go.mod h1:xpnFELMwJABBLVhffcfd1MZx6VsNRFpEugbxziKVo7w=
 github.com/nats-io/nkeys v0.1.4/go.mod h1:XdZpAbhgyyODYqjTawOnIOI7VlbKSarI9Gfy1tqEu/s=
 github.com/nats-io/nkeys v0.2.0 h1:WXKF7diOaPU9cJdLD7nuzwasQy9vT1tBqzXZZf3AMJM=
@@ -35,6 +41,9 @@ golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20191022100944-742c48ecaeb7 h1:HmbHVPwrPEKPGLAcHSrMe6+hqSUlvZU0rab6x5EXfGU=
 golang.org/x/sys v0.0.0-20191022100944-742c48ecaeb7/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/time v0.0.0-20200416051211-89c76fbcd5d1 h1:NusfzzA6yGQ+ua51ck7E3omNUX/JuqbFSaRGqU8CcLI=
+golang.org/x/time v0.0.0-20200416051211-89c76fbcd5d1/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
 google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0=
diff --git a/vendor/github.com/nats-io/nats.go/jetstream.go b/vendor/github.com/nats-io/nats.go/jetstream.go
new file mode 100644
index 00000000..4d2691ca
--- /dev/null
+++ b/vendor/github.com/nats-io/nats.go/jetstream.go
@@ -0,0 +1,366 @@
+// Copyright 2020 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 nats
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"strconv"
+	"time"
+)
+
+// JetStreamMsgMetaData is metadata related to a JetStream originated message
+type JetStreamMsgMetaData struct {
+	Stream      string
+	Consumer    string
+	Parsed      bool
+	Delivered   int
+	StreamSeq   int
+	ConsumerSeq int
+	TimeStamp   time.Time
+}
+
+func (m *Msg) JetStreamMetaData() (*JetStreamMsgMetaData, error) {
+	var err error
+
+	if m.jsMeta != nil && m.jsMeta.Parsed {
+		return m.jsMeta, nil
+	}
+
+	m.jsMeta, err = m.parseJSMsgMetadata()
+
+	return m.jsMeta, err
+}
+
+func (m *Msg) parseJSMsgMetadata() (*JetStreamMsgMetaData, error) {
+	if m.jsMeta != nil {
+		return m.jsMeta, nil
+	}
+
+	if len(m.Reply) == 0 {
+		return nil, ErrNotJSMessage
+	}
+
+	meta := &JetStreamMsgMetaData{}
+
+	tsa := [32]string{}
+	parts := tsa[:0]
+	start := 0
+	btsep := byte('.')
+	for i := 0; i < len(m.Reply); i++ {
+		if m.Reply[i] == btsep {
+			parts = append(parts, m.Reply[start:i])
+			start = i + 1
+		}
+	}
+	parts = append(parts, m.Reply[start:])
+
+	if len(parts) != 8 || parts[0] != "$JS" || parts[1] != "ACK" {
+		return nil, ErrNotJSMessage
+	}
+
+	var err error
+
+	meta.Stream = parts[2]
+	meta.Consumer = parts[3]
+	meta.Delivered, err = strconv.Atoi(parts[4])
+	if err != nil {
+		return nil, ErrNotJSMessage
+	}
+
+	meta.StreamSeq, err = strconv.Atoi(parts[5])
+	if err != nil {
+		return nil, ErrNotJSMessage
+	}
+
+	meta.ConsumerSeq, err = strconv.Atoi(parts[6])
+	if err != nil {
+		return nil, ErrNotJSMessage
+	}
+
+	tsi, err := strconv.Atoi(parts[7])
+	if err != nil {
+		return nil, ErrNotJSMessage
+	}
+	meta.TimeStamp = time.Unix(0, int64(tsi))
+
+	meta.Parsed = true
+
+	return meta, nil
+}
+
+const jsStreamUnspecified = "not.set"
+
+type jsOpts struct {
+	timeout time.Duration
+	ctx     context.Context
+
+	ackstr     string
+	consumer   *ConsumerConfig
+	streamName string
+}
+
+func newJsOpts() *jsOpts {
+	return &jsOpts{ackstr: jsStreamUnspecified, consumer: &ConsumerConfig{}}
+}
+
+func (j *jsOpts) context(dftl time.Duration) (context.Context, context.CancelFunc) {
+	if j.ctx != nil {
+		return context.WithCancel(j.ctx)
+	}
+
+	if j.timeout == 0 {
+		j.timeout = dftl
+	}
+
+	return context.WithTimeout(context.Background(), j.timeout)
+}
+
+// AckOption configures the various JetStream message acknowledgement helpers
+type AckOption func(opts *jsOpts) error
+
+// PublishOption configures publishing messages
+type PublishOption func(opts *jsOpts) error
+
+// SubscribeOption configures JetStream consumer behavior
+type SubscribeOption func(opts *jsOpts) error
+
+// Consumer creates a JetStream Consumer on a Stream
+func Consumer(stream string, cfg ConsumerConfig) SubscribeOption {
+	return func(jopts *jsOpts) error {
+		jopts.consumer = &cfg
+		jopts.streamName = stream
+		return nil
+	}
+}
+
+// PublishExpectsStream waits for an ack after publishing and ensure it's from a specific stream, empty arguments waits for any valid acknowledgement
+func PublishExpectsStream(stream ...string) PublishOption {
+	return func(opts *jsOpts) error {
+		switch len(stream) {
+		case 0:
+			opts.ackstr = ""
+		case 1:
+			opts.ackstr = stream[0]
+			if !isValidJSName(opts.ackstr) {
+				return ErrInvalidStreamName
+			}
+		default:
+			return ErrMultiStreamUnsupported
+		}
+
+		return nil
+	}
+}
+
+// PublishStreamTimeout sets the period of time to wait for JetStream to acknowledge receipt, defaults to JetStreamTimeout option
+func PublishStreamTimeout(t time.Duration) PublishOption {
+	return func(opts *jsOpts) error {
+		opts.timeout = t
+		return nil
+	}
+}
+
+// PublishCtx sets an interrupt context for waiting on a stream to reply
+func PublishCtx(ctx context.Context) PublishOption {
+	return func(opts *jsOpts) error {
+		opts.ctx = ctx
+		return nil
+	}
+}
+
+// AckWaitDuration waits for confirmation from the JetStream server
+func AckWaitDuration(d time.Duration) AckOption {
+	return func(opts *jsOpts) error {
+		opts.timeout = d
+		return nil
+	}
+}
+
+func (m *Msg) jsAck(body []byte, opts ...AckOption) error {
+	if m.Reply == "" {
+		return ErrMsgNoReply
+	}
+
+	if m == nil || m.Sub == nil {
+		return ErrMsgNotBound
+	}
+
+	m.Sub.mu.Lock()
+	nc := m.Sub.conn
+	m.Sub.mu.Unlock()
+
+	var err error
+	var aopts *jsOpts
+
+	if len(opts) > 0 {
+		aopts = newJsOpts()
+		for _, f := range opts {
+			if err = f(aopts); err != nil {
+				return err
+			}
+		}
+	}
+
+	if aopts == nil || aopts.timeout == 0 {
+		return m.Respond(body)
+	}
+
+	_, err = nc.Request(m.Reply, body, aopts.timeout)
+
+	return err
+}
+
+// Ack acknowledges a JetStream messages received from a Consumer, indicating the message
+// should not be received again later
+func (m *Msg) Ack(opts ...AckOption) error {
+	return m.jsAck(AckAck, opts...)
+}
+
+// Nak acknowledges a JetStream message received from a Consumer, indicating that the message
+// is not completely processed and should be sent again later
+func (m *Msg) Nak(opts ...AckOption) error {
+	return m.jsAck(AckNak, opts...)
+}
+
+// AckProgress acknowledges a Jetstream message received from a Consumer, indicating that work is
+// ongoing and further processing time is required equal to the configured AckWait of the Consumer
+func (m *Msg) AckProgress(opts ...AckOption) error {
+	return m.jsAck(AckProgress, opts...)
+}
+
+// AckNext performs an Ack() and request that the next message be sent to subject ib
+func (m *Msg) AckNext(ib string) error {
+	return m.RespondMsg(&Msg{Subject: m.Reply, Reply: ib, Data: AckNext})
+}
+
+// AckAndFetch performs an AckNext() and returns the next message from the stream
+func (m *Msg) AckAndFetch(opts ...AckOption) (*Msg, error) {
+	if m.Reply == "" {
+		return nil, ErrMsgNoReply
+	}
+
+	if m == nil || m.Sub == nil {
+		return nil, ErrMsgNotBound
+	}
+
+	m.Sub.mu.Lock()
+	nc := m.Sub.conn
+	m.Sub.mu.Unlock()
+
+	var err error
+
+	aopts := newJsOpts()
+	for _, f := range opts {
+		if err = f(aopts); err != nil {
+			return nil, err
+		}
+	}
+
+	ctx, cancel := aopts.context(nc.Opts.JetStreamTimeout)
+	defer cancel()
+
+	sub, err := nc.SubscribeSync(NewInbox())
+	if err != nil {
+		return nil, err
+	}
+	sub.AutoUnsubscribe(1)
+	defer sub.Unsubscribe()
+
+	err = m.RespondMsg(&Msg{Reply: sub.Subject, Data: AckNext, Subject: m.Reply})
+	if err != nil {
+		return nil, err
+	}
+	nc.Flush()
+
+	return sub.NextMsgWithContext(ctx)
+}
+
+// AckTerm acknowledges a message received from JetStream indicating the message will not be processed
+// and should not be sent to another consumer
+func (m *Msg) AckTerm(opts ...AckOption) error {
+	return m.jsAck(AckTerm, opts...)
+}
+
+// JetStreamPublishAck metadata received from JetStream when publishing messages
+type JetStreamPublishAck struct {
+	Stream   string `json:"stream"`
+	Sequence int    `json:"seq"`
+}
+
+// ParsePublishAck parses the publish acknowledgement sent by JetStream
+func ParsePublishAck(m []byte) (*JetStreamPublishAck, error) {
+	if bytes.HasPrefix([]byte("-ERR"), m) {
+		if len(m) > 7 {
+			return nil, fmt.Errorf(string(m[6 : len(m)-1]))
+		}
+
+		return nil, fmt.Errorf(string(m))
+	}
+
+	if !bytes.HasPrefix(m, []byte("+OK {")) {
+		return nil, fmt.Errorf("invalid JetStream Ack: %v", string(m))
+	}
+
+	ack := &JetStreamPublishAck{}
+	err := json.Unmarshal(m[3:], ack)
+	return ack, err
+}
+
+func (nc *Conn) jsPublish(subj string, data []byte, opts []PublishOption) error {
+	var err error
+	var aopts *jsOpts
+
+	if len(opts) > 0 {
+		aopts = newJsOpts()
+		for _, f := range opts {
+			if err = f(aopts); err != nil {
+				return err
+			}
+		}
+	}
+
+	if aopts == nil || aopts.timeout == 0 && aopts.ctx == nil && aopts.ackstr == jsStreamUnspecified {
+		return nc.publish(subj, _EMPTY_, nil, data)
+	}
+
+	ctx, cancel := aopts.context(nc.Opts.JetStreamTimeout)
+	defer cancel()
+
+	resp, err := nc.RequestWithContext(ctx, subj, data)
+	if err != nil {
+		return err
+	}
+
+	ack, err := ParsePublishAck(resp.Data)
+	if err != nil {
+		return err
+	}
+
+	if ack.Stream == "" || ack.Sequence == 0 {
+		return ErrInvalidJSAck
+	}
+
+	if aopts.ackstr == jsStreamUnspecified || aopts.ackstr == "" {
+		return nil
+	}
+
+	if ack.Stream == aopts.ackstr {
+		return nil
+	}
+
+	return fmt.Errorf("received ack from stream %q", ack.Stream)
+}
diff --git a/vendor/github.com/nats-io/nats.go/jetstream_consumer.go b/vendor/github.com/nats-io/nats.go/jetstream_consumer.go
new file mode 100644
index 00000000..9c598594
--- /dev/null
+++ b/vendor/github.com/nats-io/nats.go/jetstream_consumer.go
@@ -0,0 +1,268 @@
+// Copyright 2020 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 nats
+
+import (
+	"encoding/json"
+	"fmt"
+	"strings"
+	"time"
+)
+
+func (nc *Conn) createOrUpdateConsumer(opts *jsOpts, delivery string) (*ConsumerInfo, error) {
+	if opts.streamName == "" {
+		return nil, ErrStreamNameRequired
+	}
+	if opts.consumer == nil {
+		return nil, ErrConsumerConfigRequired
+	}
+
+	crj, err := json.Marshal(&jSApiConsumerCreateRequest{
+		Stream: opts.streamName,
+		Config: consumerConfig{DeliverSubject: delivery, ConsumerConfig: opts.consumer},
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	ctx, cancel := opts.context(nc.Opts.JetStreamTimeout)
+	defer cancel()
+
+	var subj string
+	switch len(opts.consumer.Durable) {
+	case 0:
+		subj = fmt.Sprintf(jSApiConsumerCreateT, opts.streamName)
+	default:
+		subj = fmt.Sprintf(jSApiDurableCreateT, opts.streamName, opts.consumer.Durable)
+	}
+
+	resp, err := nc.RequestWithContext(ctx, subj, crj)
+	if err != nil {
+		return nil, err
+	}
+
+	cresp := &jSApiConsumerCreateResponse{}
+	err = json.Unmarshal(resp.Data, cresp)
+	if err != nil {
+		return nil, err
+	}
+
+	if cresp.Error != nil {
+		return nil, cresp.Error
+	}
+
+	return cresp.ConsumerInfo, nil
+}
+
+const (
+	jSApiConsumerCreateT = "$JS.API.CONSUMER.CREATE.%s"
+	jSApiDurableCreateT  = "$JS.API.CONSUMER.DURABLE.CREATE.%s.%s"
+)
+
+type apiError struct {
+	Code        int    `json:"code"`
+	Description string `json:"description,omitempty"`
+}
+
+// Error implements error
+func (e apiError) Error() string {
+	switch {
+	case e.Description == "" && e.Code == 0:
+		return "unknown JetStream Error"
+	case e.Description == "" && e.Code > 0:
+		return fmt.Sprintf("unknown JetStream %d Error", e.Code)
+	default:
+		return e.Description
+	}
+}
+
+type jSApiResponse struct {
+	Type  string    `json:"type"`
+	Error *apiError `json:"error,omitempty"`
+}
+
+// io.nats.jetstream.api.v1.consumer_create_request
+type jSApiConsumerCreateRequest struct {
+	Stream string         `json:"stream_name"`
+	Config consumerConfig `json:"config"`
+}
+
+// io.nats.jetstream.api.v1.consumer_create_response
+type jSApiConsumerCreateResponse struct {
+	jSApiResponse
+	*ConsumerInfo
+}
+
+type AckPolicy int
+
+const (
+	AckNone AckPolicy = iota
+	AckAll
+	AckExplicit
+)
+
+func (p *AckPolicy) UnmarshalJSON(data []byte) error {
+	switch string(data) {
+	case jsonString("none"):
+		*p = AckNone
+	case jsonString("all"):
+		*p = AckAll
+	case jsonString("explicit"):
+		*p = AckExplicit
+	default:
+		return fmt.Errorf("can not unmarshal %q", data)
+	}
+
+	return nil
+}
+
+func (p AckPolicy) MarshalJSON() ([]byte, error) {
+	switch p {
+	case AckNone:
+		return json.Marshal("none")
+	case AckAll:
+		return json.Marshal("all")
+	case AckExplicit:
+		return json.Marshal("explicit")
+	default:
+		return nil, fmt.Errorf("unknown acknowlegement policy %v", p)
+	}
+}
+
+type ReplayPolicy int
+
+const (
+	ReplayInstant ReplayPolicy = iota
+	ReplayOriginal
+)
+
+func (p *ReplayPolicy) UnmarshalJSON(data []byte) error {
+	switch string(data) {
+	case jsonString("instant"):
+		*p = ReplayInstant
+	case jsonString("original"):
+		*p = ReplayOriginal
+	default:
+		return fmt.Errorf("can not unmarshal %q", data)
+	}
+
+	return nil
+}
+
+func (p ReplayPolicy) MarshalJSON() ([]byte, error) {
+	switch p {
+	case ReplayOriginal:
+		return json.Marshal("original")
+	case ReplayInstant:
+		return json.Marshal("instant")
+	default:
+		return nil, fmt.Errorf("unknown replay policy %v", p)
+	}
+}
+
+var (
+	AckAck      = []byte("+ACK")
+	AckNak      = []byte("-NAK")
+	AckProgress = []byte("+WPI")
+	AckNext     = []byte("+NXT")
+	AckTerm     = []byte("+TERM")
+)
+
+type DeliverPolicy int
+
+const (
+	DeliverAll DeliverPolicy = iota
+	DeliverLast
+	DeliverNew
+	DeliverByStartSequence
+	DeliverByStartTime
+)
+
+func (p *DeliverPolicy) UnmarshalJSON(data []byte) error {
+	switch string(data) {
+	case jsonString("all"), jsonString("undefined"):
+		*p = DeliverAll
+	case jsonString("last"):
+		*p = DeliverLast
+	case jsonString("new"):
+		*p = DeliverNew
+	case jsonString("by_start_sequence"):
+		*p = DeliverByStartSequence
+	case jsonString("by_start_time"):
+		*p = DeliverByStartTime
+	}
+
+	return nil
+}
+
+func (p DeliverPolicy) MarshalJSON() ([]byte, error) {
+	switch p {
+	case DeliverAll:
+		return json.Marshal("all")
+	case DeliverLast:
+		return json.Marshal("last")
+	case DeliverNew:
+		return json.Marshal("new")
+	case DeliverByStartSequence:
+		return json.Marshal("by_start_sequence")
+	case DeliverByStartTime:
+		return json.Marshal("by_start_time")
+	default:
+		return nil, fmt.Errorf("unknown deliver policy %v", p)
+	}
+}
+
+// ConsumerConfig is the configuration for a JetStream consumes
+type ConsumerConfig struct {
+	Durable         string        `json:"durable_name,omitempty"`
+	DeliverPolicy   DeliverPolicy `json:"deliver_policy"`
+	OptStartSeq     uint64        `json:"opt_start_seq,omitempty"`
+	OptStartTime    *time.Time    `json:"opt_start_time,omitempty"`
+	AckPolicy       AckPolicy     `json:"ack_policy"`
+	AckWait         time.Duration `json:"ack_wait,omitempty"`
+	MaxDeliver      int           `json:"max_deliver,omitempty"`
+	FilterSubject   string        `json:"filter_subject,omitempty"`
+	ReplayPolicy    ReplayPolicy  `json:"replay_policy"`
+	SampleFrequency string        `json:"sample_freq,omitempty"`
+	RateLimit       uint64        `json:"rate_limit_bps,omitempty"`
+}
+
+type consumerConfig struct {
+	DeliverSubject string `json:"deliver_subject,omitempty"`
+	*ConsumerConfig
+}
+
+type SequencePair struct {
+	ConsumerSeq uint64 `json:"consumer_seq"`
+	StreamSeq   uint64 `json:"stream_seq"`
+}
+
+type ConsumerInfo struct {
+	Stream         string         `json:"stream_name"`
+	Name           string         `json:"name"`
+	Config         ConsumerConfig `json:"config"`
+	Created        time.Time      `json:"created"`
+	Delivered      SequencePair   `json:"delivered"`
+	AckFloor       SequencePair   `json:"ack_floor"`
+	NumPending     int            `json:"num_pending"`
+	NumRedelivered int            `json:"num_redelivered"`
+}
+
+func jsonString(s string) string {
+	return "\"" + s + "\""
+}
+
+func isValidJSName(n string) bool {
+	return !(n == "" || strings.ContainsAny(n, ">*. "))
+}
diff --git a/vendor/github.com/nats-io/nats.go/nats.go b/vendor/github.com/nats-io/nats.go/nats.go
index daa56fdb..6b0bb267 100644
--- a/vendor/github.com/nats-io/nats.go/nats.go
+++ b/vendor/github.com/nats-io/nats.go/nats.go
@@ -54,6 +54,7 @@ const (
 	DefaultReconnectJitter    = 100 * time.Millisecond
 	DefaultReconnectJitterTLS = time.Second
 	DefaultTimeout            = 2 * time.Second
+	DefaultJetStreamTimeout   = 2 * time.Second
 	DefaultPingInterval       = 2 * time.Minute
 	DefaultMaxPingOut         = 2
 	DefaultMaxChanLen         = 8192            // 8k
@@ -123,6 +124,14 @@ var (
 	ErrDisconnected           = errors.New("nats: server is disconnected")
 	ErrHeadersNotSupported    = errors.New("nats: headers not supported by this server")
 	ErrBadHeaderMsg           = errors.New("nats: message could not decode headers")
+	ErrNoResponders           = errors.New("nats: no responders available for request")
+	ErrNoContextOrTimeout     = errors.New("nats: no context or timeout given")
+	ErrNotJSMessage           = errors.New("nats: not a JetStream message")
+	ErrInvalidStreamName      = errors.New("nats: invalid stream name")
+	ErrInvalidJSAck           = errors.New("nats: invalid JetStream publish acknowledgement")
+	ErrMultiStreamUnsupported = errors.New("nats: multiple streams are not supported")
+	ErrStreamNameRequired     = errors.New("nats: Stream name is required")
+	ErrConsumerConfigRequired = errors.New("nats: Consumer configuration is required")
 )
 
 func init() {
@@ -138,6 +147,7 @@ func GetDefaultOptions() Options {
 		ReconnectJitter:    DefaultReconnectJitter,
 		ReconnectJitterTLS: DefaultReconnectJitterTLS,
 		Timeout:            DefaultTimeout,
+		JetStreamTimeout:   DefaultJetStreamTimeout,
 		PingInterval:       DefaultPingInterval,
 		MaxPingsOut:        DefaultMaxPingOut,
 		SubChanLen:         DefaultMaxChanLen,
@@ -292,6 +302,9 @@ type Options struct {
 	// Timeout sets the timeout for a Dial operation on a connection.
 	Timeout time.Duration
 
+	// JetStreamTimeout set the default timeout for the JetStream API
+	JetStreamTimeout time.Duration
+
 	// DrainTimeout sets the timeout for a Drain Operation to complete.
 	DrainTimeout time.Duration
 
@@ -391,6 +404,15 @@ type Options struct {
 	// gradually disconnect all its connections before shuting down. This is
 	// often used in deployments when upgrading NATS Servers.
 	LameDuckModeHandler ConnHandler
+
+	// RetryOnFailedConnect sets the connection in reconnecting state right
+	// away if it can't connect to a server in the initial set. The
+	// MaxReconnect and ReconnectWait options are used for this process,
+	// similarly to when an established connection is disconnected.
+	// If a ReconnectHandler is set, it will be invoked when the connection
+	// is established, and if a ClosedHandler is set, it will be invoked if
+	// it fails to connect (after exhausting the MaxReconnect attempts).
+	RetryOnFailedConnect bool
 }
 
 const (
@@ -472,6 +494,10 @@ type Subscription struct {
 	// only be processed by one member of the group.
 	Queue string
 
+	// ConsumerConfig is the configuration for the JetStream consumer if one was created
+	// or updated using the subscription options
+	ConsumerConfig *ConsumerConfig
+
 	delivered  uint64
 	max        uint64
 	conn       *Conn
@@ -508,6 +534,7 @@ type Msg struct {
 	Sub     *Subscription
 	next    *Msg
 	barrier *barrierInfo
+	jsMeta  *JetStreamMsgMetaData
 }
 
 func (m *Msg) headerBytes() ([]byte, error) {
@@ -589,21 +616,22 @@ const (
 )
 
 type connectInfo struct {
-	Verbose   bool   `json:"verbose"`
-	Pedantic  bool   `json:"pedantic"`
-	UserJWT   string `json:"jwt,omitempty"`
-	Nkey      string `json:"nkey,omitempty"`
-	Signature string `json:"sig,omitempty"`
-	User      string `json:"user,omitempty"`
-	Pass      string `json:"pass,omitempty"`
-	Token     string `json:"auth_token,omitempty"`
-	TLS       bool   `json:"tls_required"`
-	Name      string `json:"name"`
-	Lang      string `json:"lang"`
-	Version   string `json:"version"`
-	Protocol  int    `json:"protocol"`
-	Echo      bool   `json:"echo"`
-	Headers   bool   `json:"headers"`
+	Verbose      bool   `json:"verbose"`
+	Pedantic     bool   `json:"pedantic"`
+	UserJWT      string `json:"jwt,omitempty"`
+	Nkey         string `json:"nkey,omitempty"`
+	Signature    string `json:"sig,omitempty"`
+	User         string `json:"user,omitempty"`
+	Pass         string `json:"pass,omitempty"`
+	Token        string `json:"auth_token,omitempty"`
+	TLS          bool   `json:"tls_required"`
+	Name         string `json:"name"`
+	Lang         string `json:"lang"`
+	Version      string `json:"version"`
+	Protocol     int    `json:"protocol"`
+	Echo         bool   `json:"echo"`
+	Headers      bool   `json:"headers"`
+	NoResponders bool   `json:"no_responders"`
 }
 
 // MsgHandler is a callback function that processes messages delivered to
@@ -791,6 +819,14 @@ func Timeout(t time.Duration) Option {
 	}
 }
 
+// JetStreamTimeout is an Option to set the timeout for access to the JetStream API
+func JetStreamTimeout(t time.Duration) Option {
+	return func(o *Options) error {
+		o.JetStreamTimeout = t
+		return nil
+	}
+}
+
 // FlusherTimeout is an Option to set the write (and flush) timeout on a connection.
 func FlusherTimeout(t time.Duration) Option {
 	return func(o *Options) error {
@@ -848,7 +884,7 @@ func DiscoveredServersHandler(cb ConnHandler) Option {
 	}
 }
 
-// ErrorHandler is an Option to set the async error  handler.
+// ErrorHandler is an Option to set the async error handler.
 func ErrorHandler(cb ErrHandler) Option {
 	return func(o *Options) error {
 		o.AsyncErrorCB = cb
@@ -998,6 +1034,16 @@ func LameDuckModeHandler(cb ConnHandler) Option {
 	}
 }
 
+// RetryOnFailedConnect sets the connection in reconnecting state right away
+// if it can't connect to a server in the initial set.
+// See RetryOnFailedConnect option for more details.
+func RetryOnFailedConnect(retry bool) Option {
+	return func(o *Options) error {
+		o.RetryOnFailedConnect = retry
+		return nil
+	}
+}
+
 // Handler processing
 
 // SetDisconnectHandler will set the disconnect event handler.
@@ -1576,7 +1622,9 @@ func (nc *Conn) connect() error {
 				nc.mu.Unlock()
 				nc.close(DISCONNECTED, false, err)
 				nc.mu.Lock()
-				nc.current = nil
+				// Do not reset nc.current here since it would prevent
+				// RetryOnFailedConnect to work should this be the last server
+				// to try before starting doReconnect().
 			}
 		} else {
 			// Cancel out default connection refused, will trigger the
@@ -1586,11 +1634,27 @@ func (nc *Conn) connect() error {
 			}
 		}
 	}
-	nc.initc = false
+
 	if returnedErr == nil && nc.status != CONNECTED {
 		returnedErr = ErrNoServers
 	}
 
+	if returnedErr == nil {
+		nc.initc = false
+	} else if nc.Opts.RetryOnFailedConnect {
+		nc.setup()
+		nc.status = RECONNECTING
+		nc.pending = new(bytes.Buffer)
+		if nc.bw == nil {
+			nc.bw = nc.newBuffer()
+		}
+		nc.bw.Reset(nc.pending)
+		go nc.doReconnect(ErrNoServers)
+		returnedErr = nil
+	} else {
+		nc.current = nil
+	}
+
 	return returnedErr
 }
 
@@ -1711,8 +1775,10 @@ func (nc *Conn) connectProto() (string, error) {
 		token = nc.Opts.TokenHandler()
 	}
 
+	// If our server does not support headers then we can't do them or no responders.
+	hdrs := nc.info.Headers
 	cinfo := connectInfo{o.Verbose, o.Pedantic, ujwt, nkey, sig, user, pass, token,
-		o.Secure, o.Name, LangString, Version, clientProtoInfo, !o.NoEcho, true}
+		o.Secure, o.Name, LangString, Version, clientProtoInfo, !o.NoEcho, hdrs, hdrs}
 
 	b, err := json.Marshal(cinfo)
 	if err != nil {
@@ -1908,10 +1974,12 @@ func (nc *Conn) doReconnect(err error) {
 	nc.err = nil
 	// Perform appropriate callback if needed for a disconnect.
 	// DisconnectedErrCB has priority over deprecated DisconnectedCB
-	if nc.Opts.DisconnectedErrCB != nil {
-		nc.ach.push(func() { nc.Opts.DisconnectedErrCB(nc, err) })
-	} else if nc.Opts.DisconnectedCB != nil {
-		nc.ach.push(func() { nc.Opts.DisconnectedCB(nc) })
+	if !nc.initc {
+		if nc.Opts.DisconnectedErrCB != nil {
+			nc.ach.push(func() { nc.Opts.DisconnectedErrCB(nc, err) })
+		} else if nc.Opts.DisconnectedCB != nil {
+			nc.ach.push(func() { nc.Opts.DisconnectedCB(nc) })
+		}
 	}
 
 	// This is used to wait on go routines exit if we start them in the loop
@@ -2052,6 +2120,10 @@ func (nc *Conn) doReconnect(err error) {
 		// This is where we are truly connected.
 		nc.status = CONNECTED
 
+		// If we are here with a retry on failed connect, indicate that the
+		// initial connect is now complete.
+		nc.initc = false
+
 		// Queue up the reconnect callback.
 		if nc.Opts.ReconnectedCB != nil {
 			nc.ach.push(func() { nc.Opts.ReconnectedCB(nc) })
@@ -2528,7 +2600,7 @@ func (nc *Conn) processInfo(info string) error {
 	// did not include themselves in the async INFO protocol.
 	// If empty, do not remove the implicit servers from the pool.
 	if len(ncInfo.ConnectURLs) == 0 {
-		if ncInfo.LameDuckMode && nc.Opts.LameDuckModeHandler != nil {
+		if !nc.initc && ncInfo.LameDuckMode && nc.Opts.LameDuckModeHandler != nil {
 			nc.ach.push(func() { nc.Opts.LameDuckModeHandler(nc) })
 		}
 		return nil
@@ -2591,7 +2663,7 @@ func (nc *Conn) processInfo(info string) error {
 			nc.ach.push(func() { nc.Opts.DiscoveredServersCB(nc) })
 		}
 	}
-	if ncInfo.LameDuckMode && nc.Opts.LameDuckModeHandler != nil {
+	if !nc.initc && ncInfo.LameDuckMode && nc.Opts.LameDuckModeHandler != nil {
 		nc.ach.push(func() { nc.Opts.LameDuckModeHandler(nc) })
 	}
 	return nil
@@ -2676,7 +2748,11 @@ func (nc *Conn) kickFlusher() {
 // Publish publishes the data argument to the given subject. The data
 // argument is left untouched and needs to be correctly interpreted on
 // the receiver.
-func (nc *Conn) Publish(subj string, data []byte) error {
+func (nc *Conn) Publish(subj string, data []byte, opts ...PublishOption) error {
+	if len(opts) > 0 {
+		return nc.jsPublish(subj, data, opts)
+	}
+
 	return nc.publish(subj, _EMPTY_, nil, data)
 }
 
@@ -2689,21 +2765,39 @@ func NewMsg(subject string) *Msg {
 }
 
 const (
-	hdrLine   = "NATS/1.0\r\n"
-	crlf      = "\r\n"
-	hdrPreEnd = len(hdrLine) - len(crlf)
+	hdrLine      = "NATS/1.0\r\n"
+	crlf         = "\r\n"
+	hdrPreEnd    = len(hdrLine) - len(crlf)
+	statusHdr    = "Status"
+	descrHdr     = "Description"
+	noResponders = "503"
+	statusLen    = 3 // e.g. 20x, 40x, 50x
 )
 
 // decodeHeadersMsg will decode and headers.
 func decodeHeadersMsg(data []byte) (http.Header, error) {
 	tp := textproto.NewReader(bufio.NewReader(bytes.NewReader(data)))
-	if l, err := tp.ReadLine(); err != nil || l != hdrLine[:hdrPreEnd] {
+	l, err := tp.ReadLine()
+	if err != nil || len(l) < hdrPreEnd || l[:hdrPreEnd] != hdrLine[:hdrPreEnd] {
 		return nil, ErrBadHeaderMsg
 	}
 	mh, err := tp.ReadMIMEHeader()
 	if err != nil {
 		return nil, ErrBadHeaderMsg
 	}
+	// Check if we have an inlined status.
+	if len(l) > hdrPreEnd {
+		var description string
+		status := strings.TrimSpace(l[hdrPreEnd:])
+		if len(status) != statusLen {
+			description = strings.TrimSpace(status[statusLen:])
+			status = status[:statusLen]
+		}
+		mh.Add(statusHdr, status)
+		if len(description) > 0 {
+			mh.Add(descrHdr, description)
+		}
+	}
 	return http.Header(mh), nil
 }
 
@@ -2765,7 +2859,8 @@ func (nc *Conn) publish(subj, reply string, hdr, data []byte) error {
 
 	// Proactively reject payloads over the threshold set by server.
 	msgSize := int64(len(data) + len(hdr))
-	if msgSize > nc.info.MaxPayload {
+	// Skip this check if we are not yet connected (RetryOnFailedConnect)
+	if !nc.initc && msgSize > nc.info.MaxPayload {
 		nc.mu.Unlock()
 		return ErrMaxPayload
 	}
@@ -2904,6 +2999,7 @@ func (nc *Conn) respHandler(m *Msg) {
 
 // Helper to setup and send new request style requests. Return the chan to receive the response.
 func (nc *Conn) createNewRequestAndSend(subj string, hdr, data []byte) (chan *Msg, string, error) {
+	nc.mu.Lock()
 	// Do setup for the new style if needed.
 	if nc.respMap == nil {
 		nc.initNewResp()
@@ -2944,7 +3040,6 @@ func (nc *Conn) RequestMsg(msg *Msg, timeout time.Duration) (*Msg, error) {
 		if !nc.info.Headers {
 			return nil, ErrHeadersNotSupported
 		}
-
 		hdr, err = msg.headerBytes()
 		if err != nil {
 			return nil, err
@@ -2960,18 +3055,32 @@ func (nc *Conn) Request(subj string, data []byte, timeout time.Duration) (*Msg,
 	return nc.request(subj, nil, data, timeout)
 }
 
+func (nc *Conn) useOldRequestStyle() bool {
+	nc.mu.RLock()
+	r := nc.Opts.UseOldRequestStyle
+	nc.mu.RUnlock()
+	return r
+}
+
 func (nc *Conn) request(subj string, hdr, data []byte, timeout time.Duration) (*Msg, error) {
 	if nc == nil {
 		return nil, ErrInvalidConnection
 	}
 
-	nc.mu.Lock()
-	if nc.Opts.UseOldRequestStyle {
-		nc.mu.Unlock()
-		return nc.oldRequest(subj, hdr, data, timeout)
+	var m *Msg
+	var err error
+
+	if nc.useOldRequestStyle() {
+		m, err = nc.oldRequest(subj, hdr, data, timeout)
+	} else {
+		m, err = nc.newRequest(subj, hdr, data, timeout)
 	}
 
-	return nc.newRequest(subj, hdr, data, timeout)
+	// Check for no responder status.
+	if err == nil && len(m.Data) == 0 && m.Header.Get(statusHdr) == noResponders {
+		m, err = nil, ErrNoResponders
+	}
+	return m, err
 }
 
 func (nc *Conn) newRequest(subj string, hdr, data []byte, timeout time.Duration) (*Msg, error) {
@@ -3095,15 +3204,15 @@ func (nc *Conn) respToken(respInbox string) string {
 // Subscribe will express interest in the given subject. The subject
 // can have wildcards (partial:*, full:>). Messages will be delivered
 // to the associated MsgHandler.
-func (nc *Conn) Subscribe(subj string, cb MsgHandler) (*Subscription, error) {
-	return nc.subscribe(subj, _EMPTY_, cb, nil, false)
+func (nc *Conn) Subscribe(subj string, cb MsgHandler, opts ...SubscribeOption) (*Subscription, error) {
+	return nc.subscribe(subj, _EMPTY_, cb, nil, false, opts...)
 }
 
 // ChanSubscribe will express interest in the given subject and place
 // all messages received on the channel.
 // You should not close the channel until sub.Unsubscribe() has been called.
-func (nc *Conn) ChanSubscribe(subj string, ch chan *Msg) (*Subscription, error) {
-	return nc.subscribe(subj, _EMPTY_, nil, ch, false)
+func (nc *Conn) ChanSubscribe(subj string, ch chan *Msg, opts ...SubscribeOption) (*Subscription, error) {
+	return nc.subscribe(subj, _EMPTY_, nil, ch, false, opts...)
 }
 
 // ChanQueueSubscribe will express interest in the given subject.
@@ -3112,18 +3221,18 @@ func (nc *Conn) ChanSubscribe(subj string, ch chan *Msg) (*Subscription, error)
 // which will be placed on the channel.
 // You should not close the channel until sub.Unsubscribe() has been called.
 // Note: This is the same than QueueSubscribeSyncWithChan.
-func (nc *Conn) ChanQueueSubscribe(subj, group string, ch chan *Msg) (*Subscription, error) {
-	return nc.subscribe(subj, group, nil, ch, false)
+func (nc *Conn) ChanQueueSubscribe(subj, group string, ch chan *Msg, opts ...SubscribeOption) (*Subscription, error) {
+	return nc.subscribe(subj, group, nil, ch, false, opts...)
 }
 
 // SubscribeSync will express interest on the given subject. Messages will
 // be received synchronously using Subscription.NextMsg().
-func (nc *Conn) SubscribeSync(subj string) (*Subscription, error) {
+func (nc *Conn) SubscribeSync(subj string, opts ...SubscribeOption) (*Subscription, error) {
 	if nc == nil {
 		return nil, ErrInvalidConnection
 	}
 	mch := make(chan *Msg, nc.Opts.SubChanLen)
-	s, e := nc.subscribe(subj, _EMPTY_, nil, mch, true)
+	s, e := nc.subscribe(subj, _EMPTY_, nil, mch, true, opts...)
 	return s, e
 }
 
@@ -3131,17 +3240,17 @@ func (nc *Conn) SubscribeSync(subj string) (*Subscription, error) {
 // All subscribers with the same queue name will form the queue group and
 // only one member of the group will be selected to receive any given
 // message asynchronously.
-func (nc *Conn) QueueSubscribe(subj, queue string, cb MsgHandler) (*Subscription, error) {
-	return nc.subscribe(subj, queue, cb, nil, false)
+func (nc *Conn) QueueSubscribe(subj, queue string, cb MsgHandler, opts ...SubscribeOption) (*Subscription, error) {
+	return nc.subscribe(subj, queue, cb, nil, false, opts...)
 }
 
 // QueueSubscribeSync creates a synchronous queue subscriber on the given
 // subject. All subscribers with the same queue name will form the queue
 // group and only one member of the group will be selected to receive any
 // given message synchronously using Subscription.NextMsg().
-func (nc *Conn) QueueSubscribeSync(subj, queue string) (*Subscription, error) {
+func (nc *Conn) QueueSubscribeSync(subj, queue string, opts ...SubscribeOption) (*Subscription, error) {
 	mch := make(chan *Msg, nc.Opts.SubChanLen)
-	s, e := nc.subscribe(subj, queue, nil, mch, true)
+	s, e := nc.subscribe(subj, queue, nil, mch, true, opts...)
 	return s, e
 }
 
@@ -3151,8 +3260,8 @@ func (nc *Conn) QueueSubscribeSync(subj, queue string) (*Subscription, error) {
 // which will be placed on the channel.
 // You should not close the channel until sub.Unsubscribe() has been called.
 // Note: This is the same than ChanQueueSubscribe.
-func (nc *Conn) QueueSubscribeSyncWithChan(subj, queue string, ch chan *Msg) (*Subscription, error) {
-	return nc.subscribe(subj, queue, nil, ch, false)
+func (nc *Conn) QueueSubscribeSyncWithChan(subj, queue string, ch chan *Msg, opts ...SubscribeOption) (*Subscription, error) {
+	return nc.subscribe(subj, queue, nil, ch, false, opts...)
 }
 
 // badSubject will do quick test on whether a subject is acceptable.
@@ -3176,17 +3285,47 @@ func badQueue(qname string) bool {
 }
 
 // subscribe is the internal subscribe function that indicates interest in a subject.
-func (nc *Conn) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync bool) (*Subscription, error) {
+func (nc *Conn) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync bool, opts ...SubscribeOption) (*Subscription, error) {
 	if nc == nil {
 		return nil, ErrInvalidConnection
 	}
+
+	var aopts *jsOpts
+	if len(opts) > 0 {
+		aopts = newJsOpts()
+		for _, f := range opts {
+			if err := f(aopts); err != nil {
+				return nil, err
+			}
+		}
+
+		if subj == "" {
+			subj = NewInbox()
+		}
+	}
+
 	nc.mu.Lock()
-	s, err := nc.subscribeLocked(subj, queue, cb, ch, isSync)
+	s, err := nc.subscribeLocked(subj, queue, cb, ch, isSync, opts...)
 	nc.mu.Unlock()
-	return s, err
+	if err != nil {
+		return nil, err
+	}
+
+	// here so that interest exist already when doing ephemerals
+	if aopts != nil {
+		nfo, err := nc.createOrUpdateConsumer(aopts, subj)
+		if err != nil {
+			s.Unsubscribe()
+			return nil, fmt.Errorf("nats: JetStream consumer creation failed: %s", err)
+		}
+
+		s.ConsumerConfig = &nfo.Config
+	}
+
+	return s, nil
 }
 
-func (nc *Conn) subscribeLocked(subj, queue string, cb MsgHandler, ch chan *Msg, isSync bool) (*Subscription, error) {
+func (nc *Conn) subscribeLocked(subj, queue string, cb MsgHandler, ch chan *Msg, isSync bool, opts ...SubscribeOption) (*Subscription, error) {
 	if nc == nil {
 		return nil, ErrInvalidConnection
 	}
@@ -4238,6 +4377,13 @@ func (nc *Conn) MaxPayload() int64 {
 	return nc.info.MaxPayload
 }
 
+// HeadersSupported will return if the server supports headers
+func (nc *Conn) HeadersSupported() bool {
+	nc.mu.RLock()
+	defer nc.mu.RUnlock()
+	return nc.info.Headers
+}
+
 // AuthRequired will return if the connected server requires authorization.
 func (nc *Conn) AuthRequired() bool {
 	nc.mu.RLock()
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 0a54e824..d5b9bf72 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -2,7 +2,7 @@
 github.com/minio/highwayhash
 # github.com/nats-io/jwt/v2 v2.0.0-20201015190852-e11ce317263c
 github.com/nats-io/jwt/v2
-# github.com/nats-io/nats.go v1.10.1-0.20200606002146-fc6fed82929a
+# github.com/nats-io/nats.go v1.10.1-0.20201021145452-94be476ad6e0
 github.com/nats-io/nats.go
 github.com/nats-io/nats.go/encoders/builtin
 github.com/nats-io/nats.go/util
@@ -10,7 +10,7 @@ github.com/nats-io/nats.go/util
 github.com/nats-io/nkeys
 # github.com/nats-io/nuid v1.0.1
 github.com/nats-io/nuid
-# golang.org/x/crypto v0.0.0-20200323165209-0ec3e9974c59
+# golang.org/x/crypto v0.0.0-20201016220609-9e8e0b390897
 golang.org/x/crypto/bcrypt
 golang.org/x/crypto/blowfish
 golang.org/x/crypto/ed25519


Test patch:
diff --git a/server/accounts_test.go b/server/accounts_test.go
index 35d74277..eaf78302 100644
--- a/server/accounts_test.go
+++ b/server/accounts_test.go
@@ -2640,6 +2640,8 @@ func TestAccountRouteMappingsConfiguration(t *testing.T) {
 	if len(az.Account.Mappings) != 3 {
 		t.Fatalf("Expected %d mappings, saw %d", 3, len(az.Account.Mappings))
 	}
+	b, _ := json.MarshalIndent(az, "", "  ")
+	fmt.Printf("%s", b)
 }
 
 func TestAccountRouteMappingsWithLossInjection(t *testing.T) {
diff --git a/server/dirstore_test.go b/server/dirstore_test.go
index c5bdc720..9d13e6dc 100644
--- a/server/dirstore_test.go
+++ b/server/dirstore_test.go
@@ -426,7 +426,7 @@ func TestExpiration(t *testing.T) {
 	dir, err := ioutil.TempDir(os.TempDir(), "jwtstore_test")
 	require_NoError(t, err)
 
-	dirStore, err := NewExpiringDirJWTStore(dir, false, false, NoDelete, time.Millisecond*100, 10, true, 0, nil)
+	dirStore, err := NewExpiringDirJWTStore(dir, false, false, time.Millisecond*100, 10, true, 0, nil)
 	require_NoError(t, err)
 	defer dirStore.Close()
 
@@ -464,7 +464,7 @@ func TestLimit(t *testing.T) {
 	dir, err := ioutil.TempDir(os.TempDir(), "jwtstore_test")
 	require_NoError(t, err)
 
-	dirStore, err := NewExpiringDirJWTStore(dir, false, false, NoDelete, time.Millisecond*100, 5, true, 0, nil)
+	dirStore, err := NewExpiringDirJWTStore(dir, false, false, time.Millisecond*100, 5, true, 0, nil)
 	require_NoError(t, err)
 	defer dirStore.Close()
 
@@ -506,7 +506,7 @@ func TestLimitNoEvict(t *testing.T) {
 	dir, err := ioutil.TempDir(os.TempDir(), "jwtstore_test")
 	require_NoError(t, err)
 
-	dirStore, err := NewExpiringDirJWTStore(dir, false, false, NoDelete, time.Millisecond*100, 2, false, 0, nil)
+	dirStore, err := NewExpiringDirJWTStore(dir, false, false, time.Millisecond*100, 2, false, 0, nil)
 	require_NoError(t, err)
 	defer dirStore.Close()
 
@@ -557,7 +557,7 @@ func TestLimitNoEvict(t *testing.T) {
 func TestLruLoad(t *testing.T) {
 	dir, err := ioutil.TempDir(os.TempDir(), "jwtstore_test")
 	require_NoError(t, err)
-	dirStore, err := NewExpiringDirJWTStore(dir, false, false, NoDelete, time.Millisecond*100, 2, true, 0, nil)
+	dirStore, err := NewExpiringDirJWTStore(dir, false, false, time.Millisecond*100, 2, true, 0, nil)
 	require_NoError(t, err)
 	defer dirStore.Close()
 
@@ -590,7 +590,7 @@ func TestLru(t *testing.T) {
 	dir, err := ioutil.TempDir(os.TempDir(), "jwtstore_test")
 	require_NoError(t, err)
 
-	dirStore, err := NewExpiringDirJWTStore(dir, false, false, NoDelete, time.Millisecond*100, 2, true, 0, nil)
+	dirStore, err := NewExpiringDirJWTStore(dir, false, false, time.Millisecond*100, 2, true, 0, nil)
 	require_NoError(t, err)
 	defer dirStore.Close()
 
@@ -638,7 +638,7 @@ func TestReload(t *testing.T) {
 	dir, err := ioutil.TempDir(os.TempDir(), "jwtstore_test")
 	require_NoError(t, err)
 	notificationChan := make(chan struct{}, 5)
-	dirStore, err := NewExpiringDirJWTStore(dir, false, false, NoDelete, time.Millisecond*100, 2, true, 0, func(publicKey string) {
+	dirStore, err := NewExpiringDirJWTStore(dir, false, false, time.Millisecond*100, 2, true, 0, func(publicKey string) {
 		notificationChan <- struct{}{}
 	})
 	require_NoError(t, err)
@@ -699,7 +699,7 @@ func TestExpirationUpdate(t *testing.T) {
 	dir, err := ioutil.TempDir(os.TempDir(), "jwtstore_test")
 	require_NoError(t, err)
 
-	dirStore, err := NewExpiringDirJWTStore(dir, false, false, NoDelete, time.Millisecond*100, 10, true, 0, nil)
+	dirStore, err := NewExpiringDirJWTStore(dir, false, false, time.Millisecond*100, 10, true, 0, nil)
 	require_NoError(t, err)
 	defer dirStore.Close()
 
@@ -762,7 +762,7 @@ func TestTTL(t *testing.T) {
 		require_NoError(t, err)
 		require_Len(t, len(f), 1)
 	}
-	dirStore, err := NewExpiringDirJWTStore(dir, false, false, NoDelete, 100*time.Millisecond, 10, true, time.Second, nil)
+	dirStore, err := NewExpiringDirJWTStore(dir, false, false, 100*time.Millisecond, 10, true, time.Second, nil)
 	require_NoError(t, err)
 	defer dirStore.Close()
 
@@ -794,41 +794,28 @@ func TestTTL(t *testing.T) {
 }
 
 func TestRemove(t *testing.T) {
-	for deleteType, test := range map[deleteType]struct {
-		expected int
-		moved    int
-	}{
-		HardDelete:    {0, 0},
-		RenameDeleted: {0, 1},
-		NoDelete:      {1, 0},
-	} {
-		t.Run("", func(t *testing.T) {
-			dir, err := ioutil.TempDir(os.TempDir(), "jwtstore_test")
-			require_NoError(t, err)
-			require_OneJWT := func() {
-				t.Helper()
-				f, err := ioutil.ReadDir(dir)
-				require_NoError(t, err)
-				require_Len(t, len(f), 1)
-			}
-			dirStore, err := NewExpiringDirJWTStore(dir, false, false, deleteType, 0, 10, true, 0, nil)
-			require_NoError(t, err)
-			defer dirStore.Close()
-			accountKey, err := nkeys.CreateAccount()
-			require_NoError(t, err)
-			pubKey, err := accountKey.PublicKey()
-			require_NoError(t, err)
-			createTestAccount(t, dirStore, 0, accountKey)
-			require_OneJWT()
-			dirStore.delete(pubKey)
-			f, err := filepath.Glob(dir + string(os.PathSeparator) + "/*.jwt")
-			require_NoError(t, err)
-			require_Len(t, len(f), test.expected)
-			f, err = filepath.Glob(dir + string(os.PathSeparator) + "/*.jwt.deleted")
-			require_NoError(t, err)
-			require_Len(t, len(f), test.moved)
-		})
+	dir, err := ioutil.TempDir(os.TempDir(), "jwtstore_test")
+	require_NoError(t, err)
+	require_OneJWT := func() {
+		t.Helper()
+		f, err := ioutil.ReadDir(dir)
+		require_NoError(t, err)
+		require_Len(t, len(f), 1)
 	}
+	dirStore, err := NewExpiringDirJWTStore(dir, false, false, 0, 10, true, 0, nil)
+	require_NoError(t, err)
+	defer dirStore.Close()
+
+	accountKey, err := nkeys.CreateAccount()
+	require_NoError(t, err)
+	pubKey, err := accountKey.PublicKey()
+	require_NoError(t, err)
+	createTestAccount(t, dirStore, 0, accountKey)
+	require_OneJWT()
+	dirStore.delete(pubKey)
+	f, err := ioutil.ReadDir(dir)
+	require_NoError(t, err)
+	require_Len(t, len(f), 0)
 }
 
 const infDur = time.Duration(math.MaxInt64)
@@ -849,7 +836,7 @@ func TestNotificationOnPack(t *testing.T) {
 	}
 	dirPack, err := ioutil.TempDir(os.TempDir(), "jwtstore_test")
 	require_NoError(t, err)
-	packStore, err := NewExpiringDirJWTStore(dirPack, false, false, NoDelete, infDur, 0, true, 0, notification)
+	packStore, err := NewExpiringDirJWTStore(dirPack, false, false, infDur, 0, true, 0, notification)
 	require_NoError(t, err)
 	// prefill the store with data
 	for k, v := range jwts {
@@ -865,7 +852,7 @@ func TestNotificationOnPack(t *testing.T) {
 	for _, shard := range []bool{true, false, true, false} {
 		dirMerge, err := ioutil.TempDir(os.TempDir(), "jwtstore_test")
 		require_NoError(t, err)
-		mergeStore, err := NewExpiringDirJWTStore(dirMerge, shard, false, NoDelete, infDur, 0, true, 0, notification)
+		mergeStore, err := NewExpiringDirJWTStore(dirMerge, shard, false, infDur, 0, true, 0, notification)
 		require_NoError(t, err)
 		// set
 		err = mergeStore.Merge(msg)
@@ -905,7 +892,7 @@ func TestNotificationOnPackWalk(t *testing.T) {
 	for i := 0; i < storeCnt; i++ {
 		dirMerge, err := ioutil.TempDir(os.TempDir(), "jwtstore_test")
 		require_NoError(t, err)
-		mergeStore, err := NewExpiringDirJWTStore(dirMerge, true, false, NoDelete, infDur, 0, true, 0, nil)
+		mergeStore, err := NewExpiringDirJWTStore(dirMerge, true, false, infDur, 0, true, 0, nil)
 		require_NoError(t, err)
 		store[i] = mergeStore
 	}
diff --git a/server/events_test.go b/server/events_test.go
index 43d0719e..9138822e 100644
--- a/server/events_test.go
+++ b/server/events_test.go
@@ -882,27 +882,6 @@ func TestSystemAccountConnectionLimits(t *testing.T) {
 	})
 }
 
-func TestBadAccountUpdate(t *testing.T) {
-	sa, _ := runTrustedServer(t)
-	defer sa.Shutdown()
-	akp1, _ := nkeys.CreateAccount()
-	pub, _ := akp1.PublicKey()
-	nac := jwt.NewAccountClaims(pub)
-	ajwt1, err := nac.Encode(oKp)
-	require_NoError(t, err)
-	addAccountToMemResolver(sa, pub, ajwt1)
-	akp2, _ := nkeys.CreateAccount()
-	pub2, _ := akp2.PublicKey()
-	nac.Subject = pub2 // maliciously use a different subject but pretend to remain pub
-	ajwt2, err := nac.Encode(oKp)
-	require_NoError(t, err)
-	acc, err := sa.fetchAccount(pub)
-	require_NoError(t, err)
-	if err := sa.updateAccountWithClaimJWT(acc, ajwt2); err != ErrAccountValidation {
-		t.Fatalf("expected %v but got %v", ErrAccountValidation, err)
-	}
-}
-
 // Make sure connection limits apply to the system account itself.
 func TestSystemAccountSystemConnectionLimitsHonored(t *testing.T) {
 	sa, optsA, sb, optsB, sakp := runTrustedCluster(t)
diff --git a/server/jwt_test.go b/server/jwt_test.go
index 0355de4e..c5c6b492 100644
--- a/server/jwt_test.go
+++ b/server/jwt_test.go
@@ -2130,7 +2130,7 @@ func TestAccountURLResolverFetchFailureInCluster(t *testing.T) {
 		cluster {
 			name: clust
 			no_advertise: true
-			listen: -1 
+			listen: -1
 			routes [
 				nats-route://localhost:%d
 			]
@@ -3407,7 +3407,7 @@ func TestAccountNATSResolverFetch(t *testing.T) {
 		}
 		cluster {
 			name: clust
-			listen: -1 
+			listen: -1
 			no_advertise: true
 			routes [
 				nats-route://localhost:%d
@@ -3431,7 +3431,7 @@ func TestAccountNATSResolverFetch(t *testing.T) {
 		}
 		cluster {
 			name: clust
-			listen: -1 
+			listen: -1
 			no_advertise: true
 			routes [
 				nats-route://localhost:%d
@@ -3650,7 +3650,7 @@ func TestAccountNATSResolverCrossClusterFetch(t *testing.T) {
 		}
 		cluster {
 			name: clust-A
-			listen: -1 
+			listen: -1
 			no_advertise: true
 			routes [
 				nats-route://localhost:%d
@@ -4107,7 +4107,7 @@ func TestJWTJetStreamLimits(t *testing.T) {
 	}
 	expect_JSDisabledForAccount := func(c *nats.Conn) {
 		t.Helper()
-		if _, err := c.Request("$JS.API.INFO", nil, time.Second); err != nats.ErrTimeout {
+		if _, err := c.Request("$JS.API.INFO", nil, time.Second); err != nats.ErrTimeout && err != nats.ErrNoResponders {
 			t.Fatalf("Unexpected error: %v", err)
 		}
 	}
@@ -4379,73 +4379,46 @@ func TestJWTAccountOps(t *testing.T) {
 		require_NoError(t, err)
 		*creds1 = genCredsFile(t, ujwt1, seed)
 	}
-	generateRequest := func(accs []string) []byte {
-		t.Helper()
-		opk, _ := oKp.PublicKey()
-		c := jwt.NewGenericClaims(opk)
-		c.Data["accounts"] = accs
-		cJwt, err := c.Encode(oKp)
-		if err != nil {
-			t.Fatalf("Expected no error %v", err)
-		}
-		return []byte(cJwt)
-	}
-	for _, cfg := range []string{
-		`type: full
- 		allow_delete: true`,
-		`type: cache`,
-	} {
-		t.Run("", func(t *testing.T) {
-			var syspub, sysjwt, sysCreds string
-			createAccountAndUser(&syspub, &sysjwt, &sysCreds)
-			var apub, ajwt1, aCreds1 string
-			createAccountAndUser(&apub, &ajwt1, &aCreds1)
-			defer os.Remove(sysCreds)
-			defer os.Remove(aCreds1)
-			dirSrv := createDir(t, "srv")
-			defer os.RemoveAll(dirSrv)
-			conf := createConfFile(t, []byte(fmt.Sprintf(`
+	var syspub, sysjwt, sysCreds string
+	createAccountAndUser(&syspub, &sysjwt, &sysCreds)
+	var apub, ajwt1, aCreds1 string
+	createAccountAndUser(&apub, &ajwt1, &aCreds1)
+	defer os.Remove(sysCreds)
+	defer os.Remove(aCreds1)
+	dirSrv := createDir(t, "srv")
+	defer os.RemoveAll(dirSrv)
+	conf := createConfFile(t, []byte(fmt.Sprintf(`
 		listen: -1
 		operator: %s
 		system_account: %s
 		resolver: {
-			%s
+			type: full
 			dir: %s
 		}
-    `, ojwt, syspub, cfg, dirSrv)))
-			defer os.Remove(conf)
-			srv, _ := RunServerWithConfig(conf)
-			defer srv.Shutdown()
-			updateJwt(t, srv.ClientURL(), sysCreds, syspub, sysjwt, 1) // update system account jwt
-			// push jwt (for full resolver)
-			updateJwt(t, srv.ClientURL(), sysCreds, apub, ajwt1, 1) // set jwt
-			nc := natsConnect(t, srv.ClientURL(), nats.UserCredentials(sysCreds))
-			defer nc.Close()
-			// simulate nas resolver in case of a lookup request (cache)
-			nc.Subscribe(fmt.Sprintf(accLookupReqSubj, apub), func(msg *nats.Msg) {
-				msg.Respond([]byte(ajwt1))
-			})
-			// connect so there is a reason to cache the request
-			ncA := natsConnect(t, srv.ClientURL(), nats.UserCredentials(aCreds1))
-			ncA.Close()
-			resp, err := nc.Request(accListReqSubj, nil, time.Second)
-			require_NoError(t, err)
-			require_True(t, strings.Contains(string(resp.Data), apub))
-			require_True(t, strings.Contains(string(resp.Data), syspub))
-			// delete nothing
-			resp, err = nc.Request(accDeleteReqSubj, generateRequest([]string{}), time.Second)
-			require_NoError(t, err)
-			require_True(t, strings.Contains(string(resp.Data), `"message": "deleted 0 accounts"`))
-			// issue delete, twice to also delete a non existing account
-			for i := 0; i < 2; i++ {
-				resp, err = nc.Request(accDeleteReqSubj, generateRequest([]string{apub}), time.Second)
-				require_NoError(t, err)
-				require_True(t, strings.Contains(string(resp.Data), `"message": "deleted 1 accounts"`))
-				resp, err = nc.Request(accListReqSubj, nil, time.Second)
-				require_False(t, strings.Contains(string(resp.Data), apub))
-				require_True(t, strings.Contains(string(resp.Data), syspub))
-				require_NoError(t, err)
-			}
-		})
+    `, ojwt, syspub, dirSrv)))
+	defer os.Remove(conf)
+	srv, _ := RunServerWithConfig(conf)
+	defer srv.Shutdown()
+	updateJwt(t, srv.ClientURL(), sysCreds, syspub, sysjwt, 1) // update system account jwt
+	updateJwt(t, srv.ClientURL(), sysCreds, apub, ajwt1, 1)    // set jwt
+	nc := natsConnect(t, srv.ClientURL(), nats.UserCredentials(sysCreds))
+	defer nc.Close()
+	resp, err := nc.Request(accListReqSubj, nil, time.Second)
+	require_NoError(t, err)
+	require_True(t, strings.Contains(string(resp.Data), apub))
+	require_True(t, strings.Contains(string(resp.Data), syspub))
+	// delete nothing
+	resp, err = nc.Request(accDeleteReqSubj, nil, time.Second)
+	require_NoError(t, err)
+	require_True(t, strings.Contains(string(resp.Data), `"message": "deleted 0 accounts"`))
+	// issue delete, twice to also delete a non existing account
+	for i := 0; i < 2; i++ {
+		resp, err = nc.Request(accDeleteReqSubj, []byte(apub), time.Second)
+		require_NoError(t, err)
+		require_True(t, strings.Contains(string(resp.Data), `"message": "deleted 1 accounts"`))
+		resp, err = nc.Request(accListReqSubj, nil, time.Second)
+		require_False(t, strings.Contains(string(resp.Data), apub))
+		require_True(t, strings.Contains(string(resp.Data), syspub))
+		require_NoError(t, err)
 	}
 }
diff --git a/server/reload_test.go b/server/reload_test.go
index 0d46fe51..50d11efd 100644
--- a/server/reload_test.go
+++ b/server/reload_test.go
@@ -3180,7 +3180,7 @@ func TestConfigReloadAccountServicesImportExport(t *testing.T) {
 			if string(msg.Data) != reply {
 				t.Fatalf("Expected reply %s on subject %s, got %s", reply, subj, msg.Data)
 			}
-		} else if err != nats.ErrTimeout {
+		} else if err != nats.ErrTimeout && err != nats.ErrNoResponders {
 			t.Fatalf("Expected timeout on subject %s, got %v", subj, err)
 		}
 	}
diff --git a/test/jetstream_test.go b/test/jetstream_test.go
index 0a0aa4d4..485d1b2e 100644
--- a/test/jetstream_test.go
+++ b/test/jetstream_test.go
@@ -1364,7 +1364,7 @@ func TestJetStreamBasicWorkQueue(t *testing.T) {
 
 			go func() {
 				time.Sleep(nextDelay)
-				nc.Request(sendSubj, []byte("Hello World!"), 100*time.Millisecond)
+				sendStreamMsg(t, nc, sendSubj, "Hello World!")
 			}()
 
 			start := time.Now()
@@ -1388,6 +1388,301 @@ func TestJetStreamBasicWorkQueue(t *testing.T) {
 	}
 }
 
+func TestJetStreamWorkQueueMaxWaiting(t *testing.T) {
+	cases := []struct {
+		name    string
+		mconfig *server.StreamConfig
+	}{
+		{"MemoryStore", &server.StreamConfig{Name: "MY_MSG_SET", Storage: server.MemoryStorage, Subjects: []string{"foo", "bar"}}},
+		{"FileStore", &server.StreamConfig{Name: "MY_MSG_SET", Storage: server.FileStorage, Subjects: []string{"foo", "bar"}}},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			s := RunBasicJetStreamServer()
+			defer s.Shutdown()
+
+			if config := s.JetStreamConfig(); config != nil {
+				defer os.RemoveAll(config.StoreDir)
+			}
+
+			mset, err := s.GlobalAccount().AddStream(c.mconfig)
+			if err != nil {
+				t.Fatalf("Unexpected error adding stream: %v", err)
+			}
+			defer mset.Delete()
+
+			// Make sure these cases fail
+			cfg := &server.ConsumerConfig{Durable: "foo", AckPolicy: server.AckExplicit, MaxWaiting: 10, DeliverSubject: "_INBOX.22"}
+			if _, err := mset.AddConsumer(cfg); err == nil {
+				t.Fatalf("Expected an error with MaxWaiting set on non-pull based consumer")
+			}
+			cfg = &server.ConsumerConfig{Durable: "foo", AckPolicy: server.AckExplicit, MaxWaiting: -1}
+			if _, err := mset.AddConsumer(cfg); err == nil {
+				t.Fatalf("Expected an error with MaxWaiting being negative")
+			}
+
+			// Create basic work queue mode consumer.
+			wcfg := workerModeConfig("MAXWQ")
+			o, err := mset.AddConsumer(wcfg)
+			if err != nil {
+				t.Fatalf("Expected no error with registered interest, got %v", err)
+			}
+			defer o.Delete()
+
+			// Make sure we set default correctly.
+			if cfg := o.Config(); cfg.MaxWaiting != server.JSWaitQueueDefaultMax {
+				t.Fatalf("Expected default max waiting to have been set to %d, got %d", server.JSWaitQueueDefaultMax, cfg.MaxWaiting)
+			}
+
+			expectWaiting := func(expected int) {
+				t.Helper()
+				checkFor(t, time.Second, 25*time.Millisecond, func() error {
+					if oi := o.Info(); oi.NumWaiting != expected {
+						return fmt.Errorf("Expected %d waiting, got %d", expected, oi.NumWaiting)
+					}
+					return nil
+				})
+			}
+
+			nc := clientConnectWithOldRequest(t, s)
+			defer nc.Close()
+
+			// Like muxed new INBOX.
+			sub, _ := nc.SubscribeSync("req.*")
+			defer sub.Unsubscribe()
+			nc.Flush()
+
+			checkSubPending := func(numExpected int) {
+				t.Helper()
+				checkFor(t, 200*time.Millisecond, 10*time.Millisecond, func() error {
+					if nmsgs, _, _ := sub.Pending(); err != nil || nmsgs != numExpected {
+						return fmt.Errorf("Did not receive correct number of messages: %d vs %d", nmsgs, numExpected)
+					}
+					return nil
+				})
+			}
+
+			getSubj := o.RequestNextMsgSubject()
+			// Queue up JSWaitQueueDefaultMax requests.
+			for i := 0; i < server.JSWaitQueueDefaultMax; i++ {
+				nc.PublishRequest(getSubj, fmt.Sprintf("req.%d", i), nil)
+			}
+			expectWaiting(server.JSWaitQueueDefaultMax)
+
+			// So when we submit our next request this one should succeed since we do not want these to fail.
+			// We should get notified that the first request is now stale and has been removed.
+			if _, err := nc.Request(getSubj, nil, 10*time.Millisecond); err != nats.ErrTimeout {
+				t.Fatalf("Expected timeout error, got: %v", err)
+			}
+			checkSubPending(1)
+			m, _ := sub.NextMsg(0)
+			// Make sure this is an alert that tells us our request is now stale.
+			if m.Header.Get("Status") != "408" {
+				t.Fatalf("Expected a 408 status code, got %q", m.Header.Get("Status"))
+			}
+			sendStreamMsg(t, nc, "foo", "Hello World!")
+			sendStreamMsg(t, nc, "bar", "Hello World!")
+			expectWaiting(server.JSWaitQueueDefaultMax - 2)
+		})
+	}
+}
+
+func TestJetStreamWorkQueueWrapWaiting(t *testing.T) {
+	cases := []struct {
+		name    string
+		mconfig *server.StreamConfig
+	}{
+		{"MemoryStore", &server.StreamConfig{Name: "MY_MSG_SET", Storage: server.MemoryStorage, Subjects: []string{"foo", "bar"}}},
+		{"FileStore", &server.StreamConfig{Name: "MY_MSG_SET", Storage: server.FileStorage, Subjects: []string{"foo", "bar"}}},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			s := RunBasicJetStreamServer()
+			defer s.Shutdown()
+
+			if config := s.JetStreamConfig(); config != nil {
+				defer os.RemoveAll(config.StoreDir)
+			}
+
+			mset, err := s.GlobalAccount().AddStream(c.mconfig)
+			if err != nil {
+				t.Fatalf("Unexpected error adding stream: %v", err)
+			}
+			defer mset.Delete()
+
+			maxWaiting := 8
+			wcfg := workerModeConfig("WRAP")
+			wcfg.MaxWaiting = maxWaiting
+
+			o, err := mset.AddConsumer(wcfg)
+			if err != nil {
+				t.Fatalf("Expected no error with registered interest, got %v", err)
+			}
+			defer o.Delete()
+
+			getSubj := o.RequestNextMsgSubject()
+
+			expectWaiting := func(expected int) {
+				t.Helper()
+				checkFor(t, time.Second, 25*time.Millisecond, func() error {
+					if oi := o.Info(); oi.NumWaiting != expected {
+						return fmt.Errorf("Expected %d waiting, got %d", expected, oi.NumWaiting)
+					}
+					return nil
+				})
+			}
+
+			nc := clientConnectToServer(t, s)
+			defer nc.Close()
+
+			sub, _ := nc.SubscribeSync("req.*")
+			defer sub.Unsubscribe()
+			nc.Flush()
+
+			// Fill up waiting.
+			for i := 0; i < maxWaiting; i++ {
+				nc.PublishRequest(getSubj, fmt.Sprintf("req.%d", i), nil)
+			}
+			expectWaiting(maxWaiting)
+
+			// Now use 1/2 of the waiting.
+			for i := 0; i < maxWaiting/2; i++ {
+				sendStreamMsg(t, nc, "foo", "Hello World!")
+			}
+			expectWaiting(maxWaiting / 2)
+
+			// Now add in two (2) more pull requests.
+			for i := maxWaiting; i < maxWaiting+2; i++ {
+				nc.PublishRequest(getSubj, fmt.Sprintf("req.%d", i), nil)
+			}
+			expectWaiting(maxWaiting/2 + 2)
+
+			// Now use second 1/2 of the waiting and the 2 extra.
+			for i := 0; i < maxWaiting/2+2; i++ {
+				sendStreamMsg(t, nc, "bar", "Hello World!")
+			}
+			expectWaiting(0)
+
+			if nmsgs, _, _ := sub.Pending(); err != nil || nmsgs != maxWaiting+2 {
+				t.Fatalf("Expected sub to have %d pending, got %d", maxWaiting+2, nmsgs)
+			}
+		})
+	}
+}
+
+func TestJetStreamWorkQueueRequest(t *testing.T) {
+	cases := []struct {
+		name    string
+		mconfig *server.StreamConfig
+	}{
+		{"MemoryStore", &server.StreamConfig{Name: "MY_MSG_SET", Storage: server.MemoryStorage, Subjects: []string{"foo", "bar"}}},
+		{"FileStore", &server.StreamConfig{Name: "MY_MSG_SET", Storage: server.FileStorage, Subjects: []string{"foo", "bar"}}},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			s := RunBasicJetStreamServer()
+			defer s.Shutdown()
+
+			if config := s.JetStreamConfig(); config != nil {
+				defer os.RemoveAll(config.StoreDir)
+			}
+
+			mset, err := s.GlobalAccount().AddStream(c.mconfig)
+			if err != nil {
+				t.Fatalf("Unexpected error adding stream: %v", err)
+			}
+			defer mset.Delete()
+
+			o, err := mset.AddConsumer(workerModeConfig("WRAP"))
+			if err != nil {
+				t.Fatalf("Expected no error with registered interest, got %v", err)
+			}
+			defer o.Delete()
+
+			nc := clientConnectToServer(t, s)
+			defer nc.Close()
+
+			toSend := 25
+			for i := 0; i < toSend; i++ {
+				sendStreamMsg(t, nc, "bar", "Hello World!")
+			}
+
+			reply := "_.consumer._"
+			sub, _ := nc.SubscribeSync(reply)
+			defer sub.Unsubscribe()
+
+			getSubj := o.RequestNextMsgSubject()
+
+			checkSubPending := func(numExpected int) {
+				t.Helper()
+				checkFor(t, 200*time.Millisecond, 10*time.Millisecond, func() error {
+					if nmsgs, _, _ := sub.Pending(); err != nil || nmsgs != numExpected {
+						return fmt.Errorf("Did not receive correct number of messages: %d vs %d", nmsgs, numExpected)
+					}
+					return nil
+				})
+			}
+
+			// Create a formal request object.
+			req := &server.JSApiConsumerGetNextRequest{Batch: toSend}
+			jreq, _ := json.Marshal(req)
+			nc.PublishRequest(getSubj, reply, jreq)
+
+			checkSubPending(toSend)
+
+			// Now check that we can ask for NoWait
+			req.Batch = 1
+			req.NoWait = true
+			jreq, _ = json.Marshal(req)
+
+			resp, err := nc.Request(getSubj, jreq, 50*time.Millisecond)
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			if status := resp.Header.Get("Status"); !strings.HasPrefix(status, "404") {
+				t.Fatalf("Expected status code of 404")
+			}
+			// Load up more messages.
+			for i := 0; i < toSend; i++ {
+				sendStreamMsg(t, nc, "foo", "Hello World!")
+			}
+			// Now we will ask for a batch larger then what is queued up.
+			req.Batch = toSend + 10
+			req.NoWait = true
+			jreq, _ = json.Marshal(req)
+			nc.PublishRequest(getSubj, reply, jreq)
+			// We should now have 2 * toSend + the 404 message.
+			checkSubPending(2*toSend + 1)
+			for i := 0; i < 2*toSend+1; i++ {
+				sub.NextMsg(time.Millisecond)
+			}
+			checkSubPending(0)
+			mset.Purge()
+
+			// Now do expiration
+			req.Batch = 1
+			req.NoWait = false
+			req.Expires = time.Now().Add(10 * time.Millisecond)
+			jreq, _ = json.Marshal(req)
+
+			nc.PublishRequest(getSubj, reply, jreq)
+			// Let it expire
+			time.Sleep(20 * time.Millisecond)
+
+			// Send a few more messages. These should not be delivered to the sub.
+			sendStreamMsg(t, nc, "foo", "Hello World!")
+			sendStreamMsg(t, nc, "bar", "Hello World!")
+			// We will have an alert here.
+			checkSubPending(1)
+			m, _ := sub.NextMsg(0)
+			// Make sure this is an alert that tells us our request is now stale.
+			if m.Header.Get("Status") != "408" {
+				t.Fatalf("Expected a 408 status code, got %q", m.Header.Get("Status"))
+			}
+		})
+	}
+}
+
 func TestJetStreamSubjectFiltering(t *testing.T) {
 	cases := []struct {
 		name    string
@@ -1791,6 +2086,26 @@ func TestJetStreamWorkQueueRequestBatch(t *testing.T) {
 				}
 				return nil
 			})
+
+			// Now queue up the request without messages and add them after.
+			sub, _ = nc.SubscribeSync(nats.NewInbox())
+			defer sub.Unsubscribe()
+			mset.Purge()
+
+			nc.PublishRequest(o.RequestNextMsgSubject(), sub.Subject, []byte(strconv.Itoa(batchSize)))
+			nc.Flush() // Make sure its registered.
+
+			for i := 0; i < toSend; i++ {
+				sendStreamMsg(t, nc, sendSubj, "Hello World!")
+			}
+
+			// We should receive batchSize with no acks or additional requests.
+			checkFor(t, 250*time.Millisecond, 10*time.Millisecond, func() error {
+				if nmsgs, _, _ := sub.Pending(); err != nil || nmsgs != batchSize {
+					return fmt.Errorf("Did not receive correct number of messages: %d vs %d", nmsgs, batchSize)
+				}
+				return nil
+			})
 		})
 	}
 }
diff --git a/test/leafnode_test.go b/test/leafnode_test.go
index a8055d8d..c6833713 100644
--- a/test/leafnode_test.go
+++ b/test/leafnode_test.go
@@ -1204,39 +1204,6 @@ func TestLeafNodeTLS(t *testing.T) {
 	checkLeafNodeConnected(t, s)
 }
 
-func TestLeafNodeTLSConnCloseEarly(t *testing.T) {
-	content := `
-	listen: "127.0.0.1:-1"
-
-	leafnodes {
-		listen: "127.0.0.1:-1"
-		tls {
-			cert_file: "./configs/certs/server-cert.pem"
-			key_file: "./configs/certs/server-key.pem"
-			timeout: 2.0
-		}
-	}
-	`
-	conf := createConfFile(t, []byte(content))
-	defer os.Remove(conf)
-
-	s, opts := RunServerWithConfig(conf)
-	defer s.Shutdown()
-
-	lc, err := net.Dial("tcp", fmt.Sprintf("127.0.0.1:%d", opts.LeafNode.Port))
-	if err != nil {
-		t.Fatalf("Unable to connect: %v", err)
-	}
-	// Then close right away
-	lc.Close()
-
-	// Check server does not crash...
-	time.Sleep(250 * time.Millisecond)
-	if s.ID() == "" {
-		t.Fatalf("should not happen")
-	}
-}
-
 type captureLeafNodeErrLogger struct {
 	dummyLogger
 	ch chan string