mqtt.go

// Copyright 2020-2024 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"
"cmp"
"crypto/tls"
"encoding/binary"
"encoding/json"
"errors"
"fmt"
"io"
"net"
"net/http"
"slices"
"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(0x1<<1)
mqttPubQoS2=byte(0x2<<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)
mqttConnAckRCQoS2WillRejected=byte(0x10)

// Maximum payload size of a control packet
mqttMaxPayloadSize=0xFFFFFFF

// Topic/Filter characters
mqttTopicLevelSep='/'
mqttSingleLevelWC='+'
mqttMultiLevelWC='#'
mqttReservedPre='$'

// 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."

// Stream name for MQTT messages on a given account
mqttStreamName="$MQTT_msgs"
mqttStreamSubjectPrefix="$MQTT.msgs."

// Stream name for MQTT retained messages on a given account
mqttRetainedMsgsStreamName="$MQTT_rmsgs"
mqttRetainedMsgsStreamSubject="$MQTT.rmsgs."

// Stream name for MQTT sessions on a given account
mqttSessStreamName="$MQTT_sess"
mqttSessStreamSubjectPrefix="$MQTT.sess."

// Stream name prefix for MQTT sessions on a given account
mqttSessionsStreamNamePrefix="$MQTT_sess_"

// Stream name and subject for incoming MQTT QoS2 messages
mqttQoS2IncomingMsgsStreamName="$MQTT_qos2in"
mqttQoS2IncomingMsgsStreamSubjectPrefix="$MQTT.qos2.in."

// Stream name and subjects for outgoing MQTT QoS (PUBREL) messages
mqttOutStreamName="$MQTT_out"
mqttOutSubjectPrefix="$MQTT.out."
mqttPubRelSubjectPrefix="$MQTT.out.pubrel."
mqttPubRelDeliverySubjectPrefix="$MQTT.deliver.pubrel."
mqttPubRelConsumerDurablePrefix="$MQTT_PUBREL_"

// As per spec, MQTT server may not redeliver QoS 1 and 2 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

// A session's list of subscriptions cannot have a cumulative MaxAckPending
// of more than this limit.
mqttMaxAckTotalLimit=0xFFFF

// Prefix of the reply subject for JS API requests.
mqttJSARepliesPrefix="$MQTT.JSA."

// Those are tokens that are used for the reply subject of JS API requests.
// For instance "$MQTT.JSA.<node id>.SC.<number>" is the reply subject
// for a request to create a stream (where <node id> is the server name hash),
// while "$MQTT.JSA.<node id>.SL.<number>" is for a stream lookup, etc...
mqttJSAIdTokenPos=3
mqttJSATokenPos=4
mqttJSAClientIDPos=5
mqttJSAStreamCreate="SC"
mqttJSAStreamUpdate="SU"
mqttJSAStreamLookup="SL"
mqttJSAStreamDel="SD"
mqttJSAConsumerCreate="CC"
mqttJSAConsumerLookup="CL"
mqttJSAConsumerDel="CD"
mqttJSAMsgStore="MS"
mqttJSAMsgLoad="ML"
mqttJSAMsgDelete="MD"
mqttJSASessPersist="SP"
mqttJSARetainedMsgDel="RD"
mqttJSAStreamNames="SN"

// This is how long to keep a client in the flappers map before closing the
// connection. This prevent quick reconnect from those clients that keep
// wanting to connect with a client ID already in use.
mqttSessFlappingJailDur=time.Second

// This is how frequently the timer to cleanup the sessions flappers map is firing.
mqttSessFlappingCleanupInterval=5*time.Second

// Default retry delay if transfer of old session streams to new one fails
mqttDefaultTransferRetry=5*time.Second

// For Websocket URLs
mqttWSPath="/mqtt"

mqttInitialPubHeader=16// An overkill, should need 7 bytes max
mqttProcessSubTooLong=100*time.Millisecond
mqttDefaultRetainedCacheTTL=2*time.Minute
mqttRetainedTransferTimeout=10*time.Second
)

