client.go

// Copyright 2012-2025 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"
"crypto/x509"
"encoding/json"
"errors"
"fmt"
"io"
"math/rand"
"net"
"net/http"
"net/url"
"regexp"
"runtime"
"strconv"
"strings"
"sync"
"sync/atomic"
"time"

"github.com/klauspost/compress/s2"
"github.com/nats-io/jwt/v2"
"github.com/nats-io/nats-server/v2/internal/fastrand"
)

// Type of client connection.
const (
// CLIENT is an end user.
CLIENT=iota
// ROUTER represents another server in the cluster.
ROUTER
// GATEWAY is a link between 2 clusters.
GATEWAY
// SYSTEM is an internal system client.
SYSTEM
// LEAF is for leaf node connections.
LEAF
// JETSTREAM is an internal jetstream client.
JETSTREAM
// ACCOUNT is for the internal client for accounts.
ACCOUNT
)

// Internal clients. kind should be SYSTEM, JETSTREAM or ACCOUNT
funcisInternalClient(kindint) bool {
returnkind==SYSTEM||kind==JETSTREAM||kind==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/
ClientProtoZero=iota
// ClientProtoInfo signals a client can receive more then the original INFO block.
// This can be used to update clients on other cluster members, etc.
ClientProtoInfo
)

const (
pingProto="PING"+_CRLF_
pongProto="PONG"+_CRLF_
errProto="-ERR '%s'"+_CRLF_
okProto="+OK"+_CRLF_
)

// TLS Hanshake client types
const (
tlsHandshakeLeaf="leafnode"
tlsHandshakeMQTT="mqtt"
)

const (
// Scratch buffer size for the processMsg() calls.
msgScratchSize=1024
msgHeadProto="RMSG "
msgHeadProtoLen=len(msgHeadProto)

// For controlling dynamic buffer sizes.
startBufSize=512// For INFO/CONNECT block
minBufSize=64// Smallest to shrink to for PING/PONG
maxBufSize=65536// 64k
shortsToShrink=2// Trigger to shrink dynamic buffers
maxFlushPending=10// Max fsps to have in order to wait for writeLoop
readLoopReport=2*time.Second

// Server should not send a PING (for RTT) before the first PONG has
// been sent to the client. However, in case some client libs don't
// send CONNECT+PING, cap the maximum time before server can send
// the RTT PING.
maxNoRTTPingBeforeFirstPong=2*time.Second

// For stalling fast producers
stallClientMinDuration=2*time.Millisecond
stallClientMaxDuration=5*time.Millisecond
stallTotalAllowed=10*time.Millisecond
)

varreadLoopReportThreshold=readLoopReport

// Represent client booleans with a bitmask
typeclientFlaguint16

const (
hdrLine="NATS/1.0\r\n"
emptyHdrLine="NATS/1.0\r\n\r\n"
)

// Some client state represented as flags
const (
connectReceivedclientFlag=1<<iota// The CONNECT proto has been received
infoReceived// The INFO protocol has been received
firstPongSent// The first PONG has been sent
handshakeComplete// For TLS clients, indicate that the handshake is complete
flushOutbound// Marks client as having a flushOutbound call in progress.
noReconnect// Indicate that on close, this connection should not attempt a reconnect
closeConnection// Marks that closeConnection has already been called.
connMarkedClosed// Marks that markConnAsClosed has already been called.
writeLoopStarted// Marks that the writeLoop has been started.
skipFlushOnClose// Marks that flushOutbound() should not be called on connection close.
expectConnect// Marks if this connection is expected to send a CONNECT
connectProcessFinished// Marks if this connection has finished the connect process.
compressionNegotiated// Marks if this connection has negotiated compression level with remote.
didTLSFirst// Marks if this connection requested and was accepted doing the TLS handshake first (prior to INFO).
isSlowConsumer// Marks connection as a slow consumer.
)

// set the flag (would be equivalent to set the boolean to true)
func (cf*clientFlag) set(cclientFlag) {
*cf|=c
}

// clear the flag (would be equivalent to set the boolean to false)
func (cf*clientFlag) clear(cclientFlag) {
*cf&=^c
}

