Repository: nats-io/nsc
Pull request #335 (2020-12-16 02:32:50)

[Added] command to validate jwt in a file
Input file can be a jwt, list of jwt (separated by new line)
If the jwt contains a | only the suffix will be interpreted as jwt
(result of pack call)
Also updated the jwt library.

Signed-off-by: Matthias Hanel <mh@synadia.com>


Patch:
diff --git a/cmd/addexport.go b/cmd/addexport.go
index 2ce8332..fd0d847 100644
--- a/cmd/addexport.go
+++ b/cmd/addexport.go
@@ -238,7 +238,7 @@ func (p *AddExportParams) Validate(ctx ActionCtx) error {
 
 	// if we have a latency report subject create it
 	if p.latSubject != "" {
-		p.export.Latency = &jwt.ServiceLatency{Results: jwt.Subject(p.latSubject), Sampling: p.latSampling}
+		p.export.Latency = &jwt.ServiceLatency{Results: jwt.Subject(p.latSubject), Sampling: jwt.SamplingRate(p.latSampling)}
 	}
 
 	// add the new export
diff --git a/cmd/editexport.go b/cmd/editexport.go
index 79bb8df..c0cb7e9 100644
--- a/cmd/editexport.go
+++ b/cmd/editexport.go
@@ -206,7 +206,7 @@ func (p *EditExportParams) PostInteractive(ctx ActionCtx) error {
 			return err
 		}
 		if ok {
-			cls := 0
+			cls := jwt.SamplingRate(0)
 			results := jwt.Subject("")
 			if sel.Latency != nil {
 				cls = sel.Latency.Sampling
@@ -301,7 +301,7 @@ func (p *EditExportParams) syncOptions(ctx ActionCtx) {
 	sampling := 0
 	latency := ""
 	if old.Latency != nil {
-		sampling = old.Latency.Sampling
+		sampling = int(old.Latency.Sampling)
 		latency = string(old.Latency.Results)
 	}
 	if !(cmd.Flag("latency").Changed) {
@@ -369,14 +369,14 @@ func (p *EditExportParams) Run(ctx ActionCtx) (store.Status, error) {
 				r.AddOK("no need to remove latency tracking as it was not set")
 			}
 		} else {
-			oldSampling := 0
+			oldSampling := jwt.SamplingRate(0)
 			oldReport := jwt.Subject("")
 			if old.Latency != nil {
 				oldSampling = old.Latency.Sampling
 				oldReport = old.Latency.Results
 			}
 			if p.latSubject != "" {
-				export.Latency = &jwt.ServiceLatency{Results: jwt.Subject(p.latSubject), Sampling: p.latSampling}
+				export.Latency = &jwt.ServiceLatency{Results: jwt.Subject(p.latSubject), Sampling: jwt.SamplingRate(p.latSampling)}
 				if oldSampling != export.Latency.Sampling {
 					r.AddOK("changed service latency to %d%%", export.Latency.Sampling)
 				}
diff --git a/cmd/root.go b/cmd/root.go
index 3b8dcc8..e80f375 100644
--- a/cmd/root.go
+++ b/cmd/root.go
@@ -118,7 +118,7 @@ var rootCmd = &cobra.Command{
 					return fmt.Errorf("the store %#q is at version %d. To upgrade nsc - type `%s update`",
 						store.GetName(), c.Version, os.Args[0])
 				} else if c.Version == 1 {
-					allowCmdWithJWTV1Store := cmd.Name() == "upgrade-jwt" || cmd.Name() == "env" || cmd.Name() == "help"  || cmd.Name() == "update"
+					allowCmdWithJWTV1Store := cmd.Name() == "upgrade-jwt" || cmd.Name() == "env" || cmd.Name() == "help"
 					if !allowCmdWithJWTV1Store && cmd.Name() == "operator" {
 						for _, v := range addCmd.Commands() {
 							if v == cmd {
diff --git a/cmd/validate.go b/cmd/validate.go
index 7f553c6..4d5931a 100644
--- a/cmd/validate.go
+++ b/cmd/validate.go
@@ -18,7 +18,9 @@ package cmd
 import (
 	"errors"
 	"fmt"
+	"io/ioutil"
 	"sort"
+	"strings"
 
 	"github.com/nats-io/jwt/v2"
 	"github.com/nats-io/nsc/cmd/store"
@@ -34,7 +36,8 @@ func createValidateCommand() *cobra.Command {
 		Example: "validate",
 		Use: `validate (current operator/current account/account users)
 validate -a <accountName> (current operator/<accountName>/account users)
-validate -A (current operator/all accounts/all users)`,
+validate -A (current operator/all accounts/all users)
+validate -f <file>`,
 		Args: MaxArgs(0),
 		RunE: func(cmd *cobra.Command, args []string) error {
 			cmd.SilenceUsage = false
@@ -55,7 +58,8 @@ validate -A (current operator/all accounts/all users)`,
 			return nil
 		},
 	}
-	cmd.Flags().BoolVarP(&params.allAccounts, "all-accounts", "A", false, "validate all accounts under the current operator (exclusive of -a)")
+	cmd.Flags().BoolVarP(&params.allAccounts, "all-accounts", "A", false, "validate all accounts under the current operator (exclusive of -a and -f)")
+	cmd.Flags().StringVarP(&params.file, "file", "f", "", "validate all jwt (separated by newline) in the provided file (exclusive of -a and -A)")
 	params.AccountContextParams.BindFlags(cmd)
 	return cmd
 }
@@ -67,6 +71,7 @@ func init() {
 type ValidateCmdParams struct {
 	AccountContextParams
 	allAccounts        bool
+	file               string
 	operator           *jwt.ValidationResults
 	accounts           []string
 	accountValidations map[string]*jwt.ValidationResults
@@ -77,10 +82,15 @@ func (p *ValidateCmdParams) SetDefaults(ctx ActionCtx) error {
 	if p.allAccounts && p.Name != "" {
 		return errors.New("specify only one of --account or --all-accounts")
 	}
-
-	// if they specified an account name, this will validate it
-	if err := p.AccountContextParams.SetDefaults(ctx); err != nil {
-		return err
+	if p.file != "" {
+		if p.allAccounts || p.Name != "" {
+			return errors.New("specify only one of --account or --all-accounts or --file")
+		}
+	} else {
+		// if they specified an account name, this will validate it
+		if err := p.AccountContextParams.SetDefaults(ctx); err != nil {
+			return err
+		}
 	}
 
 	return nil
@@ -88,7 +98,7 @@ func (p *ValidateCmdParams) SetDefaults(ctx ActionCtx) error {
 
 func (p *ValidateCmdParams) PreInteractive(ctx ActionCtx) error {
 	var err error
-	if !p.allAccounts {
+	if !p.allAccounts && p.file == "" {
 		if err = p.AccountContextParams.Edit(ctx); err != nil {
 			return err
 		}
@@ -119,6 +129,67 @@ func (p *ValidateCmdParams) validateJWT(claim jwt.Claims) *jwt.ValidationResults
 }
 
 func (p *ValidateCmdParams) Validate(ctx ActionCtx) error {
+	if p.file != "" {
+		return p.validateFile(ctx)
+	}
+	return p.validate(ctx)
+}
+
+func (p *ValidateCmdParams) validateFile(ctx ActionCtx) error {
+	f, err := ioutil.ReadFile(p.file)
+	if err != nil {
+		return err
+	}
+	type entry struct {
+		issue jwt.ValidationIssue
+		cnt   int
+	}
+	summary := map[string]*entry{}
+	lines := strings.Split(string(f), "\n")
+	for _, l := range lines {
+		l = strings.TrimSpace(l)
+		if l == "" {
+			continue
+		}
+		// cut off what is a result of pack
+		if i := strings.Index(l, "|"); i != -1 {
+			l = l[i+1:]
+		}
+		c, err := jwt.Decode(l)
+		if err != nil {
+			return fmt.Errorf("claim decoding error: '%v' claim: '%v'", err, l)
+		}
+		subj := c.Claims().Subject
+		vr := jwt.ValidationResults{}
+		c.Validate(&vr)
+		if _, ok := p.accountValidations[subj]; !ok {
+			p.accountValidations[subj] = &jwt.ValidationResults{}
+			p.accounts = append(p.accounts, subj)
+		}
+		for _, vi := range vr.Issues {
+			p.accountValidations[subj].Add(vi)
+			if val, ok := summary[vi.Description]; !ok {
+				summary[vi.Description] = &entry{*vi, 1}
+			} else {
+				val.cnt++
+			}
+		}
+	}
+	if len(p.accounts) > 1 {
+		summaryAcc := "summary of all accounts"
+		p.accounts = append(p.accounts, summaryAcc)
+		vr := &jwt.ValidationResults{}
+		p.accountValidations[summaryAcc] = vr
+		for _, v := range summary {
+			iss := v.issue
+			iss.Description = fmt.Sprintf("%s (%d occurrences)", iss.Description, v.cnt)
+			vr.Add(&iss)
+		}
+	}
+	return nil
+}
+
+func (p *ValidateCmdParams) validate(ctx ActionCtx) error {
 	var err error
 	oc, err := ctx.StoreCtx().Store.ReadOperatorClaim()
 	if err != nil {
diff --git a/go.mod b/go.mod
index e73e668..6b0aa60 100644
--- a/go.mod
+++ b/go.mod
@@ -11,8 +11,8 @@ require (
 	github.com/mitchellh/go-homedir v1.1.0
 	github.com/nats-io/cliprompts/v2 v2.0.0-20191226174129-372d79b36768
 	github.com/nats-io/jwt v1.2.2
-	github.com/nats-io/jwt/v2 v2.0.0-20201207222029-6d7e74b660de
-	github.com/nats-io/nats-server/v2 v2.1.8-0.20201125160748-de0c992ca6db
+	github.com/nats-io/jwt/v2 v2.0.0-20201211164018-2e78446f4e6f
+	github.com/nats-io/nats-server/v2 v2.1.8-0.20201216194404-d7741b9aa16a
 	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
diff --git a/go.sum b/go.sum
index e37b85d..074745a 100644
--- a/go.sum
+++ b/go.sum
@@ -82,15 +82,13 @@ github.com/nats-io/jwt v1.1.0/go.mod h1:n3cvmLfBfnpV4JJRN7lRYCyZnw48ksGsbThGXEk4
 github.com/nats-io/jwt v1.2.2 h1:w3GMTO969dFg+UOKTmmyuu7IGdusK+7Ytlt//OYH/uU=
 github.com/nats-io/jwt v1.2.2/go.mod h1:/xX356yQA6LuXI9xWW7mZNpxgF2mBmGecH+Fj34sP5Q=
 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/jwt/v2 v2.0.0-20201207222029-6d7e74b660de h1:AZER/vBA93FmJg6yQEDf1yGY/5lhujx7AcWub1Q41Go=
-github.com/nats-io/jwt/v2 v2.0.0-20201207222029-6d7e74b660de/go.mod h1:PuO5FToRL31ecdFqVjc794vK0Bj0CwzveQEDvkb7MoQ=
+github.com/nats-io/jwt/v2 v2.0.0-20201211164018-2e78446f4e6f h1:+F8Vkc9pDf2x6O+csstCXaubPu2njJkbXeTm5wfDzGI=
+github.com/nats-io/jwt/v2 v2.0.0-20201211164018-2e78446f4e6f/go.mod h1:PuO5FToRL31ecdFqVjc794vK0Bj0CwzveQEDvkb7MoQ=
 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-server/v2 v2.1.8-0.20201125160748-de0c992ca6db h1:x+Sv0WsmnoJd2VdsuJUNQED8MojO7d3j/ZRP9dLzi7k=
-github.com/nats-io/nats-server/v2 v2.1.8-0.20201125160748-de0c992ca6db/go.mod h1:XD0zHR/jTXdZvWaQfS5mQgsXj6x12kMjKLyAk/cOGgY=
+github.com/nats-io/nats-server/v2 v2.1.8-0.20201216194404-d7741b9aa16a h1:xVy7aD956W69q/RBpSpXJDhO5u7hQS5RyLKtC253mKw=
+github.com/nats-io/nats-server/v2 v2.1.8-0.20201216194404-d7741b9aa16a/go.mod h1:p6YiXCODGJlTl0q8AlXWoyCkIYItOd3ZdQuP2m2KR6w=
 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=
diff --git a/vendor/github.com/nats-io/jwt/v2/activation_claims.go b/vendor/github.com/nats-io/jwt/v2/activation_claims.go
index 3ba8c4c..eb0c322 100644
--- a/vendor/github.com/nats-io/jwt/v2/activation_claims.go
+++ b/vendor/github.com/nats-io/jwt/v2/activation_claims.go
@@ -105,7 +105,14 @@ func (a *ActivationClaims) Payload() interface{} {
 
 // Validate checks the claims
 func (a *ActivationClaims) Validate(vr *ValidationResults) {
-	a.ClaimsData.Validate(vr)
+	a.validateWithTimeChecks(vr, true)
+}
+
+// Validate checks the claims
+func (a *ActivationClaims) validateWithTimeChecks(vr *ValidationResults, timeChecks bool) {
+	if timeChecks {
+		a.ClaimsData.Validate(vr)
+	}
 	a.Activation.Validate(vr)
 	if a.IssuerAccount != "" && !nkeys.IsValidPublicAccountKey(a.IssuerAccount) {
 		vr.AddError("account_id is not an account public key")
diff --git a/vendor/github.com/nats-io/jwt/v2/exports.go b/vendor/github.com/nats-io/jwt/v2/exports.go
index 04d900c..23065fb 100644
--- a/vendor/github.com/nats-io/jwt/v2/exports.go
+++ b/vendor/github.com/nats-io/jwt/v2/exports.go
@@ -16,6 +16,7 @@
 package jwt
 
 import (
+	"encoding/json"
 	"fmt"
 	"strings"
 	"time"
@@ -56,13 +57,49 @@ const (
 // }
 //
 type ServiceLatency struct {
-	Sampling int     `json:"sampling,omitempty"`
-	Results  Subject `json:"results"`
+	Sampling SamplingRate `json:"sampling"`
+	Results  Subject      `json:"results"`
+}
+
+type SamplingRate int
+
+const Headers = SamplingRate(0)
+
+// MarshalJSON marshals the field as "headers" or percentages
+func (r *SamplingRate) MarshalJSON() ([]byte, error) {
+	sr := *r
+	if sr == 0 {
+		return []byte(`"headers"`), nil
+	}
+	if sr >= 1 && sr <= 100 {
+		return []byte(fmt.Sprintf("%d", sr)), nil
+	}
+	return nil, fmt.Errorf("unknown sampling rate")
+}
+
+// UnmarshalJSON unmashals numbers as percentages or "headers"
+func (t *SamplingRate) UnmarshalJSON(b []byte) error {
+	if len(b) == 0 {
+		return fmt.Errorf("empty sampling rate")
+	}
+	if strings.ToLower(string(b)) == `"headers"` {
+		*t = Headers
+		return nil
+	}
+	var j int
+	err := json.Unmarshal(b, &j)
+	if err != nil {
+		return err
+	}
+	*t = SamplingRate(j)
+	return nil
 }
 
 func (sl *ServiceLatency) Validate(vr *ValidationResults) {
-	if sl.Sampling < 1 || sl.Sampling > 100 {
-		vr.AddError("sampling percentage needs to be between 1-100")
+	if sl.Sampling != 0 {
+		if sl.Sampling < 1 || sl.Sampling > 100 {
+			vr.AddError("sampling percentage needs to be between 1-100")
+		}
 	}
 	sl.Results.Validate(vr)
 	if sl.Results.HasWildCards() {
diff --git a/vendor/github.com/nats-io/jwt/v2/imports.go b/vendor/github.com/nats-io/jwt/v2/imports.go
index e4838a4..478b6de 100644
--- a/vendor/github.com/nats-io/jwt/v2/imports.go
+++ b/vendor/github.com/nats-io/jwt/v2/imports.go
@@ -37,8 +37,9 @@ type Import struct {
 	// from the perspective of a service, it is the subscription waiting for
 	// requests (the exporter). If the field is empty, it will default to the
 	// value in the Subject field.
-	To   Subject    `json:"to,omitempty"`
-	Type ExportType `json:"type,omitempty"`
+	To    Subject    `json:"to,omitempty"`
+	Type  ExportType `json:"type,omitempty"`
+	Share bool       `json:"share,omitempty"`
 }
 
 // IsService returns true if the import is of type service
@@ -73,7 +74,9 @@ func (i *Import) Validate(actPubKey string, vr *ValidationResults) {
 	if i.IsStream() && i.To.HasWildCards() {
 		vr.AddError("streams cannot have wildcard to subject: %q", i.Subject)
 	}
-
+	if i.Share && !i.IsService() {
+		vr.AddError("sharing information (for latency tracking) is only valid for services: %q", i.Subject)
+	}
 	var act *ActivationClaims
 
 	if i.Token != "" {
@@ -114,7 +117,7 @@ func (i *Import) Validate(actPubKey string, vr *ValidationResults) {
 		if act.ClaimsData.Subject != actPubKey {
 			vr.AddError("activation token doesn't match account it is being included in, %q", i.Subject)
 		}
-		act.Validate(vr)
+		act.validateWithTimeChecks(vr, false)
 	} else {
 		vr.AddWarning("no activation provided for import %s", i.Subject)
 	}
diff --git a/vendor/github.com/nats-io/jwt/v2/revocation_list.go b/vendor/github.com/nats-io/jwt/v2/revocation_list.go
index 5840061..c582896 100644
--- a/vendor/github.com/nats-io/jwt/v2/revocation_list.go
+++ b/vendor/github.com/nats-io/jwt/v2/revocation_list.go
@@ -23,18 +23,43 @@ const All = "*"
 
 // RevocationList is used to store a mapping of public keys to unix timestamps
 type RevocationList map[string]int64
+type RevocationEntry struct {
+	PublicKey string
+	TimeStamp int64
+}
 
 // Revoke enters a revocation by publickey and timestamp into this export
 // If there is already a revocation for this public key that is newer, it is kept.
 func (r RevocationList) Revoke(pubKey string, timestamp time.Time) {
 	newTS := timestamp.Unix()
+	// cannot move a revocation into the future - only into the past
 	if ts, ok := r[pubKey]; ok && ts > newTS {
 		return
 	}
-
 	r[pubKey] = newTS
 }
 
+// MaybeCompact will compact the revocation list if jwt.All is found. Any
+// revocation that is covered by a jwt.All revocation will be deleted, thus
+// reducing the size of the JWT. Returns a slice of entries that were removed
+// during the process.
+func (r RevocationList) MaybeCompact() []RevocationEntry {
+	var deleted []RevocationEntry
+	ats, ok := r[All]
+	if ok {
+		for k, ts := range r {
+			if k != All && ats >= ts {
+				deleted = append(deleted, RevocationEntry{
+					PublicKey: k,
+					TimeStamp: ts,
+				})
+				delete(r, k)
+			}
+		}
+	}
+	return deleted
+}
+
 // ClearRevocation removes any revocation for the public key
 func (r RevocationList) ClearRevocation(pubKey string) {
 	delete(r, pubKey)
diff --git a/vendor/github.com/nats-io/nats-server/v2/internal/ldap/dn.go b/vendor/github.com/nats-io/nats-server/v2/internal/ldap/dn.go
index 2bd46d8..878ffbc 100644
--- a/vendor/github.com/nats-io/nats-server/v2/internal/ldap/dn.go
+++ b/vendor/github.com/nats-io/nats-server/v2/internal/ldap/dn.go
@@ -5,6 +5,7 @@ package ldap
 import (
 	"bytes"
 	"crypto/x509/pkix"
+	"encoding/asn1"
 	enchex "encoding/hex"
 	"errors"
 	"fmt"
@@ -12,14 +13,14 @@ import (
 )
 
 var attributeTypeNames = map[string]string{
-	"2.5.4.6":  "C",
-	"2.5.4.10": "O",
-	"2.5.4.11": "OU",
 	"2.5.4.3":  "CN",
 	"2.5.4.5":  "SERIALNUMBER",
+	"2.5.4.6":  "C",
 	"2.5.4.7":  "L",
 	"2.5.4.8":  "ST",
 	"2.5.4.9":  "STREET",
+	"2.5.4.10": "O",
+	"2.5.4.11": "OU",
 	"2.5.4.17": "POSTALCODE",
 	// FIXME: Add others.
 	"0.9.2342.19200300.100.1.25": "DC",
@@ -44,7 +45,7 @@ type DN struct {
 }
 
 // FromCertSubject takes a pkix.Name from a cert and returns a DN
-// that uses the same set.
+// that uses the same set.  Does not support multi value RDNs.
 func FromCertSubject(subject pkix.Name) (*DN, error) {
 	dn := &DN{
 		RDNs: make([]*RelativeDN, 0),
@@ -73,6 +74,53 @@ func FromCertSubject(subject pkix.Name) (*DN, error) {
 	return dn, nil
 }
 
+// FromRawCertSubject takes a raw subject from a certificate
+// and uses asn1.Unmarshal to get the individual RDNs in the
+// original order, including multi-value RDNs.
+func FromRawCertSubject(rawSubject []byte) (*DN, error) {
+	dn := &DN{
+		RDNs: make([]*RelativeDN, 0),
+	}
+	var rdns pkix.RDNSequence
+	_, err := asn1.Unmarshal(rawSubject, &rdns)
+	if err != nil {
+		return nil, err
+	}
+
+	for i := len(rdns) - 1; i >= 0; i-- {
+		rdn := rdns[i]
+		if len(rdn) == 0 {
+			continue
+		}
+
+		r := &RelativeDN{}
+		attrs := make([]*AttributeTypeAndValue, 0)
+		for j := len(rdn) - 1; j >= 0; j-- {
+			atv := rdn[j]
+
+			typeName := ""
+			name := atv.Type.String()
+			typeName, ok := attributeTypeNames[name]
+			if !ok {
+				return nil, fmt.Errorf("invalid type name: %+v", name)
+			}
+			value, ok := atv.Value.(string)
+			if !ok {
+				return nil, fmt.Errorf("invalid type value: %+v", atv.Value)
+			}
+			attr := &AttributeTypeAndValue{
+				Type:  typeName,
+				Value: value,
+			}
+			attrs = append(attrs, attr)
+		}
+		r.Attributes = attrs
+		dn.RDNs = append(dn.RDNs, r)
+	}
+
+	return dn, nil
+}
+
 // ParseDN returns a distinguishedName or an error.
 // The function respects https://tools.ietf.org/html/rfc4514
 func ParseDN(str string) (*DN, error) {
@@ -181,6 +229,29 @@ func (d *DN) Equal(other *DN) bool {
 	return true
 }
 
+// RDNsMatch returns true if the individual RDNs of the DNs
+// are the same regardless of ordering.
+func (d *DN) RDNsMatch(other *DN) bool {
+	if len(d.RDNs) != len(other.RDNs) {
+		return false
+	}
+
+CheckNextRDN:
+	for _, irdn := range d.RDNs {
+		for _, ordn := range other.RDNs {
+			if (len(irdn.Attributes) == len(ordn.Attributes)) &&
+				(irdn.hasAllAttributes(ordn.Attributes) && ordn.hasAllAttributes(irdn.Attributes)) {
+				// Found the RDN, check if next one matches.
+				continue CheckNextRDN
+			}
+		}
+
+		// Could not find a matching individual RDN, auth fails.
+		return false
+	}
+	return true
+}
+
 // AncestorOf returns true if the other DN consists of at least one RDN followed by all the RDNs of the current DN.
 // "ou=widgets,o=acme.com" is an ancestor of "ou=sprockets,ou=widgets,o=acme.com"
 // "ou=widgets,o=acme.com" is not an ancestor of "ou=sprockets,ou=widgets,o=foo.com"
diff --git a/vendor/github.com/nats-io/nats-server/v2/server/accounts.go b/vendor/github.com/nats-io/nats-server/v2/server/accounts.go
index 9a68c03..2802fb9 100644
--- a/vendor/github.com/nats-io/nats-server/v2/server/accounts.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/accounts.go
@@ -118,7 +118,6 @@ type serviceImport struct {
 	sid         []byte
 	from        string
 	to          string
-	exsub       string
 	tr          *transform
 	ts          int64
 	rt          ServiceRespType
@@ -893,6 +892,7 @@ func (a *Account) AddServiceExportWithResponse(subject string, respType ServiceR
 	if a == nil {
 		return ErrMissingAccount
 	}
+
 	a.mu.Lock()
 	defer a.mu.Unlock()
 
@@ -1344,26 +1344,94 @@ func (a *Account) AddServiceImportWithClaim(destination *Account, from, to strin
 		return ErrServiceImportAuthorization
 	}
 
-	if a.importFormsCycle(destination, from, to) {
-		return ErrServiceImportFormsCycle
+	// Check if this introduces a cycle before proceeding.
+	if err := a.serviceImportFormsCycle(destination, from); err != nil {
+		return err
 	}
 
 	_, err := a.addServiceImport(destination, from, to, imClaim)
 	return err
 }
 
-// Detects if we have a cycle.
-func (a *Account) importFormsCycle(destination *Account, from, to string) bool {
-	// Check that what we are importing is not something we also export.
-	if a.serviceExportOverlaps(to) {
-		// So at this point if destination account is also importing from us, that forms a cycle.
-		if destination.serviceImportOverlaps(from) {
+const MaxAccountCycleSearchDepth = 1024
+
+func (a *Account) serviceImportFormsCycle(dest *Account, from string) error {
+	return dest.checkServiceImportsForCycles(from, map[string]bool{a.Name: true})
+}
+
+func (a *Account) checkServiceImportsForCycles(from string, visited map[string]bool) error {
+	if len(visited) >= MaxAccountCycleSearchDepth {
+		return ErrCycleSearchDepth
+	}
+	a.mu.RLock()
+	for _, si := range a.imports.services {
+		if SubjectsCollide(from, si.to) {
+			a.mu.RUnlock()
+			if visited[si.acc.Name] {
+				return ErrImportFormsCycle
+			}
+			// Push ourselves and check si.acc
+			visited[a.Name] = true
+			if subjectIsSubsetMatch(si.from, from) {
+				from = si.from
+			}
+			if err := si.acc.checkServiceImportsForCycles(from, visited); err != nil {
+				return err
+			}
+			a.mu.RLock()
+		}
+	}
+	a.mu.RUnlock()
+	return nil
+}
+
+func (a *Account) streamImportFormsCycle(dest *Account, to string) error {
+	return dest.checkStreamImportsForCycles(to, map[string]bool{a.Name: true})
+}
+
+// Lock should be held.
+func (a *Account) hasStreamExportMatching(to string) bool {
+	for subj := range a.exports.streams {
+		if subjectIsSubsetMatch(to, subj) {
 			return true
 		}
 	}
 	return false
 }
 
+func (a *Account) checkStreamImportsForCycles(to string, visited map[string]bool) error {
+	if len(visited) >= MaxAccountCycleSearchDepth {
+		return ErrCycleSearchDepth
+	}
+
+	a.mu.RLock()
+
+	if !a.hasStreamExportMatching(to) {
+		a.mu.RUnlock()
+		return nil
+	}
+
+	for _, si := range a.imports.streams {
+		if SubjectsCollide(to, si.to) {
+			a.mu.RUnlock()
+			if visited[si.acc.Name] {
+				return ErrImportFormsCycle
+			}
+			// Push ourselves and check si.acc
+			visited[a.Name] = true
+			if subjectIsSubsetMatch(si.to, to) {
+				to = si.to
+			}
+			if err := si.acc.checkStreamImportsForCycles(to, visited); err != nil {
+				return err
+			}
+			a.mu.RLock()
+		}
+	}
+	a.mu.RUnlock()
+	return nil
+}
+
 // SetServiceImportSharing will allow sharing of information about requests with the export account.
 // Used for service latency tracking at the moment.
 func (a *Account) SetServiceImportSharing(destination *Account, to string, allow bool) error {
@@ -1602,30 +1670,6 @@ func (a *Account) checkForReverseEntry(reply string, si *serviceImport, checkInt
 	}
 }
 
-// Internal check to see if the to subject overlaps with another export.
-func (a *Account) serviceExportOverlaps(to string) bool {
-	a.mu.RLock()
-	defer a.mu.RUnlock()
-	for subj := range a.exports.services {
-		if to == subj || SubjectsCollide(to, subj) {
-			return true
-		}
-	}
-	return false
-}
-
-// Internal check to see if the from subject overlaps with another import.
-func (a *Account) serviceImportOverlaps(from string) bool {
-	a.mu.RLock()
-	defer a.mu.RUnlock()
-	for subj := range a.imports.services {
-		if from == subj || SubjectsCollide(from, subj) {
-			return true
-		}
-	}
-	return false
-}
-
 // Internal check to see if a service import exists.
 func (a *Account) serviceImportExists(dest *Account, from string) bool {
 	a.mu.RLock()
@@ -1692,7 +1736,11 @@ func (a *Account) addServiceImport(dest *Account, from, to string, claim *jwt.Im
 			}
 		}
 	}
-	si := &serviceImport{dest, claim, se, nil, from, to, "", tr, 0, rt, lat, nil, nil, usePub, false, false, false, false, false, isSysAcc, nil}
+	share := false
+	if claim != nil {
+		share = claim.Share
+	}
+	si := &serviceImport{dest, claim, se, nil, from, to, tr, 0, rt, lat, nil, nil, usePub, false, false, share, false, false, isSysAcc, nil}
 	a.imports.services[from] = si
 	a.mu.Unlock()
 
@@ -2107,7 +2155,7 @@ func (a *Account) addRespServiceImport(dest *Account, to string, osi *serviceImp
 
 	// dest is the requestor's account. a is the service responder with the export.
 	// Marked as internal here, that is how we distinguish.
-	si := &serviceImport{dest, nil, osi.se, nil, nrr, to, osi.to, nil, 0, rt, nil, nil, nil, false, true, false, osi.share, false, false, false, nil}
+	si := &serviceImport{dest, nil, osi.se, nil, nrr, to, nil, 0, rt, nil, nil, nil, false, true, false, osi.share, false, false, false, nil}
 
 	if a.exports.responses == nil {
 		a.exports.responses = make(map[string]*serviceImport)
@@ -2159,6 +2207,7 @@ func (a *Account) AddStreamImportWithClaim(account *Account, from, prefix string
 			prefix = prefix + string(btsep)
 		}
 	}
+
 	return a.AddMappedStreamImportWithClaim(account, from, prefix+from, imClaim)
 }
 
@@ -2181,6 +2230,12 @@ func (a *Account) AddMappedStreamImportWithClaim(account *Account, from, to stri
 	if to == "" {
 		to = from
 	}
+
+	// Check if this forms a cycle.
+	if err := a.streamImportFormsCycle(account, to); err != nil {
+		return err
+	}
+
 	var (
 		usePub bool
 		tr     *transform
@@ -2489,7 +2544,7 @@ func (a *Account) checkActivation(importAcc *Account, claim *jwt.Import, expTime
 		clone.Token = fetchActivation(url.String())
 	}
 	vr := jwt.CreateValidationResults()
-	clone.Validate(a.Name, vr)
+	clone.Validate(importAcc.Name, vr)
 	if vr.IsBlocking(true) {
 		return false
 	}
@@ -2692,8 +2747,8 @@ func (a *Account) hasIssuer(issuer string) bool {
 
 // hasIssuerNoLock is the unlocked version of hasIssuer
 func (a *Account) hasIssuerNoLock(issuer string) bool {
-	// same issuer
-	if a.Issuer == issuer {
+	// same issuer -- keep this for safety on the calling code
+	if a.Name == issuer {
 		return true
 	}
 	for i := 0; i < len(a.signingKeys); i++ {
@@ -2846,8 +2901,12 @@ func (s *Server) updateAccountClaimsWithRefresh(a *Account, ac *jwt.AccountClaim
 				s.Debugf("Error adding service export to account [%s]: %v", a.Name, err)
 			}
 			if e.Latency != nil {
-				if err := a.TrackServiceExportWithSampling(string(e.Subject), string(e.Latency.Results), e.Latency.Sampling); err != nil {
-					s.Debugf("Error adding latency tracking for service export to account [%s]: %v", a.Name, err)
+				if err := a.TrackServiceExportWithSampling(string(e.Subject), string(e.Latency.Results), int(e.Latency.Sampling)); err != nil {
+					hdrNote := ""
+					if e.Latency.Sampling == jwt.Headers {
+						hdrNote = " (using headers)"
+					}
+					s.Debugf("Error adding latency tracking%s for service export to account [%s]: %v", hdrNote, a.Name, err)
 				}
 			}
 		}
diff --git a/vendor/github.com/nats-io/nats-server/v2/server/auth.go b/vendor/github.com/nats-io/nats-server/v2/server/auth.go
index 7043b2c..0e04304 100644
--- a/vendor/github.com/nats-io/nats-server/v2/server/auth.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/auth.go
@@ -253,6 +253,8 @@ func (s *Server) configureAuthorization() {
 
 	// Do similar for websocket config
 	s.wsConfigAuth(&opts.Websocket)
+	// And for mqtt config
+	s.mqttConfigAuth(&opts.MQTT)
 }
 
 // Takes the given slices of NkeyUser and User options and build
@@ -343,11 +345,15 @@ func (s *Server) processClientOrLeafAuthentication(c *client, opts *Options) boo
 	)
 	s.mu.Lock()
 	authRequired := s.info.AuthRequired
-	// c.ws is immutable, but may need lock if we get race reports.
-	if !authRequired && c.ws != nil {
+	if !authRequired {
 		// If no auth required for regular clients, then check if
-		// we have an override for websocket clients.
-		authRequired = s.websocket.authOverride
+		// we have an override for MQTT or Websocket clients.
+		switch c.clientType() {
+		case MQTT:
+			authRequired = s.mqtt.authOverride
+		case WS:
+			authRequired = s.websocket.authOverride
+		}
 	}
 	if !authRequired {
 		// TODO(dlc) - If they send us credentials should we fail?
@@ -361,20 +367,36 @@ func (s *Server) processClientOrLeafAuthentication(c *client, opts *Options) boo
 		noAuthUser string
 	)
 	tlsMap := opts.TLSMap
-	if c.ws != nil {
-		wo := &opts.Websocket
-		// Always override TLSMap.
-		tlsMap = wo.TLSMap
-		// The rest depends on if there was any auth override in
-		// the websocket's config.
-		if s.websocket.authOverride {
-			noAuthUser = wo.NoAuthUser
-			username = wo.Username
-			password = wo.Password
-			token = wo.Token
-			ao = true
+	if c.kind == CLIENT {
+		switch c.clientType() {
+		case MQTT:
+			mo := &opts.MQTT
+			// Always override TLSMap.
+			tlsMap = mo.TLSMap
+			// The rest depends on if there was any auth override in
+			// the mqtt's config.
+			if s.mqtt.authOverride {
+				noAuthUser = mo.NoAuthUser
+				username = mo.Username
+				password = mo.Password
+				token = mo.Token
+				ao = true
+			}
+		case WS:
+			wo := &opts.Websocket
+			// Always override TLSMap.
+			tlsMap = wo.TLSMap
+			// The rest depends on if there was any auth override in
+			// the websocket's config.
+			if s.websocket.authOverride {
+				noAuthUser = wo.NoAuthUser
+				username = wo.Username
+				password = wo.Password
+				token = wo.Token
+				ao = true
+			}
 		}
-	} else if c.kind == LEAF {
+	} else {
 		tlsMap = opts.LeafNode.TLSMap
 	}
 	if !ao {
@@ -419,7 +441,7 @@ func (s *Server) processClientOrLeafAuthentication(c *client, opts *Options) boo
 	} else if hasUsers {
 		// Check if we are tls verify and are mapping users from the client_certificate.
 		if tlsMap {
-			authorized := checkClientTLSCertSubject(c, func(u string, certRDN *ldap.DN, _ bool) (string, bool) {
+			authorized := checkClientTLSCertSubject(c, func(u string, certDN *ldap.DN, _ bool) (string, bool) {
 				// First do literal lookup using the resulting string representation
 				// of RDNSequence as implemented by the pkix package from Go.
 				if u != "" {
@@ -431,23 +453,36 @@ func (s *Server) processClientOrLeafAuthentication(c *client, opts *Options) boo
 					return usr.Username, ok
 				}
 
-				if certRDN == nil {
+				if certDN == nil {
 					return "", false
 				}
 
-				// Look through the accounts for an RDN that is equal to the one
+				// Look through the accounts for a DN that is equal to the one
 				// presented by the certificate.
+				dns := make(map[*User]*ldap.DN)
 				for _, usr := range s.users {
 					if !c.connectionTypeAllowed(usr.AllowedConnectionTypes) {
 						continue
 					}
 					// TODO: Use this utility to make a full validation pass
 					// on start in case tlsmap feature is being used.
-					inputRDN, err := ldap.ParseDN(usr.Username)
+					inputDN, err := ldap.ParseDN(usr.Username)
 					if err != nil {
 						continue
 					}
-					if inputRDN.Equal(certRDN) {
+					if inputDN.Equal(certDN) {
+						user = usr
+						return usr.Username, true
+					}
+
+					// In case it did not match exactly, then collect the DNs
+					// and try to match later in case the DN was reordered.
+					dns[usr] = inputDN
+				}
+
+				// Check in case the DN was reordered.
+				for usr, inputDN := range dns {
+					if inputDN.RDNsMatch(certDN) {
 						user = usr
 						return usr.Username, true
 					}
@@ -525,6 +560,7 @@ func (s *Server) processClientOrLeafAuthentication(c *client, opts *Options) boo
 			c.Debugf("Account JWT not signed by trusted operator")
 			return false
 		}
+		// this only executes IF there's an issuer on the Juc - otherwise the account is already vetted
 		if juc.IssuerAccount != "" && !acc.hasIssuer(juc.Issuer) {
 			c.Debugf("User JWT issuer is not known")
 			return false
@@ -724,8 +760,9 @@ func checkClientTLSCertSubject(c *client, fn tlsMapAuthFn) bool {
 	// the domain components in case there are any.
 	rdn := cert.Subject.ToRDNSequence().String()
 
-	// Match that follows original order from the subject takes precedence.
-	dn, err := ldap.FromCertSubject(cert.Subject)
+	// Match using the raw subject to avoid ignoring attributes.
+	// https://github.com/golang/go/issues/12342
+	dn, err := ldap.FromRawCertSubject(cert.RawSubject)
 	if err == nil {
 		if match, ok := fn("", dn, false); ok {
 			c.Debugf("Using DistinguishedNameMatch for auth [%q]", match)
@@ -998,7 +1035,7 @@ func validateAllowedConnectionTypes(m map[string]struct{}) error {
 	for ct := range m {
 		ctuc := strings.ToUpper(ct)
 		switch ctuc {
-		case jwt.ConnectionTypeStandard, jwt.ConnectionTypeWebsocket, jwt.ConnectionTypeLeafnode:
+		case jwt.ConnectionTypeStandard, jwt.ConnectionTypeWebsocket, jwt.ConnectionTypeLeafnode, jwt.ConnectionTypeMqtt:
 		default:
 			return fmt.Errorf("unknown connection type %q", ct)
 		}
diff --git a/vendor/github.com/nats-io/nats-server/v2/server/client.go b/vendor/github.com/nats-io/nats-server/v2/server/client.go
index 11fe670..e6ad311 100644
--- a/vendor/github.com/nats-io/nats-server/v2/server/client.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/client.go
@@ -51,6 +51,20 @@ const (
 	ACCOUNT
 )
 
+// Extended type of a CLIENT connection. This is returned by c.clientType()
+// and indicate what type of client connection we are dealing with.
+// If invoked on a non CLIENT connection, NON_CLIENT type is returned.
+const (
+	// If the connection is not a CLIENT connection.
+	NON_CLIENT = iota
+	// Regular NATS client.
+	NATS
+	// MQTT client.
+	MQTT
+	// Websocket client.
+	WS
+)
+
 const (
 	// ClientProtoZero is the original Client protocol from 2009.
 	// http://nats.io/documentation/internals/nats-protocol/
@@ -178,6 +192,7 @@ const (
 	NoRespondersRequiresHeaders
 	ClusterNameConflict
 	DuplicateRemoteLeafnodeConnection
+	DuplicateClientID
 )
 
 // Some flags passed to processMsgResults
@@ -237,6 +252,7 @@ type client struct {
 	gw    *gateway
 	leaf  *leaf
 	ws    *websocket
+	mqtt  *mqtt
 
 	// To keep track of gateway replies mapping
 	gwrm map[string]*gwReplyMap
@@ -425,6 +441,26 @@ func (c *client) GetTLSConnectionState() *tls.ConnectionState {
 	return &state
 }
 
+// For CLIENT connections, this function returns the client type, that is,
+// NATS (for regular clients), MQTT or WS for websocket.
+// If this is invoked for a non CLIENT connection, NON_CLIENT is returned.
+//
+// This function does not lock the client and accesses fields that are supposed
+// to be immutable and therefore it can be invoked outside of the client's lock.
+func (c *client) clientType() int {
+	switch c.kind {
+	case CLIENT:
+		if c.isMqtt() {
+			return MQTT
+		} else if c.isWebsocket() {
+			return WS
+		}
+		return NATS
+	default:
+		return NON_CLIENT
+	}
+}
+
 // This is the main subscription struct that indicates
 // interest in published messages.
 // FIXME(dlc) - This is getting bloated for normal subs, need
@@ -442,6 +478,7 @@ type subscription struct {
 	max     int64
 	qw      int32
 	closed  int32
+	mqtt    *mqttSub
 }
 
 // Indicate that this subscription is closed.
@@ -537,11 +574,14 @@ func (c *client) initClient() {
 
 	switch c.kind {
 	case CLIENT:
-		name := "cid"
-		if c.ws != nil {
-			name = "wid"
+		switch c.clientType() {
+		case NATS:
+			c.ncs.Store(fmt.Sprintf("%s - cid:%d", conn, c.cid))
+		case WS:
+			c.ncs.Store(fmt.Sprintf("%s - wid:%d", conn, c.cid))
+		case MQTT:
+			c.ncs.Store(fmt.Sprintf("%s - mid:%d", conn, c.cid))
 		}
-		c.ncs.Store(fmt.Sprintf("%s - %s:%d", conn, name, c.cid))
 	case ROUTER:
 		c.ncs.Store(fmt.Sprintf("%s - rid:%d", conn, c.cid))
 	case GATEWAY:
@@ -963,7 +1003,10 @@ func (c *client) readLoop(pre []byte) {
 		return
 	}
 	nc := c.nc
-	ws := c.ws != nil
+	ws := c.isWebsocket()
+	if c.isMqtt() {
+		c.mqtt.r = &mqttReader{reader: nc}
+	}
 	c.in.rsz = startBufSize
 	// Snapshot max control line since currently can not be changed on reload and we
 	// were checking it on each call to parse. If this changes and we allow MaxControlLine
@@ -983,6 +1026,9 @@ func (c *client) readLoop(pre []byte) {
 	c.mu.Unlock()
 
 	defer func() {
+		if c.isMqtt() {
+			s.mqttHandleWill(c)
+		}
 		// These are used only in the readloop, so we can set them to nil
 		// on exit of the readLoop.
 		c.in.results, c.in.pacache = nil, nil
@@ -1144,7 +1190,7 @@ func closedStateForErr(err error) ClosedState {
 // collapsePtoNB will place primary onto nb buffer as needed in prep for WriteTo.
 // This will return a copy on purpose.
 func (c *client) collapsePtoNB() (net.Buffers, int64) {
-	if c.ws != nil {
+	if c.isWebsocket() {
 		return c.wsCollapsePtoNB()
 	}
 	if c.out.p != nil {
@@ -1158,7 +1204,7 @@ func (c *client) collapsePtoNB() (net.Buffers, int64) {
 // This will handle the fixup needed on a partial write.
 // Assume pending has been already calculated correctly.
 func (c *client) handlePartialWrite(pnb net.Buffers) {
-	if c.ws != nil {
+	if c.isWebsocket() {
 		c.ws.frames = append(pnb, c.ws.frames...)
 		return
 	}
@@ -1252,7 +1298,7 @@ func (c *client) flushOutbound() bool {
 
 	// Subtract from pending bytes and messages.
 	c.out.pb -= n
-	if c.ws != nil {
+	if c.isWebsocket() {
 		c.ws.fs -= n
 	}
 	c.out.pm -= apm // FIXME(dlc) - this will not be totally accurate on partials.
@@ -1371,7 +1417,7 @@ func (c *client) markConnAsClosed(reason ClosedState) {
 	c.flags.set(connMarkedClosed)
 	// For a websocket client, unless we are told not to flush, enqueue
 	// a websocket CloseMessage based on the reason.
-	if !skipFlush && c.ws != nil && !c.ws.closeSent {
+	if !skipFlush && c.isWebsocket() && !c.ws.closeSent {
 		c.wsEnqueueCloseMessage(reason)
 	}
 	// Be consistent with the creation: for routes and gateways,
@@ -1683,7 +1729,6 @@ func (c *client) processConnect(arg []byte) error {
 			// By default register with the global account.
 			c.registerWithAccount(srv.globalAccount())
 		}
-
 	}
 
 	switch kind {
@@ -1703,7 +1748,6 @@ func (c *client) processConnect(arg []byte) error {
 			c.sendErr(ErrNoRespondersRequiresHeaders.Error())
 			c.closeConnection(NoRespondersRequiresHeaders)
 			return ErrNoRespondersRequiresHeaders
-
 		}
 		if verbose {
 			c.sendOK()
@@ -1933,6 +1977,9 @@ func (c *client) sendRTTPing() bool {
 // the c.rtt is 0 and wants to force an update by sending a PING.
 // Client lock held on entry.
 func (c *client) sendRTTPingLocked() bool {
+	if c.isMqtt() {
+		return false
+	}
 	// Most client libs send a CONNECT+PING and wait for a PONG from the
 	// server. So if firstPongSent flag is set, it is ok for server to
 	// send the PING. But in case we have client libs that don't do that,
@@ -1963,7 +2010,7 @@ func (c *client) generateClientInfoJSON(info Info) []byte {
 	info.CID = c.cid
 	info.ClientIP = c.host
 	info.MaxPayload = c.mpay
-	if c.ws != nil {
+	if c.isWebsocket() {
 		info.ClientConnectURLs = info.WSConnectURLs
 	}
 	info.WSConnectURLs = nil
@@ -1978,7 +2025,9 @@ func (c *client) sendErr(err string) {
 	if c.trace {
 		c.traceOutOp("-ERR", []byte(err))
 	}
-	c.enqueueProto([]byte(fmt.Sprintf(errProto, err)))
+	if !c.isMqtt() {
+		c.enqueueProto([]byte(fmt.Sprintf(errProto, err)))
+	}
 	c.mu.Unlock()
 }
 
@@ -2254,7 +2303,7 @@ func (c *client) processSub(subject, queue, bsid []byte, cb msgHandler, noForwar
 	// This check does not apply to SYSTEM or JETSTREAM or ACCOUNT clients (because they don't have a `nc`...)
 	if c.isClosed() && (kind != SYSTEM && kind != JETSTREAM && kind != ACCOUNT) {
 		c.mu.Unlock()
-		return sub, nil
+		return nil, ErrConnectionClosed
 	}
 
 	// Check permissions if applicable.
@@ -2942,7 +2991,12 @@ func (c *client) deliverMsg(sub *subscription, subject, reply, mh, msg []byte, g
 	// Update statistics
 
 	// The msg includes the CR_LF, so pull back out for accounting.
-	msgSize := int64(len(msg) - LEN_CR_LF)
+	msgSize := int64(len(msg))
+	prodIsMQTT := c.isMqtt()
+	// MQTT producers send messages without CR_LF, so don't remove it for them.
+	if !prodIsMQTT {
+		msgSize -= int64(LEN_CR_LF)
+	}
 
 	// No atomic needed since accessed under client lock.
 	// Monitor is reading those also under client's lock.
@@ -2951,6 +3005,13 @@ func (c *client) deliverMsg(sub *subscription, subject, reply, mh, msg []byte, g
 
 	// Check for internal subscriptions.
 	if sub.icb != nil {
+		if gwrply {
+			// Note that we keep track of the GW routed reply in the destination
+			// connection (`client`). The routed reply subject is in `c.pa.reply`,
+			// should that change, we would have to pass the GW routed reply as
+			// a parameter of deliverMsg().
+			srv.trackGWReply(client, c.pa.reply)
+		}
 		client.mu.Unlock()
 		// Internal account clients are for service imports and need the '\r\n'.
 		if client.kind == ACCOUNT {
@@ -2988,9 +3049,10 @@ func (c *client) deliverMsg(sub *subscription, subject, reply, mh, msg []byte, g
 	// header (which is c.pa.reply without the GNR routing prefix).
 	if client.kind == CLIENT && len(c.pa.reply) > minReplyLen {
 		if gwrply {
-			// Note we keep track "in" the destination client (`client`) but the
-			// routed reply subject is in `c.pa.reply`. Should that change, we
-			// would have to pass the "reply" in deliverMsg().
+			// Note that we keep track of the GW routed reply in the destination
+			// connection (`client`). The routed reply subject is in `c.pa.reply`,
+			// should that change, we would have to pass the GW routed reply as
+			// a parameter of deliverMsg().
 			srv.trackGWReply(client, c.pa.reply)
 		}
 
@@ -3009,6 +3071,10 @@ func (c *client) deliverMsg(sub *subscription, subject, reply, mh, msg []byte, g
 	// Queue to outbound buffer
 	client.queueOutbound(mh)
 	client.queueOutbound(msg)
+	if prodIsMQTT {
+		// Need to add CR_LF since MQTT producers don't send CR_LF
+		client.queueOutbound([]byte(CR_LF))
+	}
 
 	client.out.pm++
 
@@ -3303,6 +3369,11 @@ func (c *client) processInboundClientMsg(msg []byte) bool {
 		c.sendOK()
 	}
 
+	// If MQTT client, check for retain flag now that we have passed permissions check
+	if c.isMqtt() {
+		c.mqttHandlePubRetain()
+	}
+
 	// Check if this client's gateway replies map is not empty
 	if atomic.LoadInt32(&c.cgwrt) > 0 && c.handleGWReplyMap(msg) {
 		return true
@@ -3483,7 +3554,14 @@ func (c *client) setupResponseServiceImport(acc *Account, si *serviceImport, tra
 // processServiceImport is an internal callback when a subscription matches an imported service
 // from another account. This includes response mappings as well.
 func (c *client) processServiceImport(si *serviceImport, acc *Account, msg []byte) {
-	if c.kind == GATEWAY && !si.isRespServiceImport() {
+	// If we are a GW and this is not a direct serviceImport ignore.
+	isResponse := si.isRespServiceImport()
+	if c.kind == GATEWAY && !isResponse {
+		return
+	}
+	// If we are here and we are a serviceImport response make sure we are not matching back
+	// to the import/export pair that started the request. If so ignore.
+	if isResponse && c.pa.psi != nil && c.pa.psi.se == si.se {
 		return
 	}
 
@@ -3535,8 +3613,11 @@ func (c *client) processServiceImport(si *serviceImport, acc *Account, msg []byt
 	// so we only lock once.
 	to, _ = si.acc.selectMappedSubject(to)
 
-	oreply, oacc := c.pa.reply, c.acc
+	// Save off some of our state.
+	oreply, oacc, opsi := c.pa.reply, c.acc, c.pa.psi
 	c.pa.reply = nrr
+	c.pa.psi = si
+
 	if !si.isSysAcc {
 		c.mu.Lock()
 		c.acc = si.acc
@@ -3577,6 +3658,7 @@ func (c *client) processServiceImport(si *serviceImport, acc *Account, msg []byt
 	// Put what was there back now.
 	c.in.rts = orts
 	c.pa.reply = oreply
+	c.pa.psi = opsi
 
 	if !si.isSysAcc {
 		c.mu.Lock()
@@ -3746,7 +3828,7 @@ func (c *client) processMsgResults(acc *Account, r *SublistResult, msg, deliver,
 			}
 		}
 
-		// Remap to original if internal.
+		// Remap to the original subject if internal.
 		if sub.icb != nil {
 			subj = subject
 		}
@@ -4588,7 +4670,7 @@ func convertAllowedConnectionTypes(cts []string) (map[string]struct{}, error) {
 	for _, i := range cts {
 		i = strings.ToUpper(i)
 		switch i {
-		case jwt.ConnectionTypeStandard, jwt.ConnectionTypeWebsocket, jwt.ConnectionTypeLeafnode:
+		case jwt.ConnectionTypeStandard, jwt.ConnectionTypeWebsocket, jwt.ConnectionTypeLeafnode, jwt.ConnectionTypeMqtt:
 			m[i] = struct{}{}
 		default:
 			unknown = append(unknown, i)
@@ -4610,15 +4692,20 @@ func (c *client) connectionTypeAllowed(acts map[string]struct{}) bool {
 	if len(acts) == 0 {
 		return true
 	}
-	// Assume standard client, then update based on presence of websocket
-	// or other type.
-	want := jwt.ConnectionTypeStandard
-	if c.kind == LEAF {
+	var want string
+	switch c.kind {
+	case CLIENT:
+		switch c.clientType() {
+		case NATS:
+			want = jwt.ConnectionTypeStandard
+		case WS:
+			want = jwt.ConnectionTypeWebsocket
+		case MQTT:
+			want = jwt.ConnectionTypeMqtt
+		}
+	case LEAF:
 		want = jwt.ConnectionTypeLeafnode
 	}
-	if c.ws != nil {
-		want = jwt.ConnectionTypeWebsocket
-	}
 	_, ok := acts[want]
 	return ok
 }
diff --git a/vendor/github.com/nats-io/nats-server/v2/server/const.go b/vendor/github.com/nats-io/nats-server/v2/server/const.go
index 24dc72d..bcb1f43 100644
--- a/vendor/github.com/nats-io/nats-server/v2/server/const.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/const.go
@@ -40,7 +40,7 @@ var (
 
 const (
 	// VERSION is the current version for the server.
-	VERSION = "2.2.0-beta.33"
+	VERSION = "2.2.0-beta.35"
 
 	// PROTO is the currently supported protocol.
 	// 0 was the original
diff --git a/vendor/github.com/nats-io/nats-server/v2/server/consumer.go b/vendor/github.com/nats-io/nats-server/v2/server/consumer.go
index 605e15b..897bddd 100644
--- a/vendor/github.com/nats-io/nats-server/v2/server/consumer.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/consumer.go
@@ -60,6 +60,11 @@ type ConsumerConfig struct {
 	SampleFrequency string        `json:"sample_freq,omitempty"`
 	MaxWaiting      int           `json:"max_waiting,omitempty"`
 	MaxAckPending   int           `json:"max_ack_pending,omitempty"`
+
+	// These are non public configuration options.
+	// If you add new options, check fileConsumerInfoJSON in order for them to
+	// be properly persisted/recovered, if needed.
+	allowNoInterest bool
 }
 
 type CreateConsumerRequest struct {
@@ -265,21 +270,23 @@ func (mset *Stream) AddConsumer(config *ConsumerConfig) (*Consumer, error) {
 
 	// Make sure any partition subject is also a literal.
 	if config.FilterSubject != "" {
-		// If this is a direct match for the streams only subject clear the filter.
+		var checkSubject bool
+
 		mset.mu.RLock()
-		if len(mset.config.Subjects) == 1 && mset.config.Subjects[0] == config.FilterSubject {
-			config.FilterSubject = _EMPTY_
+		// If the stream was created with no subject, then skip the checks
+		if !mset.config.allowNoSubject {
+			// If this is a direct match for the streams only subject clear the filter.
+			if len(mset.config.Subjects) == 1 && mset.config.Subjects[0] == config.FilterSubject {
+				config.FilterSubject = _EMPTY_
+			} else {
+				checkSubject = true
+			}
 		}
 		mset.mu.RUnlock()
 
-		if config.FilterSubject != "" {
-			// Make sure this is a valid partition of the interest subjects.
-			if !mset.validSubject(config.FilterSubject) {
-				return nil, fmt.Errorf("consumer filter subject is not a valid subset of the interest subjects")
-			}
-			if config.AckPolicy == AckAll {
-				return nil, fmt.Errorf("consumer with filter subject can not have an ack policy of ack all")
-			}
+		// Make sure this is a valid partition of the interest subjects.
+		if checkSubject && !mset.validSubject(config.FilterSubject) {
+			return nil, fmt.Errorf("consumer filter subject is not a valid subset of the interest subjects")
 		}
 	}
 
@@ -353,7 +360,7 @@ func (mset *Stream) AddConsumer(config *ConsumerConfig) (*Consumer, error) {
 			} else {
 				// If we are a push mode and not active and the only difference
 				// is deliver subject then update and return.
-				if configsEqualSansDelivery(ocfg, *config) && eo.hasNoLocalInterest() {
+				if configsEqualSansDelivery(ocfg, *config) && (config.allowNoInterest || eo.hasNoLocalInterest()) {
 					eo.updateDeliverSubject(config.DeliverSubject)
 					return eo, nil
 				} else {
@@ -762,15 +769,15 @@ func (o *Consumer) sendAckReply(subj string) {
 
 // Process a message for the ack reply subject delivered with a message.
 func (o *Consumer) processAck(_ *subscription, _ *client, subject, reply string, msg []byte) {
-	sseq, dseq, dcount := ackReplyInfo(subject)
+	sseq, dseq, dc := ackReplyInfo(subject)
 
 	skipAckReply := sseq == 0
 
 	switch {
 	case len(msg) == 0, bytes.Equal(msg, AckAck), bytes.Equal(msg, AckOK):
-		o.ackMsg(sseq, dseq, dcount)
+		o.ackMsg(sseq, dseq, dc)
 	case bytes.HasPrefix(msg, AckNext):
-		o.ackMsg(sseq, dseq, dcount)
+		o.ackMsg(sseq, dseq, dc)
 		o.processNextMsgReq(nil, nil, subject, reply, msg[len(AckNext):])
 		skipAckReply = true
 	case bytes.Equal(msg, AckNak):
@@ -778,7 +785,7 @@ func (o *Consumer) processAck(_ *subscription, _ *client, subject, reply string,
 	case bytes.Equal(msg, AckProgress):
 		o.progressUpdate(sseq)
 	case bytes.Equal(msg, AckTerm):
-		o.processTerm(sseq, dseq, dcount)
+		o.processTerm(sseq, dseq, dc)
 	}
 
 	// Ack the ack if requested.
@@ -824,9 +831,9 @@ func (o *Consumer) processNak(sseq, dseq uint64) {
 }
 
 // Process a TERM
-func (o *Consumer) processTerm(sseq, dseq, dcount uint64) {
+func (o *Consumer) processTerm(sseq, dseq, dc uint64) {
 	// Treat like an ack to suppress redelivery.
-	o.processAckMsg(sseq, dseq, dcount, false)
+	o.processAckMsg(sseq, dseq, dc, false)
 
 	o.mu.Lock()
 	defer o.mu.Unlock()
@@ -842,7 +849,7 @@ func (o *Consumer) processTerm(sseq, dseq, dcount uint64) {
 		Consumer:    o.name,
 		ConsumerSeq: dseq,
 		StreamSeq:   sseq,
-		Deliveries:  dcount,
+		Deliveries:  dc,
 	}
 
 	j, err := json.MarshalIndent(e, "", "  ")
@@ -1031,7 +1038,7 @@ func (o *Consumer) shouldSample() bool {
 	return mrand.Int31n(100) <= o.sfreq
 }
 
-func (o *Consumer) sampleAck(sseq, dseq, dcount uint64) {
+func (o *Consumer) sampleAck(sseq, dseq, dc uint64) {
 	if !o.shouldSample() {
 		return
 	}
@@ -1050,7 +1057,7 @@ func (o *Consumer) sampleAck(sseq, dseq, dcount uint64) {
 		ConsumerSeq: dseq,
 		StreamSeq:   sseq,
 		Delay:       unow - o.pending[sseq].Timestamp,
-		Deliveries:  dcount,
+		Deliveries:  dc,
 	}
 
 	j, err := json.MarshalIndent(e, "", "  ")
@@ -1062,11 +1069,11 @@ func (o *Consumer) sampleAck(sseq, dseq, dcount uint64) {
 }
 
 // Process an ack for a message.
-func (o *Consumer) ackMsg(sseq, dseq, dcount uint64) {
-	o.processAckMsg(sseq, dseq, dcount, true)
+func (o *Consumer) ackMsg(sseq, dseq, dc uint64) {
+	o.processAckMsg(sseq, dseq, dc, true)
 }
 
-func (o *Consumer) processAckMsg(sseq, dseq, dcount uint64, doSample bool) {
+func (o *Consumer) processAckMsg(sseq, dseq, dc uint64, doSample bool) {
 	o.mu.Lock()
 	var sagap uint64
 	var needSignal bool
@@ -1075,7 +1082,7 @@ func (o *Consumer) processAckMsg(sseq, dseq, dcount uint64, doSample bool) {
 	case AckExplicit:
 		if p, ok := o.pending[sseq]; ok {
 			if doSample {
-				o.sampleAck(sseq, dseq, dcount)
+				o.sampleAck(sseq, dseq, dc)
 			}
 			if o.maxp > 0 && len(o.pending) >= o.maxp {
 				needSignal = true
@@ -1363,7 +1370,7 @@ func (o *Consumer) incDeliveryCount(sseq uint64) uint64 {
 }
 
 // send a delivery exceeded advisory.
-func (o *Consumer) notifyDeliveryExceeded(sseq, dcount uint64) {
+func (o *Consumer) notifyDeliveryExceeded(sseq, dc uint64) {
 	e := JSConsumerDeliveryExceededAdvisory{
 		TypedEvent: TypedEvent{
 			Type: JSConsumerDeliveryExceededAdvisoryType,
@@ -1373,7 +1380,7 @@ func (o *Consumer) notifyDeliveryExceeded(sseq, dcount uint64) {
 		Stream:     o.stream,
 		Consumer:   o.name,
 		StreamSeq:  sseq,
-		Deliveries: dcount,
+		Deliveries: dc,
 	}
 
 	j, err := json.MarshalIndent(e, "", "  ")
@@ -1405,20 +1412,20 @@ var errBadConsumer = errors.New("consumer not valid")
 // Get next available message from underlying store.
 // Is partition aware and redeliver aware.
 // Lock should be held.
-func (o *Consumer) getNextMsg() (subj string, hdr, msg []byte, seq uint64, dcount uint64, ts int64, err error) {
+func (o *Consumer) getNextMsg() (subj string, hdr, msg []byte, seq uint64, dc uint64, ts int64, err error) {
 	if o.mset == nil {
 		return _EMPTY_, nil, nil, 0, 0, 0, errBadConsumer
 	}
 	for {
-		seq, dcount := o.sseq, uint64(1)
+		seq, dc := o.sseq, uint64(1)
 		if len(o.rdq) > 0 {
 			seq = o.rdq[0]
 			o.rdq = append(o.rdq[:0], o.rdq[1:]...)
-			dcount = o.incDeliveryCount(seq)
-			if o.maxdc > 0 && dcount > o.maxdc {
+			dc = o.incDeliveryCount(seq)
+			if o.maxdc > 0 && dc > o.maxdc {
 				// Only send once
-				if dcount == o.maxdc+1 {
-					o.notifyDeliveryExceeded(seq, dcount-1)
+				if dc == o.maxdc+1 {
+					o.notifyDeliveryExceeded(seq, dc-1)
 				}
 				// Make sure to remove from pending.
 				delete(o.pending, seq)
@@ -1432,14 +1439,14 @@ func (o *Consumer) getNextMsg() (subj string, hdr, msg []byte, seq uint64, dcoun
 
 		subj, hdr, msg, ts, err := o.mset.store.LoadMsg(seq)
 		if err == nil {
-			if dcount == 1 { // First delivery.
+			if dc == 1 { // First delivery.
 				o.sseq++
 				if o.config.FilterSubject != _EMPTY_ && !o.isFilteredMatch(subj) {
 					continue
 				}
 			}
 			// We have the msg here.
-			return subj, hdr, msg, seq, dcount, ts, nil
+			return subj, hdr, msg, seq, dc, ts, nil
 		}
 		// We got an error here. If this is an EOF we will return, otherwise
 		// we can continue looking.
@@ -1517,7 +1524,7 @@ func (o *Consumer) loopAndGatherMsgs(s *Server, a *Account) {
 	// Deliver all the msgs we have now, once done or on a condition, we wait for new ones.
 	for {
 		var (
-			seq, dcount uint64
+			seq, dc     uint64
 			subj, dsubj string
 			hdr         []byte
 			msg         []byte
@@ -1543,7 +1550,7 @@ func (o *Consumer) loopAndGatherMsgs(s *Server, a *Account) {
 			goto waitForMsgs
 		}
 
-		subj, hdr, msg, seq, dcount, ts, err = o.getNextMsg()
+		subj, hdr, msg, seq, dc, ts, err = o.getNextMsg()
 
 		// On error either wait or return.
 		if err != nil {
@@ -1595,7 +1602,7 @@ func (o *Consumer) loopAndGatherMsgs(s *Server, a *Account) {
 			}
 		}
 
-		o.deliverMsg(dsubj, subj, hdr, msg, seq, dcount, ts)
+		o.deliverMsg(dsubj, subj, hdr, msg, seq, dc, ts)
 
 		o.mu.Unlock()
 		continue
@@ -1619,8 +1626,8 @@ func (o *Consumer) loopAndGatherMsgs(s *Server, a *Account) {
 	}
 }
 
-func (o *Consumer) ackReply(sseq, dseq, dcount uint64, ts int64, pending uint64) string {
-	return fmt.Sprintf(o.ackReplyT, dcount, sseq, dseq, ts, pending)
+func (o *Consumer) ackReply(sseq, dseq, dc uint64, ts int64, pending uint64) string {
+	return fmt.Sprintf(o.ackReplyT, dc, sseq, dseq, ts, pending)
 }
 
 // deliverCurrentMsg is the hot path to deliver a message that was just received.
@@ -1676,12 +1683,12 @@ func (o *Consumer) deliverCurrentMsg(subj string, hdr, msg []byte, seq uint64, t
 
 // Deliver a msg to the consumer.
 // Lock should be held and o.mset validated to be non-nil.
-func (o *Consumer) deliverMsg(dsubj, subj string, hdr, msg []byte, seq, dcount uint64, ts int64) {
+func (o *Consumer) deliverMsg(dsubj, subj string, hdr, msg []byte, seq, dc uint64, ts int64) {
 	if o.mset == nil {
 		return
 	}
 	// Update pending on first attempt
-	if dcount == 1 && o.sgap > 0 {
+	if dc == 1 && o.sgap > 0 {
 		o.sgap--
 	}
 
@@ -1693,7 +1700,7 @@ func (o *Consumer) deliverMsg(dsubj, subj string, hdr, msg []byte, seq, dcount u
 	}
 
 	dseq := o.dseq
-	pmsg := &jsPubMsg{dsubj, subj, o.ackReply(seq, dseq, dcount, ts, o.sgap), hdr, msg, o, seq}
+	pmsg := &jsPubMsg{dsubj, subj, o.ackReply(seq, dseq, dc, ts, o.sgap), hdr, msg, o, seq}
 	mset := o.mset
 	ap := o.config.AckPolicy
 
@@ -1718,7 +1725,7 @@ func (o *Consumer) deliverMsg(dsubj, subj string, hdr, msg []byte, seq, dcount u
 	o.dseq++
 
 	// FIXME(dlc) - Capture errors?
-	o.store.UpdateDelivered(dseq, seq, dcount, ts)
+	o.store.UpdateDelivered(dseq, seq, dc, ts)
 }
 
 // Tracks our outstanding pending acks. Only applicable to AckExplicit mode.
@@ -1868,7 +1875,7 @@ func parseAckReplyNum(d string) (n int64) {
 const expectedNumReplyTokens = 9
 
 // Grab encoded information in the reply subject for a delivered message.
-func (o *Consumer) ReplyInfo(subject string) (sseq, dseq, dcount uint64, ts int64, pending uint64) {
+func (o *Consumer) ReplyInfo(subject string) (sseq, dseq, dc uint64, ts int64, pending uint64) {
 	tsa := [expectedNumReplyTokens]string{}
 	start, tokens := 0, tsa[:0]
 	for i := 0; i < len(subject); i++ {
@@ -1883,15 +1890,15 @@ func (o *Consumer) ReplyInfo(subject string) (sseq, dseq, dcount uint64, ts int6
 	}
 	// TODO(dlc) - Should we error if we do not match consumer name?
 	// stream is tokens[2], consumer is 3.
-	dcount = uint64(parseAckReplyNum(tokens[4]))
+	dc = uint64(parseAckReplyNum(tokens[4]))
 	sseq, dseq = uint64(parseAckReplyNum(tokens[5])), uint64(parseAckReplyNum(tokens[6]))
 	ts = parseAckReplyNum(tokens[7])
 	pending = uint64(parseAckReplyNum(tokens[8]))
 
-	return sseq, dseq, dcount, ts, pending
+	return sseq, dseq, dc, ts, pending
 }
 
-func ackReplyInfo(subject string) (sseq, dseq, dcount uint64) {
+func ackReplyInfo(subject string) (sseq, dseq, dc uint64) {
 	tsa := [expectedNumReplyTokens]string{}
 	start, tokens := 0, tsa[:0]
 	for i := 0; i < len(subject); i++ {
@@ -1904,10 +1911,10 @@ func ackReplyInfo(subject string) (sseq, dseq, dcount uint64) {
 	if len(tokens) != expectedNumReplyTokens || tokens[0] != "$JS" || tokens[1] != "ACK" {
 		return 0, 0, 0
 	}
-	dcount = uint64(parseAckReplyNum(tokens[4]))
+	dc = uint64(parseAckReplyNum(tokens[4]))
 	sseq, dseq = uint64(parseAckReplyNum(tokens[5])), uint64(parseAckReplyNum(tokens[6]))
 
-	return sseq, dseq, dcount
+	return sseq, dseq, dc
 }
 
 // NextSeq returns the next delivered sequence number for this observable.
@@ -2169,8 +2176,8 @@ func (o *Consumer) stop(dflag, doSignal, advisory bool) error {
 // Check that we do not form a cycle by delivering to a delivery subject
 // that is part of the interest group.
 func (mset *Stream) deliveryFormsCycle(deliverySubject string) bool {
-	mset.mu.Lock()
-	defer mset.mu.Unlock()
+	mset.mu.RLock()
+	defer mset.mu.RUnlock()
 
 	for _, subject := range mset.config.Subjects {
 		if subjectIsSubsetMatch(deliverySubject, subject) {
@@ -2268,13 +2275,9 @@ func (o *Consumer) incStreamPending(sseq uint64, subj string) {
 
 func (o *Consumer) decStreamPending(sseq uint64, subj string) {
 	o.mu.Lock()
-	defer o.mu.Unlock()
-
 	// Ignore if we have already seen this one.
-	if sseq < o.sseq || o.sgap == 0 {
-		return
-	}
-	if o.isFilteredMatch(subj) {
+	if sseq >= o.sseq && o.sgap > 0 && o.isFilteredMatch(subj) {
 		o.sgap--
 	}
+	o.mu.Unlock()
 }
diff --git a/vendor/github.com/nats-io/nats-server/v2/server/dirstore.go b/vendor/github.com/nats-io/nats-server/v2/server/dirstore.go
index 5a6af63..c1fb8b7 100644
--- a/vendor/github.com/nats-io/nats-server/v2/server/dirstore.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/dirstore.go
@@ -607,7 +607,7 @@ func (pq *expirationTracker) updateTrack(publicKey string) {
 		i := e.Value.(*jwtItem)
 		if pq.ttl != 0 {
 			// only update expiration when set
-			i.expiration = time.Now().Round(time.Second).Add(pq.ttl).Unix()
+			i.expiration = time.Now().Add(pq.ttl).UnixNano()
 			heap.Fix(pq, i.index)
 		}
 		if pq.evictOnLimit {
@@ -631,11 +631,11 @@ func (pq *expirationTracker) track(publicKey string, hash *[sha256.Size]byte, th
 		if pq.ttl == time.Duration(math.MaxInt64) {
 			exp = math.MaxInt64
 		} else {
-			exp = time.Now().Round(time.Second).Add(pq.ttl).Unix()
+			exp = time.Now().Add(pq.ttl).UnixNano()
 		}
 	} else {
 		if g, err := jwt.DecodeGeneric(theJWT); err == nil {
-			exp = g.Expires
+			exp = time.Unix(g.Expires, 0).UnixNano()
 		}
 		if exp == 0 {
 			exp = math.MaxInt64 // default to indefinite
@@ -683,21 +683,18 @@ func (store *DirJWTStore) startExpiring(reCheck time.Duration, limit int64, evic
 		defer t.Stop()
 		defer pq.wg.Done()
 		for {
-			now := time.Now()
+			now := time.Now().UnixNano()
 			store.Lock()
 			if pq.Len() > 0 {
-				if it := heap.Pop(pq).(*jwtItem); it.expiration <= now.Unix() {
+				if it := pq.heap[0]; it.expiration <= now {
 					path := store.pathForKey(it.publicKey)
-					if err := os.Remove(path); err != nil {
-						heap.Push(pq, it) // retry later
-					} else {
+					if err := os.Remove(path); err == nil {
+						heap.Pop(pq)
 						pq.unTrack(it.publicKey)
 						xorAssign(&pq.hash, it.hash)
 						store.Unlock()
-						continue // we removed an entry, check next one
+						continue // we removed an entry, check next one right away
 					}
-				} else {
-					heap.Push(pq, it)
 				}
 			}
 			store.Unlock()
diff --git a/vendor/github.com/nats-io/nats-server/v2/server/errors.go b/vendor/github.com/nats-io/nats-server/v2/server/errors.go
index 28f12fd..3610352 100644
--- a/vendor/github.com/nats-io/nats-server/v2/server/errors.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/errors.go
@@ -125,8 +125,11 @@ var (
 	// ErrServiceImportAuthorization is returned when a service import is not authorized.
 	ErrServiceImportAuthorization = errors.New("service import not authorized")
 
-	// ErrServiceImportFormsCycle is returned when a service import forms a cycle.
-	ErrServiceImportFormsCycle = errors.New("service import forms cycle")
+	// ErrImportFormsCycle is returned when an import would form a cycle.
+	ErrImportFormsCycle = errors.New("import forms a cycle")
+
+	// ErrCycleSearchDepth is returned when we have exceeded our maximum search depth..
+	ErrCycleSearchDepth = errors.New("search cycle depth exhausted")
 
 	// ErrClientOrRouteConnectedToGatewayPort represents an error condition when
 	// a client or route attempted to connect to the Gateway port.
diff --git a/vendor/github.com/nats-io/nats-server/v2/server/events.go b/vendor/github.com/nats-io/nats-server/v2/server/events.go
index 449198c..9d3862d 100644
--- a/vendor/github.com/nats-io/nats-server/v2/server/events.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/events.go
@@ -1410,6 +1410,7 @@ func (s *Server) systemSubscribe(subject, queue string, internalOnly bool, cb ms
 	if queue != "" {
 		q = []byte(queue)
 	}
+	// Now create the subscription
 	return c.processSub([]byte(subject), q, []byte(sid), cb, internalOnly)
 }
 
diff --git a/vendor/github.com/nats-io/nats-server/v2/server/filestore.go b/vendor/github.com/nats-io/nats-server/v2/server/filestore.go
index fd1520f..1aaf65a 100644
--- a/vendor/github.com/nats-io/nats-server/v2/server/filestore.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/filestore.go
@@ -56,6 +56,38 @@ type FileStreamInfo struct {
 	StreamConfig
 }
 
+// Need an alias (which does not have MarshalJSON/UnmarshalJSON) to avoid
+// recursive calls which would lead to stack overflow.
+type fileStreamInfoAlias FileStreamInfo
+
+// We will use this struct definition to serialize/deserialize FileStreamInfo
+// object. This embeds FileStreamInfo (the alias to prevent recursive calls)
+// and makes the non-public options public so they can be persisted/recovered.
+type fileStreamInfoJSON struct {
+	fileStreamInfoAlias
+	Internal       bool `json:"internal,omitempty"`
+	AllowNoSubject bool `json:"allow_no_subject,omitempty"`
+}
+
+func (fsi FileStreamInfo) MarshalJSON() ([]byte, error) {
+	return json.Marshal(&fileStreamInfoJSON{
+		fileStreamInfoAlias(fsi),
+		fsi.internal,
+		fsi.allowNoSubject,
+	})
+}
+
+func (fsi *FileStreamInfo) UnmarshalJSON(b []byte) error {
+	fsiJSON := &fileStreamInfoJSON{}
+	if err := json.Unmarshal(b, &fsiJSON); err != nil {
+		return err
+	}
+	*fsi = FileStreamInfo(fsiJSON.fileStreamInfoAlias)
+	fsi.internal = fsiJSON.Internal
+	fsi.allowNoSubject = fsiJSON.AllowNoSubject
+	return nil
+}
+
 // File ConsumerInfo is used for creating consumer stores.
 type FileConsumerInfo struct {
 	Created time.Time
@@ -63,6 +95,31 @@ type FileConsumerInfo struct {
 	ConsumerConfig
 }
 
+// See fileStreamInfoAlias, etc.. for details on how this all work.
+type fileConsumerInfoAlias FileConsumerInfo
+
+type fileConsumerInfoJSON struct {
+	fileConsumerInfoAlias
+	AllowNoInterest bool `json:"allow_no_interest,omitempty"`
+}
+
+func (fci FileConsumerInfo) MarshalJSON() ([]byte, error) {
+	return json.Marshal(&fileConsumerInfoJSON{
+		fileConsumerInfoAlias(fci),
+		fci.allowNoInterest,
+	})
+}
+
+func (fci *FileConsumerInfo) UnmarshalJSON(b []byte) error {
+	fciJSON := &fileConsumerInfoJSON{}
+	if err := json.Unmarshal(b, &fciJSON); err != nil {
+		return err
+	}
+	*fci = FileConsumerInfo(fciJSON.fileConsumerInfoAlias)
+	fci.allowNoInterest = fciJSON.AllowNoInterest
+	return nil
+}
+
 type fileStore struct {
 	mu       sync.RWMutex
 	state    StreamState
@@ -2232,6 +2289,12 @@ func (mb *msgBlock) readIndexInfo() error {
 	mb.last.ts = readTimeStamp()
 	dmapLen := readCount()
 
+	// Check if this is a short write index file.
+	if bi < 0 || bi+checksumSize > len(buf) {
+		defer os.Remove(mb.ifn)
+		return fmt.Errorf("short index file")
+	}
+
 	// Checksum
 	copy(mb.lchk[0:], buf[bi:bi+checksumSize])
 	bi += checksumSize
@@ -2319,11 +2382,11 @@ func (fs *fileStore) dmapEntries() int {
 
 // Purge will remove all messages from this store.
 // Will return the number of purged messages.
-func (fs *fileStore) Purge() uint64 {
+func (fs *fileStore) Purge() (uint64, error) {
 	fs.mu.Lock()
 	if fs.closed {
 		fs.mu.Unlock()
-		return 0
+		return 0, ErrStoreClosed
 	}
 
 	purged := fs.state.Msgs
@@ -2370,7 +2433,43 @@ func (fs *fileStore) Purge() uint64 {
 		cb(-int64(purged), -rbytes, 0, _EMPTY_)
 	}
 
-	return purged
+	return purged, nil
+}
+
+// Compact will remove all messages from this store up to
+// but not including the seq parameter.
+// Will return the number of purged messages.
+func (fs *fileStore) Compact(seq uint64) (uint64, error) {
+	if seq == 0 {
+		return fs.Purge()
+	}
+
+	if _, err := fs.msgForSeq(seq); err != nil {
+		return 0, err
+	}
+
+	var purged uint64
+	for fseq := fs.firstSeq(); fseq < seq; fseq = fs.firstSeq() {
+		if found, err := fs.removeMsg(fseq, false); err != nil {
+			if err == ErrStoreMsgNotFound {
+				continue
+			} else if err == ErrStoreEOF {
+				err = nil
+			}
+			return purged, err
+		} else if found {
+			purged++
+		}
+	}
+
+	return purged, nil
+}
+
+func (fs *fileStore) firstSeq() uint64 {
+	fs.mu.RLock()
+	fseq := fs.state.FirstSeq
+	fs.mu.RUnlock()
+	return fseq
 }
 
 // Returns number of msg blks.
@@ -2488,7 +2587,6 @@ func (fs *fileStore) Delete() error {
 	if fs.isClosed() {
 		return ErrStoreClosed
 	}
-	// TODO(dlc) - check error here?
 	fs.Purge()
 	if err := fs.Stop(); err != nil {
 		return err
@@ -2897,11 +2995,12 @@ func (o *consumerFileStore) UpdateDelivered(dseq, sseq, dc uint64, ts int64) err
 			o.state.Delivered.Consumer = dseq
 			o.state.Delivered.Stream = sseq
 			p = &Pending{dseq, ts}
-		} else if dc > 1 {
+		}
+		if dc > 1 {
 			if o.state.Redelivered == nil {
 				o.state.Redelivered = make(map[uint64]uint64)
 			}
-			o.state.Redelivered[sseq] = dc
+			o.state.Redelivered[sseq] = dc - 1
 		}
 		o.state.Pending[sseq] = &Pending{dseq, ts}
 	} else {
@@ -2932,27 +3031,33 @@ func (o *consumerFileStore) UpdateAcks(dseq, sseq uint64) error {
 	if p == nil {
 		return ErrStoreMsgNotFound
 	}
+	// Delete from our state.
 	delete(o.state.Pending, sseq)
+	if len(o.state.Redelivered) > 0 {
+		delete(o.state.Redelivered, sseq)
+		if len(o.state.Redelivered) == 0 {
+			o.state.Redelivered = nil
+		}
+	}
 
-	// TODO(dlc) - Check to see if we move ack floor.
 	if len(o.state.Pending) == 0 {
 		o.state.Pending = nil
 		o.state.AckFloor.Consumer = o.state.Delivered.Consumer
 		o.state.AckFloor.Stream = o.state.Delivered.Stream
-	} else if o.state.AckFloor.Consumer == 0 {
-		o.state.AckFloor.Consumer = dseq
-		o.state.AckFloor.Stream = sseq
 	} else if o.state.AckFloor.Consumer == dseq-1 {
+		notFirst := o.state.AckFloor.Consumer != 0
 		o.state.AckFloor.Consumer = dseq
 		o.state.AckFloor.Stream = sseq
-		// Close gap if needed.
-		for ss := sseq + 1; ss < o.state.Delivered.Stream; ss++ {
-			if p, ok := o.state.Pending[ss]; ok {
-				if p.Sequence > 0 {
-					o.state.AckFloor.Consumer = p.Sequence - 1
-					o.state.AckFloor.Stream = ss - 1
+		// Close the gap if needed.
+		if notFirst && o.state.Delivered.Consumer > dseq {
+			for ss := sseq + 1; ss < o.state.Delivered.Stream; ss++ {
+				if p, ok := o.state.Pending[ss]; ok {
+					if p.Sequence > 0 {
+						o.state.AckFloor.Consumer = p.Sequence - 1
+						o.state.AckFloor.Stream = ss - 1
+					}
+					break
 				}
-				break
 			}
 		}
 	}
@@ -3028,6 +3133,7 @@ func (o *consumerFileStore) encodeState() ([]byte, error) {
 			n += binary.PutUvarint(buf[n:], v)
 		}
 	}
+
 	return buf[:n], nil
 }
 
@@ -3323,6 +3429,23 @@ func (o *consumerFileStore) State() (*ConsumerState, error) {
 			state.Redelivered[seq] = n
 		}
 	}
+
+	// Copy this state into our own.
+	o.state.Delivered = state.Delivered
+	o.state.AckFloor = state.AckFloor
+	if len(state.Pending) > 0 {
+		o.state.Pending = make(map[uint64]*Pending, len(state.Pending))
+		for seq, p := range state.Pending {
+			o.state.Pending[seq] = &Pending{p.Sequence, p.Timestamp}
+		}
+	}
+	if len(state.Redelivered) > 0 {
+		o.state.Redelivered = make(map[uint64]uint64, len(state.Redelivered))
+		for seq, dc := range state.Redelivered {
+			o.state.Redelivered[seq] = dc
+		}
+	}
+
 	return state, nil
 }
 
diff --git a/vendor/github.com/nats-io/nats-server/v2/server/gateway.go b/vendor/github.com/nats-io/nats-server/v2/server/gateway.go
index 5f7a290..1629998 100644
--- a/vendor/github.com/nats-io/nats-server/v2/server/gateway.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/gateway.go
@@ -370,6 +370,29 @@ func (s *Server) newGateway(opts *Options) error {
 	return nil
 }
 
+// Update remote gateways TLS configurations after a config reload.
+func (g *srvGateway) updateRemotesTLSConfig(opts *Options) {
+	g.Lock()
+	defer g.Unlock()
+
+	for _, ro := range opts.Gateway.Gateways {
+		if ro.Name == g.name {
+			continue
+		}
+		if cfg, ok := g.remotes[ro.Name]; ok {
+			cfg.Lock()
+			// If TLS config is in remote, use that one, otherwise,
+			// use the TLS config from the main block.
+			if ro.TLSConfig != nil {
+				cfg.TLSConfig = ro.TLSConfig.Clone()
+			} else if opts.Gateway.TLSConfig != nil {
+				cfg.TLSConfig = opts.Gateway.TLSConfig.Clone()
+			}
+			cfg.Unlock()
+		}
+	}
+}
+
 // Returns the Gateway's name of this server.
 func (g *srvGateway) getName() string {
 	g.RLock()
diff --git a/vendor/github.com/nats-io/nats-server/v2/server/jetstream.go b/vendor/github.com/nats-io/nats-server/v2/server/jetstream.go
index 88f62a7..9a59147 100644
--- a/vendor/github.com/nats-io/nats-server/v2/server/jetstream.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/jetstream.go
@@ -258,7 +258,15 @@ func (s *Server) configJetStream(acc *Account) error {
 
 // configAllJetStreamAccounts walk all configured accounts and turn on jetstream if requested.
 func (s *Server) configAllJetStreamAccounts() error {
-	var jsAccounts []*Account
+	// Check to see if system account has been enabled. We could arrive here via reload and
+	// a non-default system account.
+	if sacc := s.SystemAccount(); sacc != nil && !sacc.IsExportService(JSApiAccountInfo) {
+		for _, export := range allJsExports {
+			if err := sacc.AddServiceExport(export, nil); err != nil {
+				return fmt.Errorf("Error setting up jetstream service exports: %v", err)
+			}
+		}
+	}
 
 	// Snapshot into our own list. Might not be needed.
 	s.mu.Lock()
@@ -268,11 +276,15 @@ func (s *Server) configAllJetStreamAccounts() error {
 		s.mu.Unlock()
 		return nil
 	}
+
+	var jsAccounts []*Account
+
 	s.accounts.Range(func(k, v interface{}) bool {
 		jsAccounts = append(jsAccounts, v.(*Account))
 		return true
 	})
 	s.mu.Unlock()
+
 	// Process any jetstream enabled accounts here.
 	for _, acc := range jsAccounts {
 		if err := s.configJetStream(acc); err != nil {
@@ -382,14 +394,14 @@ func (a *Account) EnableJetStream(limits *JetStreamAccountLimits) error {
 
 	js.mu.Lock()
 	// Check the limits against existing reservations.
-	if err := js.sufficientResources(limits); err != nil {
-		js.mu.Unlock()
-		return err
-	}
 	if _, ok := js.accounts[a]; ok {
 		js.mu.Unlock()
 		return fmt.Errorf("jetstream already enabled for account")
 	}
+	if err := js.sufficientResources(limits); err != nil {
+		js.mu.Unlock()
+		return err
+	}
 	jsa := &jsAccount{js: js, account: a, limits: *limits, streams: make(map[string]*Stream)}
 	jsa.storeDir = path.Join(js.config.StoreDir, a.Name)
 	js.accounts[a] = jsa
@@ -926,6 +938,7 @@ func (js *jetStream) reserveResources(limits *JetStreamAccountLimits) error {
 	return nil
 }
 
+// Lock should be held.
 func (js *jetStream) releaseResources(limits *JetStreamAccountLimits) error {
 	if limits == nil {
 		return nil
@@ -939,6 +952,17 @@ func (js *jetStream) releaseResources(limits *JetStreamAccountLimits) error {
 	return nil
 }
 
+// Will clear the resource reservations. Mostly for reload of a config.
+func (js *jetStream) clearResources() {
+	if js == nil {
+		return
+	}
+	js.mu.Lock()
+	js.memReserved = 0
+	js.storeReserved = 0
+	js.mu.Unlock()
+}
+
 const (
 	// JetStreamStoreDir is the prefix we use.
 	JetStreamStoreDir = "jetstream"
diff --git a/vendor/github.com/nats-io/nats-server/v2/server/jetstream_api.go b/vendor/github.com/nats-io/nats-server/v2/server/jetstream_api.go
index 48ce47a..7d27c69 100644
--- a/vendor/github.com/nats-io/nats-server/v2/server/jetstream_api.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/jetstream_api.go
@@ -143,7 +143,7 @@ const (
 	jsSnapshotAckT    = "$JS.SNAPSHOT.ACK.%s.%s"
 	jsRestoreDeliverT = "$JS.SNAPSHOT.RESTORE.%s.%s"
 
-	// JetStreamAckT is the template for the ack message stream coming back from a consumer
+	// jsAckT is the template for the ack message stream coming back from a consumer
 	// when they ACK/NAK, etc a message.
 	jsAckT   = "$JS.ACK.%s.%s"
 	jsAckPre = "$JS.ACK."
@@ -952,7 +952,14 @@ func (s *Server) jsStreamInfoRequest(sub *subscription, c *client, subject, repl
 		s.sendAPIResponse(c, subject, reply, string(msg), s.jsonResponse(&resp))
 		return
 	}
-	resp.StreamInfo = &StreamInfo{Created: mset.Created(), State: mset.State(), Config: mset.Config()}
+	config := mset.Config()
+	// Some streams are created without subject (for instance MQTT streams),
+	// but "nats" tooling would then fail to display them since it uses
+	// validation and expect the config's Subjects to not be empty.
+	if config.allowNoSubject && len(config.Subjects) == 0 {
+		config.Subjects = []string{">"}
+	}
+	resp.StreamInfo = &StreamInfo{Created: mset.Created(), State: mset.State(), Config: config}
 	s.sendAPIResponse(c, subject, reply, string(msg), s.jsonResponse(resp))
 }
 
@@ -999,6 +1006,11 @@ func (s *Server) jsStreamDeleteRequest(sub *subscription, c *client, subject, re
 		s.sendAPIResponse(c, subject, reply, string(msg), s.jsonResponse(&resp))
 		return
 	}
+	if mset.Config().internal {
+		resp.Error = &ApiError{Code: 403, Description: "not allowed to delete internal stream"}
+		s.sendAPIResponse(c, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
 	if err := mset.Delete(); err != nil {
 		resp.Error = jsError(err)
 		s.sendAPIResponse(c, subject, reply, string(msg), s.jsonResponse(&resp))
@@ -1127,8 +1139,13 @@ func (s *Server) jsStreamPurgeRequest(sub *subscription, c *client, subject, rep
 		s.sendAPIResponse(c, subject, reply, string(msg), s.jsonResponse(&resp))
 		return
 	}
-	resp.Purged = mset.Purge()
-	resp.Success = true
+	purged, err := mset.Purge()
+	if err != nil {
+		resp.Error = jsError(err)
+	} else {
+		resp.Purged = purged
+		resp.Success = true
+	}
 	s.sendAPIResponse(c, subject, reply, string(msg), s.jsonResponse(resp))
 }
 
@@ -1706,6 +1723,11 @@ func (s *Server) jsConsumerDeleteRequest(sub *subscription, c *client, subject,
 		s.sendAPIResponse(c, subject, reply, string(msg), s.jsonResponse(&resp))
 		return
 	}
+	if mset.Config().internal {
+		resp.Error = &ApiError{Code: 403, Description: "not allowed to delete consumer of internal stream"}
+		s.sendAPIResponse(c, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
 	consumer := consumerNameFromSubject(subject)
 	obs := mset.LookupConsumer(consumer)
 	if obs == nil {
diff --git a/vendor/github.com/nats-io/nats-server/v2/server/leafnode.go b/vendor/github.com/nats-io/nats-server/v2/server/leafnode.go
index 61264f7..a448c33 100644
--- a/vendor/github.com/nats-io/nats-server/v2/server/leafnode.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/leafnode.go
@@ -105,7 +105,10 @@ func (c *client) isHubLeafNode() bool {
 // This will spin up go routines to solicit the remote leaf node connections.
 func (s *Server) solicitLeafNodeRemotes(remotes []*RemoteLeafOpts) {
 	for _, r := range remotes {
+		s.mu.Lock()
 		remote := newLeafNodeCfg(r)
+		s.leafRemoteCfgs = append(s.leafRemoteCfgs, remote)
+		s.mu.Unlock()
 		s.startGoRoutine(func() { s.connectToRemoteLeafNode(remote, true) })
 	}
 }
@@ -209,6 +212,32 @@ func validateLeafNodeAuthOptions(o *Options) error {
 	return nil
 }
 
+// Update remote LeafNode TLS configurations after a config reload.
+func (s *Server) updateRemoteLeafNodesTLSConfig(opts *Options) {
+	max := len(opts.LeafNode.Remotes)
+	if max == 0 {
+		return
+	}
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	// Changes in the list of remote leaf nodes is not supported.
+	// However, make sure that we don't go over the arrays.
+	if len(s.leafRemoteCfgs) < max {
+		max = len(s.leafRemoteCfgs)
+	}
+	for i := 0; i < max; i++ {
+		ro := opts.LeafNode.Remotes[i]
+		cfg := s.leafRemoteCfgs[i]
+		if ro.TLSConfig != nil {
+			cfg.Lock()
+			cfg.TLSConfig = ro.TLSConfig.Clone()
+			cfg.Unlock()
+		}
+	}
+}
+
 func (s *Server) reConnectToRemoteLeafNode(remote *leafNodeCfg) {
 	delay := s.getOpts().LeafNode.ReconnectInterval
 	select {
@@ -724,13 +753,16 @@ func (s *Server) createLeafNode(conn net.Conn, remote *leafNodeCfg) *client {
 		}
 
 		// Do TLS here as needed.
-		tlsRequired := remote.TLS || remote.TLSConfig != nil
+		remote.RLock()
+		remoteTLSConfig := remote.TLSConfig
+		tlsRequired := remote.TLS || remoteTLSConfig != nil
+		remote.RUnlock()
 		if tlsRequired {
 			c.Debugf("Starting TLS leafnode client handshake")
 			// Specify the ServerName we are expecting.
 			var tlsConfig *tls.Config
-			if remote.TLSConfig != nil {
-				tlsConfig = remote.TLSConfig.Clone()
+			if remoteTLSConfig != nil {
+				tlsConfig = remoteTLSConfig.Clone()
 			} else {
 				tlsConfig = &tls.Config{MinVersion: tls.VersionTLS12}
 			}
diff --git a/vendor/github.com/nats-io/nats-server/v2/server/memstore.go b/vendor/github.com/nats-io/nats-server/v2/server/memstore.go
index 971ed45..9d93229 100644
--- a/vendor/github.com/nats-io/nats-server/v2/server/memstore.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/memstore.go
@@ -254,7 +254,7 @@ func (ms *memStore) expireMsgs() {
 
 // Purge will remove all messages from this store.
 // Will return the number of purged messages.
-func (ms *memStore) Purge() uint64 {
+func (ms *memStore) Purge() (uint64, error) {
 	ms.mu.Lock()
 	purged := uint64(len(ms.msgs))
 	cb := ms.scb
@@ -270,7 +270,46 @@ func (ms *memStore) Purge() uint64 {
 		cb(-int64(purged), -bytes, 0, _EMPTY_)
 	}
 
-	return purged
+	return purged, nil
+}
+
+// Compact will remove all messages from this store up to
+// but not including the seq parameter.
+// Will return the number of purged messages.
+func (ms *memStore) Compact(seq uint64) (uint64, error) {
+	if seq == 0 {
+		return ms.Purge()
+	}
+	ms.mu.Lock()
+	sm, ok := ms.msgs[seq]
+	if !ok {
+		ms.mu.Unlock()
+		return 0, ErrStoreMsgNotFound
+	}
+	ms.state.FirstSeq = seq
+	ms.state.FirstTime = time.Unix(0, sm.ts).UTC()
+
+	var purged, bytes uint64
+	for seq := seq - 1; seq > 0; seq-- {
+		sm := ms.msgs[seq]
+		if sm == nil {
+			continue
+		}
+		bytes += memStoreMsgSize(sm.subj, sm.hdr, sm.msg)
+		purged++
+		delete(ms.msgs, seq)
+	}
+	ms.state.Msgs -= purged
+	ms.state.Bytes -= bytes
+
+	cb := ms.scb
+	ms.mu.Unlock()
+
+	if cb != nil {
+		cb(-int64(purged), -int64(bytes), 0, _EMPTY_)
+	}
+
+	return purged, nil
 }
 
 func (ms *memStore) deleteFirstMsgOrPanic() {
diff --git a/vendor/github.com/nats-io/nats-server/v2/server/monitor.go b/vendor/github.com/nats-io/nats-server/v2/server/monitor.go
index 4bae0fd..1c79816 100644
--- a/vendor/github.com/nats-io/nats-server/v2/server/monitor.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/monitor.go
@@ -1915,6 +1915,8 @@ func (reason ClosedState) String() string {
 		return "Cluster Name Conflict"
 	case DuplicateRemoteLeafnodeConnection:
 		return "Duplicate Remote LeafNode Connection"
+	case DuplicateClientID:
+		return "Duplicate Client ID"
 	}
 
 	return "Unknown State"
@@ -1931,7 +1933,7 @@ func newExtServiceLatency(l *serviceLatency) *jwt.ServiceLatency {
 		return nil
 	}
 	return &jwt.ServiceLatency{
-		Sampling: int(l.sampling),
+		Sampling: jwt.SamplingRate(l.sampling),
 		Results:  jwt.Subject(l.subject),
 	}
 }
diff --git a/vendor/github.com/nats-io/nats-server/v2/server/mqtt.go b/vendor/github.com/nats-io/nats-server/v2/server/mqtt.go
new file mode 100644
index 0000000..65fef7f
--- /dev/null
+++ b/vendor/github.com/nats-io/nats-server/v2/server/mqtt.go
@@ -0,0 +1,2724 @@
+// 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 server
+
+import (
+	"bytes"
+	"crypto/tls"
+	"encoding/binary"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+	"unicode/utf8"
+
+	"github.com/nats-io/nuid"
+)
+
+// References to "spec" here is from https://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.pdf
+
+const (
+	mqttPacketConnect    = byte(0x10)
+	mqttPacketConnectAck = byte(0x20)
+	mqttPacketPub        = byte(0x30)
+	mqttPacketPubAck     = byte(0x40)
+	mqttPacketPubRec     = byte(0x50)
+	mqttPacketPubRel     = byte(0x60)
+	mqttPacketPubComp    = byte(0x70)
+	mqttPacketSub        = byte(0x80)
+	mqttPacketSubAck     = byte(0x90)
+	mqttPacketUnsub      = byte(0xa0)
+	mqttPacketUnsubAck   = byte(0xb0)
+	mqttPacketPing       = byte(0xc0)
+	mqttPacketPingResp   = byte(0xd0)
+	mqttPacketDisconnect = byte(0xe0)
+	mqttPacketMask       = byte(0xf0)
+	mqttPacketFlagMask   = byte(0x0f)
+
+	mqttProtoLevel = byte(0x4)
+
+	// Connect flags
+	mqttConnFlagReserved     = byte(0x1)
+	mqttConnFlagCleanSession = byte(0x2)
+	mqttConnFlagWillFlag     = byte(0x04)
+	mqttConnFlagWillQoS      = byte(0x18)
+	mqttConnFlagWillRetain   = byte(0x20)
+	mqttConnFlagPasswordFlag = byte(0x40)
+	mqttConnFlagUsernameFlag = byte(0x80)
+
+	// Publish flags
+	mqttPubFlagRetain = byte(0x01)
+	mqttPubFlagQoS    = byte(0x06)
+	mqttPubFlagDup    = byte(0x08)
+	mqttPubQos1       = byte(0x2) // 1 << 1
+
+	// Subscribe flags
+	mqttSubscribeFlags = byte(0x2)
+	mqttSubAckFailure  = byte(0x80)
+
+	// Unsubscribe flags
+	mqttUnsubscribeFlags = byte(0x2)
+
+	// ConnAck returned codes
+	mqttConnAckRCConnectionAccepted          = byte(0x0)
+	mqttConnAckRCUnacceptableProtocolVersion = byte(0x1)
+	mqttConnAckRCIdentifierRejected          = byte(0x2)
+	mqttConnAckRCServerUnavailable           = byte(0x3)
+	mqttConnAckRCBadUserOrPassword           = byte(0x4)
+	mqttConnAckRCNotAuthorized               = byte(0x5)
+
+	// Maximum payload size of a control packet
+	mqttMaxPayloadSize = 0xFFFFFFF
+
+	// Topic/Filter characters
+	mqttTopicLevelSep = '/'
+	mqttSingleLevelWC = '+'
+	mqttMultiLevelWC  = '#'
+
+	// This is appended to the sid of a subscription that is
+	// created on the upper level subject because of the MQTT
+	// wildcard '#' semantic.
+	mqttMultiLevelSidSuffix = " fwc"
+
+	// This is the prefix for NATS subscriptions subjects associated as delivery
+	// subject of JS consumer. We want to make them unique so will prevent users
+	// MQTT subscriptions to start with this.
+	mqttSubPrefix = "$MQTT.sub."
+
+	// MQTT Stream names prefix. Will be used to prevent users from creating
+	// JS consumers on those.
+	mqttStreamNamePrefix = "$MQTT_"
+
+	// Stream name for MQTT messages on a given account
+	mqttStreamName = mqttStreamNamePrefix + "msgs"
+
+	// Stream name for MQTT retained messages on a given account
+	mqttRetainedMsgsStreamName = mqttStreamNamePrefix + "rmsgs"
+
+	// Stream name for MQTT sessions on a given account
+	mqttSessionsStreamName = mqttStreamNamePrefix + "sessions"
+
+	// Normally, MQTT server should not redeliver QoS 1 messages to clients,
+	// except after client reconnects. However, NATS Server will redeliver
+	// unacknowledged messages after this default interval. This can be
+	// changed with the server.Options.MQTT.AckWait option.
+	mqttDefaultAckWait = 30 * time.Second
+
+	// This is the default for the outstanding number of pending QoS 1
+	// messages sent to a session with QoS 1 subscriptions.
+	mqttDefaultMaxAckPending = 1024
+)
+
+var (
+	mqttPingResponse = []byte{mqttPacketPingResp, 0x0}
+	mqttProtoName    = []byte("MQTT")
+	mqttOldProtoName = []byte("MQIsdp")
+)
+
+type srvMQTT struct {
+	listener     net.Listener
+	authOverride bool
+	sessmgr      mqttSessionManager
+}
+
+type mqttSessionManager struct {
+	mu       sync.RWMutex
+	sessions map[string]*mqttAccountSessionManager // key is account name
+}
+
+type mqttAccountSessionManager struct {
+	mu       sync.RWMutex
+	sstream  *Stream                     // stream where sessions are recorded
+	mstream  *Stream                     // messages stream
+	rstream  *Stream                     // retained messages stream
+	sessions map[string]*mqttSession     // key is MQTT client ID
+	sl       *Sublist                    // sublist allowing to find retained messages for given subscription
+	retmsgs  map[string]*mqttRetainedMsg // retained messages
+}
+
+type mqttSession struct {
+	mu       sync.Mutex
+	c        *client
+	subs     map[string]byte
+	cons     map[string]*Consumer
+	stream   *Stream
+	sseq     uint64                          // stream sequence where this sesion is recorded
+	pending  map[uint16]*mqttPending         // Key is the PUBLISH packet identifier sent to client and maps to a mqttPending record
+	cpending map[*Consumer]map[uint64]uint16 // For each JS consumer, the key is the stream sequence and maps to the PUBLISH packet identifier
+	ppi      uint16                          // publish packet identifier
+	maxp     uint16
+	stalled  bool
+	clean    bool
+}
+
+type mqttPersistedSession struct {
+	ID    string            `json:"id,omitempty"`
+	Clean bool              `json:"clean,omitempty"`
+	Subs  map[string]byte   `json:"subs,omitempty"`
+	Cons  map[string]string `json:"cons,omitempty"`
+}
+
+type mqttRetainedMsg struct {
+	Topic  []byte `json:"topic,omitempty"`
+	Msg    []byte `json:"msg,omitempty"`
+	Flags  byte   `json:"flags,omitempty"`
+	Source string `json:"source,omitempty"`
+
+	// non exported
+	sseq uint64
+	sub  *subscription
+}
+
+type mqttSub struct {
+	qos byte
+	// Pending serialization of retained messages to be sent when subscription is registered
+	prm *mqttWriter
+	// This is the corresponding JS consumer. This is applicable to a subscription that is
+	// done for QoS > 0 (the subscription attached to a JS consumer's delivery subject).
+	jsCons *Consumer
+}
+
+type mqtt struct {
+	r    *mqttReader
+	cp   *mqttConnectProto
+	pp   *mqttPublish
+	asm  *mqttAccountSessionManager // quick reference to account session manager, immutable after processConnect()
+	sess *mqttSession               // quick reference to session, immutable after processConnect()
+}
+
+type mqttPending struct {
+	sseq   uint64    // stream sequence
+	dseq   uint64    // consumer delivery sequence
+	dcount uint64    // consumer delivery count
+	jsCons *Consumer // pointer to JS consumer (to which we will call ackMsg() on with above info)
+}
+
+type mqttConnectProto struct {
+	clientID string
+	rd       time.Duration
+	will     *mqttWill
+	flags    byte
+}
+
+type mqttIOReader interface {
+	io.Reader
+	SetReadDeadline(time.Time) error
+}
+
+type mqttReader struct {
+	reader mqttIOReader
+	buf    []byte
+	pos    int
+}
+
+type mqttWriter struct {
+	bytes.Buffer
+}
+
+type mqttWill struct {
+	topic   []byte
+	subject []byte
+	message []byte
+	qos     byte
+	retain  bool
+}
+
+type mqttFilter struct {
+	filter string
+	qos    byte
+	// Used only for tracing and should not be used after parsing of (un)sub protocols.
+	ttopic []byte
+}
+
+type mqttPublish struct {
+	topic   []byte
+	subject []byte
+	msg     []byte
+	sz      int
+	pi      uint16
+	flags   byte
+	szb     [9]byte // MQTT max payload size is 268,435,455
+}
+
+func (s *Server) startMQTT() {
+	sopts := s.getOpts()
+	o := &sopts.MQTT
+
+	var hl net.Listener
+	var err error
+
+	port := o.Port
+	if port == -1 {
+		port = 0
+	}
+	hp := net.JoinHostPort(o.Host, strconv.Itoa(port))
+	s.mu.Lock()
+	if s.shutdown {
+		s.mu.Unlock()
+		return
+	}
+	s.mqtt.sessmgr.sessions = make(map[string]*mqttAccountSessionManager)
+	hl, err = net.Listen("tcp", hp)
+	if err != nil {
+		s.mu.Unlock()
+		s.Fatalf("Unable to listen for MQTT connections: %v", err)
+		return
+	}
+	if port == 0 {
+		o.Port = hl.Addr().(*net.TCPAddr).Port
+	}
+	s.mqtt.listener = hl
+	scheme := "mqtt"
+	if o.TLSConfig != nil {
+		scheme = "tls"
+	}
+	s.Noticef("Listening for MQTT clients on %s://%s:%d", scheme, o.Host, o.Port)
+	go s.acceptConnections(hl, "MQTT", func(conn net.Conn) { s.createMQTTClient(conn) }, nil)
+	s.mu.Unlock()
+}
+
+// This is similar to createClient() but has some modifications specifi to MQTT clients.
+// The comments have been kept to minimum to reduce code size. Check createClient() for
+// more details.
+func (s *Server) createMQTTClient(conn net.Conn) *client {
+	opts := s.getOpts()
+
+	maxPay := int32(opts.MaxPayload)
+	maxSubs := int32(opts.MaxSubs)
+	if maxSubs == 0 {
+		maxSubs = -1
+	}
+	now := time.Now()
+
+	c := &client{srv: s, nc: conn, mpay: maxPay, msubs: maxSubs, start: now, last: now, mqtt: &mqtt{}}
+	c.mqtt.pp = &mqttPublish{}
+	// MQTT clients don't send NATS CONNECT protocols. So make it an "echo"
+	// client, but disable verbose and pedantic (by not setting them).
+	c.opts.Echo = true
+
+	c.registerWithAccount(s.globalAccount())
+
+	s.mu.Lock()
+	// Check auth, override if applicable.
+	authRequired := s.info.AuthRequired || s.mqtt.authOverride
+	s.totalClients++
+	s.mu.Unlock()
+
+	c.mu.Lock()
+	if authRequired {
+		c.flags.set(expectConnect)
+	}
+	c.initClient()
+	c.Debugf("Client connection created")
+	c.mu.Unlock()
+
+	s.mu.Lock()
+	if !s.running || s.ldm {
+		if s.shutdown {
+			conn.Close()
+		}
+		s.mu.Unlock()
+		return c
+	}
+
+	if opts.MaxConn > 0 && len(s.clients) >= opts.MaxConn {
+		s.mu.Unlock()
+		c.maxConnExceeded()
+		return nil
+	}
+	s.clients[c.cid] = c
+
+	tlsRequired := opts.MQTT.TLSConfig != nil
+	s.mu.Unlock()
+
+	c.mu.Lock()
+
+	isClosed := c.isClosed()
+
+	var pre []byte
+	if !isClosed && tlsRequired && opts.AllowNonTLS {
+		pre = make([]byte, 4)
+		c.nc.SetReadDeadline(time.Now().Add(secondsToDuration(opts.MQTT.TLSTimeout)))
+		n, _ := io.ReadFull(c.nc, pre[:])
+		c.nc.SetReadDeadline(time.Time{})
+		pre = pre[:n]
+		if n > 0 && pre[0] == 0x16 {
+			tlsRequired = true
+		} else {
+			tlsRequired = false
+		}
+	}
+
+	if !isClosed && tlsRequired {
+		c.Debugf("Starting TLS client connection handshake")
+		if len(pre) > 0 {
+			c.nc = &tlsMixConn{c.nc, bytes.NewBuffer(pre)}
+			pre = nil
+		}
+
+		c.nc = tls.Server(c.nc, opts.MQTT.TLSConfig)
+		conn := c.nc.(*tls.Conn)
+
+		ttl := secondsToDuration(opts.MQTT.TLSTimeout)
+		time.AfterFunc(ttl, func() { tlsTimeout(c, conn) })
+		conn.SetReadDeadline(time.Now().Add(ttl))
+
+		c.mu.Unlock()
+		if err := conn.Handshake(); err != nil {
+			c.Errorf("TLS handshake error: %v", err)
+			c.closeConnection(TLSHandshakeError)
+			return nil
+		}
+		conn.SetReadDeadline(time.Time{})
+
+		c.mu.Lock()
+
+		c.flags.set(handshakeComplete)
+
+		isClosed = c.isClosed()
+	}
+
+	if isClosed {
+		c.mu.Unlock()
+		c.closeConnection(WriteError)
+		return nil
+	}
+
+	if authRequired {
+		timeout := opts.AuthTimeout
+		// Possibly override with MQTT specific value.
+		if opts.MQTT.AuthTimeout != 0 {
+			timeout = opts.MQTT.AuthTimeout
+		}
+		c.setAuthTimer(secondsToDuration(timeout))
+	}
+
+	// No Ping timer for MQTT clients...
+
+	s.startGoRoutine(func() { c.readLoop(pre) })
+	s.startGoRoutine(func() { c.writeLoop() })
+
+	if tlsRequired {
+		c.Debugf("TLS handshake complete")
+		cs := c.nc.(*tls.Conn).ConnectionState()
+		c.Debugf("TLS version %s, cipher suite %s", tlsVersion(cs.Version), tlsCipher(cs.CipherSuite))
+	}
+
+	c.mu.Unlock()
+
+	return c
+}
+
+// Given the mqtt options, we check if any auth configuration
+// has been provided. If so, possibly create users/nkey users and
+// store them in s.mqtt.users/nkeys.
+// Also update a boolean that indicates if auth is required for
+// mqtt clients.
+// Server lock is held on entry.
+func (s *Server) mqttConfigAuth(opts *MQTTOpts) {
+	mqtt := &s.mqtt
+	// If any of those is specified, we consider that there is an override.
+	mqtt.authOverride = opts.Username != _EMPTY_ || opts.Token != _EMPTY_ || opts.NoAuthUser != _EMPTY_
+}
+
+// Validate the mqtt related options.
+func validateMQTTOptions(o *Options) error {
+	mo := &o.MQTT
+	// If no port is defined, we don't care about other options
+	if mo.Port == 0 {
+		return nil
+	}
+	// If there is a NoAuthUser, we need to have Users defined and
+	// the user to be present.
+	if mo.NoAuthUser != _EMPTY_ {
+		if err := validateNoAuthUser(o, mo.NoAuthUser); err != nil {
+			return err
+		}
+	}
+	// Token/Username not possible if there are users/nkeys
+	if len(o.Users) > 0 || len(o.Nkeys) > 0 {
+		if mo.Username != _EMPTY_ {
+			return fmt.Errorf("mqtt authentication username not compatible with presence of users/nkeys")
+		}
+		if mo.Token != _EMPTY_ {
+			return fmt.Errorf("mqtt authentication token not compatible with presence of users/nkeys")
+		}
+	}
+	if mo.AckWait < 0 {
+		return fmt.Errorf("ack wait must be a positive value")
+	}
+	return nil
+}
+
+// Returns true if this connection is from a MQTT client.
+// Lock held on entry.
+func (c *client) isMqtt() bool {
+	return c.mqtt != nil
+}
+
+// Parse protocols inside the given buffer.
+// This is invoked from the readLoop.
+func (c *client) mqttParse(buf []byte) error {
+	c.mu.Lock()
+	s := c.srv
+	trace := c.trace
+	connected := c.flags.isSet(connectReceived)
+	mqtt := c.mqtt
+	r := mqtt.r
+	var rd time.Duration
+	if mqtt.cp != nil {
+		rd = mqtt.cp.rd
+		if rd > 0 {
+			r.reader.SetReadDeadline(time.Time{})
+		}
+	}
+	c.mu.Unlock()
+
+	r.reset(buf)
+
+	var err error
+	var b byte
+	var pl int
+
+	for err == nil && r.hasMore() {
+
+		// Read packet type and flags
+		if b, err = r.readByte("packet type"); err != nil {
+			break
+		}
+
+		// Packet type
+		pt := b & mqttPacketMask
+
+		// If client was not connected yet, the first packet must be
+		// a mqttPacketConnect otherwise we fail the connection.
+		if !connected && pt != mqttPacketConnect {
+			err = errors.New("not connected")
+			break
+		}
+
+		if pl, err = r.readPacketLen(); err != nil {
+			break
+		}
+
+		switch pt {
+		case mqttPacketPub:
+			pp := c.mqtt.pp
+			pp.flags = b & mqttPacketFlagMask
+			err = c.mqttParsePub(r, pl, pp)
+			if trace {
+				c.traceInOp("PUBLISH", errOrTrace(err, mqttPubTrace(pp)))
+				if err == nil {
+					c.mqttTraceMsg(pp.msg)
+				}
+			}
+			if err == nil {
+				s.mqttProcessPub(c, pp)
+				if pp.pi > 0 {
+					c.mqttEnqueuePubAck(pp.pi)
+					if trace {
+						c.traceOutOp("PUBACK", []byte(fmt.Sprintf("pi=%v", pp.pi)))
+					}
+				}
+			}
+		case mqttPacketPubAck:
+			var pi uint16
+			pi, err = mqttParsePubAck(r, pl)
+			if trace {
+				c.traceInOp("PUBACK", errOrTrace(err, fmt.Sprintf("pi=%v", pi)))
+			}
+			if err == nil {
+				c.mqttProcessPubAck(pi)
+			}
+		case mqttPacketSub:
+			var pi uint16 // packet identifier
+			var filters []*mqttFilter
+			var subs []*subscription
+			pi, filters, err = c.mqttParseSubs(r, b, pl)
+			if trace {
+				c.traceInOp("SUBSCRIBE", errOrTrace(err, mqttSubscribeTrace(pi, filters)))
+			}
+			if err == nil {
+				subs, err = c.mqttProcessSubs(filters)
+				if err == nil && trace {
+					c.traceOutOp("SUBACK", []byte(fmt.Sprintf("pi=%v", pi)))
+				}
+			}
+			if err == nil {
+				c.mqttEnqueueSubAck(pi, filters)
+				c.mqttSendRetainedMsgsToNewSubs(subs)
+			}
+		case mqttPacketUnsub:
+			var pi uint16 // packet identifier
+			var filters []*mqttFilter
+			pi, filters, err = c.mqttParseUnsubs(r, b, pl)
+			if trace {
+				c.traceInOp("UNSUBSCRIBE", errOrTrace(err, mqttUnsubscribeTrace(pi, filters)))
+			}
+			if err == nil {
+				err = c.mqttProcessUnsubs(filters)
+				if err == nil && trace {
+					c.traceOutOp("UNSUBACK", []byte(fmt.Sprintf("pi=%v", pi)))
+				}
+			}
+			if err == nil {
+				c.mqttEnqueueUnsubAck(pi)
+			}
+		case mqttPacketPing:
+			if trace {
+				c.traceInOp("PINGREQ", nil)
+			}
+			c.mqttEnqueuePingResp()
+			if trace {
+				c.traceOutOp("PINGRESP", nil)
+			}
+		case mqttPacketConnect:
+			// It is an error to receive a second connect packet
+			if connected {
+				err = errors.New("second connect packet")
+				break
+			}
+			var rc byte
+			var cp *mqttConnectProto
+			var sessp bool
+			rc, cp, err = c.mqttParseConnect(r, pl)
+			if trace && cp != nil {
+				c.traceInOp("CONNECT", errOrTrace(err, c.mqttConnectTrace(cp)))
+			}
+			if rc != 0 {
+				c.mqttEnqueueConnAck(rc, sessp)
+				if trace {
+					c.traceOutOp("CONNACK", []byte(fmt.Sprintf("sp=%v rc=%v", sessp, rc)))
+				}
+			} else if err == nil {
+				if err = s.mqttProcessConnect(c, cp, trace); err == nil {
+					connected = true
+					rd = cp.rd
+				}
+			}
+		case mqttPacketDisconnect:
+			if trace {
+				c.traceInOp("DISCONNECT", nil)
+			}
+			// Normal disconnect, we need to discard the will.
+			// Spec [MQTT-3.1.2-8]
+			c.mu.Lock()
+			if c.mqtt.cp != nil {
+				c.mqtt.cp.will = nil
+			}
+			c.mu.Unlock()
+			c.closeConnection(ClientClosed)
+			return nil
+		case mqttPacketPubRec:
+			fallthrough
+		case mqttPacketPubRel:
+			fallthrough
+		case mqttPacketPubComp:
+			err = fmt.Errorf("protocol %d not supported", pt>>4)
+		default:
+			err = fmt.Errorf("received unknown packet type %d", pt>>4)
+		}
+	}
+	if err == nil && rd > 0 {
+		r.reader.SetReadDeadline(time.Now().Add(rd))
+	}
+	return err
+}
+
+func (c *client) mqttTraceMsg(msg []byte) {
+	maxTrace := c.srv.getOpts().MaxTracedMsgLen
+	if maxTrace > 0 && len(msg) > maxTrace {
+		c.Tracef("<<- MSG_PAYLOAD: [\"%s...\"]", msg[:maxTrace])
+	} else {
+		c.Tracef("<<- MSG_PAYLOAD: [%q]", msg)
+	}
+}
+
+// Update the session (possibly remove it) of this disconnected client.
+func (s *Server) mqttHandleClosedClient(c *client) {
+	c.mu.Lock()
+	cp := c.mqtt.cp
+	accName := c.acc.Name
+	c.mu.Unlock()
+	if cp == nil {
+		return
+	}
+	sm := &s.mqtt.sessmgr
+	sm.mu.RLock()
+	asm, ok := sm.sessions[accName]
+	sm.mu.RUnlock()
+	if !ok {
+		return
+	}
+
+	asm.mu.Lock()
+	defer asm.mu.Unlock()
+	es, ok := asm.sessions[cp.clientID]
+	// If not found, ignore.
+	if !ok {
+		return
+	}
+	es.mu.Lock()
+	defer es.mu.Unlock()
+	// If this client is not the currently registered client, ignore.
+	if es.c != c {
+		return
+	}
+	// It the session was created with "clean session" flag, we cleanup
+	// when the client disconnects.
+	if es.clean {
+		es.clear()
+		delete(asm.sessions, cp.clientID)
+	} else {
+		// Clear the client from the session, but session stays.
+		es.c = nil
+	}
+}
+
+// Updates the MaxAckPending for all MQTT sessions, updating the
+// JetStream consumers and updating their max ack pending and forcing
+// a expiration of pending messages.
+func (s *Server) mqttUpdateMaxAckPending(newmaxp uint16) {
+	msm := &s.mqtt.sessmgr
+	s.accounts.Range(func(k, _ interface{}) bool {
+		accName := k.(string)
+		msm.mu.RLock()
+		asm := msm.sessions[accName]
+		msm.mu.RUnlock()
+		if asm == nil {
+			// Move to next account
+			return true
+		}
+		asm.mu.RLock()
+		for _, sess := range asm.sessions {
+			sess.mu.Lock()
+			sess.maxp = newmaxp
+			// Do not check for sess.stalled here because due to original
+			// consumer's maxp it is possible that the consumer was stalled
+			// and MQTT code did not have to stall the session.
+			for _, cons := range sess.cons {
+				cons.mu.Lock()
+				cons.maxp = int(newmaxp)
+				cons.forceExpirePending()
+				cons.mu.Unlock()
+			}
+			sess.mu.Unlock()
+		}
+		asm.mu.RUnlock()
+		return true
+	})
+}
+
+//////////////////////////////////////////////////////////////////////////////
+//
+// Sessions Manager related functions
+//
+//////////////////////////////////////////////////////////////////////////////
+
+// Returns the MQTT sessions manager for a given account.
+// If new, creates the required JetStream streams/consumers
+// for handling of sessions and messages.
+func (sm *mqttSessionManager) getOrCreateAccountSessionManager(clientID string, c *client) (*mqttAccountSessionManager, error) {
+	c.mu.Lock()
+	acc := c.acc
+	accName := acc.GetName()
+	c.mu.Unlock()
+
+	sm.mu.RLock()
+	asm, ok := sm.sessions[accName]
+	sm.mu.RUnlock()
+
+	if ok {
+		return asm, nil
+	}
+
+	// Not found, now take the write lock and check again
+	sm.mu.Lock()
+	defer sm.mu.Unlock()
+	asm, ok = sm.sessions[accName]
+	if ok {
+		return asm, nil
+	}
+	// First check that we have JS enabled for this account.
+	// TODO: Since we check only when creating a session manager for this
+	// account, would probably need to do some cleanup if JS can be disabled
+	// on config reload.
+	if !acc.JetStreamEnabled() {
+		return nil, fmt.Errorf("JetStream must be enabled for account %q used by MQTT client ID %q",
+			accName, clientID)
+	}
+	// Need to create one here.
+	asm = &mqttAccountSessionManager{sessions: make(map[string]*mqttSession)}
+	if err := asm.init(acc, c); err != nil {
+		return nil, err
+	}
+	sm.sessions[accName] = asm
+	return asm, nil
+}
+
+//////////////////////////////////////////////////////////////////////////////
+//
+// Account Sessions Manager related functions
+//
+//////////////////////////////////////////////////////////////////////////////
+
+// Creates JS streams/consumers for handling of sessions and messages for this
+// account. Note that lookup are performed in case we are in a restart
+// situation and the account is loaded for the first time but had state on disk.
+//
+// Global session manager lock is held on entry.
+func (as *mqttAccountSessionManager) init(acc *Account, c *client) error {
+	opts := c.srv.getOpts()
+	var err error
+	// Start with sessions stream
+	as.sstream, err = acc.LookupStream(mqttSessionsStreamName)
+	if err != nil {
+		as.sstream, err = acc.AddStream(&StreamConfig{
+			Subjects:       []string{},
+			Name:           mqttSessionsStreamName,
+			Storage:        FileStorage,
+			Retention:      InterestPolicy,
+			internal:       true,
+			allowNoSubject: true,
+		})
+		if err != nil {
+			return fmt.Errorf("unable to create sessions stream for MQTT account %q: %v", acc.GetName(), err)
+		}
+	}
+	// Create the stream for the messages.
+	as.mstream, err = acc.LookupStream(mqttStreamName)
+	if err != nil {
+		as.mstream, err = acc.AddStream(&StreamConfig{
+			Subjects:       []string{},
+			Name:           mqttStreamName,
+			Storage:        FileStorage,
+			Retention:      InterestPolicy,
+			internal:       true,
+			allowNoSubject: true,
+		})
+		if err != nil {
+			return fmt.Errorf("unable to create messages stream for MQTT account %q: %v", acc.GetName(), err)
+		}
+	}
+	// Create the stream for retained messages.
+	as.rstream, err = acc.LookupStream(mqttRetainedMsgsStreamName)
+	if err != nil {
+		as.rstream, err = acc.AddStream(&StreamConfig{
+			Subjects:       []string{},
+			Name:           mqttRetainedMsgsStreamName,
+			Storage:        FileStorage,
+			Retention:      InterestPolicy,
+			internal:       true,
+			allowNoSubject: true,
+		})
+		if err != nil {
+			return fmt.Errorf("unable to create retained messages stream for MQTT account %q: %v", acc.GetName(), err)
+		}
+	}
+	// Now recover all sessions (in case it did already exist)
+	if state := as.sstream.State(); state.Msgs > 0 {
+		for seq := state.FirstSeq; seq <= state.LastSeq; seq++ {
+			_, _, content, _, err := as.sstream.store.LoadMsg(seq)
+			if err != nil {
+				if err != errDeletedMsg {
+					c.Errorf("Error loading session record at sequence %v: %v", seq, err)
+				}
+				continue
+			}
+			ps := &mqttPersistedSession{}
+			if err := json.Unmarshal(content, ps); err != nil {
+				c.Errorf("Error unmarshaling session record at sequence %v: %v", seq, err)
+				continue
+			}
+			if as.sessions == nil {
+				as.sessions = make(map[string]*mqttSession)
+			}
+			es, ok := as.sessions[ps.ID]
+			if ok && es.sseq != 0 {
+				as.sstream.DeleteMsg(es.sseq)
+			} else if !ok {
+				es = mqttSessionCreate(opts)
+				es.stream = as.sstream
+				as.sessions[ps.ID] = es
+			}
+			es.sseq = seq
+			es.clean = ps.Clean
+			es.subs = ps.Subs
+			if l := len(ps.Cons); l > 0 {
+				if es.cons == nil {
+					es.cons = make(map[string]*Consumer, l)
+				}
+				for sid, name := range ps.Cons {
+					if cons := as.mstream.LookupConsumer(name); cons != nil {
+						es.cons[sid] = cons
+					}
+				}
+			}
+		}
+	}
+	// Finally, recover retained messages.
+	if state := as.rstream.State(); state.Msgs > 0 {
+		for seq := state.FirstSeq; seq <= state.LastSeq; seq++ {
+			subject, _, content, _, err := as.rstream.store.LoadMsg(seq)
+			if err != nil {
+				if err != errDeletedMsg {
+					c.Errorf("Error loading retained message at sequence %v: %v", seq, err)
+				}
+				continue
+			}
+			rm := &mqttRetainedMsg{}
+			if err := json.Unmarshal(content, &rm); err != nil {
+				c.Errorf("Error unmarshaling retained message on subject %q, sequence %v: %v", subject, seq, err)
+				continue
+			}
+			rm = as.handleRetainedMsg(subject, rm)
+			rm.sseq = seq
+		}
+	}
+	return nil
+}
+
+// Add/Replace this message from the retained messages map.
+// Returns the retained message actually stored in the map, which means that
+// it may be different from the given `rm`.
+//
+// Account session manager lock held on entry.
+func (as *mqttAccountSessionManager) handleRetainedMsg(key string, rm *mqttRetainedMsg) *mqttRetainedMsg {
+	if as.retmsgs == nil {
+		as.retmsgs = make(map[string]*mqttRetainedMsg)
+		as.sl = NewSublistWithCache()
+	} else {
+		// Check if we already had one. If so, update the existing one.
+		if erm, exists := as.retmsgs[key]; exists {
+			erm.Msg = rm.Msg
+			erm.Flags = rm.Flags
+			erm.Source = rm.Source
+			return erm
+		}
+	}
+	rm.sub = &subscription{subject: []byte(key)}
+	as.retmsgs[key] = rm
+	as.sl.Insert(rm.sub)
+	return rm
+}
+
+// Process subscriptions for the given session/client.
+//
+// When `fromSubProto` is false, it means that this is invoked from the CONNECT
+// protocol, when restoring subscriptions that were saved for this session.
+// In that case, there is no need to update the session record.
+//
+// When `fromSubProto` is true, it means that this call is invoked from the
+// processing of the SUBSCRIBE protocol, which means that the session needs to
+// be updated. It also means that if a subscription on same subject with same
+// QoS already exist, we should not be recreating the subscription/JS durable,
+// since it was already done when processing the CONNECT protocol.
+//
+// Account session manager lock held on entry.
+// Session lock held when `fromSubProto` is false.
+func (as *mqttAccountSessionManager) processSubs(sess *mqttSession, clientID string, c *client,
+	filters []*mqttFilter, fromSubProto, trace bool) ([]*subscription, error) {
+
+	if fromSubProto {
+		sess.mu.Lock()
+		defer sess.mu.Unlock()
+		if sess.c != c {
+			return nil, fmt.Errorf("client %q no longer registered with MQTT session", clientID)
+		}
+	}
+
+	addJSConsToSess := func(sid string, cons *Consumer) {
+		if cons == nil {
+			return
+		}
+		if sess.cons == nil {
+			sess.cons = make(map[string]*Consumer)
+		}
+		sess.cons[sid] = cons
+	}
+
+	setupSub := func(sub *subscription, qos byte) {
+		if sub.mqtt == nil {
+			sub.mqtt = &mqttSub{}
+		}
+		sub.mqtt.qos = qos
+		if fromSubProto {
+			as.serializeRetainedMsgsForSub(sess, c, sub, trace)
+		}
+	}
+
+	subs := make([]*subscription, 0, len(filters))
+	for _, f := range filters {
+		if f.qos > 1 {
+			f.qos = 1
+		}
+		subject := f.filter
+		sid := subject
+
+		if strings.HasPrefix(subject, mqttSubPrefix) {
+			f.qos = mqttSubAckFailure
+			continue
+		}
+
+		var jscons *Consumer
+		var jssub *subscription
+
+		// Note that if a subscription already exists on this subject,
+		// the existing sub is returned. Need to update the qos.
+		sub, err := c.processSub([]byte(subject), nil, []byte(sid), mqttDeliverMsgCb, false)
+		if err == nil {
+			setupSub(sub, f.qos)
+			// This will create (if not already exist) a JS consumer for subscriptions
+			// of QoS >= 1. But if a JS consumer already exists and the subscription
+			// for same subject is now a QoS==0, then the JS consumer will be deleted.
+			jscons, jssub, err = c.mqttProcessJSConsumer(sess, as.mstream,
+				subject, sid, f.qos, fromSubProto)
+		}
+		if err != nil {
+			c.Errorf("error subscribing to %q: err=%v", subject, err)
+			f.qos = mqttSubAckFailure
+			c.mqttCleanupFailedSub(sub, jscons, jssub)
+			continue
+		}
+		if mqttNeedSubForLevelUp(subject) {
+			var fwjscons *Consumer
+			var fwjssub *subscription
+			var fwcsub *subscription
+
+			// Say subject is "foo.>", remove the ".>" so that it becomes "foo"
+			fwcsubject := subject[:len(subject)-2]
+			// Change the sid to "foo fwc"
+			fwcsid := fwcsubject + mqttMultiLevelSidSuffix
+			// See note above about existing subscription.
+			fwcsub, err = c.processSub([]byte(fwcsubject), nil, []byte(fwcsid), mqttDeliverMsgCb, false)
+			if err == nil {
+				setupSub(fwcsub, f.qos)
+				fwjscons, fwjssub, err = c.mqttProcessJSConsumer(sess, as.mstream,
+					fwcsubject, fwcsid, f.qos, fromSubProto)
+			}
+			if err != nil {
+				c.Errorf("error subscribing to %q: err=%v", fwcsubject, err)
+				f.qos = mqttSubAckFailure
+				c.mqttCleanupFailedSub(sub, jscons, jssub)
+				c.mqttCleanupFailedSub(fwcsub, fwjscons, fwjssub)
+				continue
+			}
+			subs = append(subs, fwcsub)
+			addJSConsToSess(fwcsid, fwjscons)
+		}
+		subs = append(subs, sub)
+		addJSConsToSess(sid, jscons)
+	}
+	var err error
+	if fromSubProto {
+		err = sess.update(clientID, filters, true)
+	}
+	return subs, err
+}
+
+// Retained publish messages matching this subscription are serialized in the
+// subscription's `prm` mqtt writer. This buffer will be queued for outbound
+// after the subscription is processed and SUBACK is sent or possibly when
+// server processes an incoming published message matching the newly
+// registered subscription.
+//
+// Account session manager lock held on entry.
+// Session lock held on entry
+func (as *mqttAccountSessionManager) serializeRetainedMsgsForSub(sess *mqttSession, c *client, sub *subscription, trace bool) {
+	if len(as.retmsgs) > 0 {
+		var rmsa [64]*mqttRetainedMsg
+		rms := rmsa[:0]
+
+		as.getRetainedPublishMsgs(string(sub.subject), &rms)
+		for _, rm := range rms {
+			if sub.mqtt.prm == nil {
+				sub.mqtt.prm = &mqttWriter{}
+			}
+			prm := sub.mqtt.prm
+			pi := sess.getPubAckIdentifier(mqttGetQoS(rm.Flags), sub)
+			// Need to use the subject for the retained message, not the `sub` subject.
+			// We can find the published retained message in rm.sub.subject.
+			flags := mqttSerializePublishMsg(prm, pi, false, true, rm.Topic, rm.Msg)
+			if trace {
+				pp := mqttPublish{
+					topic: rm.Topic,
+					flags: flags,
+					pi:    pi,
+					sz:    len(rm.Msg),
+				}
+				c.traceOutOp("PUBLISH", []byte(mqttPubTrace(&pp)))
+			}
+		}
+	}
+}
+
+// Returns in the provided slice all publish retained message records that
+// match the given subscription's `subject` (which could have wildcards).
+//
+// Account session manager lock held on entry.
+func (as *mqttAccountSessionManager) getRetainedPublishMsgs(subject string, rms *[]*mqttRetainedMsg) {
+	result := as.sl.ReverseMatch(subject)
+	if len(result.psubs) == 0 {
+		return
+	}
+	for _, sub := range result.psubs {
+		// Since this is a reverse match, the subscription objects here
+		// contain literals corresponding to the published subjects.
+		if rm, ok := as.retmsgs[string(sub.subject)]; ok {
+			*rms = append(*rms, rm)
+		}
+	}
+}
+
+//////////////////////////////////////////////////////////////////////////////
+//
+// MQTT session related functions
+//
+//////////////////////////////////////////////////////////////////////////////
+
+// Returns a new mqttSession object with max ack pending set based on
+// option or use mqttDefaultMaxAckPending if no option set.
+func mqttSessionCreate(opts *Options) *mqttSession {
+	maxp := opts.MQTT.MaxAckPending
+	if maxp == 0 {
+		maxp = mqttDefaultMaxAckPending
+	}
+	return &mqttSession{maxp: maxp}
+}
+
+// Persists a session. Note that if the session's current client does not match
+// the given client, nothing is done.
+//
+// Lock held on entry.
+func (sess *mqttSession) save(clientID string) error {
+	ps := mqttPersistedSession{
+		ID:    clientID,
+		Clean: sess.clean,
+		Subs:  sess.subs,
+	}
+	if l := len(sess.cons); l > 0 {
+		cons := make(map[string]string, l)
+		for sid, jscons := range sess.cons {
+			cons[sid] = jscons.Name()
+		}
+		ps.Cons = cons
+	}
+	sessBytes, _ := json.Marshal(&ps)
+	newSeq, _, err := sess.stream.store.StoreMsg("sessions", nil, sessBytes)
+	if err != nil {
+		return err
+	}
+	if sess.sseq != 0 {
+		sess.stream.DeleteMsg(sess.sseq)
+	}
+	sess.sseq = newSeq
+	return nil
+}
+
+// Delete JS consumers for this session and delete the persisted session from
+// the stream.
+//
+// Lock held on entry.
+func (sess *mqttSession) clear() {
+	for consName, cons := range sess.cons {
+		delete(sess.cons, consName)
+		cons.Delete()
+	}
+	if sess.stream != nil && sess.sseq != 0 {
+		sess.stream.DeleteMsg(sess.sseq)
+		sess.sseq = 0
+	}
+	sess.subs, sess.pending, sess.cpending = nil, nil, nil
+}
+
+// This will update the session record for this client in the account's MQTT
+// sessions stream if the session had any change in the subscriptions.
+//
+// Lock held on entry.
+func (sess *mqttSession) update(clientID string, filters []*mqttFilter, add bool) error {
+	// Evaluate if we need to persist anything.
+	var needUpdate bool
+	for _, f := range filters {
+		if add {
+			if f.qos == mqttSubAckFailure {
+				continue
+			}
+			if qos, ok := sess.subs[f.filter]; !ok || qos != f.qos {
+				if sess.subs == nil {
+					sess.subs = make(map[string]byte)
+				}
+				sess.subs[f.filter] = f.qos
+				needUpdate = true
+			}
+		} else {
+			if _, ok := sess.subs[f.filter]; ok {
+				delete(sess.subs, f.filter)
+				needUpdate = true
+			}
+		}
+	}
+	var err error
+	if needUpdate {
+		err = sess.save(clientID)
+	}
+	return err
+}
+
+// If both pQos and sub.mqtt.qos are > 0, then this will return the next
+// packet identifier to use for a published message.
+//
+// Lock held on entry
+func (sess *mqttSession) getPubAckIdentifier(pQos byte, sub *subscription) uint16 {
+	pi, _ := sess.trackPending(pQos, _EMPTY_, sub)
+	return pi
+}
+
+// If publish message QoS (pQos) and the subscription's QoS are both at least 1,
+// this function will assign a packet identifier (pi) and will keep track of
+// the pending ack. If the message has already been redelivered (reply != ""),
+// the returned boolean will be `true`.
+//
+// Lock held on entry
+func (sess *mqttSession) trackPending(pQos byte, reply string, sub *subscription) (uint16, bool) {
+	if pQos == 0 || sub.mqtt.qos == 0 {
+		return 0, false
+	}
+	var dup bool
+	var pi uint16
+
+	bumpPI := func() uint16 {
+		var avail bool
+		next := sess.ppi
+		for i := 0; i < 0xFFFF; i++ {
+			next++
+			if next == 0 {
+				next = 1
+			}
+			if _, used := sess.pending[next]; !used {
+				sess.ppi = next
+				avail = true
+				break
+			}
+		}
+		if !avail {
+			return 0
+		}
+		return sess.ppi
+	}
+
+	if reply == _EMPTY_ || sub.mqtt.jsCons == nil {
+		return bumpPI(), false
+	}
+
+	// Here, we have an ACK subject and a JS consumer...
+	jsCons := sub.mqtt.jsCons
+	if sess.pending == nil {
+		sess.pending = make(map[uint16]*mqttPending)
+		sess.cpending = make(map[*Consumer]map[uint64]uint16)
+	}
+	// Get the stream sequence and other from the ack reply subject
+	sseq, dseq, dcount := ackReplyInfo(reply)
+
+	var pending *mqttPending
+	// For this JS consumer, check to see if we already have sseq->pi
+	sseqToPi, ok := sess.cpending[jsCons]
+	if !ok {
+		sseqToPi = make(map[uint64]uint16)
+		sess.cpending[jsCons] = sseqToPi
+	} else if pi, ok = sseqToPi[sseq]; ok {
+		// If we already have a pi, get the ack so we update it.
+		// We will reuse the save packet identifier (pi).
+		pending = sess.pending[pi]
+	}
+	if pi == 0 {
+		// sess.maxp will always have a value > 0.
+		if len(sess.pending) >= int(sess.maxp) {
+			// Indicate that we did not assign a packet identifier.
+			// The caller will not send the message to the subscription
+			// and JS will redeliver later, based on consumer's AckWait.
+			sess.stalled = true
+			return 0, false
+		}
+		pi = bumpPI()
+		sseqToPi[sseq] = pi
+	}
+	if pending == nil {
+		pending = &mqttPending{jsCons: jsCons, sseq: sseq}
+		sess.pending[pi] = pending
+	}
+	// Update pending with consumer delivery sequence and count
+	pending.dseq, pending.dcount = dseq, dcount
+	// If redelivery, return DUP flag
+	if dcount > 1 {
+		dup = true
+	}
+	return pi, dup
+}
+
+//////////////////////////////////////////////////////////////////////////////
+//
+// CONNECT protocol related functions
+//
+//////////////////////////////////////////////////////////////////////////////
+
+// Parse the MQTT connect protocol
+func (c *client) mqttParseConnect(r *mqttReader, pl int) (byte, *mqttConnectProto, error) {
+
+	// Make sure that we have the expected length in the buffer,
+	// and if not, this will read it from the underlying reader.
+	if err := r.ensurePacketInBuffer(pl); err != nil {
+		return 0, nil, err
+	}
+
+	// Protocol name
+	proto, err := r.readBytes("protocol name", false)
+	if err != nil {
+		return 0, nil, err
+	}
+
+	// Spec [MQTT-3.1.2-1]
+	if !bytes.Equal(proto, mqttProtoName) {
+		// Check proto name against v3.1 to report better error
+		if bytes.Equal(proto, mqttOldProtoName) {
+			return 0, nil, fmt.Errorf("older protocol %q not supported", proto)
+		}
+		return 0, nil, fmt.Errorf("expected connect packet with protocol name %q, got %q", mqttProtoName, proto)
+	}
+
+	// Protocol level
+	level, err := r.readByte("protocol level")
+	if err != nil {
+		return 0, nil, err
+	}
+	// Spec [MQTT-3.1.2-2]
+	if level != mqttProtoLevel {
+		return mqttConnAckRCUnacceptableProtocolVersion, nil, fmt.Errorf("unacceptable protocol version of %v", level)
+	}
+
+	cp := &mqttConnectProto{}
+	// Connect flags
+	cp.flags, err = r.readByte("flags")
+	if err != nil {
+		return 0, nil, err
+	}
+
+	// Spec [MQTT-3.1.2-3]
+	if cp.flags&mqttConnFlagReserved != 0 {
+		return 0, nil, fmt.Errorf("connect flags reserved bit not set to 0")
+	}
+
+	var hasWill bool
+	wqos := (cp.flags & mqttConnFlagWillQoS) >> 3
+	wretain := cp.flags&mqttConnFlagWillRetain != 0
+	// Spec [MQTT-3.1.2-11]
+	if cp.flags&mqttConnFlagWillFlag == 0 {
+		// Spec [MQTT-3.1.2-13]
+		if wqos != 0 {
+			return 0, nil, fmt.Errorf("if Will flag is set to 0, Will QoS must be 0 too, got %v", wqos)
+		}
+		// Spec [MQTT-3.1.2-15]
+		if wretain {
+			return 0, nil, fmt.Errorf("if Will flag is set to 0, Will Retain flag must be 0 too")
+		}
+	} else {
+		// Spec [MQTT-3.1.2-14]
+		if wqos == 3 {
+			return 0, nil, fmt.Errorf("if Will flag is set to 1, Will QoS can be 0, 1 or 2, got %v", wqos)
+		}
+		hasWill = true
+	}
+
+	// Spec [MQTT-3.1.2-19]
+	hasUser := cp.flags&mqttConnFlagUsernameFlag != 0
+	// Spec [MQTT-3.1.2-21]
+	hasPassword := cp.flags&mqttConnFlagPasswordFlag != 0
+	// Spec [MQTT-3.1.2-22]
+	if !hasUser && hasPassword {
+		return 0, nil, fmt.Errorf("password flag set but username flag is not")
+	}
+
+	// Keep alive
+	var ka uint16
+	ka, err = r.readUint16("keep alive")
+	if err != nil {
+		return 0, nil, err
+	}
+	// Spec [MQTT-3.1.2-24]
+	if ka > 0 {
+		cp.rd = time.Duration(float64(ka)*1.5) * time.Second
+	}
+
+	// Payload starts here and order is mandated by:
+	// Spec [MQTT-3.1.3-1]: client ID, will topic, will message, username, password
+
+	// Client ID
+	cp.clientID, err = r.readString("client ID")
+	if err != nil {
+		return 0, nil, err
+	}
+	// Spec [MQTT-3.1.3-7]
+	if cp.clientID == _EMPTY_ {
+		if cp.flags&mqttConnFlagCleanSession == 0 {
+			return mqttConnAckRCIdentifierRejected, nil, fmt.Errorf("when client ID is empty, clean session flag must be set to 1")
+		}
+		// Spec [MQTT-3.1.3-6]
+		cp.clientID = nuid.Next()
+	}
+	// Spec [MQTT-3.1.3-4] and [MQTT-3.1.3-9]
+	if !utf8.ValidString(cp.clientID) {
+		return mqttConnAckRCIdentifierRejected, nil, fmt.Errorf("invalid utf8 for client ID: %q", cp.clientID)
+	}
+
+	if hasWill {
+		cp.will = &mqttWill{
+			qos:    wqos,
+			retain: wretain,
+		}
+		var topic []byte
+		// Need to make a copy since we need to hold to this topic after the
+		// parsing of this protocol.
+		topic, err = r.readBytes("Will topic", true)
+		if err != nil {
+			return 0, nil, err
+		}
+		if len(topic) == 0 {
+			return 0, nil, fmt.Errorf("empty Will topic not allowed")
+		}
+		if !utf8.Valid(topic) {
+			return 0, nil, fmt.Errorf("invalide utf8 for Will topic %q", topic)
+		}
+		cp.will.topic = topic
+		// Convert MQTT topic to NATS subject
+		if cp.will.subject, err = mqttTopicToNATSPubSubject(topic); err != nil {
+			return 0, nil, err
+		}
+		// Now "will" message.
+		// Ask for a copy since we need to hold to this after parsing of this protocol.
+		cp.will.message, err = r.readBytes("Will message", true)
+		if err != nil {
+			return 0, nil, err
+		}
+	}
+
+	if hasUser {
+		c.opts.Username, err = r.readString("user name")
+		if err != nil {
+			return 0, nil, err
+		}
+		if c.opts.Username == _EMPTY_ {
+			return mqttConnAckRCBadUserOrPassword, nil, fmt.Errorf("empty user name not allowed")
+		}
+		// Spec [MQTT-3.1.3-11]
+		if !utf8.ValidString(c.opts.Username) {
+			return mqttConnAckRCBadUserOrPassword, nil, fmt.Errorf("invalid utf8 for user name %q", c.opts.Username)
+		}
+	}
+
+	if hasPassword {
+		c.opts.Password, err = r.readString("password")
+		if err != nil {
+			return 0, nil, err
+		}
+		c.opts.Token = c.opts.Password
+	}
+	return 0, cp, nil
+}
+
+func (c *client) mqttConnectTrace(cp *mqttConnectProto) string {
+	trace := fmt.Sprintf("clientID=%s", cp.clientID)
+	if cp.rd > 0 {
+		trace += fmt.Sprintf(" keepAlive=%v", cp.rd)
+	}
+	if cp.will != nil {
+		trace += fmt.Sprintf(" will=(topic=%s QoS=%v retain=%v)",
+			cp.will.topic, cp.will.qos, cp.will.retain)
+	}
+	if c.opts.Username != _EMPTY_ {
+		trace += fmt.Sprintf(" username=%s", c.opts.Username)
+	}
+	if c.opts.Password != _EMPTY_ {
+		trace += " password=****"
+	}
+	return trace
+}
+
+func (s *Server) mqttProcessConnect(c *client, cp *mqttConnectProto, trace bool) error {
+	sendConnAck := func(rc byte, sessp bool) {
+		c.mqttEnqueueConnAck(rc, sessp)
+		if trace {
+			c.traceOutOp("CONNACK", []byte(fmt.Sprintf("sp=%v rc=%v", sessp, rc)))
+		}
+	}
+
+	c.mu.Lock()
+	c.clearAuthTimer()
+	c.mu.Unlock()
+	if !s.isClientAuthorized(c) {
+		sendConnAck(mqttConnAckRCNotAuthorized, false)
+		c.closeConnection(AuthenticationViolation)
+		return ErrAuthentication
+	}
+	// Now that we are are authenticated, we have the client bound to the account.
+	// Get the account's level MQTT sessions manager. If it does not exists yet,
+	// this will create it along with the streams where sessions and messages
+	// are stored.
+	sm := &s.mqtt.sessmgr
+	asm, err := sm.getOrCreateAccountSessionManager(cp.clientID, c)
+	if err != nil {
+		return err
+	}
+
+	// Rest of code runs under the account's sessions manager write lock.
+	asm.mu.Lock()
+	defer asm.mu.Unlock()
+
+	// Is the client requesting a clean session or not.
+	cleanSess := cp.flags&mqttConnFlagCleanSession != 0
+	// Session present? Assume false, will be set to true only when applicable.
+	sessp := false
+	// Do we have an existing session for this client ID
+	es, ok := asm.sessions[cp.clientID]
+	if !ok {
+		es = mqttSessionCreate(s.getOpts())
+	}
+	es.mu.Lock()
+	defer es.mu.Unlock()
+	if ok {
+		// Clear the session if client wants a clean session.
+		// Also, Spec [MQTT-3.2.2-1]: don't report session present
+		if cleanSess || es.clean {
+			// Spec [MQTT-3.1.2-6]: If CleanSession is set to 1, the Client and
+			// Server MUST discard any previous Session and start a new one.
+			// This Session lasts as long as the Network Connection. State data
+			// associated with this Session MUST NOT be reused in any subsequent
+			// Session.
+			es.clear()
+		} else {
+			// Report to the client that the session was present
+			sessp = true
+		}
+		// Spec [MQTT-3.1.4-2]. If the ClientId represents a Client already
+		// connected to the Server then the Server MUST disconnect the existing
+		// client.
+		if ec := es.c; ec != nil {
+			ec.mu.Lock()
+			// Remove will before closing
+			ec.mqtt.cp.will = nil
+			ec.mu.Unlock()
+			// Close old client in separate go routine
+			go ec.closeConnection(DuplicateClientID)
+		}
+		// Bind with the new client
+		es.c = c
+		es.clean = cleanSess
+	} else {
+		// Spec [MQTT-3.2.2-3]: if the Server does not have stored Session state,
+		// it MUST set Session Present to 0 in the CONNACK packet.
+		es.c, es.clean, es.stream = c, cleanSess, asm.sstream
+		es.save(cp.clientID)
+		// Now save this new session into the account sessions
+		asm.sessions[cp.clientID] = es
+	}
+	c.mu.Lock()
+	c.flags.set(connectReceived)
+	c.mqtt.cp = cp
+	c.mqtt.asm = asm
+	c.mqtt.sess = es
+	c.mu.Unlock()
+	//
+	// Spec [MQTT-3.2.0-1]: CONNACK must be the first protocol sent to the
+	// session. However, we are going to possibly restore the subscriptions
+	// first and then send the CONNACK. This will help tests that restore
+	// a MQTT connection with subs and immediately use NATS to publish.
+	// In that case, message would not be received because the pub could
+	// occur before the subscriptions are processed here. It would be
+	// easy to fix test with doing a PINGREQ/PINGRESP before doing NATS pub,
+	// but it seems better to ensure that everything is setup before sending
+	// back the CONNACK.
+	// Note that since we are under the session lock, the subs callback will
+	// have to wait to acquire the lock, so we are still guaranteed to enqueue
+	// the CONNACK before any message.
+	//
+	// Process possible saved subscriptions.
+	if l := len(es.subs); l > 0 {
+		filters := make([]*mqttFilter, 0, l)
+		for subject, qos := range es.subs {
+			filters = append(filters, &mqttFilter{filter: subject, qos: qos})
+		}
+		if _, err := asm.processSubs(es, cp.clientID, c, filters, false, trace); err != nil {
+			return err
+		}
+	}
+	// Now send the CONNACK
+	sendConnAck(mqttConnAckRCConnectionAccepted, sessp)
+	return nil
+}
+
+func (c *client) mqttEnqueueConnAck(rc byte, sessionPresent bool) {
+	proto := [4]byte{mqttPacketConnectAck, 2, 0, rc}
+	c.mu.Lock()
+	// Spec [MQTT-3.2.2-4]. If return code is different from 0, then
+	// session present flag must be set to 0.
+	if rc == 0 {
+		if sessionPresent {
+			proto[2] = 1
+		}
+	}
+	c.enqueueProto(proto[:])
+	c.mu.Unlock()
+}
+
+func (s *Server) mqttHandleWill(c *client) {
+	c.mu.Lock()
+	if c.mqtt.cp == nil {
+		c.mu.Unlock()
+		return
+	}
+	will := c.mqtt.cp.will
+	if will == nil {
+		c.mu.Unlock()
+		return
+	}
+	pp := c.mqtt.pp
+	pp.topic = will.topic
+	pp.subject = will.subject
+	pp.msg = will.message
+	pp.sz = len(will.message)
+	pp.pi = 0
+	pp.flags = will.qos << 1
+	if will.retain {
+		pp.flags |= mqttPubFlagRetain
+	}
+	c.mu.Unlock()
+	s.mqttProcessPub(c, pp)
+	c.flushClients(0)
+}
+
+//////////////////////////////////////////////////////////////////////////////
+//
+// PUBLISH protocol related functions
+//
+//////////////////////////////////////////////////////////////////////////////
+
+func (c *client) mqttParsePub(r *mqttReader, pl int, pp *mqttPublish) error {
+	qos := mqttGetQoS(pp.flags)
+	if qos > 1 {
+		return fmt.Errorf("publish QoS=%v not supported", qos)
+	}
+	if err := r.ensurePacketInBuffer(pl); err != nil {
+		return err
+	}
+	// Keep track of where we are when starting to read the variable header
+	start := r.pos
+
+	var err error
+	pp.topic, err = r.readBytes("topic", false)
+	if err != nil {
+		return err
+	}
+	if len(pp.topic) == 0 {
+		return fmt.Errorf("topic cannot be empty")
+	}
+	// Convert the topic to a NATS subject. This call will also check that
+	// there is no MQTT wildcards (Spec [MQTT-3.3.2-2] and [MQTT-4.7.1-1])
+	// Note that this may not result in a copy if there is no conversion.
+	// It is good because after the message is processed we won't have a
+	// reference to the buffer and we save a copy.
+	pp.subject, err = mqttTopicToNATSPubSubject(pp.topic)
+	if err != nil {
+		return err
+	}
+
+	if qos > 0 {
+		pp.pi, err = r.readUint16("packet identifier")
+		if err != nil {
+			return err
+		}
+		if pp.pi == 0 {
+			return fmt.Errorf("with QoS=%v, packet identifier cannot be 0", qos)
+		}
+	} else {
+		pp.pi = 0
+	}
+
+	// The message payload will be the total packet length minus
+	// what we have consumed for the variable header
+	pp.sz = pl - (r.pos - start)
+	if pp.sz > 0 {
+		start = r.pos
+		r.pos += pp.sz
+		pp.msg = r.buf[start:r.pos]
+	} else {
+		pp.msg = nil
+	}
+	return nil
+}
+
+func mqttPubTrace(pp *mqttPublish) string {
+	dup := pp.flags&mqttPubFlagDup != 0
+	qos := mqttGetQoS(pp.flags)
+	retain := mqttIsRetained(pp.flags)
+	var piStr string
+	if pp.pi > 0 {
+		piStr = fmt.Sprintf(" pi=%v", pp.pi)
+	}
+	return fmt.Sprintf("%s dup=%v QoS=%v retain=%v size=%v%s",
+		pp.topic, dup, qos, retain, pp.sz, piStr)
+}
+
+func (s *Server) mqttProcessPub(c *client, pp *mqttPublish) {
+	c.pa.subject, c.pa.hdr, c.pa.size = pp.subject, -1, pp.sz
+
+	// Convert size into bytes.
+	i := len(pp.szb)
+	if pp.sz > 0 {
+		for l := pp.sz; l > 0; l /= 10 {
+			i--
+			pp.szb[i] = digits[l%10]
+		}
+	} else {
+		i--
+		pp.szb[i] = digits[0]
+	}
+	c.pa.szb = pp.szb[i:]
+
+	// This will work for QoS 0 but mqtt msg delivery callback will ignore
+	// delivery for QoS > 0 published messages (since it is handled specifically
+	// with call to directProcessInboundJetStreamMsg).
+	// However, this needs to be invoked before directProcessInboundJetStreamMsg()
+	// in case we are dealing with publish retained messages.
+	c.processInboundClientMsg(pp.msg)
+	if mqttGetQoS(pp.flags) > 0 {
+		// Since this is the fast path, we access the messages stream directly here
+		// without locking. All the fields mqtt.asm.mstream are immutable.
+		c.mqtt.asm.mstream.processInboundJetStreamMsg(nil, c, string(c.pa.subject), _EMPTY_, pp.msg)
+	}
+	c.pa.subject, c.pa.hdr, c.pa.size, c.pa.szb = nil, -1, 0, nil
+}
+
+// Invoked when processing an inbound client message. If the "retain" flag is
+// set, the message is stored so it can be later resent to (re)starting
+// subscriptions that match the subject.
+//
+// Invoked from the publisher's readLoop. No client lock is held on entry.
+func (c *client) mqttHandlePubRetain() {
+	pp := c.mqtt.pp
+	if mqttIsRetained(pp.flags) {
+		key := string(pp.subject)
+		asm := c.mqtt.asm
+		asm.mu.Lock()
+		// Spec [MQTT-3.3.1-11]. Payload of size 0 removes the retained message,
+		// but should still be delivered as a normal message.
+		if pp.sz == 0 {
+			if asm.retmsgs != nil {
+				if erm, ok := asm.retmsgs[key]; ok {
+					delete(asm.retmsgs, key)
+					asm.sl.Remove(erm.sub)
+					if erm.sseq != 0 {
+						asm.rstream.DeleteMsg(erm.sseq)
+					}
+				}
+			}
+		} else {
+			// Spec [MQTT-3.3.1-5]. Store the retained message with its QoS.
+			// When coming from a publish protocol, `pp` is referencing a stack
+			// variable that itself possibly references the read buffer.
+			rm := &mqttRetainedMsg{
+				Topic:  copyBytes(pp.topic),
+				Msg:    copyBytes(pp.msg),
+				Flags:  pp.flags,
+				Source: c.opts.Username,
+			}
+			rm = asm.handleRetainedMsg(key, rm)
+			rmBytes, _ := json.Marshal(rm)
+			// TODO: For now we will report the error but continue...
+			seq, _, err := asm.rstream.store.StoreMsg(key, nil, rmBytes)
+			if err != nil {
+				c.mu.Lock()
+				acc := c.acc
+				c.mu.Unlock()
+				c.Errorf("unable to store retained message for account %q, subject %q: %v",
+					acc.GetName(), key, err)
+			}
+			// If it has been replaced, rm.sseq will be != 0
+			if rm.sseq != 0 {
+				asm.rstream.DeleteMsg(rm.sseq)
+			}
+			// Keep track of current stream sequence (possibly 0 if failed to store)
+			rm.sseq = seq
+		}
+
+		asm.mu.Unlock()
+
+		// Clear the retain flag for a normal published message.
+		pp.flags &= ^mqttPubFlagRetain
+	}
+}
+
+// After a config reload, it is possible that the source of a publish retained
+// message is no longer allowed to publish on the given topic. If that is the
+// case, the retained message is removed from the map and will no longer be
+// sent to (re)starting subscriptions.
+//
+// Server lock is held on entry
+func (s *Server) mqttCheckPubRetainedPerms() {
+	sm := &s.mqtt.sessmgr
+	sm.mu.RLock()
+	defer sm.mu.RUnlock()
+
+	for _, asm := range sm.sessions {
+		perms := map[string]*perm{}
+		asm.mu.Lock()
+		for subject, rm := range asm.retmsgs {
+			if rm.Source == _EMPTY_ {
+				continue
+			}
+			// Lookup source from global users.
+			u := s.users[rm.Source]
+			if u != nil {
+				p, ok := perms[rm.Source]
+				if !ok {
+					p = generatePubPerms(u.Permissions)
+					perms[rm.Source] = p
+				}
+				// If there is permission and no longer allowed to publish in
+				// the subject, remove the publish retained message from the map.
+				if p != nil && !pubAllowed(p, subject) {
+					u = nil
+				}
+			}
+
+			// Not present or permissions have changed such that the source can't
+			// publish on that subject anymore: remove it from the map.
+			if u == nil {
+				delete(asm.retmsgs, subject)
+				asm.rstream.DeleteMsg(rm.sseq)
+				asm.sl.Remove(rm.sub)
+			}
+		}
+		asm.mu.Unlock()
+	}
+}
+
+// Helper to generate only pub permissions from a Permissions object
+func generatePubPerms(perms *Permissions) *perm {
+	var p *perm
+	if perms.Publish.Allow != nil {
+		p = &perm{}
+		p.allow = NewSublistWithCache()
+	}
+	for _, pubSubject := range perms.Publish.Allow {
+		sub := &subscription{subject: []byte(pubSubject)}
+		p.allow.Insert(sub)
+	}
+	if len(perms.Publish.Deny) > 0 {
+		if p == nil {
+			p = &perm{}
+		}
+		p.deny = NewSublistWithCache()
+	}
+	for _, pubSubject := range perms.Publish.Deny {
+		sub := &subscription{subject: []byte(pubSubject)}
+		p.deny.Insert(sub)
+	}
+	return p
+}
+
+// Helper that checks if given `perms` allow to publish on the given `subject`
+func pubAllowed(perms *perm, subject string) bool {
+	allowed := true
+	if perms.allow != nil {
+		r := perms.allow.Match(subject)
+		allowed = len(r.psubs) != 0
+	}
+	// If we have a deny list and are currently allowed, check that as well.
+	if allowed && perms.deny != nil {
+		r := perms.deny.Match(subject)
+		allowed = len(r.psubs) == 0
+	}
+	return allowed
+}
+
+func mqttWritePublish(w *mqttWriter, qos byte, dup, retain bool, subject string, pi uint16, payload []byte) {
+	flags := qos << 1
+	if dup {
+		flags |= mqttPubFlagDup
+	}
+	if retain {
+		flags |= mqttPubFlagRetain
+	}
+	w.WriteByte(mqttPacketPub | flags)
+	pkLen := 2 + len(subject) + len(payload)
+	if qos > 0 {
+		pkLen += 2
+	}
+	w.WriteVarInt(pkLen)
+	w.WriteString(subject)
+	if qos > 0 {
+		w.WriteUint16(pi)
+	}
+	w.Write([]byte(payload))
+}
+
+func (c *client) mqttEnqueuePubAck(pi uint16) {
+	proto := [4]byte{mqttPacketPubAck, 0x2, 0, 0}
+	proto[2] = byte(pi >> 8)
+	proto[3] = byte(pi)
+	c.mu.Lock()
+	c.enqueueProto(proto[:4])
+	c.mu.Unlock()
+}
+
+func mqttParsePubAck(r *mqttReader, pl int) (uint16, error) {
+	if err := r.ensurePacketInBuffer(pl); err != nil {
+		return 0, err
+	}
+	pi, err := r.readUint16("packet identifier")
+	if err != nil {
+		return 0, err
+	}
+	if pi == 0 {
+		return 0, fmt.Errorf("packet identifier cannot be 0")
+	}
+	return pi, nil
+}
+
+func (c *client) mqttProcessPubAck(pi uint16) {
+	sess := c.mqtt.sess
+	if sess == nil {
+		return
+	}
+	sess.mu.Lock()
+	defer sess.mu.Unlock()
+	if sess.c != c {
+		return
+	}
+	if ack, ok := sess.pending[pi]; ok {
+		delete(sess.pending, pi)
+		jsCons := ack.jsCons
+		if sseqToPi, ok := sess.cpending[jsCons]; ok {
+			delete(sseqToPi, ack.sseq)
+		}
+		jsCons.ackMsg(ack.sseq, ack.dseq, ack.dcount)
+		if len(sess.pending) == 0 {
+			sess.ppi = 0
+		}
+		if sess.stalled && len(sess.pending) < int(sess.maxp) {
+			sess.stalled = false
+			for _, cons := range sess.cons {
+				cons.mu.Lock()
+				cons.forceExpirePending()
+				cons.mu.Unlock()
+			}
+		}
+	}
+}
+
+// Return the QoS from the given PUBLISH protocol's flags
+func mqttGetQoS(flags byte) byte {
+	return flags & mqttPubFlagQoS >> 1
+}
+
+func mqttIsRetained(flags byte) bool {
+	return flags&mqttPubFlagRetain != 0
+}
+
+//////////////////////////////////////////////////////////////////////////////
+//
+// SUBSCRIBE related functions
+//
+//////////////////////////////////////////////////////////////////////////////
+
+func (c *client) mqttParseSubs(r *mqttReader, b byte, pl int) (uint16, []*mqttFilter, error) {
+	return c.mqttParseSubsOrUnsubs(r, b, pl, true)
+}
+
+func (c *client) mqttParseSubsOrUnsubs(r *mqttReader, b byte, pl int, sub bool) (uint16, []*mqttFilter, error) {
+	var expectedFlag byte
+	var action string
+	if sub {
+		expectedFlag = mqttSubscribeFlags
+	} else {
+		expectedFlag = mqttUnsubscribeFlags
+		action = "un"
+	}
+	// Spec [MQTT-3.8.1-1], [MQTT-3.10.1-1]
+	if rf := b & 0xf; rf != expectedFlag {
+		return 0, nil, fmt.Errorf("wrong %ssubscribe reserved flags: %x", action, rf)
+	}
+	if err := r.ensurePacketInBuffer(pl); err != nil {
+		return 0, nil, err
+	}
+	pi, err := r.readUint16("packet identifier")
+	if err != nil {
+		return 0, nil, fmt.Errorf("reading packet identifier: %v", err)
+	}
+	end := r.pos + (pl - 2)
+	var filters []*mqttFilter
+	for r.pos < end {
+		// Don't make a copy now because, this will happen during conversion
+		// or when processing the sub.
+		filter, err := r.readBytes("topic filter", false)
+		if err != nil {
+			return 0, nil, err
+		}
+		if len(filter) == 0 {
+			return 0, nil, errors.New("topic filter cannot be empty")
+		}
+		// Spec [MQTT-3.8.3-1], [MQTT-3.10.3-1]
+		if !utf8.Valid(filter) {
+			return 0, nil, fmt.Errorf("invalid utf8 for topic filter %q", filter)
+		}
+		var qos byte
+		topic := filter
+		// This won't return an error. We will find out if the subject
+		// is valid or not when trying to create the subscription.
+		filter, _ = mqttFilterToNATSSubject(filter)
+		if sub {
+			qos, err = r.readByte("QoS")
+			if err != nil {
+				return 0, nil, err
+			}
+			// Spec [MQTT-3-8.3-4].
+			if qos > 2 {
+				return 0, nil, fmt.Errorf("subscribe QoS value must be 0, 1 or 2, got %v", qos)
+			}
+		}
+		f := &mqttFilter{ttopic: topic, filter: string(filter), qos: qos}
+		filters = append(filters, f)
+	}
+	// Spec [MQTT-3.8.3-3], [MQTT-3.10.3-2]
+	if len(filters) == 0 {
+		return 0, nil, fmt.Errorf("%ssubscribe protocol must contain at least 1 topic filter", action)
+	}
+	return pi, filters, nil
+}
+
+func mqttSubscribeTrace(pi uint16, filters []*mqttFilter) string {
+	var sep string
+	sb := &strings.Builder{}
+	sb.WriteString("[")
+	for i, f := range filters {
+		sb.WriteString(sep)
+		sb.Write(f.ttopic)
+		sb.WriteString(" (")
+		sb.WriteString(f.filter)
+		sb.WriteString(") QoS=")
+		sb.WriteString(fmt.Sprintf("%v", f.qos))
+		if i == 0 {
+			sep = ", "
+		}
+	}
+	sb.WriteString(fmt.Sprintf("] pi=%v", pi))
+	return sb.String()
+}
+
+func mqttDeliverMsgCb(sub *subscription, pc *client, subject, reply string, msg []byte) {
+	var ppFlags byte
+	var pQoS byte
+	var pi uint16
+	var dup bool
+	var retained bool
+	var topic []byte
+
+	// This is the client associated with the subscription.
+	cc := sub.client
+
+	// This is immutable
+	sess := cc.mqtt.sess
+	// We lock to check some of the subscription's fields and if we need to
+	// keep track of pending acks, etc..
+	sess.mu.Lock()
+	if sess.c != cc || sub.mqtt == nil {
+		sess.mu.Unlock()
+		return
+	}
+
+	// Check the publisher's kind. If JETSTREAM it means that this is a persisted message
+	// that is being delivered.
+	if pc.kind == JETSTREAM {
+		// If there is no JS consumer attached to this subscription, it means that we are
+		// dealing with a bare NATS subscription, in which case we simply return to avoid
+		// duplicate delivery.
+		if sub.mqtt.jsCons == nil {
+			sess.mu.Unlock()
+			return
+		}
+		ppFlags = mqttPubQos1
+		pQoS = 1
+		// This is a QoS1 message for a QoS1 subscription, so get the pi and keep
+		// track of ack subject.
+		pi, dup = sess.trackPending(pQoS, reply, sub)
+		if pi == 0 {
+			// We have reached max pending, don't send the message now.
+			// JS will cause a redelivery and if by then the number of pending
+			// messages has fallen below threshold, the message will be resent.
+			sess.mu.Unlock()
+			return
+		}
+		// TODO: We hae to conver the subject back to topic. One approach
+		// was the pass the parsed MQTT topic in jsPubMsg but it was decided
+		// to not do that for now.
+		topic = natsSubjectToMQTTTopic(string(pc.pa.deliver))
+	} else if pc.isMqtt() {
+		// This is a MQTT publisher...
+		topic = pc.mqtt.pp.topic
+		ppFlags = pc.mqtt.pp.flags
+		pQoS = mqttGetQoS(ppFlags)
+		// If the QoS of published message and subscription is 1, then we return here to
+		// avoid duplicate delivery. The JetStream publisher will handle that case.
+		if pQoS > 0 && sub.mqtt.qos > 0 {
+			sess.mu.Unlock()
+			return
+		}
+		retained = mqttIsRetained(ppFlags)
+	} else {
+		// This is coming from a non MQTT publisher, so Qos 0, no dup nor retain flag, etc..
+		// Should probably reject, for now just truncate.
+		if len(msg) > mqttMaxPayloadSize {
+			msg = msg[:mqttMaxPayloadSize]
+		}
+		// See TODO for JS producer. Right now, we need to do the conversion
+		// back to a subject everytime.
+		topic = natsSubjectToMQTTTopic(subject)
+	}
+	sess.mu.Unlock()
+
+	sw := mqttWriter{}
+	w := &sw
+
+	flags := mqttSerializePublishMsg(w, pi, dup, retained, topic, msg)
+
+	cc.mu.Lock()
+	if sub.mqtt.prm != nil {
+		cc.queueOutbound(sub.mqtt.prm.Bytes())
+		sub.mqtt.prm = nil
+	}
+	cc.queueOutbound(w.Bytes())
+	pc.addToPCD(cc)
+	if cc.trace {
+		pp := mqttPublish{
+			topic: topic,
+			flags: flags,
+			pi:    pi,
+			sz:    len(msg),
+		}
+		cc.traceOutOp("PUBLISH", []byte(mqttPubTrace(&pp)))
+	}
+	cc.mu.Unlock()
+}
+
+// Serializes to the given writer the message for the given subject.
+func mqttSerializePublishMsg(w *mqttWriter, pi uint16, dup, retained bool, topic, msg []byte) byte {
+
+	// Compute len (will have to add packet id if message is sent as QoS>=1)
+	pkLen := 2 + len(topic) + len(msg)
+
+	var flags byte
+
+	// Set flags for dup/retained/qos1
+	if dup {
+		flags |= mqttPubFlagDup
+	}
+	if retained {
+		flags |= mqttPubFlagRetain
+	}
+	// For now, we have only QoS 1
+	if pi > 0 {
+		pkLen += 2
+		flags |= mqttPubQos1
+	}
+
+	w.WriteByte(mqttPacketPub | flags)
+	w.WriteVarInt(pkLen)
+	w.WriteBytes(topic)
+	if pi > 0 {
+		w.WriteUint16(pi)
+	}
+	w.Write(msg)
+
+	return flags
+}
+
+// Process the list of subscriptions and update the given filter
+// with the QoS that has been accepted (or failure).
+//
+// Spec [MQTT-3.8.4-3] says that if an exact same subscription is
+// found, it needs to be replaced with the new one (possibly updating
+// the qos) and that the flow of publications must not be interrupted,
+// which I read as the replacement cannot be a "remove then add" if there
+// is a chance that in between the 2 actions, published messages
+// would be "lost" because there would not be any matching subscription.
+func (c *client) mqttProcessSubs(filters []*mqttFilter) ([]*subscription, error) {
+	// Those things are immutable, but since processing subs is not
+	// really in the fast path, let's get them under the client lock.
+	c.mu.Lock()
+	asm := c.mqtt.asm
+	sess := c.mqtt.sess
+	clientID := c.mqtt.cp.clientID
+	trace := c.trace
+	c.mu.Unlock()
+
+	asm.mu.RLock()
+	defer asm.mu.RUnlock()
+	return asm.processSubs(sess, clientID, c, filters, true, trace)
+}
+
+func (c *client) mqttCleanupFailedSub(sub *subscription, jscons *Consumer, jssub *subscription) {
+	c.mu.Lock()
+	acc := c.acc
+	c.mu.Unlock()
+
+	if sub != nil {
+		c.unsubscribe(acc, sub, true, true)
+	}
+	if jssub != nil {
+		c.unsubscribe(acc, jssub, true, true)
+	}
+	if jscons != nil {
+		jscons.Delete()
+	}
+}
+
+// When invoked with a QoS of 0, looks for an existing JS durable consumer for
+// the given sid and if one is found, delete the JS durable consumer and unsub
+// the NATS subscription on the delivery subject.
+// With a QoS > 0, creates or update the existing JS durable consumer along with
+// its NATS subscription on a delivery subject.
+//
+// Account session manager lock held on entry.
+func (c *client) mqttProcessJSConsumer(sess *mqttSession, stream *Stream, subject,
+	sid string, qos byte, fromSubProto bool) (*Consumer, *subscription, error) {
+
+	// Check if we are already a JS consumer for this SID.
+	cons, exists := sess.cons[sid]
+	if exists {
+		// If current QoS is 0, it means that we need to delete the existing
+		// one (that was QoS > 0)
+		if qos == 0 {
+			// The JS durable consumer's delivery subject is on a NUID of
+			// the form: mqttSubPrefix + <nuid>. It is also used as the sid
+			// for the NATS subscription, so use that for the lookup.
+			sub := c.subs[cons.Config().DeliverSubject]
+			delete(sess.cons, sid)
+			cons.Delete()
+			if sub != nil {
+				c.mu.Lock()
+				acc := c.acc
+				c.mu.Unlock()
+				c.unsubscribe(acc, sub, true, true)
+			}
+			return nil, nil, nil
+		}
+		// If this is called when processing SUBSCRIBE protocol, then if
+		// the JS consumer already exists, we are done (it was created
+		// during the processing of CONNECT).
+		if fromSubProto {
+			return nil, nil, nil
+		}
+	}
+	// Here it means we don't have a JS consumer and if we are QoS 0,
+	// we have nothing to do.
+	if qos == 0 {
+		return nil, nil, nil
+	}
+	var err error
+	inbox := mqttSubPrefix + nuid.Next()
+	if exists {
+		cons.updateDeliverSubject(inbox)
+	} else {
+		durName := nuid.Next()
+		opts := c.srv.getOpts()
+		ackWait := opts.MQTT.AckWait
+		if ackWait == 0 {
+			ackWait = mqttDefaultAckWait
+		}
+		maxAckPending := opts.MQTT.MaxAckPending
+		if maxAckPending == 0 {
+			maxAckPending = mqttDefaultMaxAckPending
+		}
+		cc := &ConsumerConfig{
+			DeliverSubject:  inbox,
+			Durable:         durName,
+			AckPolicy:       AckExplicit,
+			DeliverPolicy:   DeliverNew,
+			FilterSubject:   subject,
+			AckWait:         ackWait,
+			MaxAckPending:   int(maxAckPending),
+			allowNoInterest: true,
+		}
+		cons, err = stream.AddConsumer(cc)
+		if err != nil {
+			c.Errorf("Unable to add JetStream consumer for subscription on %q: err=%v", subject, err)
+			return nil, nil, err
+		}
+	}
+	// This is an internal subscription on subject like "$MQTT.sub.<nuid>" that is setup
+	// for the JS durable's deliver subject. I don't think that there is any need to
+	// forward this subscription in the cluster/super cluster.
+	sub, err := c.processSub([]byte(inbox), nil, []byte(inbox), mqttDeliverMsgCb, true)
+	if err != nil {
+		if !exists {
+			cons.Delete()
+		}
+		c.Errorf("Unable to create subscription for JetStream consumer on %q: %v", subject, err)
+		return nil, nil, err
+	}
+	if sub.mqtt == nil {
+		sub.mqtt = &mqttSub{}
+	}
+	sub.mqtt.qos = qos
+	sub.mqtt.jsCons = cons
+	return cons, sub, nil
+}
+
+// Queues the published retained messages for each subscription and signals
+// the writeLoop.
+func (c *client) mqttSendRetainedMsgsToNewSubs(subs []*subscription) {
+	c.mu.Lock()
+	for _, sub := range subs {
+		if sub.mqtt != nil && sub.mqtt.prm != nil {
+			c.queueOutbound(sub.mqtt.prm.Bytes())
+			sub.mqtt.prm = nil
+		}
+	}
+	c.flushSignal()
+	c.mu.Unlock()
+}
+
+func (c *client) mqttEnqueueSubAck(pi uint16, filters []*mqttFilter) {
+	w := &mqttWriter{}
+	w.WriteByte(mqttPacketSubAck)
+	// packet length is 2 (for packet identifier) and 1 byte per filter.
+	w.WriteVarInt(2 + len(filters))
+	w.WriteUint16(pi)
+	for _, f := range filters {
+		w.WriteByte(f.qos)
+	}
+	c.mu.Lock()
+	c.enqueueProto(w.Bytes())
+	c.mu.Unlock()
+}
+
+//////////////////////////////////////////////////////////////////////////////
+//
+// UNSUBSCRIBE related functions
+//
+//////////////////////////////////////////////////////////////////////////////
+
+func (c *client) mqttParseUnsubs(r *mqttReader, b byte, pl int) (uint16, []*mqttFilter, error) {
+	return c.mqttParseSubsOrUnsubs(r, b, pl, false)
+}
+
+func (c *client) mqttProcessUnsubs(filters []*mqttFilter) error {
+	// Those things are immutable, but since processing unsubs is not
+	// really in the fast path, let's get them under the client lock.
+	c.mu.Lock()
+	sess := c.mqtt.sess
+	clientID := c.mqtt.cp.clientID
+	c.mu.Unlock()
+
+	sess.mu.Lock()
+	defer sess.mu.Unlock()
+	if sess.c != c {
+		return fmt.Errorf("client %q no longer registered with MQTT session", clientID)
+	}
+
+	removeJSCons := func(sid string) {
+		if jscons, ok := sess.cons[sid]; ok {
+			delete(sess.cons, sid)
+			jscons.Delete()
+			if seqPis, ok := sess.cpending[jscons]; ok {
+				delete(sess.cpending, jscons)
+				for _, pi := range seqPis {
+					delete(sess.pending, pi)
+				}
+				if len(sess.pending) == 0 {
+					sess.ppi = 0
+				}
+			}
+		}
+	}
+	for _, f := range filters {
+		sid := f.filter
+		// Remove JS Consumer if one exists for this sid
+		removeJSCons(sid)
+		if err := c.processUnsub([]byte(sid)); err != nil {
+			c.Errorf("error unsubscribing from %q: %v", sid, err)
+		}
+		if mqttNeedSubForLevelUp(sid) {
+			subject := sid[:len(sid)-2]
+			sid = subject + mqttMultiLevelSidSuffix
+			removeJSCons(sid)
+			if err := c.processUnsub([]byte(sid)); err != nil {
+				c.Errorf("error unsubscribing from %q: %v", subject, err)
+			}
+		}
+	}
+	return sess.update(clientID, filters, false)
+}
+
+func (c *client) mqttEnqueueUnsubAck(pi uint16) {
+	w := &mqttWriter{}
+	w.WriteByte(mqttPacketUnsubAck)
+	w.WriteVarInt(2)
+	w.WriteUint16(pi)
+	c.mu.Lock()
+	c.enqueueProto(w.Bytes())
+	c.mu.Unlock()
+}
+
+func mqttUnsubscribeTrace(pi uint16, filters []*mqttFilter) string {
+	var sep string
+	sb := strings.Builder{}
+	sb.WriteString("[")
+	for i, f := range filters {
+		sb.WriteString(sep)
+		sb.Write(f.ttopic)
+		sb.WriteString(" (")
+		sb.WriteString(f.filter)
+		sb.WriteString(")")
+		if i == 0 {
+			sep = ", "
+		}
+	}
+	sb.WriteString(fmt.Sprintf("] pi=%v", pi))
+	return sb.String()
+}
+
+//////////////////////////////////////////////////////////////////////////////
+//
+// PINGREQ/PINGRESP related functions
+//
+//////////////////////////////////////////////////////////////////////////////
+
+func (c *client) mqttEnqueuePingResp() {
+	c.mu.Lock()
+	c.enqueueProto(mqttPingResponse)
+	c.mu.Unlock()
+}
+
+//////////////////////////////////////////////////////////////////////////////
+//
+// Trace functions
+//
+//////////////////////////////////////////////////////////////////////////////
+
+func errOrTrace(err error, trace string) []byte {
+	if err != nil {
+		return []byte(err.Error())
+	}
+	return []byte(trace)
+}
+
+//////////////////////////////////////////////////////////////////////////////
+//
+// Subject/Topic conversion functions
+//
+//////////////////////////////////////////////////////////////////////////////
+
+// Converts an MQTT Topic Name to a NATS Subject (used by PUBLISH)
+// See mqttToNATSSubjectConversion() for details.
+func mqttTopicToNATSPubSubject(mt []byte) ([]byte, error) {
+	return mqttToNATSSubjectConversion(mt, false)
+}
+
+// Converts an MQTT Topic Filter to a NATS Subject (used by SUBSCRIBE)
+// See mqttToNATSSubjectConversion() for details.
+func mqttFilterToNATSSubject(filter []byte) ([]byte, error) {
+	return mqttToNATSSubjectConversion(filter, true)
+}
+
+// Converts an MQTT Topic Name or Filter to a NATS Subject.
+// In MQTT:
+// - a Topic Name does not have wildcard (PUBLISH uses only topic names).
+// - a Topic Filter can include wildcards (SUBSCRIBE uses those).
+// - '+' and '#' are wildcard characters (single and multiple levels respectively)
+// - '/' is the topic level separator.
+//
+// Conversion that occurs:
+// - '/' is replaced with '/.' if it is the first character in mt
+// - '/' is replaced with './' if the last or next character in mt is '/'
+//   For instance, foo//bar would become foo./.bar
+// - '/' is replaced with '.' for all other conditions (foo/bar -> foo.bar)
+// - '.' and ' ' cause an error to be returned.
+//
+// If there is no need to convert anything (say "foo" remains "foo"), then
+// the no memory is allocated and the returned slice is the original `mt`.
+func mqttToNATSSubjectConversion(mt []byte, wcOk bool) ([]byte, error) {
+	var cp bool
+	var j int
+	res := mt
+
+	makeCopy := func(i int) {
+		cp = true
+		res = make([]byte, 0, len(mt)+10)
+		if i > 0 {
+			res = append(res, mt[:i]...)
+		}
+	}
+
+	end := len(mt) - 1
+	for i := 0; i < len(mt); i++ {
+		switch mt[i] {
+		case mqttTopicLevelSep:
+			if i == 0 || res[j-1] == btsep {
+				if !cp {
+					makeCopy(0)
+				}
+				res = append(res, mqttTopicLevelSep, btsep)
+				j++
+			} else if i == end || mt[i+1] == mqttTopicLevelSep {
+				if !cp {
+					makeCopy(i)
+				}
+				res = append(res, btsep, mqttTopicLevelSep)
+				j++
+			} else {
+				if !cp {
+					makeCopy(i)
+				}
+				res = append(res, btsep)
+			}
+		case btsep, ' ':
+			// As of now, we cannot support '.' or ' ' in the MQTT topic/filter.
+			return nil, fmt.Errorf("characters ' ' and '.' not supported for MQTT topics")
+		case mqttSingleLevelWC, mqttMultiLevelWC:
+			if !wcOk {
+				// Spec [MQTT-3.3.2-2] and [MQTT-4.7.1-1]
+				// The wildcard characters can be used in Topic Filters, but MUST NOT be used within a Topic Name
+				return nil, fmt.Errorf("wildcards not allowed in publish's topic: %q", mt)
+			}
+			if !cp {
+				makeCopy(i)
+			}
+			if mt[i] == mqttSingleLevelWC {
+				res = append(res, pwc)
+			} else {
+				res = append(res, fwc)
+			}
+		default:
+			if cp {
+				res = append(res, mt[i])
+			}
+		}
+		j++
+	}
+	if cp && res[j-1] == btsep {
+		res = append(res, mqttTopicLevelSep)
+		j++
+	}
+	return res[:j], nil
+}
+
+// Converts a NATS subject to MQTT topic. This is for publish
+// messages only, so there is no checking for wildcards.
+// Rules are reversed of mqttToNATSSubjectConversion.
+func natsSubjectToMQTTTopic(subject string) []byte {
+	topic := []byte(subject)
+	end := len(subject) - 1
+	var j int
+	for i := 0; i < len(subject); i++ {
+		switch subject[i] {
+		case mqttTopicLevelSep:
+			if !(i == 0 && i < end && subject[i+1] == btsep) {
+				topic[j] = mqttTopicLevelSep
+				j++
+			}
+		case btsep:
+			topic[j] = mqttTopicLevelSep
+			j++
+			if i < end && subject[i+1] == mqttTopicLevelSep {
+				i++
+			}
+		default:
+			topic[j] = subject[i]
+			j++
+		}
+	}
+	return topic[:j]
+}
+
+// Returns true if the subject has more than 1 token and ends with ".>"
+func mqttNeedSubForLevelUp(subject string) bool {
+	if len(subject) < 3 {
+		return false
+	}
+	end := len(subject)
+	if subject[end-2] == '.' && subject[end-1] == fwc {
+		return true
+	}
+	return false
+}
+
+//////////////////////////////////////////////////////////////////////////////
+//
+// MQTT Reader functions
+//
+//////////////////////////////////////////////////////////////////////////////
+
+func copyBytes(b []byte) []byte {
+	if b == nil {
+		return nil
+	}
+	cbuf := make([]byte, len(b))
+	copy(cbuf, b)
+	return cbuf
+}
+
+func (r *mqttReader) reset(buf []byte) {
+	r.buf = buf
+	r.pos = 0
+}
+
+func (r *mqttReader) hasMore() bool {
+	return r.pos != len(r.buf)
+}
+
+func (r *mqttReader) readByte(field string) (byte, error) {
+	if r.pos == len(r.buf) {
+		return 0, fmt.Errorf("error reading %s: %v", field, io.EOF)
+	}
+	b := r.buf[r.pos]
+	r.pos++
+	return b, nil
+}
+
+func (r *mqttReader) readPacketLen() (int, error) {
+	m := 1
+	v := 0
+	for {
+		var b byte
+		if r.pos != len(r.buf) {
+			b = r.buf[r.pos]
+			r.pos++
+		} else {
+			var buf [1]byte
+			if _, err := r.reader.Read(buf[:1]); err != nil {
+				if err == io.EOF {
+					return 0, io.ErrUnexpectedEOF
+				}
+				return 0, fmt.Errorf("error reading packet length: %v", err)
+			}
+			b = buf[0]
+		}
+		v += int(b&0x7f) * m
+		if (b & 0x80) == 0 {
+			return v, nil
+		}
+		m *= 0x80
+		if m > 0x200000 {
+			return 0, errors.New("malformed variable int")
+		}
+	}
+}
+
+func (r *mqttReader) ensurePacketInBuffer(pl int) error {
+	rem := len(r.buf) - r.pos
+	if rem >= pl {
+		return nil
+	}
+	b := make([]byte, pl)
+	start := copy(b, r.buf[r.pos:])
+	for start != pl {
+		n, err := r.reader.Read(b[start:cap(b)])
+		if err != nil {
+			if err == io.EOF {
+				err = io.ErrUnexpectedEOF
+			}
+			return fmt.Errorf("error ensuring protocol is loaded: %v", err)
+		}
+		start += n
+	}
+	r.reset(b)
+	return nil
+}
+
+func (r *mqttReader) readString(field string) (string, error) {
+	var s string
+	bs, err := r.readBytes(field, false)
+	if err == nil {
+		s = string(bs)
+	}
+	return s, err
+}
+
+func (r *mqttReader) readBytes(field string, cp bool) ([]byte, error) {
+	luint, err := r.readUint16(field)
+	if err != nil {
+		return nil, err
+	}
+	l := int(luint)
+	if l == 0 {
+		return nil, nil
+	}
+	start := r.pos
+	if start+l > len(r.buf) {
+		return nil, fmt.Errorf("error reading %s: %v", field, io.ErrUnexpectedEOF)
+	}
+	r.pos += l
+	b := r.buf[start:r.pos]
+	if cp {
+		b = copyBytes(b)
+	}
+	return b, nil
+}
+
+func (r *mqttReader) readUint16(field string) (uint16, error) {
+	if len(r.buf)-r.pos < 2 {
+		return 0, fmt.Errorf("error reading %s: %v", field, io.ErrUnexpectedEOF)
+	}
+	start := r.pos
+	r.pos += 2
+	return binary.BigEndian.Uint16(r.buf[start:r.pos]), nil
+}
+
+//////////////////////////////////////////////////////////////////////////////
+//
+// MQTT Writer functions
+//
+//////////////////////////////////////////////////////////////////////////////
+
+func (w *mqttWriter) WriteUint16(i uint16) {
+	w.WriteByte(byte(i >> 8))
+	w.WriteByte(byte(i))
+}
+
+func (w *mqttWriter) WriteString(s string) {
+	w.WriteBytes([]byte(s))
+}
+
+func (w *mqttWriter) WriteBytes(bs []byte) {
+	w.WriteUint16(uint16(len(bs)))
+	w.Write(bs)
+}
+
+func (w *mqttWriter) WriteVarInt(value int) {
+	for {
+		b := byte(value & 0x7f)
+		value >>= 7
+		if value > 0 {
+			b |= 0x80
+		}
+		w.WriteByte(b)
+		if value == 0 {
+			break
+		}
+	}
+}
diff --git a/vendor/github.com/nats-io/nats-server/v2/server/opts.go b/vendor/github.com/nats-io/nats-server/v2/server/opts.go
index 3280213..3f44538 100644
--- a/vendor/github.com/nats-io/nats-server/v2/server/opts.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/opts.go
@@ -197,6 +197,7 @@ type Options struct {
 	JetStreamMaxStore     int64         `json:"-"`
 	StoreDir              string        `json:"-"`
 	Websocket             WebsocketOpts `json:"-"`
+	MQTT                  MQTTOpts      `json:"-"`
 	ProfPort              int           `json:"-"`
 	PidFile               string        `json:"-"`
 	PortsFileDir          string        `json:"-"`
@@ -256,7 +257,7 @@ type Options struct {
 	routeProto           int
 }
 
-// WebsocketOpts ...
+// WebsocketOpts are options for websocket
 type WebsocketOpts struct {
 	// The server will accept websocket client connections on this hostname/IP.
 	Host string
@@ -316,6 +317,49 @@ type WebsocketOpts struct {
 	HandshakeTimeout time.Duration
 }
 
+// MQTTOpts are options for MQTT
+type MQTTOpts struct {
+	// The server will accept MQTT client connections on this hostname/IP.
+	Host string
+	// The server will accept MQTT client connections on this port.
+	Port int
+
+	// If no user name is provided when a client connects, will default to the
+	// matching user from the global list of users in `Options.Users`.
+	NoAuthUser string
+
+	// Authentication section. If anything is configured in this section,
+	// it will override the authorization configuration of regular clients.
+	Username string
+	Password string
+	Token    string
+
+	// Timeout for the authentication process.
+	AuthTimeout float64
+
+	// TLS configuration is required.
+	TLSConfig *tls.Config
+	// If true, map certificate values for authentication purposes.
+	TLSMap bool
+	// Timeout for the TLS handshake
+	TLSTimeout float64
+
+	// AckWait is the amount of time after which a QoS 1 message sent to
+	// a client is redelivered as a DUPLICATE if the server has not
+	// received the PUBACK on the original Packet Identifier.
+	// The value has to be positive.
+	// Zero will cause the server to use the default value (1 hour).
+	// Note that changes to this option is applied only to new MQTT subscriptions.
+	AckWait time.Duration
+
+	// MaxAckPending is the amount of QoS 1 messages the server can send to
+	// a session without receiving any PUBACK for those messages.
+	// The valid range is [0..65535].
+	// Zero will cause the server to use the default value (1024).
+	// Note that changes to this option is applied only to new MQTT sessions.
+	MaxAckPending uint16
+}
+
 type netResolver interface {
 	LookupHost(ctx context.Context, host string) ([]string, error)
 }
@@ -1011,6 +1055,11 @@ func (o *Options) processConfigFileLine(k string, v interface{}, errors *[]error
 			*errors = append(*errors, err)
 			return
 		}
+	case "mqtt":
+		if err := parseMQTT(tk, o, errors, warnings); err != nil {
+			*errors = append(*errors, err)
+			return
+		}
 	default:
 		if au := atomic.LoadInt32(&allowUnknownTopLevelField); au == 0 && !tk.IsUsedVariable() {
 			err := &unknownConfigFieldErr{
@@ -3327,7 +3376,7 @@ func parseTLS(v interface{}, isClientCtx bool) (t *TLSConfigOpts, retErr error)
 	return &tc, nil
 }
 
-func parseAuthForWS(v interface{}, errors *[]error, warnings *[]error) *authorization {
+func parseSimpleAuth(v interface{}, errors *[]error, warnings *[]error) *authorization {
 	var (
 		am   map[string]interface{}
 		tk   token
@@ -3463,7 +3512,7 @@ func parseWebsocket(v interface{}, o *Options, errors *[]error, warnings *[]erro
 		case "compression":
 			o.Websocket.Compression = mv.(bool)
 		case "authorization", "authentication":
-			auth := parseAuthForWS(tk, errors, warnings)
+			auth := parseSimpleAuth(tk, errors, warnings)
 			o.Websocket.Username = auth.user
 			o.Websocket.Password = auth.pass
 			o.Websocket.Token = auth.token
@@ -3488,6 +3537,79 @@ func parseWebsocket(v interface{}, o *Options, errors *[]error, warnings *[]erro
 	return nil
 }
 
+func parseMQTT(v interface{}, o *Options, errors *[]error, warnings *[]error) error {
+	var lt token
+	defer convertPanicToErrorList(&lt, errors)
+
+	tk, v := unwrapValue(v, &lt)
+	gm, ok := v.(map[string]interface{})
+	if !ok {
+		return &configErr{tk, fmt.Sprintf("Expected mqtt to be a map, got %T", v)}
+	}
+	for mk, mv := range gm {
+		// Again, unwrap token value if line check is required.
+		tk, mv = unwrapValue(mv, &lt)
+		switch strings.ToLower(mk) {
+		case "listen":
+			hp, err := parseListen(mv)
+			if err != nil {
+				err := &configErr{tk, err.Error()}
+				*errors = append(*errors, err)
+				continue
+			}
+			o.MQTT.Host = hp.host
+			o.MQTT.Port = hp.port
+		case "port":
+			o.MQTT.Port = int(mv.(int64))
+		case "host", "net":
+			o.MQTT.Host = mv.(string)
+		case "tls":
+			tc, err := parseTLS(tk, true)
+			if err != nil {
+				*errors = append(*errors, err)
+				continue
+			}
+			if o.MQTT.TLSConfig, err = GenTLSConfig(tc); err != nil {
+				err := &configErr{tk, err.Error()}
+				*errors = append(*errors, err)
+				continue
+			}
+			o.MQTT.TLSTimeout = tc.Timeout
+			o.MQTT.TLSMap = tc.Map
+		case "authorization", "authentication":
+			auth := parseSimpleAuth(tk, errors, warnings)
+			o.MQTT.Username = auth.user
+			o.MQTT.Password = auth.pass
+			o.MQTT.Token = auth.token
+			o.MQTT.AuthTimeout = auth.timeout
+		case "no_auth_user":
+			o.MQTT.NoAuthUser = mv.(string)
+		case "ack_wait", "ackwait":
+			o.MQTT.AckWait = parseDuration("ack_wait", tk, mv, errors, warnings)
+		case "max_ack_pending", "max_pending", "max_inflight":
+			tmp := int(mv.(int64))
+			if tmp < 0 || tmp > 0xFFFF {
+				err := &configErr{tk, fmt.Sprintf("invalid value %v, should in [0..%d] range", tmp, 0xFFFF)}
+				*errors = append(*errors, err)
+			} else {
+				o.MQTT.MaxAckPending = uint16(tmp)
+			}
+		default:
+			if !tk.IsUsedVariable() {
+				err := &unknownConfigFieldErr{
+					field: mk,
+					configErr: configErr{
+						token: tk,
+					},
+				}
+				*errors = append(*errors, err)
+				continue
+			}
+		}
+	}
+	return nil
+}
+
 // GenTLSConfig loads TLS related configuration parameters.
 func GenTLSConfig(tc *TLSConfigOpts) (*tls.Config, error) {
 	// Create the tls.Config from our options before including the certs.
@@ -3831,6 +3953,14 @@ func setBaselineOptions(opts *Options) {
 			opts.Websocket.Host = DEFAULT_HOST
 		}
 	}
+	if opts.MQTT.Port != 0 {
+		if opts.MQTT.Host == "" {
+			opts.MQTT.Host = DEFAULT_HOST
+		}
+		if opts.MQTT.TLSTimeout == 0 {
+			opts.MQTT.TLSTimeout = float64(TLS_TIMEOUT) / float64(time.Second)
+		}
+	}
 	// JetStream
 	if opts.JetStreamMaxMemory == 0 {
 		opts.JetStreamMaxMemory = -1
diff --git a/vendor/github.com/nats-io/nats-server/v2/server/parser.go b/vendor/github.com/nats-io/nats-server/v2/server/parser.go
index d7a0d73..b6bbd53 100644
--- a/vendor/github.com/nats-io/nats-server/v2/server/parser.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/parser.go
@@ -47,6 +47,7 @@ type pubArg struct {
 	queues  [][]byte
 	size    int
 	hdr     int
+	psi     *serviceImport
 }
 
 // Parser constants
@@ -130,6 +131,11 @@ const (
 )
 
 func (c *client) parse(buf []byte) error {
+	// Branch out to mqtt clients. c.mqtt is immutable, but should it become
+	// an issue (say data race detection), we could branch outside in readLoop
+	if c.isMqtt() {
+		return c.mqttParse(buf)
+	}
 	var i int
 	var b byte
 	var lmsg bool
@@ -733,7 +739,12 @@ func (c *client) parse(buf []byte) error {
 					err = c.processUnsub(arg)
 				case ROUTER:
 					if trace && c.srv != nil {
-						c.traceInOp("RS-", arg)
+						switch c.op {
+						case 'R', 'r':
+							c.traceInOp("RS-", arg)
+						case 'L', 'l':
+							c.traceInOp("LS-", arg)
+						}
 					}
 					err = c.processRemoteUnsub(arg)
 				case GATEWAY:
diff --git a/vendor/github.com/nats-io/nats-server/v2/server/reload.go b/vendor/github.com/nats-io/nats-server/v2/server/reload.go
index 7861dcc..a533992 100644
--- a/vendor/github.com/nats-io/nats-server/v2/server/reload.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/reload.go
@@ -596,6 +596,25 @@ func (m *maxTracedMsgLenOption) Apply(server *Server) {
 	server.Noticef("Reloaded: max_traced_msg_len = %d", m.newValue)
 }
 
+type mqttAckWaitReload struct {
+	noopOption
+	newValue time.Duration
+}
+
+func (o *mqttAckWaitReload) Apply(s *Server) {
+	s.Noticef("Reloaded: MQTT ack_wait = %v", o.newValue)
+}
+
+type mqttMaxAckPendingReload struct {
+	noopOption
+	newValue uint16
+}
+
+func (o *mqttMaxAckPendingReload) Apply(s *Server) {
+	s.mqttUpdateMaxAckPending(o.newValue)
+	s.Noticef("Reloaded: MQTT max_ack_pending = %v", o.newValue)
+}
+
 // Reload reads the current configuration file and applies any supported
 // changes. This returns an error if the server was not started with a config
 // file or an option which doesn't support hot-swapping was changed.
@@ -633,6 +652,7 @@ func (s *Server) Reload() error {
 	gatewayOrgPort := curOpts.Gateway.Port
 	leafnodesOrgPort := curOpts.LeafNode.Port
 	websocketOrgPort := curOpts.Websocket.Port
+	mqttOrgPort := curOpts.MQTT.Port
 
 	s.mu.Unlock()
 
@@ -665,6 +685,9 @@ func (s *Server) Reload() error {
 	if newOpts.Websocket.Port == -1 {
 		newOpts.Websocket.Port = websocketOrgPort
 	}
+	if newOpts.MQTT.Port == -1 {
+		newOpts.MQTT.Port = mqttOrgPort
+	}
 
 	if err := s.reloadOptions(curOpts, newOpts); err != nil {
 		return err
@@ -766,7 +789,7 @@ func imposeOrder(value interface{}) error {
 	case WebsocketOpts:
 		sort.Strings(value.AllowedOrigins)
 	case string, bool, int, int32, int64, time.Duration, float64, nil,
-		LeafNodeOpts, ClusterOpts, *tls.Config, *URLAccResolver, *MemAccResolver, *DirAccResolver, *CacheDirAccResolver, Authentication:
+		LeafNodeOpts, ClusterOpts, *tls.Config, *URLAccResolver, *MemAccResolver, *DirAccResolver, *CacheDirAccResolver, Authentication, MQTTOpts:
 		// explicitly skipped types
 	default:
 		// this will fail during unit tests
@@ -893,6 +916,13 @@ func (s *Server) diffOptions(newOpts *Options) ([]option, error) {
 			tmpNew := newValue.(GatewayOpts)
 			tmpOld.TLSConfig = nil
 			tmpNew.TLSConfig = nil
+
+			// Need to do the same for remote gateways' TLS configs.
+			// But we can't just set remotes' TLSConfig to nil otherwise this
+			// would lose the real TLS configuration.
+			tmpOld.Gateways = copyRemoteGWConfigsWithoutTLSConfig(tmpOld.Gateways)
+			tmpNew.Gateways = copyRemoteGWConfigsWithoutTLSConfig(tmpNew.Gateways)
+
 			// If there is really a change prevents reload.
 			if !reflect.DeepEqual(tmpOld, tmpNew) {
 				// See TODO(ik) note below about printing old/new values.
@@ -906,6 +936,12 @@ func (s *Server) diffOptions(newOpts *Options) ([]option, error) {
 			tmpOld.TLSConfig = nil
 			tmpNew.TLSConfig = nil
 
+			// Need to do the same for remote leafnodes' TLS configs.
+			// But we can't just set remotes' TLSConfig to nil otherwise this
+			// would lose the real TLS configuration.
+			tmpOld.Remotes = copyRemoteLNConfigWithoutTLSConfig(tmpOld.Remotes)
+			tmpNew.Remotes = copyRemoteLNConfigWithoutTLSConfig(tmpNew.Remotes)
+
 			// Special check for leafnode remotes changes which are not supported right now.
 			leafRemotesChanged := func(a, b LeafNodeOpts) bool {
 				if len(tmpOld.Remotes) != len(tmpNew.Remotes) {
@@ -973,6 +1009,20 @@ func (s *Server) diffOptions(newOpts *Options) ([]option, error) {
 				return nil, fmt.Errorf("config reload not supported for %s: old=%v, new=%v",
 					field.Name, oldValue, newValue)
 			}
+		case "mqtt":
+			diffOpts = append(diffOpts, &mqttAckWaitReload{newValue: newValue.(MQTTOpts).AckWait})
+			diffOpts = append(diffOpts, &mqttMaxAckPendingReload{newValue: newValue.(MQTTOpts).MaxAckPending})
+			// Nil out/set to 0 the options that we allow to be reloaded so that
+			// we only fail reload if some that we don't support are changed.
+			tmpOld := oldValue.(MQTTOpts)
+			tmpNew := newValue.(MQTTOpts)
+			tmpOld.TLSConfig, tmpOld.AckWait, tmpOld.MaxAckPending = nil, 0, 0
+			tmpNew.TLSConfig, tmpNew.AckWait, tmpNew.MaxAckPending = nil, 0, 0
+			if !reflect.DeepEqual(tmpOld, tmpNew) {
+				// See TODO(ik) note below about printing old/new values.
+				return nil, fmt.Errorf("config reload not supported for %s: old=%v, new=%v",
+					field.Name, oldValue, newValue)
+			}
 		case "connecterrorreports":
 			diffOpts = append(diffOpts, &connectErrorReports{newValue: newValue.(int)})
 		case "reconnecterrorreports":
@@ -1008,6 +1058,37 @@ func (s *Server) diffOptions(newOpts *Options) ([]option, error) {
 	return diffOpts, nil
 }
 
+func copyRemoteGWConfigsWithoutTLSConfig(current []*RemoteGatewayOpts) []*RemoteGatewayOpts {
+	l := len(current)
+	if l == 0 {
+		return nil
+	}
+	rgws := make([]*RemoteGatewayOpts, 0, l)
+	for _, rcfg := range current {
+		cp := *rcfg
+		cp.TLSConfig = nil
+		rgws = append(rgws, &cp)
+	}
+	return rgws
+}
+
+func copyRemoteLNConfigWithoutTLSConfig(current []*RemoteLeafOpts) []*RemoteLeafOpts {
+	l := len(current)
+	if l == 0 {
+		return nil
+	}
+	rlns := make([]*RemoteLeafOpts, 0, l)
+	for _, rcfg := range current {
+		cp := *rcfg
+		cp.TLSConfig = nil
+		// This is set only when processing a CONNECT, so reset here so that we
+		// don't fail the DeepEqual comparison.
+		cp.TLS = false
+		rlns = append(rlns, &cp)
+	}
+	return rlns
+}
+
 func (s *Server) applyOptions(ctx *reloadContext, opts []option) {
 	var (
 		reloadLogging      = false
@@ -1043,6 +1124,16 @@ func (s *Server) applyOptions(ctx *reloadContext, opts []option) {
 	if reloadClusterPerms {
 		s.reloadClusterPermissions(ctx.oldClusterPerms)
 	}
+	// For remote gateways and leafnodes, make sure that their TLS configuration
+	// is updated (since the config is "captured" early and changes would otherwise
+	// not be visible).
+	newOpts := s.getOpts()
+	if s.gateway.enabled {
+		s.gateway.updateRemotesTLSConfig(newOpts)
+	}
+	if len(newOpts.LeafNode.Remotes) > 0 {
+		s.updateRemoteLeafNodesTLSConfig(newOpts)
+	}
 
 	s.Noticef("Reloaded server configuration")
 }
@@ -1172,7 +1263,9 @@ func (s *Server) reloadAuthorization() {
 			s.accounts.Store(s.sys.account.Name, s.sys.account)
 		}
 		// Double check any JetStream configs.
-		checkJetStream = true
+		// For now, seems like JS as a whole cannot be enabled/disabled with
+		// config reload, so check only if was started with JS enabled.
+		checkJetStream = s.js != nil
 	} else if s.opts.AccountResolver != nil {
 		s.configureResolver()
 		if _, ok := s.accResolver.(*MemAccResolver); ok {
@@ -1233,6 +1326,8 @@ func (s *Server) reloadAuthorization() {
 		// can't hold the lock as go routine reading it may be waiting for lock as well
 		resetCh = s.sys.resetCh
 	}
+	// Check that publish retained messages sources are still allowed to publish.
+	s.mqttCheckPubRetainedPerms()
 	s.mu.Unlock()
 
 	if resetCh != nil {
@@ -1269,7 +1364,16 @@ func (s *Server) reloadAuthorization() {
 
 	// We will double check all JetStream configs on a reload.
 	if checkJetStream {
-		s.configAllJetStreamAccounts()
+		var err error
+		s.getJetStream().clearResources()
+		if s.globalAccountOnly() {
+			err = s.GlobalAccount().EnableJetStream(nil)
+		} else {
+			err = s.configAllJetStreamAccounts()
+		}
+		if err != nil {
+			s.Errorf(err.Error())
+		}
 	}
 
 	if res := s.AccountResolver(); res != nil {
diff --git a/vendor/github.com/nats-io/nats-server/v2/server/route.go b/vendor/github.com/nats-io/nats-server/v2/server/route.go
index 5852721..9ad5242 100644
--- a/vendor/github.com/nats-io/nats-server/v2/server/route.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/route.go
@@ -726,7 +726,7 @@ func (s *Server) sendAsyncInfoToClients(regCli, wsCli bool) {
 		// registered (server has received CONNECT and first PING). For
 		// clients that are not at this stage, this will happen in the
 		// processing of the first PING (see client.processPing)
-		if ((regCli && c.ws == nil) || (wsCli && c.ws != nil)) &&
+		if ((regCli && !c.isWebsocket()) || (wsCli && c.isWebsocket())) &&
 			c.opts.Protocol >= ClientProtoInfo &&
 			c.flags.isSet(firstPongSent) {
 			// sendInfo takes care of checking if the connection is still
@@ -1236,10 +1236,10 @@ func (c *client) sendRouteSubOrUnSubProtos(subs []*subscription, isSubProto, tra
 		if len(sub.origin) > 0 && c.route.lnoc {
 			if isSubProto {
 				buf = append(buf, lSubBytes...)
+				buf = append(buf, sub.origin...)
 			} else {
 				buf = append(buf, lUnsubBytes...)
 			}
-			buf = append(buf, sub.origin...)
 			buf = append(buf, ' ')
 		} else {
 			if isSubProto {
diff --git a/vendor/github.com/nats-io/nats-server/v2/server/server.go b/vendor/github.com/nats-io/nats-server/v2/server/server.go
index 455457a..24b5115 100644
--- a/vendor/github.com/nats-io/nats-server/v2/server/server.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/server.go
@@ -151,6 +151,7 @@ type Server struct {
 		resolver    netResolver
 		dialTimeout time.Duration
 	}
+	leafRemoteCfgs []*leafNodeCfg
 
 	quitCh           chan struct{}
 	shutdownComplete chan struct{}
@@ -222,6 +223,9 @@ type Server struct {
 	// Websocket structure
 	websocket srvWebsocket
 
+	// MQTT structure
+	mqtt srvMQTT
+
 	// exporting account name the importer experienced issues with
 	incompleteAccExporterMap sync.Map
 
@@ -533,6 +537,9 @@ func validateOptions(o *Options) error {
 	if err := validateClusterName(o); err != nil {
 		return err
 	}
+	if err := validateMQTTOptions(o); err != nil {
+		return err
+	}
 	// Finally check websocket options.
 	return validateWebsocketOptions(o)
 }
@@ -1516,6 +1523,11 @@ func (s *Server) Start() {
 		s.startWebsocketServer()
 	}
 
+	// MQTT
+	if opts.MQTT.Port != 0 {
+		s.startMQTT()
+	}
+
 	// Start up routing as well if needed.
 	if opts.Cluster.Port != 0 {
 		s.startGoRoutine(func() {
@@ -1611,6 +1623,13 @@ func (s *Server) Shutdown() {
 		s.websocket.listener = nil
 	}
 
+	// Kick MQTT accept loop
+	if s.mqtt.listener != nil {
+		doneExpected++
+		s.mqtt.listener.Close()
+		s.mqtt.listener = nil
+	}
+
 	// Kick leafnodes AcceptLoop()
 	if s.leafNodeListener != nil {
 		doneExpected++
@@ -1747,7 +1766,7 @@ func (s *Server) AcceptLoop(clr chan struct{}) {
 	s.clientConnectURLs = s.getClientConnectURLs()
 	s.listener = l
 
-	go s.acceptConnections(l, "Client", func(conn net.Conn) { s.createClient(conn, nil) },
+	go s.acceptConnections(l, "Client", func(conn net.Conn) { s.createClient(conn) },
 		func(_ error) bool {
 			if s.isLameDuckMode() {
 				// Signal that we are not accepting new clients
@@ -2073,7 +2092,7 @@ func (c *tlsMixConn) Read(b []byte) (int, error) {
 	return c.Conn.Read(b)
 }
 
-func (s *Server) createClient(conn net.Conn, ws *websocket) *client {
+func (s *Server) createClient(conn net.Conn) *client {
 	// Snapshot server options.
 	opts := s.getOpts()
 
@@ -2085,19 +2104,16 @@ func (s *Server) createClient(conn net.Conn, ws *websocket) *client {
 	}
 	now := time.Now()
 
-	c := &client{srv: s, nc: conn, opts: defaultOpts, mpay: maxPay, msubs: maxSubs, start: now, last: now, ws: ws}
+	c := &client{srv: s, nc: conn, opts: defaultOpts, mpay: maxPay, msubs: maxSubs, start: now, last: now}
 
 	c.registerWithAccount(s.globalAccount())
 
-	// Grab JSON info string
+	var info Info
+	var authRequired bool
+
 	s.mu.Lock()
-	info := s.copyInfo()
-	// If this is a websocket client and there is no top-level auth specified,
-	// then we use the websocket's specific boolean that will be set to true
-	// if there is any auth{} configured in websocket{}.
-	if ws != nil && !info.AuthRequired {
-		info.AuthRequired = s.websocket.authOverride
-	}
+	// Grab JSON info string
+	info = s.copyInfo()
 	if s.nonceRequired() {
 		// Nonce handling
 		var raw [nonceLen]byte
@@ -2106,12 +2122,14 @@ func (s *Server) createClient(conn net.Conn, ws *websocket) *client {
 		info.Nonce = string(nonce)
 	}
 	c.nonce = []byte(info.Nonce)
+	authRequired = info.AuthRequired
+
 	s.totalClients++
 	s.mu.Unlock()
 
 	// Grab lock
 	c.mu.Lock()
-	if info.AuthRequired {
+	if authRequired {
 		c.flags.set(expectConnect)
 	}
 
@@ -2155,6 +2173,8 @@ func (s *Server) createClient(conn net.Conn, ws *websocket) *client {
 		return nil
 	}
 	s.clients[c.cid] = c
+
+	tlsRequired := info.TLSRequired
 	s.mu.Unlock()
 
 	// Re-Grab lock
@@ -2163,7 +2183,6 @@ func (s *Server) createClient(conn net.Conn, ws *websocket) *client {
 	// Connection could have been closed while sending the INFO proto.
 	isClosed := c.isClosed()
 
-	tlsRequired := ws == nil && info.TLSRequired
 	var pre []byte
 	// If we have both TLS and non-TLS allowed we need to see which
 	// one the client wants.
@@ -2231,16 +2250,8 @@ func (s *Server) createClient(conn net.Conn, ws *websocket) *client {
 	// Check for Auth. We schedule this timer after the TLS handshake to avoid
 	// the race where the timer fires during the handshake and causes the
 	// server to write bad data to the socket. See issue #432.
-	if info.AuthRequired {
-		timeout := opts.AuthTimeout
-		// For websocket, possibly override only if set. We make sure that
-		// opts.AuthTimeout is set to a default value if not configured,
-		// but we don't do the same for websocket's one so that we know
-		// if user has explicitly set or not.
-		if ws != nil && opts.Websocket.AuthTimeout != 0 {
-			timeout = opts.Websocket.AuthTimeout
-		}
-		c.setAuthTimer(secondsToDuration(timeout))
+	if authRequired {
+		c.setAuthTimer(secondsToDuration(opts.AuthTimeout))
 	}
 
 	// Do final client initialization
@@ -2421,7 +2432,11 @@ func (s *Server) removeClient(c *client) {
 		if updateProtoInfoCount {
 			s.cproto--
 		}
+		mqtt := c.isMqtt()
 		s.mu.Unlock()
+		if mqtt {
+			s.mqttHandleClosedClient(c)
+		}
 	case ROUTER:
 		s.removeRoute(c)
 	case GATEWAY:
diff --git a/vendor/github.com/nats-io/nats-server/v2/server/store.go b/vendor/github.com/nats-io/nats-server/v2/server/store.go
index bfa41bc..943c6a5 100644
--- a/vendor/github.com/nats-io/nats-server/v2/server/store.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/store.go
@@ -66,7 +66,8 @@ type StreamStore interface {
 	LoadMsg(seq uint64) (subj string, hdr, msg []byte, ts int64, err error)
 	RemoveMsg(seq uint64) (bool, error)
 	EraseMsg(seq uint64) (bool, error)
-	Purge() uint64
+	Purge() (uint64, error)
+	Compact(seq uint64) (uint64, error)
 	GetSeqFromTime(t time.Time) uint64
 	State() StreamState
 	RegisterStorageUpdates(StorageUpdateHandler)
diff --git a/vendor/github.com/nats-io/nats-server/v2/server/stream.go b/vendor/github.com/nats-io/nats-server/v2/server/stream.go
index 014b640..fa9ab96 100644
--- a/vendor/github.com/nats-io/nats-server/v2/server/stream.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/stream.go
@@ -18,6 +18,7 @@ import (
 	"bytes"
 	"compress/gzip"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"io/ioutil"
@@ -50,6 +51,12 @@ type StreamConfig struct {
 	NoAck        bool            `json:"no_ack,omitempty"`
 	Template     string          `json:"template_owner,omitempty"`
 	Duplicates   time.Duration   `json:"duplicate_window,omitempty"`
+
+	// These are non public configuration options.
+	// If you add new options, check fileStreamInfoJSON in order for them to
+	// be properly persisted/recovered, if needed.
+	internal       bool
+	allowNoSubject bool
 }
 
 const JSApiPubAckResponseType = "io.nats.jetstream.api.v1.pub_ack_response"
@@ -466,7 +473,9 @@ func checkStreamCfg(config *StreamConfig) (StreamConfig, error) {
 	}
 
 	if len(cfg.Subjects) == 0 {
-		cfg.Subjects = append(cfg.Subjects, cfg.Name)
+		if !cfg.allowNoSubject {
+			cfg.Subjects = append(cfg.Subjects, cfg.Name)
+		}
 	} else {
 		// We can allow overlaps, but don't allow direct duplicates.
 		dset := make(map[string]struct{}, len(cfg.Subjects))
@@ -610,11 +619,11 @@ func (mset *Stream) Update(config *StreamConfig) error {
 }
 
 // Purge will remove all messages from the stream and underlying store.
-func (mset *Stream) Purge() uint64 {
+func (mset *Stream) Purge() (uint64, error) {
 	mset.mu.Lock()
 	if mset.client == nil {
 		mset.mu.Unlock()
-		return 0
+		return 0, errors.New("stream closed")
 	}
 	// Purge dedupe.
 	mset.ddmap = nil
@@ -625,12 +634,15 @@ func (mset *Stream) Purge() uint64 {
 	}
 	mset.mu.Unlock()
 
-	purged := mset.store.Purge()
+	purged, err := mset.store.Purge()
+	if err != nil {
+		return purged, err
+	}
 	stats := mset.store.State()
 	for _, o := range obs {
 		o.purge(stats.FirstSeq)
 	}
-	return purged
+	return purged, nil
 }
 
 // RemoveMsg will remove a message from a stream.
@@ -775,11 +787,11 @@ func (mset *Stream) setupStore(fsCfg *FileStoreConfig) error {
 // Called for any updates to the underlying stream. We pass through the bytes to the
 // jetstream account. We do local processing for stream pending for consumers, but only
 // for removals.
-// Lock should not ne held.
+// Lock should not be held.
 func (mset *Stream) storeUpdates(md, bd int64, seq uint64, subj string) {
 	// If we have a single negative update then we will process our consumers for stream pending.
 	// Purge and Store handled separately inside individual calls.
-	if md == -1 {
+	if md == -1 && seq > 0 {
 		mset.mu.RLock()
 		for _, o := range mset.consumers {
 			o.decStreamPending(seq, subj)
@@ -940,6 +952,7 @@ func (mset *Stream) processInboundJetStreamMsg(_ *subscription, pc *client, subj
 		if sname := getExpectedStream(hdr); sname != _EMPTY_ && sname != name {
 			mset.mu.Unlock()
 			if doAck && len(reply) > 0 {
+				resp.PubAck = &PubAck{Stream: name}
 				resp.Error = &ApiError{Code: 400, Description: "expected stream does not match"}
 				b, _ := json.Marshal(resp)
 				sendq <- &jsPubMsg{reply, _EMPTY_, _EMPTY_, nil, b, nil, 0}
@@ -951,6 +964,7 @@ func (mset *Stream) processInboundJetStreamMsg(_ *subscription, pc *client, subj
 			lseq := mset.lseq
 			mset.mu.Unlock()
 			if doAck && len(reply) > 0 {
+				resp.PubAck = &PubAck{Stream: name}
 				resp.Error = &ApiError{Code: 400, Description: fmt.Sprintf("wrong last sequence: %d", lseq)}
 				b, _ := json.Marshal(resp)
 				sendq <- &jsPubMsg{reply, _EMPTY_, _EMPTY_, nil, b, nil, 0}
@@ -962,6 +976,7 @@ func (mset *Stream) processInboundJetStreamMsg(_ *subscription, pc *client, subj
 			last := mset.lmsgId
 			mset.mu.Unlock()
 			if doAck && len(reply) > 0 {
+				resp.PubAck = &PubAck{Stream: name}
 				resp.Error = &ApiError{Code: 400, Description: fmt.Sprintf("wrong last msg ID: %s", last)}
 				b, _ := json.Marshal(resp)
 				sendq <- &jsPubMsg{reply, _EMPTY_, _EMPTY_, nil, b, nil, 0}
@@ -990,6 +1005,7 @@ func (mset *Stream) processInboundJetStreamMsg(_ *subscription, pc *client, subj
 	if maxMsgSize >= 0 && len(msg) > maxMsgSize {
 		mset.mu.Unlock()
 		if doAck && len(reply) > 0 {
+			resp.PubAck = &PubAck{Stream: name}
 			resp.Error = &ApiError{Code: 400, Description: "message size exceeds maximum allowed"}
 			b, _ := json.Marshal(resp)
 			mset.sendq <- &jsPubMsg{reply, _EMPTY_, _EMPTY_, nil, b, nil, 0}
@@ -1039,26 +1055,39 @@ func (mset *Stream) processInboundJetStreamMsg(_ *subscription, pc *client, subj
 		hdr = msg[:pc.pa.hdr]
 		msg = msg[pc.pa.hdr:]
 	}
-	seq, ts, err = store.StoreMsg(subject, hdr, msg)
-	if err == nil && seq > 0 {
-		mset.lseq = seq
-		mset.lmsgId = msgId
-	}
+
+	// Assume this will succeed.
+	olseq, olmsgId := mset.lseq, mset.lmsgId
+	mset.lseq++
+	mset.lmsgId = msgId
 
 	// We hold the lock to this point to make sure nothing gets between us since we check for pre-conditions.
 	mset.mu.Unlock()
 
+	// Store actual msg.
+	seq, ts, err = store.StoreMsg(subject, hdr, msg)
+
+	// If we did not succeed put those values back.
+	if err != nil || seq == 0 {
+		mset.mu.Lock()
+		mset.lseq = olseq
+		mset.lmsgId = olmsgId
+		mset.mu.Unlock()
+	}
+
 	if err != nil {
 		if err != ErrStoreClosed {
 			c.Errorf("JetStream failed to store a msg on account: %q stream: %q -  %v", accName, name, err)
 		}
 		if doAck && len(reply) > 0 {
+			resp.PubAck = &PubAck{Stream: name}
 			resp.Error = &ApiError{Code: 400, Description: err.Error()}
 			response, _ = json.Marshal(resp)
 		}
 	} else if jsa.limitsExceeded(stype) {
 		c.Warnf("JetStream resource limits exceeded for account: %q", accName)
 		if doAck && len(reply) > 0 {
+			resp.PubAck = &PubAck{Stream: name}
 			resp.Error = &ApiError{Code: 400, Description: "resource limits exceeded for account"}
 			response, _ = json.Marshal(resp)
 		}
@@ -1142,16 +1171,16 @@ func (mset *Stream) Name() string {
 }
 
 func (mset *Stream) internalSendLoop() {
-	mset.mu.Lock()
+	mset.mu.RLock()
 	c := mset.client
 	if c == nil {
-		mset.mu.Unlock()
+		mset.mu.RUnlock()
 		return
 	}
 	s := c.srv
 	sendq := mset.sendq
 	name := mset.config.Name
-	mset.mu.Unlock()
+	mset.mu.RUnlock()
 
 	// Warn when internal send queue is backed up past 75%
 	warnThresh := 3 * msetSendQSize / 4
diff --git a/vendor/github.com/nats-io/nats-server/v2/server/sublist.go b/vendor/github.com/nats-io/nats-server/v2/server/sublist.go
index 1c860a8..c5785e6 100644
--- a/vendor/github.com/nats-io/nats-server/v2/server/sublist.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/sublist.go
@@ -1101,6 +1101,9 @@ func tokensCanMatch(t1, t2 string) bool {
 
 // SubjectsCollide will determine if two subjects could both match a single literal subject.
 func SubjectsCollide(subj1, subj2 string) bool {
+	if subj1 == subj2 {
+		return true
+	}
 	toks1 := strings.Split(subj1, tsep)
 	toks2 := strings.Split(subj2, tsep)
 	pwc1, fwc1 := analyzeTokens(toks1)
@@ -1392,3 +1395,72 @@ func (s *Sublist) collectAllSubs(l *level, subs *[]*subscription) {
 		s.collectAllSubs(l.fwc.next, subs)
 	}
 }
+
+// For a given subject (which may contain wildcards), this call returns all
+// subscriptions that would match that subject. For instance, suppose that
+// the sublist contains: foo.bar, foo.bar.baz and foo.baz, ReverseMatch("foo.*")
+// would return foo.bar and foo.baz.
+// This is used in situations where the sublist is likely to contain only
+// literals and one wants to get all the subjects that would have been a match
+// to a subscription on `subject`.
+func (s *Sublist) ReverseMatch(subject string) *SublistResult {
+	tsa := [32]string{}
+	tokens := tsa[:0]
+	start := 0
+	for i := 0; i < len(subject); i++ {
+		if subject[i] == btsep {
+			tokens = append(tokens, subject[start:i])
+			start = i + 1
+		}
+	}
+	tokens = append(tokens, subject[start:])
+
+	result := &SublistResult{}
+
+	s.Lock()
+	reverseMatchLevel(s.root, tokens, nil, result)
+	// Check for empty result.
+	if len(result.psubs) == 0 && len(result.qsubs) == 0 {
+		result = emptyResult
+	}
+	s.Unlock()
+
+	return result
+}
+
+func reverseMatchLevel(l *level, toks []string, n *node, results *SublistResult) {
+	for i, t := range toks {
+		if l == nil {
+			return
+		}
+		if len(t) == 1 {
+			if t[0] == fwc {
+				getAllNodes(l, results)
+				return
+			} else if t[0] == pwc {
+				for _, n := range l.nodes {
+					reverseMatchLevel(n.next, toks[i+1:], n, results)
+				}
+				return
+			}
+		}
+		n = l.nodes[t]
+		if n == nil {
+			break
+		}
+		l = n.next
+	}
+	if n != nil {
+		addNodeToResults(n, results)
+	}
+}
+
+func getAllNodes(l *level, results *SublistResult) {
+	if l == nil {
+		return
+	}
+	for _, n := range l.nodes {
+		addNodeToResults(n, results)
+		getAllNodes(n.next, results)
+	}
+}
diff --git a/vendor/github.com/nats-io/nats-server/v2/server/websocket.go b/vendor/github.com/nats-io/nats-server/v2/server/websocket.go
index 7b10877..901b401 100644
--- a/vendor/github.com/nats-io/nats-server/v2/server/websocket.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/websocket.go
@@ -152,6 +152,12 @@ func wsGet(r io.Reader, buf []byte, pos, needed int) ([]byte, int, error) {
 	return b, pos + avail, nil
 }
 
+// Returns true if this connection is from a Websocket client.
+// Lock held on entry.
+func (c *client) isWebsocket() bool {
+	return c.ws != nil
+}
+
 // Returns a slice of byte slices corresponding to payload of websocket frames.
 // The byte slice `buf` is filled with bytes from the connection's read loop.
 // This function will decode the frame headers and unmask the payload(s).
@@ -856,6 +862,7 @@ func (s *Server) startWebsocketServer() {
 	if port == 0 {
 		s.opts.Websocket.Port = hl.Addr().(*net.TCPAddr).Port
 	}
+	s.Noticef("Listening for websocket clients on %s://%s:%d", proto, o.Host, port)
 	s.websocket.connectURLs, err = s.getConnectURLs(o.Advertise, o.Host, o.Port)
 	if err != nil {
 		s.Fatalf("Unable to get websocket connect URLs: %v", err)
@@ -870,7 +877,7 @@ func (s *Server) startWebsocketServer() {
 			s.Errorf(err.Error())
 			return
 		}
-		s.createClient(res.conn, res.ws)
+		s.createWSClient(res.conn, res.ws)
 	})
 	hs := &http.Server{
 		Addr:        hp,
@@ -896,6 +903,102 @@ func (s *Server) startWebsocketServer() {
 	s.mu.Unlock()
 }
 
+// This is similar to createClient() but has some modifications
+// specific to handle websocket clients.
+// The comments have been kept to minimum to reduce code size.
+// Check createClient() for more details.
+func (s *Server) createWSClient(conn net.Conn, ws *websocket) *client {
+	opts := s.getOpts()
+
+	maxPay := int32(opts.MaxPayload)
+	maxSubs := int32(opts.MaxSubs)
+	if maxSubs == 0 {
+		maxSubs = -1
+	}
+	now := time.Now()
+
+	c := &client{srv: s, nc: conn, opts: defaultOpts, mpay: maxPay, msubs: maxSubs, start: now, last: now, ws: ws}
+
+	c.registerWithAccount(s.globalAccount())
+
+	var info Info
+	var authRequired bool
+
+	s.mu.Lock()
+	info = s.copyInfo()
+	// Check auth, override if applicable.
+	if !info.AuthRequired {
+		// Set info.AuthRequired since this is what is sent to the client.
+		info.AuthRequired = s.websocket.authOverride
+	}
+	if s.nonceRequired() {
+		var raw [nonceLen]byte
+		nonce := raw[:]
+		s.generateNonce(nonce)
+		info.Nonce = string(nonce)
+	}
+	c.nonce = []byte(info.Nonce)
+	authRequired = info.AuthRequired
+
+	s.totalClients++
+	s.mu.Unlock()
+
+	c.mu.Lock()
+	if authRequired {
+		c.flags.set(expectConnect)
+	}
+	c.initClient()
+	c.Debugf("Client connection created")
+	c.sendProtoNow(c.generateClientInfoJSON(info))
+	c.mu.Unlock()
+
+	s.mu.Lock()
+	if !s.running || s.ldm {
+		if s.shutdown {
+			conn.Close()
+		}
+		s.mu.Unlock()
+		return c
+	}
+
+	if opts.MaxConn > 0 && len(s.clients) >= opts.MaxConn {
+		s.mu.Unlock()
+		c.maxConnExceeded()
+		return nil
+	}
+	s.clients[c.cid] = c
+
+	// Websocket clients do TLS in the websocket http server.
+	// So no TLS here...
+	s.mu.Unlock()
+
+	c.mu.Lock()
+
+	if c.isClosed() {
+		c.mu.Unlock()
+		c.closeConnection(WriteError)
+		return nil
+	}
+
+	if authRequired {
+		timeout := opts.AuthTimeout
+		// Possibly override with Websocket specific value.
+		if opts.Websocket.AuthTimeout != 0 {
+			timeout = opts.Websocket.AuthTimeout
+		}
+		c.setAuthTimer(secondsToDuration(timeout))
+	}
+
+	c.setPingTimer()
+
+	s.startGoRoutine(func() { c.readLoop(nil) })
+	s.startGoRoutine(func() { c.writeLoop() })
+
+	c.mu.Unlock()
+
+	return c
+}
+
 type wsCaptureHTTPServerLog struct {
 	s *Server
 }
diff --git a/vendor/modules.txt b/vendor/modules.txt
index f65aa1a..e26c677 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -72,10 +72,10 @@ github.com/nats-io/cliprompts/v2
 # github.com/nats-io/jwt v1.2.2
 ## explicit
 github.com/nats-io/jwt
-# github.com/nats-io/jwt/v2 v2.0.0-20201207222029-6d7e74b660de
+# github.com/nats-io/jwt/v2 v2.0.0-20201211164018-2e78446f4e6f
 ## explicit
 github.com/nats-io/jwt/v2
-# github.com/nats-io/nats-server/v2 v2.1.8-0.20201125160748-de0c992ca6db
+# github.com/nats-io/nats-server/v2 v2.1.8-0.20201216194404-d7741b9aa16a
 ## explicit
 github.com/nats-io/nats-server/v2/conf
 github.com/nats-io/nats-server/v2/internal/ldap


Test patch:
diff --git a/cmd/addexport_test.go b/cmd/addexport_test.go
index af5cbe1..d1aa200 100644
--- a/cmd/addexport_test.go
+++ b/cmd/addexport_test.go
@@ -189,7 +189,7 @@ func TestAddServiceLatency(t *testing.T) {
 	require.Equal(t, "q", string(ac.Exports[0].Subject))
 	require.NotNil(t, ac.Exports[0].Latency)
 	require.Equal(t, "q.lat", string(ac.Exports[0].Latency.Results))
-	require.Equal(t, 100, ac.Exports[0].Latency.Sampling)
+	require.Equal(t, jwt.SamplingRate(100), ac.Exports[0].Latency.Sampling)
 	require.EqualValues(t, jwt.ResponseTypeStream, ac.Exports[0].ResponseType)
 }
 
@@ -223,6 +223,6 @@ func TestAddServiceLatencyInteractive(t *testing.T) {
 	require.Equal(t, "q", string(ac.Exports[0].Subject))
 	require.NotNil(t, ac.Exports[0].Latency)
 	require.Equal(t, "q.lat", string(ac.Exports[0].Latency.Results))
-	require.Equal(t, 100, ac.Exports[0].Latency.Sampling)
+	require.Equal(t, jwt.SamplingRate(100), ac.Exports[0].Latency.Sampling)
 	require.EqualValues(t, jwt.ResponseTypeStream, ac.Exports[0].ResponseType)
 }
diff --git a/cmd/editexport_test.go b/cmd/editexport_test.go
index 5818c07..2e5b9cd 100644
--- a/cmd/editexport_test.go
+++ b/cmd/editexport_test.go
@@ -54,7 +54,7 @@ func Test_EditExport_Latency(t *testing.T) {
 	ac, err = ts.Store.ReadAccountClaim("A")
 	require.NoError(t, err)
 	require.NotNil(t, ac.Exports[0].Latency)
-	require.Equal(t, 100, ac.Exports[0].Latency.Sampling)
+	require.Equal(t, jwt.SamplingRate(100), ac.Exports[0].Latency.Sampling)
 	require.Equal(t, jwt.Subject("lat"), ac.Exports[0].Latency.Results)
 }
 
@@ -94,7 +94,7 @@ func Test_EditExportInteractiveLatency(t *testing.T) {
 	require.Equal(t, jwt.Subject("c"), ac.Exports[0].Subject)
 	require.Equal(t, "c", ac.Exports[0].Name)
 	require.NotNil(t, ac.Exports[0].Latency)
-	require.Equal(t, 100, ac.Exports[0].Latency.Sampling)
+	require.Equal(t, jwt.SamplingRate(100), ac.Exports[0].Latency.Sampling)
 	require.Equal(t, jwt.Subject("lat"), ac.Exports[0].Latency.Results)
 	require.EqualValues(t, jwt.ResponseTypeChunked, ac.Exports[0].ResponseType)
 }