const (
sparkbNBIRTH="NBIRTH"
sparkbDBIRTH="DBIRTH"
sparkbNDEATH="NDEATH"
sparkbDDEATH="DDEATH"
)

var (
sparkbNamespaceTopicPrefix= []byte("spBv1.0/")
sparkbCertificatesTopicPrefix= []byte("$sparkplug/certificates/")
)

var (
mqttPingResponse= []byte{mqttPacketPingResp, 0x0}
mqttProtoName= []byte("MQTT")
mqttOldProtoName= []byte("MQIsdp")
mqttSessJailDur=mqttSessFlappingJailDur
mqttFlapCleanItvl=mqttSessFlappingCleanupInterval
mqttJSAPITimeout=4*time.Second
mqttRetainedCacheTTL=mqttDefaultRetainedCacheTTL
)

var (
errMQTTNotWebsocketPort=errors.New("MQTT clients over websocket must connect to the Websocket port, not the MQTT port")
errMQTTTopicFilterCannotBeEmpty=errors.New("topic filter cannot be empty")
errMQTTMalformedVarInt=errors.New("malformed variable int")
errMQTTSecondConnectPacket=errors.New("received a second CONNECT packet")
errMQTTServerNameMustBeSet=errors.New("mqtt requires server name to be explicitly set")
errMQTTUserMixWithUsersNKeys=errors.New("mqtt authentication username not compatible with presence of users/nkeys")
errMQTTTokenMixWIthUsersNKeys=errors.New("mqtt authentication token not compatible with presence of users/nkeys")
errMQTTAckWaitMustBePositive=errors.New("ack wait must be a positive value")
errMQTTStandaloneNeedsJetStream=errors.New("mqtt requires JetStream to be enabled if running in standalone mode")
errMQTTConnFlagReserved=errors.New("connect flags reserved bit not set to 0")
errMQTTWillAndRetainFlag=errors.New("if Will flag is set to 0, Will Retain flag must be 0 too")
errMQTTPasswordFlagAndNoUser=errors.New("password flag set but username flag is not")
errMQTTCIDEmptyNeedsCleanFlag=errors.New("when client ID is empty, clean session flag must be set to 1")
errMQTTEmptyWillTopic=errors.New("empty Will topic not allowed")
errMQTTEmptyUsername=errors.New("empty user name not allowed")
errMQTTTopicIsEmpty=errors.New("topic cannot be empty")
errMQTTPacketIdentifierIsZero=errors.New("packet identifier cannot be 0")
errMQTTUnsupportedCharacters=errors.New("character ' ' not supported for MQTT topics")
errMQTTInvalidSession=errors.New("invalid MQTT session")
)

typesrvMQTTstruct {
listener net.Listener
listenerErrerror
authOverridebool
sessmgrmqttSessionManager
}

typemqttSessionManagerstruct {
mu sync.RWMutex
sessionsmap[string]*mqttAccountSessionManager// key is account name
}

vartestDisableRMSCache=false

typemqttAccountSessionManagerstruct {
mu sync.RWMutex
sessionsmap[string]*mqttSession// key is MQTT client ID
sessByHashmap[string]*mqttSession// key is MQTT client ID hash
sessLockedmap[string]struct{} // key is MQTT client ID and indicate that a session can not be taken by a new client at this time
flappersmap[string]int64// When connection connects with client ID already in use
flapTimer*time.Timer// Timer to perform some cleanup of the flappers map
sl*Sublist// sublist allowing to find retained messages for given subscription
retmsgsmap[string]*mqttRetainedMsgRef// retained messages
rmsCache*sync.Map// map[subject]mqttRetainedMsg
jsamqttJSA
rrmLastSequint64// Restore retained messages expected last sequence
rrmDoneChchanstruct{} // To notify the caller that all retained messages have been loaded
domainTkstring// Domain (with trailing "."), or possibly empty. This is added to session subject.
}

typemqttJSAResponsestruct {
replystring// will be used to map to the original request in jsa.NewRequestExMulti
valueany
}