// isSet returns true if the flag is set, false otherwise
func (cfclientFlag) isSet(cclientFlag) bool {
returncf&c!=0
}

// setIfNotSet will set the flag `c` only if that flag was not already
// set and return true to indicate that the flag has been set. Returns
// false otherwise.
func (cf*clientFlag) setIfNotSet(cclientFlag) bool {
if*cf&c==0 {
*cf|=c
returntrue
 }
returnfalse
}

// ClosedState is the reason client was closed. This will
// be passed into calls to clearConnection, but will only
// be stored in ConnInfo for monitoring.
typeClosedStateint

const (
ClientClosed=ClosedState(iota+1)
AuthenticationTimeout
AuthenticationViolation
TLSHandshakeError
SlowConsumerPendingBytes
SlowConsumerWriteDeadline
WriteError
ReadError
ParseError
StaleConnection
ProtocolViolation
BadClientProtocolVersion
WrongPort
MaxAccountConnectionsExceeded
MaxConnectionsExceeded
MaxPayloadExceeded
MaxControlLineExceeded
MaxSubscriptionsExceeded
DuplicateRoute
RouteRemoved
ServerShutdown
AuthenticationExpired
WrongGateway
MissingAccount
Revocation
InternalClient
MsgHeaderViolation
NoRespondersRequiresHeaders
ClusterNameConflict
DuplicateRemoteLeafnodeConnection
DuplicateClientID
DuplicateServerName
MinimumVersionRequired
ClusterNamesIdentical
Kicked
)

// Some flags passed to processMsgResults
constpmrNoFlagint=0
const (
pmrCollectQueueNamesint=1<<iota
pmrIgnoreEmptyQueueFilter
pmrAllowSendFromRouteToRoute
pmrMsgImportedFromService
)

typeclientstruct {
// Here first because of use of atomics, and memory alignment.
stats
gwReplyMapping
kindint
srv*Server
acc*Account
perms*permissions
inreadCache
parseState
optsClientOpts
rrTracking*rrTracking
mpayint32
msubsint32
mclint32
mu sync.Mutex
ciduint64
start time.Time
nonce []byte
pubKeystring
nc net.Conn
ncs atomic.Value
outoutbound
user*NkeyUser
hoststring
portuint16
subsmap[string]*subscription
repliesmap[string]*resp
mperms*msgDeny
darray []string
pcdmap[*client]struct{}
atmr*time.Timer
expires time.Time
pingpinfo
msgb [msgScratchSize]byte
last time.Time
lastIn time.Time

repliesSincePruneuint16
lastReplyPrune time.Time

headersbool

rtt time.Duration
rttStart time.Time

route*route
gw*gateway
leaf*leaf
ws*websocket
mqtt*mqtt

flagsclientFlag// Compact booleans into a single field. Size will be increased when needed.

rrefbyte

tracebool
echobool
noIcbbool
iprocbool// In-Process connection, set at creation and immutable.

tags jwt.TagList
nameTagstring

tlsTo*time.Timer
}

typerrTrackingstruct {
rmapmap[string]*remoteLatency
ptmr*time.Timer
lrt time.Duration
}

// Struct for PING initiation from the server.
typepinfostruct {
tmr*time.Timer
outint
}

// outbound holds pending data for a socket.
typeoutboundstruct {
nb net.Buffers// Pending buffers for send, each has fixed capacity as per nbPool below.
wnb net.Buffers// Working copy of "nb", reused on each flushOutbound call, partial writes may leave entries here for next iteration.
pbint64// Total pending/queued bytes.
fspint32// Flush signals that are pending per producer from readLoop's pcd.
sg*sync.Cond// To signal writeLoop that there is data to flush.
wdl time.Duration// Snapshot of write deadline.
mpint64// Snapshot of max pending for client.
lft time.Duration// Last flush time for Write.
stcchanstruct{} // Stall chan we create to slow down producers on overrun, e.g. fan-in.
cw*s2.Writer
}