typemqttJSAstruct {
mu sync.Mutex
idstring
c*client
sendq*ipQueue[*mqttJSPubMsg]
rplyrstring
replies sync.Map// [string]chan *mqttJSAResponse
nuid*nuid.NUID
quitChchanstruct{}
domainstring// Domain or possibly empty. This is added to session subject.
domainSetbool// covers if domain was set, even to empty
}

typemqttJSPubMsgstruct {
subjstring
replystring
hdrint
msg []byte
}

typemqttRetMsgDelstruct {
Subjectstring`json:"subject"`
Sequint64`json:"seq"`
}

typemqttSessionstruct {
// subsMu is a "quick" version of the session lock, sufficient for the QoS0
// callback. It only guarantees that a new subscription is initialized, and
// its retained messages if any have been queued up for delivery. The QoS12
// callback uses the session lock.
mu sync.Mutex
subsMu sync.RWMutex

idstring// client ID
idHashstring// client ID hash
c*client
jsa*mqttJSA
subsmap[string]byte// Key is MQTT SUBSCRIBE filter, value is the subscription QoS
consmap[string]*ConsumerConfig
pubRelConsumer*ConsumerConfig
pubRelSubscribedbool
pubRelDeliverySubjectstring
pubRelDeliverySubjectB []byte
pubRelSubjectstring
sequint64

// pendingPublish maps packet identifiers (PI) to JetStream ACK subjects for
// QoS1 and 2 PUBLISH messages pending delivery to the session's client.
pendingPublishmap[uint16]*mqttPending

// pendingPubRel maps PIs to JetStream ACK subjects for QoS2 PUBREL
// messages pending delivery to the session's client.
pendingPubRelmap[uint16]*mqttPending

// cpending maps delivery attempts (that come with a JS ACK subject) to
// existing PIs.
cpendingmap[string]map[uint64]uint16// composite key: jsDur, sseq

// "Last used" publish packet identifier (PI). starting point searching for the next available.
last_piuint16

// Maximum number of pending acks for this session.
maxpuint16
tmaxackint
cleanbool
domainTkstring
}

typemqttPersistedSessionstruct {
Originstring`json:"origin,omitempty"`
IDstring`json:"id,omitempty"`
Cleanbool`json:"clean,omitempty"`
Subsmap[string]byte`json:"subs,omitempty"`
Consmap[string]*ConsumerConfig`json:"cons,omitempty"`
PubRel*ConsumerConfig`json:"pubrel,omitempty"`
}

typemqttRetainedMsgstruct {
Originstring`json:"origin,omitempty"`
Subjectstring`json:"subject,omitempty"`
Topicstring`json:"topic,omitempty"`
Msg []byte`json:"msg,omitempty"`
Flagsbyte`json:"flags,omitempty"`
Sourcestring`json:"source,omitempty"`

expiresFromCache time.Time
}

typemqttRetainedMsgRefstruct {
ssequint64
flooruint64
sub*subscription
}

// mqttSub contains fields associated with a MQTT subscription, and is added to
// the main subscription struct for MQTT message delivery subscriptions. The
// delivery callbacks may get invoked before sub.mqtt is set up, so they should
// acquire either sess.mu or sess.subsMu before accessing it.
typemqttSubstruct {
// The sub's QOS and the JS durable name. They can change when
// re-subscribing, and are used in the delivery callbacks. They can be
// quickly accessed using sess.subsMu.RLock, or under the main session lock.
qosbyte
jsDurstring

// Pending serialization of retained messages to be sent when subscription
// is registered. The sub's delivery callbacks must wait until `prm` is
// ready (can block on sess.mu for that, too).
prm [][]byte

// If this subscription needs to be checked for being reserved. E.g. '#' or
// '*' or '*/'. It is set up at the time of subscription and is immutable
// after that.
reservedbool
}

typemqttstruct {
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()
cidstring// client ID

// rejectQoS2Pub tells the MQTT client to not accept QoS2 PUBLISH, instead
// error and terminate the connection.
rejectQoS2Pubbool

// downgradeQOS2Sub tells the MQTT client to downgrade QoS2 SUBSCRIBE
// requests to QoS1.
downgradeQoS2Subbool
}

typemqttPendingstruct {
ssequint64// stream sequence
jsAckSubjectstring// the ACK subject to send the ack to
jsDurstring// JS durable name
}

typemqttConnectProtostruct {
rd time.Duration
will*mqttWill
flagsbyte
}

typemqttIOReaderinterface {
 io.Reader
SetReadDeadline(time.Time) error
}

typemqttReaderstruct {
readermqttIOReader
buf []byte
posint
pstartint
pbuf []byte
}

typemqttWriterstruct {
 bytes.Buffer
}

typemqttWillstruct {
topic []byte
subject []byte
mapped []byte
message []byte
qosbyte
retainbool
}

typemqttFilterstruct {
filterstring
qosbyte
// Used only for tracing and should not be used after parsing of (un)sub protocols.
ttopic []byte
}

typemqttPublishstruct {
topic []byte
subject []byte
mapped []byte
msg []byte
szint
piuint16
flagsbyte
}

// When we re-encode incoming MQTT PUBLISH messages for NATS delivery, we add
// the following headers:
// - "Nmqtt-Pub" (*always) indicates that the message originated from MQTT, and
// contains the original message QoS.
// - "Nmqtt-Subject" contains the original MQTT subject from mqttParsePub.
// - "Nmqtt-Mapped" contains the mapping during mqttParsePub.
//
// When we submit a PUBREL for delivery, we add a "Nmqtt-PubRel" header that
// contains the PI.
const (
// NATS header that indicates that the message originated from MQTT and
// stores the published message QOS.
mqttNatsHeader="Nmqtt-Pub"

// NATS headers to store retained message metadata (along with the original
// message as binary).
mqttNatsRetainedMessageTopic="Nmqtt-RTopic"
mqttNatsRetainedMessageOrigin="Nmqtt-ROrigin"
mqttNatsRetainedMessageFlags="Nmqtt-RFlags"
mqttNatsRetainedMessageSource="Nmqtt-RSource"

// NATS header that indicates that the message is an MQTT PubRel and stores
// the PI.
mqttNatsPubRelHeader="Nmqtt-PubRel"

// NATS headers to store the original MQTT subject and the subject mapping.
mqttNatsHeaderSubject="Nmqtt-Subject"
mqttNatsHeaderMapped="Nmqtt-Mapped"
)

typemqttParsedPublishNATSHeaderstruct {
qosbyte
subject []byte
mapped []byte
}

func (s*Server) startMQTT() {
ifs.isShuttingDown() {
return
 }

sopts:=s.getOpts()
o:=&sopts.MQTT

varhl net.Listener
varerrerror

port:=o.Port
ifport==-1 {
port=0
 }
hp:=net.JoinHostPort(o.Host, strconv.Itoa(port))
s.mu.Lock()
s.mqtt.sessmgr.sessions=make(map[string]*mqttAccountSessionManager)
hl, err=net.Listen("tcp", hp)
s.mqtt.listenerErr=err
iferr!=nil {
s.mu.Unlock()
s.Fatalf("Unable to listen for MQTT connections: %v", err)
return
 }
ifport==0 {
o.Port=hl.Addr().(*net.TCPAddr).Port
 }
s.mqtt.listener=hl
scheme:="mqtt"
ifo.TLSConfig!=nil {
scheme="tls"
 }
s.Noticef("Listening for MQTT clients on %s://%s:%d", scheme, o.Host, o.Port)
gos.acceptConnections(hl, "MQTT", func(conn net.Conn) { s.createMQTTClient(conn, nil) }, 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, ws*websocket) *client {
opts:=s.getOpts()

maxPay:=int32(opts.MaxPayload)
maxSubs:=int32(opts.MaxSubs)
ifmaxSubs==0 {
maxSubs=-1
 }
now:=time.Now()

mqtt:=&mqtt{
rejectQoS2Pub: opts.MQTT.rejectQoS2Pub,
downgradeQoS2Sub: opts.MQTT.downgradeQoS2Sub,
 }
c:=&client{srv: s, nc: conn, mpay: maxPay, msubs: maxSubs, start: now, last: now, mqtt: mqtt, ws: ws}
c.headers=true
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()
ifauthRequired {
c.flags.set(expectConnect)
 }
c.initClient()
c.Debugf("Client connection created")
c.mu.Unlock()

s.mu.Lock()
if!s.isRunning() ||s.ldm {
ifs.isShuttingDown() {
conn.Close()
 }
s.mu.Unlock()
returnc
 }

ifopts.MaxConn>0&&len(s.clients) >=opts.MaxConn {
s.mu.Unlock()
c.maxConnExceeded()
returnnil
 }
s.clients[c.cid] =c

// Websocket TLS handshake is already done when getting to this function.
tlsRequired:=opts.MQTT.TLSConfig!=nil&&ws==nil
s.mu.Unlock()

c.mu.Lock()

// In case connection has already been closed
ifc.isClosed() {
c.mu.Unlock()
c.closeConnection(WriteError)
returnnil
 }

varpre []byte
iftlsRequired&&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]
ifn>0&&pre[0] ==0x16 {
tlsRequired=true
 } else {
tlsRequired=false
 }
 }