constnbMaxVectorSize=1024// == IOV_MAX on Linux/Darwin and most other Unices (except Solaris/AIX)

constnbPoolSizeSmall=512// Underlying array size of small buffer
constnbPoolSizeMedium=4096// Underlying array size of medium buffer
constnbPoolSizeLarge=65536// Underlying array size of large buffer

varnbPoolSmall=&sync.Pool{
New: func() any {
b:= [nbPoolSizeSmall]byte{}
return&b
 },
}

varnbPoolMedium=&sync.Pool{
New: func() any {
b:= [nbPoolSizeMedium]byte{}
return&b
 },
}

varnbPoolLarge=&sync.Pool{
New: func() any {
b:= [nbPoolSizeLarge]byte{}
return&b
 },
}

// nbPoolGet returns a frame that is a best-effort match for the given size.
// Once a pooled frame is no longer needed, it should be recycled by passing
// it to nbPoolPut.
funcnbPoolGet(szint) []byte {
switch {
casesz<=nbPoolSizeSmall:
returnnbPoolSmall.Get().(*[nbPoolSizeSmall]byte)[:0]
casesz<=nbPoolSizeMedium:
returnnbPoolMedium.Get().(*[nbPoolSizeMedium]byte)[:0]
default:
returnnbPoolLarge.Get().(*[nbPoolSizeLarge]byte)[:0]
 }
}

// nbPoolPut recycles a frame that was retrieved from nbPoolGet. It is not
// safe to return multiple slices referring to chunks of the same underlying
// array as this may create overlaps when the buffers are returned to their
// original size, resulting in race conditions.
funcnbPoolPut(b []byte) {
switchcap(b) {
casenbPoolSizeSmall:
b:= (*[nbPoolSizeSmall]byte)(b[0:nbPoolSizeSmall])
nbPoolSmall.Put(b)
casenbPoolSizeMedium:
b:= (*[nbPoolSizeMedium]byte)(b[0:nbPoolSizeMedium])
nbPoolMedium.Put(b)
casenbPoolSizeLarge:
b:= (*[nbPoolSizeLarge]byte)(b[0:nbPoolSizeLarge])
nbPoolLarge.Put(b)
default:
// Ignore frames that are the wrong size, this might happen
// with WebSocket/MQTT messages as they are framed
 }
}

typepermstruct {
allow*Sublist
deny*Sublist
}

typepermissionsstruct {
// Have these 2 first for memory alignment due to the use of atomic.
pcszint32
prunint32
subperm
pubperm
resp*ResponsePermission
pcache sync.Map
}

// This is used to dynamically track responses and reply subjects
// for dynamic permissioning.
typerespstruct {
t time.Time
nint
}

// msgDeny is used when a user permission for subscriptions has a deny
// clause but a subscription could be made that is of broader scope.
// e.g. deny = "foo", but user subscribes to "*". That subscription should
// succeed but no message sent on foo should be delivered.
typemsgDenystruct {
deny*Sublist
dcachemap[string]bool
}

// routeTarget collects information regarding routes and queue groups for
// sending information to a remote.
typerouteTargetstruct {
sub*subscription
qs []byte
_qs [32]byte
}

const (
maxResultCacheSize=512
maxDenyPermCacheSize=256
maxPermCacheSize=128
pruneSize=32
routeTargetInit=8
replyPermLimit=4096
replyPruneTime=time.Second
)

// Represent read cache booleans with a bitmask
typereadCacheFlaguint16

const (
hasMappingsreadCacheFlag=1<<iota// For account subject mappings.
switchToCompressionreadCacheFlag=1<<1
)

constsysGroup="_sys_"

// Used in readloop to cache hot subject lookups and group statistics.
typereadCachestruct {
// These are for clients who are bound to a single account.
geniduint64
resultsmap[string]*SublistResult

// This is for routes and gateways to have their own L1 as well that is account aware.
pacachemap[string]*perAccountCache

// This is for when we deliver messages across a route. We use this structure
// to make sure to only send one message and properly scope to queues as needed.
rts []routeTarget

// These are all temporary totals for an invocation of a read in readloop.
msgsint32
bytesint32
subsint32

rszint32// Read buffer size
srsint32// Short reads, used for dynamic buffer resizing.

// These are for readcache flags to avoid locks.
flagsreadCacheFlag

// Capture the time we started processing our readLoop.
start time.Time

// Total time stalled so far for readLoop processing.
tst time.Duration
}