iftlsRequired {
iflen(pre) >0 {
c.nc=&tlsMixConn{c.nc, bytes.NewBuffer(pre)}
pre=nil
 }

// Perform server-side TLS handshake.
iferr:=c.doTLSServerHandshake(tlsHandshakeMQTT, opts.MQTT.TLSConfig, opts.MQTT.TLSTimeout, opts.MQTT.TLSPinnedCerts); err!=nil {
c.mu.Unlock()
returnnil
 }
 }

ifauthRequired {
timeout:=opts.AuthTimeout
// Possibly override with MQTT specific value.
ifopts.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() })

iftlsRequired {
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()

returnc
}

// 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.
funcvalidateMQTTOptions(o*Options) error {
mo:=&o.MQTT
// If no port is defined, we don't care about other options
ifmo.Port==0 {
returnnil
 }
// We have to force the server name to be explicitly set and be unique when
// in cluster mode.
ifo.ServerName==_EMPTY_&& (o.Cluster.Port!=0||o.Gateway.Port!=0) {
returnerrMQTTServerNameMustBeSet
 }
// If there is a NoAuthUser, we need to have Users defined and
// the user to be present.
ifmo.NoAuthUser!=_EMPTY_ {
iferr:=validateNoAuthUser(o, mo.NoAuthUser); err!=nil {
returnerr
 }
 }
// Token/Username not possible if there are users/nkeys
iflen(o.Users) >0||len(o.Nkeys) >0 {
ifmo.Username!=_EMPTY_ {
returnerrMQTTUserMixWithUsersNKeys
 }
ifmo.Token!=_EMPTY_ {
returnerrMQTTTokenMixWIthUsersNKeys
 }
 }
ifmo.AckWait<0 {
returnerrMQTTAckWaitMustBePositive
 }
// If strictly standalone and there is no JS enabled, then it won't work...
// For leafnodes, we could either have remote(s) and it would be ok, or no
// remote but accept from a remote side that has "hub" property set, which
// then would ok too. So we fail only if we have no leafnode config at all.
if!o.JetStream&&o.Cluster.Port==0&&o.Gateway.Port==0&&
o.LeafNode.Port==0&&len(o.LeafNode.Remotes) ==0 {
returnerrMQTTStandaloneNeedsJetStream
 }
iferr:=validatePinnedCerts(mo.TLSPinnedCerts); err!=nil {
returnfmt.Errorf("mqtt: %v", err)
 }
ifmo.ConsumerReplicas>0&&mo.StreamReplicas>0&&mo.ConsumerReplicas>mo.StreamReplicas {
returnfmt.Errorf("mqtt: consumer_replicas (%v) cannot be higher than stream_replicas (%v)",
mo.ConsumerReplicas, mo.StreamReplicas)
 }
returnnil
}

// Returns true if this connection is from a MQTT client.
// Lock held on entry.
func (c*client) isMqtt() bool {
returnc.mqtt!=nil
}

// If this is an MQTT client, returns the session client ID,
// otherwise returns the empty string.
// Lock held on entry
func (c*client) getMQTTClientID() string {
if!c.isMqtt() {
return_EMPTY_
 }
returnc.mqtt.cid
}