// set the flag (would be equivalent to set the boolean to true)
func (rcf*readCacheFlag) set(creadCacheFlag) {
*rcf|=c
}

// clear the flag (would be equivalent to set the boolean to false)
func (rcf*readCacheFlag) clear(creadCacheFlag) {
*rcf&=^c
}

// isSet returns true if the flag is set, false otherwise
func (rcfreadCacheFlag) isSet(creadCacheFlag) bool {
returnrcf&c!=0
}

const (
defaultMaxPerAccountCacheSize=8192
defaultPrunePerAccountCacheSize=1024
defaultClosedSubsCheckInterval=5*time.Minute
)

var (
maxPerAccountCacheSize=defaultMaxPerAccountCacheSize
prunePerAccountCacheSize=defaultPrunePerAccountCacheSize
closedSubsCheckInterval=defaultClosedSubsCheckInterval
)

// perAccountCache is for L1 semantics for inbound messages from a route or gateway to mimic the performance of clients.
typeperAccountCachestruct {
acc*Account
results*SublistResult
geniduint64
}

func (c*client) String() (idstring) {
loaded:=c.ncs.Load()
ifloaded!=nil {
returnloaded.(string)
 }

return_EMPTY_
}

// GetNonce returns the nonce that was presented to the user on connection
func (c*client) GetNonce() []byte {
c.mu.Lock()
deferc.mu.Unlock()

returnc.nonce
}

// GetName returns the application supplied name for the connection.
func (c*client) GetName() string {
c.mu.Lock()
name:=c.opts.Name
c.mu.Unlock()
returnname
}

// GetOpts returns the client options provided by the application.
func (c*client) GetOpts() *ClientOpts {
return&c.opts
}