// 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
varrd time.Duration
ifmqtt.cp!=nil {
rd=mqtt.cp.rd
ifrd>0 {
r.reader.SetReadDeadline(time.Time{})
 }
 }
hasMappings:=c.in.flags.isSet(hasMappings)
c.mu.Unlock()

r.reset(buf)

varerrerror
varbbyte
varplint
varcompletebool

forerr==nil&&r.hasMore() {

// Keep track of the starting of the packet, in case we have a partial
r.pstart=r.pos

// Read packet type and flags
ifb, 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 {
// If the buffer indicates that it may be a websocket handshake
// but the client is not websocket, it means that the client
// connected to the MQTT port instead of the Websocket port.
ifbytes.HasPrefix(buf, []byte("GET ")) &&!c.isWebsocket() {
err=errMQTTNotWebsocketPort
 } else {
err=fmt.Errorf("the first packet should be a CONNECT (%v), got %v", mqttPacketConnect, pt)
 }
break
 }

pl, complete, err=r.readPacketLen()
iferr!=nil||!complete {
break
 }

switchpt {
// Packets that we receive back when we act as the "sender": PUBACK,
// PUBREC, PUBCOMP.
casemqttPacketPubAck:
varpiuint16
pi, err=mqttParsePIPacket(r)
iftrace {
c.traceInOp("PUBACK", errOrTrace(err, fmt.Sprintf("pi=%v", pi)))
 }
iferr==nil {
err=c.mqttProcessPubAck(pi)
 }

casemqttPacketPubRec:
varpiuint16
pi, err=mqttParsePIPacket(r)
iftrace {
c.traceInOp("PUBREC", errOrTrace(err, fmt.Sprintf("pi=%v", pi)))
 }
iferr==nil {
err=c.mqttProcessPubRec(pi)
 }

casemqttPacketPubComp:
varpiuint16
pi, err=mqttParsePIPacket(r)
iftrace {
c.traceInOp("PUBCOMP", errOrTrace(err, fmt.Sprintf("pi=%v", pi)))
 }
iferr==nil {
c.mqttProcessPubComp(pi)
 }

// Packets where we act as the "receiver": PUBLISH, PUBREL, SUBSCRIBE, UNSUBSCRIBE.
casemqttPacketPub:
pp:=c.mqtt.pp
pp.flags=b&mqttPacketFlagMask
err=c.mqttParsePub(r, pl, pp, hasMappings)
iftrace {
c.traceInOp("PUBLISH", errOrTrace(err, mqttPubTrace(pp)))
iferr==nil {
c.mqttTraceMsg(pp.msg)
 }
 }
iferr==nil {
err=s.mqttProcessPub(c, pp, trace)
 }

casemqttPacketPubRel:
varpiuint16
pi, err=mqttParsePIPacket(r)
iftrace {
c.traceInOp("PUBREL", errOrTrace(err, fmt.Sprintf("pi=%v", pi)))
 }
iferr==nil {
err=s.mqttProcessPubRel(c, pi, trace)
 }

casemqttPacketSub:
varpiuint16// packet identifier
varfilters []*mqttFilter
varsubs []*subscription
pi, filters, err=c.mqttParseSubs(r, b, pl)
iftrace {
c.traceInOp("SUBSCRIBE", errOrTrace(err, mqttSubscribeTrace(pi, filters)))
 }
iferr==nil {
subs, err=c.mqttProcessSubs(filters)
iferr==nil&&trace {
c.traceOutOp("SUBACK", []byte(fmt.Sprintf("pi=%v", pi)))
 }
 }
iferr==nil {
c.mqttEnqueueSubAck(pi, filters)
c.mqttSendRetainedMsgsToNewSubs(subs)
 }

casemqttPacketUnsub:
varpiuint16// packet identifier
varfilters []*mqttFilter
pi, filters, err=c.mqttParseUnsubs(r, b, pl)
iftrace {
c.traceInOp("UNSUBSCRIBE", errOrTrace(err, mqttUnsubscribeTrace(pi, filters)))
 }
iferr==nil {
err=c.mqttProcessUnsubs(filters)
iferr==nil&&trace {
c.traceOutOp("UNSUBACK", []byte(fmt.Sprintf("pi=%v", pi)))
 }
 }
iferr==nil {
c.mqttEnqueueUnsubAck(pi)
 }

// Packets that we get both as a receiver and sender: PING, CONNECT, DISCONNECT
casemqttPacketPing:
iftrace {
c.traceInOp("PINGREQ", nil)
 }
c.mqttEnqueuePingResp()
iftrace {
c.traceOutOp("PINGRESP", nil)
 }

casemqttPacketConnect:
// It is an error to receive a second connect packet
ifconnected {
err=errMQTTSecondConnectPacket
break
 }
varrcbyte
varcp*mqttConnectProto
varsesspbool
rc, cp, err=c.mqttParseConnect(r, hasMappings)
// Add the client id to the client's string, regardless of error.
// We may still get the client_id if the call above fails somewhere
// after parsing the client ID itself.
c.ncs.Store(fmt.Sprintf("%s - %q", c, c.mqtt.cid))
iftrace&&cp!=nil {
c.traceInOp("CONNECT", errOrTrace(err, c.mqttConnectTrace(cp)))
 }
ifrc!=0 {
c.mqttEnqueueConnAck(rc, sessp)
iftrace {
c.traceOutOp("CONNACK", []byte(fmt.Sprintf("sp=%v rc=%v", sessp, rc)))
 }
 } elseiferr==nil {
iferr=s.mqttProcessConnect(c, cp, trace); err!=nil {
err=fmt.Errorf("unable to connect: %v", err)
 } else {
// Add this debug statement so users running in Debug mode
// will have the client id printed here for the first time.
c.Debugf("Client connected")
connected=true
rd=cp.rd
 }
 }

casemqttPacketDisconnect:
iftrace {
c.traceInOp("DISCONNECT", nil)
 }
// Normal disconnect, we need to discard the will.
// Spec [MQTT-3.1.2-8]
c.mu.Lock()
ifc.mqtt.cp!=nil {
c.mqtt.cp.will=nil
 }
c.mu.Unlock()
s.mqttHandleClosedClient(c)
c.closeConnection(ClientClosed)
returnnil

default:
err=fmt.Errorf("received unknown packet type %d", pt>>4)
 }
 }
iferr==nil&&rd>0 {
r.reader.SetReadDeadline(time.Now().Add(rd))
 }
returnerr
}

func (c*client) mqttTraceMsg(msg []byte) {
maxTrace:=c.srv.getOpts().MaxTracedMsgLen
ifmaxTrace>0&&len(msg) >maxTrace {
c.Tracef("<<- MSG_PAYLOAD: [\"%s...\"]", msg[:maxTrace])
 } else {
c.Tracef("<<- MSG_PAYLOAD: [%q]", msg)
 }
}

// The MQTT client connection has been closed, or the DISCONNECT packet was received.
// For a "clean" session, we will delete the session, otherwise, simply removing
// the binding. We will also send the "will" message if applicable.
//
// Runs from the client's readLoop.
// No lock held on entry.
func (s*Server) mqttHandleClosedClient(c*client) {
c.mu.Lock()
asm:=c.mqtt.asm
sess:=c.mqtt.sess
c.mu.Unlock()

// If asm or sess are nil, it means that we have failed a client
// before it was associated with a session, so nothing more to do.
ifasm==nil||sess==nil {
return
 }

// Add this session to the locked map for the rest of the execution.
iferr:=asm.lockSession(sess, c); err!=nil {
return
 }
deferasm.unlockSession(sess)

asm.mu.Lock()
// Clear the client from the session, but session may stay.
sess.mu.Lock()
sess.c=nil
doClean:=sess.clean
sess.mu.Unlock()
// If it was a clean session, then we remove from the account manager,
// and we will call clear() outside of any lock.
ifdoClean {
asm.removeSession(sess, false)
 }
// Remove in case it was in the flappers map.
asm.removeSessFromFlappers(sess.id)
asm.mu.Unlock()