// GetTLSConnectionState returns the TLS ConnectionState if TLS is enabled, nil
// otherwise. Implements the ClientAuth interface.
func (c*client) GetTLSConnectionState() *tls.ConnectionState {
c.mu.Lock()
deferc.mu.Unlock()
ifc.nc==nil {
returnnil
 }
tc, ok:=c.nc.(*tls.Conn)
if!ok {
returnnil
 }
state:=tc.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 {
switchc.kind {
caseCLIENT:
ifc.isMqtt() {
returnMQTT
 } elseifc.isWebsocket() {
returnWS
 }
returnNATS
default:
returnNON_CLIENT
 }
}

varclientTypeStringMap=map[int]string{
NON_CLIENT: _EMPTY_,
NATS: "nats",
WS: "websocket",
MQTT: "mqtt",
}

func (c*client) clientTypeString() string {
iftypeStringVal, ok:=clientTypeStringMap[c.clientType()]; ok {
returntypeStringVal
 }
return_EMPTY_
}

// This is the main subscription struct that indicates
// interest in published messages.
// FIXME(dlc) - This is getting bloated for normal subs, need
// to optionally have an opts section for non-normal stuff.
typesubscriptionstruct {
client*client
im*streamImport// This is for import stream support.
rsibool
sibool
shadow []*subscription// This is to track shadowed accounts.
icbmsgHandler
subject []byte
queue []byte
sid []byte
origin []byte
nmint64
maxint64
qwint32
closedint32
mqtt*mqttSub
}

// Indicate that this subscription is closed.
// This is used in pruning of route and gateway cache items.
func (s*subscription) close() {
atomic.StoreInt32(&s.closed, 1)
}

// Return true if this subscription was unsubscribed
// or its connection has been closed.
func (s*subscription) isClosed() bool {
returnatomic.LoadInt32(&s.closed) ==1
}

typeClientOptsstruct {
Echobool`json:"echo"`
Verbosebool`json:"verbose"`
Pedanticbool`json:"pedantic"`
TLSRequiredbool`json:"tls_required"`
Nkeystring`json:"nkey,omitempty"`
JWTstring`json:"jwt,omitempty"`
Sigstring`json:"sig,omitempty"`
Tokenstring`json:"auth_token,omitempty"`
Usernamestring`json:"user,omitempty"`
Passwordstring`json:"pass,omitempty"`
Namestring`json:"name"`
Langstring`json:"lang"`
Versionstring`json:"version"`
Protocolint`json:"protocol"`
Accountstring`json:"account,omitempty"`
AccountNewbool`json:"new_account,omitempty"`
Headersbool`json:"headers,omitempty"`
NoRespondersbool`json:"no_responders,omitempty"`

// Routes and Leafnodes only
Import*SubjectPermission`json:"import,omitempty"`
Export*SubjectPermission`json:"export,omitempty"`

// Leafnodes
RemoteAccountstring`json:"remote_account,omitempty"`
}

vardefaultOpts=ClientOpts{Verbose: true, Pedantic: true, Echo: true}
varinternalOpts=ClientOpts{Verbose: false, Pedantic: false, Echo: false}

func (c*client) setTraceLevel() {
ifc.kind==SYSTEM&&!(atomic.LoadInt32(&c.srv.logging.traceSysAcc) !=0) {
c.trace=false
 } else {
c.trace= (atomic.LoadInt32(&c.srv.logging.trace) !=0)
 }
}

// Lock should be held
func (c*client) initClient() {
s:=c.srv
c.cid=atomic.AddUint64(&s.gcid, 1)

// Outbound data structure setup
c.out.sg=sync.NewCond(&(c.mu))
opts:=s.getOpts()
// Snapshots to avoid mutex access in fast paths.
c.out.wdl=opts.WriteDeadline
c.out.mp=opts.MaxPending
// 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
// to be reloaded without restart, this code will need to change.
c.mcl=int32(opts.MaxControlLine)
ifc.mcl==0 {
c.mcl=MAX_CONTROL_LINE_SIZE
 }

c.subs=make(map[string]*subscription)
c.echo=true

c.setTraceLevel()

// This is a scratch buffer used for processMsg()
// The msg header starts with "RMSG ", which can be used
// for both local and routes.
// in bytes that is [82 77 83 71 32].
c.msgb= [msgScratchSize]byte{82, 77, 83, 71, 32}

// This is to track pending clients that have data to be flushed
// after we process inbound msgs from our own connection.
c.pcd=make(map[*client]struct{})

// snapshot the string version of the connection
varconnstring
ifc.nc!=nil {
ifaddr:=c.nc.RemoteAddr(); addr!=nil {
ifconn=addr.String(); conn!=_EMPTY_ {
host, port, _:=net.SplitHostPort(conn)
iPort, _:=strconv.Atoi(port)
c.host, c.port=host, uint16(iPort)
ifc.isWebsocket() &&c.ws.clientIP!=_EMPTY_ {
cip:=c.ws.clientIP
// Surround IPv6 addresses with square brackets, as
// net.JoinHostPort would do...
ifstrings.Contains(cip, ":") {
cip="["+cip+"]"
 }
conn=fmt.Sprintf("%s/%s", cip, conn)
 }
// Now that we have extracted host and port, escape
// the string because it is going to be used in Sprintf
conn=strings.ReplaceAll(conn, "%", "%%")
 }
 }
 }

switchc.kind {
caseCLIENT:
switchc.clientType() {
caseNATS:
c.ncs.Store(fmt.Sprintf("%s - cid:%d", conn, c.cid))
caseWS:
c.ncs.Store(fmt.Sprintf("%s - wid:%d", conn, c.cid))
caseMQTT:
varwsstring
ifc.isWebsocket() {
ws="_ws"
 }
c.ncs.Store(fmt.Sprintf("%s - mid%s:%d", conn, ws, c.cid))
 }
caseROUTER:
c.ncs.Store(fmt.Sprintf("%s - rid:%d", conn, c.cid))
caseGATEWAY:
c.ncs.Store(fmt.Sprintf("%s - gid:%d", conn, c.cid))
caseLEAF:
varwsstring
ifc.isWebsocket() {
ws="_ws"
 }
c.ncs.Store(fmt.Sprintf("%s - lid%s:%d", conn, ws, c.cid))
caseSYSTEM:
c.ncs.Store("SYSTEM")
caseJETSTREAM:
c.ncs.Store("JETSTREAM")
caseACCOUNT:
c.ncs.Store("ACCOUNT")
 }
}

// RemoteAddress expose the Address of the client connection,
// nil when not connected or unknown
func (c*client) RemoteAddress() net.Addr {
c.mu.Lock()
deferc.mu.Unlock()

ifc.nc==nil {
returnnil
 }

returnc.nc.RemoteAddr()
}

// Helper function to report errors.
func (c*client) reportErrRegisterAccount(acc*Account, errerror) {
iferr==ErrTooManyAccountConnections {
c.maxAccountConnExceeded()
return
 }
c.Errorf("Problem registering with account %q: %s", acc.Name, err)
c.sendErr("Failed Account Registration")
}

// Kind returns the client kind and will be one of the defined constants like CLIENT, ROUTER, GATEWAY, LEAF
func (c*client) Kind() int {
c.mu.Lock()
kind:=c.kind
c.mu.Unlock()

returnkind
}

// registerWithAccount will register the given user with a specific
// account. This will change the subject namespace.
func (c*client) registerWithAccount(acc*Account) error {
ifacc==nil {
returnErrBadAccount
 }
acc.mu.RLock()
bad:=acc.sl==nil
acc.mu.RUnlock()
ifbad {
returnErrBadAccount
 }
// If we were previously registered, usually to $G, do accounting here to remove.
ifc.acc!=nil {
ifprev:=c.acc.removeClient(c); prev==1&&c.srv!=nil {
c.srv.decActiveAccounts()
 }
 }

c.mu.Lock()
kind:=c.kind
srv:=c.srv
c.acc=acc
c.applyAccountLimits()
c.mu.Unlock()

// Check if we have a max connections violation
ifkind==CLIENT&&acc.MaxTotalConnectionsReached() {
returnErrTooManyAccountConnections
 } elseifkind==LEAF {
// Check if we are already connected to this cluster.
ifrc:=c.remoteCluster(); rc!=_EMPTY_&&acc.hasLeafNodeCluster(rc) {
returnErrLeafNodeLoop
 }
ifacc.MaxTotalLeafNodesReached() {
returnErrTooManyAccountConnections
 }
 }

// Add in new one.
ifprev:=acc.addClient(c); prev==0&&srv!=nil {
srv.incActiveAccounts()
 }

returnnil
}

// Helper to determine if we have met or exceeded max subs.
func (c*client) subsAtLimit() bool {
returnc.msubs!=jwt.NoLimit&&len(c.subs) >=int(c.msubs)
}

funcminLimit(value*int32, limitint32) bool {
v:=atomic.LoadInt32(value)
ifv!=jwt.NoLimit {
iflimit!=jwt.NoLimit {
iflimit<v {
atomic.StoreInt32(value, limit)
returntrue
 }
 }
 } elseiflimit!=jwt.NoLimit {
atomic.StoreInt32(value, limit)
returntrue
 }
returnfalse
}

// Apply account limits
// Lock is held on entry.
// FIXME(dlc) - Should server be able to override here?
func (c*client) applyAccountLimits() {
ifc.acc==nil|| (c.kind!=CLIENT&&c.kind!=LEAF) {
return
 }
atomic.StoreInt32(&c.mpay, jwt.NoLimit)
c.msubs=jwt.NoLimit
ifc.opts.JWT!=_EMPTY_ { // user jwt implies account
ifuc, _:=jwt.DecodeUserClaims(c.opts.JWT); uc!=nil {
atomic.StoreInt32(&c.mpay, int32(uc.Limits.Payload))
c.msubs=int32(uc.Limits.Subs)
ifuc.IssuerAccount!=_EMPTY_&&uc.IssuerAccount!=uc.Issuer {
ifscope, ok:=c.acc.signingKeys[uc.Issuer]; ok {
ifuserScope, ok:=scope.(*jwt.UserScope); ok {
// if signing key disappeared or changed and we don't get here, the client will be disconnected
c.mpay=int32(userScope.Template.Limits.Payload)
c.msubs=int32(userScope.Template.Limits.Subs)
 }
 }
 }
 }
 }

c.acc.mu.RLock()
minLimit(&c.mpay, c.acc.mpay)
minLimit(&c.msubs, c.acc.msubs)
c.acc.mu.RUnlock()

s:=c.srv
opts:=s.getOpts()
mPay:=opts.MaxPayload
// options encode unlimited differently
ifmPay==0 {
mPay=jwt.NoLimit
 }
mSubs:=int32(opts.MaxSubs)
ifmSubs==0 {
mSubs=jwt.NoLimit
 }
wasUnlimited:=c.mpay==jwt.NoLimit
ifminLimit(&c.mpay, mPay) &&!wasUnlimited {
c.Errorf("Max Payload set to %d from server overrides account or user config", opts.MaxPayload)
 }
wasUnlimited=c.msubs==jwt.NoLimit
ifminLimit(&c.msubs, mSubs) &&!wasUnlimited {
c.Errorf("Max Subscriptions set to %d from server overrides account or user config", opts.MaxSubs)
 }
ifc.subsAtLimit() {
gofunc() {
c.maxSubsExceeded()
time.Sleep(20*time.Millisecond)
c.closeConnection(MaxSubscriptionsExceeded)
 }()
 }
}

// RegisterUser allows auth to call back into a new client
// with the authenticated user. This is used to map
// any permissions into the client and setup accounts.
func (c*client) RegisterUser(user*User) {
// Register with proper account and sublist.
ifuser.Account!=nil {
iferr:=c.registerWithAccount(user.Account); err!=nil {
c.reportErrRegisterAccount(user.Account, err)
return
 }
 }

c.mu.Lock()

// Assign permissions.
ifuser.Permissions==nil {
// Reset perms to nil in case client previously had them.
c.perms=nil
c.mperms=nil
 } else {
c.setPermissions(user.Permissions)
 }

// allows custom authenticators to set a username to be reported in
// server events and more
ifuser.Username!=_EMPTY_ {
c.opts.Username=user.Username
 }

// if a deadline time stamp is set we start a timer to disconnect the user at that time
if!user.ConnectionDeadline.IsZero() {
c.setExpirationTimerUnlocked(time.Until(user.ConnectionDeadline))
 }

c.mu.Unlock()
}

// RegisterNkeyUser allows auth to call back into a new nkey
// client with the authenticated user. This is used to map
// any permissions into the client and setup accounts.
func (c*client) RegisterNkeyUser(user*NkeyUser) error {
// Register with proper account and sublist.
ifuser.Account!=nil {
iferr:=c.registerWithAccount(user.Account); err!=nil {
c.reportErrRegisterAccount(user.Account, err)
returnerr
 }
 }

c.mu.Lock()
c.user=user
// Assign permissions.
ifuser.Permissions==nil {
// Reset perms to nil in case client previously had them.
c.perms=nil
c.mperms=nil
 } else {
c.setPermissions(user.Permissions)
 }
c.mu.Unlock()
returnnil
}

funcsplitSubjectQueue(sqstring) ([]byte, []byte, error) {
vals:=strings.Fields(strings.TrimSpace(sq))
s:= []byte(vals[0])
varq []byte
iflen(vals) ==2 {
q= []byte(vals[1])
 } elseiflen(vals) >2 {
returnnil, nil, fmt.Errorf("invalid subject-queue %q", sq)
 }
returns, q, nil
}

// Initializes client.perms structure.
// Lock is held on entry.
func (c*client) setPermissions(perms*Permissions) {
ifperms==nil {
return
 }
c.perms=&permissions{}

// Loop over publish permissions
ifperms.Publish!=nil {
ifperms.Publish.Allow!=nil {
c.perms.pub.allow=NewSublistWithCache()