25621091956

Committed 08 May 2026 01:49PM UTC coverage: 81.505% (-1.7%) from 83.18%

Build # 25621091956

Build Type

push

github

Committed by

web-flow

Commit Message

MQTT: Return `errMQTTUnsupportedCharacters` for control characters on both pub and sub (#8112)

Replaces #8104.

Coverage Stats

75814 of 93018 relevant lines covered (81.5%)

630142.14 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

26.1

/server/msgtrace.go

// Copyright 2024-2026 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"
        "encoding/json"
        "fmt"
        "math/rand"
        "strconv"
        "strings"
        "sync/atomic"
        "time"
)

const (
        MsgTraceDest          = "Nats-Trace-Dest"
        MsgTraceDestDisabled  = "trace disabled" // This must be an invalid NATS subject
        MsgTraceHop           = "Nats-Trace-Hop"
        MsgTraceOriginAccount = "Nats-Trace-Origin-Account"
        MsgTraceOnly          = "Nats-Trace-Only"

        // External trace header. Note that this header is normally in lower
        // case (https://www.w3.org/TR/trace-context/#header-name). Vendors
        // MUST expect the header in any case (upper, lower, mixed), and
        // SHOULD send the header name in lowercase. We used to change it
        // to lower case, but no longer do that in 2.14.
        traceParentHdr = "traceparent"
)

var (
        traceDestHdrAsBytes      = stringToBytes(MsgTraceDest)
        traceDestDisabledAsBytes = stringToBytes(MsgTraceDestDisabled)
        traceParentHdrAsBytes    = stringToBytes(traceParentHdr)
        crLFAsBytes              = stringToBytes(CR_LF)
        dashAsBytes              = stringToBytes("-")
)

type MsgTraceType string

// Type of message trace events in the MsgTraceEvents list.
// This is needed to unmarshal the list.
const (
        MsgTraceIngressType        = "in"
        MsgTraceSubjectMappingType = "sm"
        MsgTraceStreamExportType   = "se"
        MsgTraceServiceImportType  = "si"
        MsgTraceJetStreamType      = "js"
        MsgTraceEgressType         = "eg"
)

type MsgTraceEvent struct {
        Server  ServerInfo      `json:"server"`
        Request MsgTraceRequest `json:"request"`
        Hops    int             `json:"hops,omitempty"`
        Events  MsgTraceEvents  `json:"events"`
}

type MsgTraceRequest struct {
        // We are not making this an http.Header so that header name case is preserved.
        Header  map[string][]string `json:"header,omitempty"`
        MsgSize int                 `json:"msgsize,omitempty"`
}

type MsgTraceEvents []MsgTrace

type MsgTrace interface {
        new() MsgTrace
        typ() MsgTraceType
}

type MsgTraceBase struct {
        Type      MsgTraceType `json:"type"`
        Timestamp time.Time    `json:"ts"`
}

type MsgTraceIngress struct {
        MsgTraceBase
        Kind    int    `json:"kind"`
        CID     uint64 `json:"cid"`
        Name    string `json:"name,omitempty"`
        Account string `json:"acc"`
        Subject string `json:"subj"`
        Error   string `json:"error,omitempty"`
}

type MsgTraceSubjectMapping struct {
        MsgTraceBase
        MappedTo string `json:"to"`
}

type MsgTraceStreamExport struct {
        MsgTraceBase
        Account string `json:"acc"`
        To      string `json:"to"`
}

type MsgTraceServiceImport struct {
        MsgTraceBase
        Account string `json:"acc"`
        From    string `json:"from"`
        To      string `json:"to"`
}

type MsgTraceJetStream struct {
        MsgTraceBase
        Stream     string `json:"stream"`
        Subject    string `json:"subject,omitempty"`
        NoInterest bool   `json:"nointerest,omitempty"`
        Error      string `json:"error,omitempty"`
}

type MsgTraceEgress struct {
        MsgTraceBase
        Kind         int    `json:"kind"`
        CID          uint64 `json:"cid"`
        Name         string `json:"name,omitempty"`
        Hop          string `json:"hop,omitempty"`
        Account      string `json:"acc,omitempty"`
        Subscription string `json:"sub,omitempty"`
        Queue        string `json:"queue,omitempty"`
        Error        string `json:"error,omitempty"`

        // This is for applications that unmarshal the trace events
        // and want to link an egress to route/leaf/gateway with
        // the MsgTraceEvent from that server.
        Link *MsgTraceEvent `json:"-"`
}

// -------------------------------------------------------------

func (t MsgTraceBase) typ() MsgTraceType     { return t.Type }
func (MsgTraceIngress) new() MsgTrace        { return &MsgTraceIngress{} }
func (MsgTraceSubjectMapping) new() MsgTrace { return &MsgTraceSubjectMapping{} }
func (MsgTraceStreamExport) new() MsgTrace   { return &MsgTraceStreamExport{} }
func (MsgTraceServiceImport) new() MsgTrace  { return &MsgTraceServiceImport{} }
func (MsgTraceJetStream) new() MsgTrace      { return &MsgTraceJetStream{} }
func (MsgTraceEgress) new() MsgTrace         { return &MsgTraceEgress{} }

var msgTraceInterfaces = map[MsgTraceType]MsgTrace{
        MsgTraceIngressType:        MsgTraceIngress{},
        MsgTraceSubjectMappingType: MsgTraceSubjectMapping{},
        MsgTraceStreamExportType:   MsgTraceStreamExport{},
        MsgTraceServiceImportType:  MsgTraceServiceImport{},
        MsgTraceJetStreamType:      MsgTraceJetStream{},
        MsgTraceEgressType:         MsgTraceEgress{},
}

func (t *MsgTraceEvents) UnmarshalJSON(data []byte) error {
        var raw []json.RawMessage
        err := json.Unmarshal(data, &raw)
        if err != nil {
                return err
        }
        *t = make(MsgTraceEvents, len(raw))
        var tt MsgTraceBase
        for i, r := range raw {
                if err = json.Unmarshal(r, &tt); err != nil {
                        return err
                }
                tr, ok := msgTraceInterfaces[tt.Type]
                if !ok {
                        return fmt.Errorf("unknown trace type %v", tt.Type)
                }
                te := tr.new()
                if err := json.Unmarshal(r, te); err != nil {
                        return err
                }
                (*t)[i] = te
        }
        return nil
}

func getTraceAs[T MsgTrace](e any) *T {
        v, ok := e.(*T)
        if ok {
                return v
        }
        return nil
}

func (t *MsgTraceEvent) Ingress() *MsgTraceIngress {
        if len(t.Events) < 1 {
                return nil
        }
        return getTraceAs[MsgTraceIngress](t.Events[0])
}

func (t *MsgTraceEvent) SubjectMapping() *MsgTraceSubjectMapping {
        for _, e := range t.Events {
                if e.typ() == MsgTraceSubjectMappingType {
                        return getTraceAs[MsgTraceSubjectMapping](e)
                }
        }
        return nil
}

func (t *MsgTraceEvent) StreamExports() []*MsgTraceStreamExport {
        var se []*MsgTraceStreamExport
        for _, e := range t.Events {
                if e.typ() == MsgTraceStreamExportType {
                        se = append(se, getTraceAs[MsgTraceStreamExport](e))
                }
        }
        return se
}

func (t *MsgTraceEvent) ServiceImports() []*MsgTraceServiceImport {
        var si []*MsgTraceServiceImport
        for _, e := range t.Events {
                if e.typ() == MsgTraceServiceImportType {
                        si = append(si, getTraceAs[MsgTraceServiceImport](e))
                }
        }
        return si
}

func (t *MsgTraceEvent) JetStream() *MsgTraceJetStream {
        for _, e := range t.Events {
                if e.typ() == MsgTraceJetStreamType {
                        return getTraceAs[MsgTraceJetStream](e)
                }
        }
        return nil
}

func (t *MsgTraceEvent) Egresses() []*MsgTraceEgress {
        var eg []*MsgTraceEgress
        for _, e := range t.Events {
                if e.typ() == MsgTraceEgressType {
                        eg = append(eg, getTraceAs[MsgTraceEgress](e))
                }
        }
        return eg
}

const (
        errMsgTraceOnlyNoSupport   = "Not delivered because remote does not support message tracing"
        errMsgTraceNoSupport       = "Message delivered but remote does not support message tracing so no trace event generated from there"
        errMsgTraceNoEcho          = "Not delivered because of no echo"
        errMsgTracePubViolation    = "Not delivered because publish denied for this subject"
        errMsgTraceSubDeny         = "Not delivered because subscription denies this subject"
        errMsgTraceSubClosed       = "Not delivered because subscription is closed"
        errMsgTraceClientClosed    = "Not delivered because client is closed"
        errMsgTraceAutoSubExceeded = "Not delivered because auto-unsubscribe exceeded"
        errMsgTraceFastProdNoStall = "Not delivered because fast producer not stalled and consumer is slow"
)

type msgTrace struct {
        ready int32
        srv   *Server
        acc   *Account
        // Origin account name, set only if acc is nil when acc lookup failed.
        oan   string
        dest  string
        event *MsgTraceEvent
        js    *MsgTraceJetStream
        hop   string
        nhop  string
        tonly bool // Will only trace the message, not do delivery.
        ct    compressionType
}

// This will be false outside of the tests, so when building the server binary,
// any code where you see `if msgTraceRunInTests` statement will be compiled
// out, so this will have no performance penalty.
var (
        msgTraceRunInTests   bool
        msgTraceCheckSupport bool
)

// Returns the message trace object, if message is being traced,
// and `true` if we want to only trace, not actually deliver the message.
func (c *client) isMsgTraceEnabled() (*msgTrace, bool) {
        t := c.pa.trace
        if t == nil {
                return nil, false
        }
        return t, t.tonly
}

// For LEAF/ROUTER/GATEWAY, return false if the remote does not support
// message tracing (important if the tracing requests trace-only).
func (c *client) msgTraceSupport() bool {
        // Exclude client connection from the protocol check.
        return c.kind == CLIENT || c.opts.Protocol >= MsgTraceProto
}

func getConnName(c *client) string {
        switch c.kind {
        case ROUTER:
                if n := c.route.remoteName; n != _EMPTY_ {
                        return n
                }
        case GATEWAY:
                if n := c.gw.remoteName; n != _EMPTY_ {
                        return n
                }
        case LEAF:
                if n := c.leaf.remoteServer; n != _EMPTY_ {
                        return n
                }
        }
        return c.opts.Name
}

func getCompressionType(cts string) compressionType {
        if cts == _EMPTY_ {
                return noCompression
        }
        cts = strings.ToLower(cts)
        if strings.Contains(cts, "snappy") || strings.Contains(cts, "s2") {
                return snappyCompression
        }
        if strings.Contains(cts, "gzip") {
                return gzipCompression
        }
        return unsupportedCompression
}

func (c *client) initMsgTrace() *msgTrace {
        // The code in the "if" statement is only running in test mode.
        if msgTraceRunInTests {
                // Check the type of client that tries to initialize a trace struct.
                if !(c.kind == CLIENT || c.kind == ROUTER || c.kind == GATEWAY || c.kind == LEAF) {
                        panic(fmt.Sprintf("Unexpected client type %q trying to initialize msgTrace", c.kindString()))
                }
                // In some tests, we want to make a server behave like an old server
                // and so even if a trace header is received, we want the server to
                // simply ignore it.
                if msgTraceCheckSupport {
                        if c.srv == nil || c.srv.getServerProto() < MsgTraceProto {
                                return nil
                        }
                }
        }
        if c.pa.hdr <= 0 {
                return nil
        }
        hdr := c.msgBuf[:c.pa.hdr]
        headers, external := genHeaderMapIfTraceHeadersPresent(hdr)
        if len(headers) == 0 {
                return nil
        }
        // Little helper to give us the first value of a given header, or _EMPTY_
        // if key is not present.
        getHdrVal := func(key string) string {
                vv, ok := headers[key]
                if !ok {
                        return _EMPTY_
                }
                return vv[0]
        }
        var (
                dest      string
                traceOnly bool
        )
        // Check for traceOnly only if not external.
        if !external {
                if to := getHdrVal(MsgTraceOnly); to != _EMPTY_ {
                        tos := strings.ToLower(to)
                        switch tos {
                        case "1", "true", "on":
                                traceOnly = true
                        }
                }
                dest = getHdrVal(MsgTraceDest)
                if c.kind == CLIENT {
                        if td, ok := c.allowedMsgTraceDest(hdr, false); !ok {
                                return nil
                        } else if td != _EMPTY_ {
                                dest = td
                        }
                }
                // Check the destination to see if this is a valid public subject.
                if !IsValidPublishSubject(dest) {
                        // We still have to return a msgTrace object (if traceOnly is set)
                        // because if we don't, the message will end-up being delivered to
                        // applications, which may break them. We report the error in any case.
                        c.Errorf("Destination %q is not valid, won't be able to trace events", dest)
                        if !traceOnly {
                                // We can bail, tracing will be disabled for this message.
                                return nil
                        }
                }
        }
        var (
                // Account to use when sending the trace event
                acc *Account
                // Ingress' account name
                ian string
                // Origin account name
                oan string
                // The hop "id", taken from headers only when not from CLIENT
                hop string
        )
        if c.kind == ROUTER || c.kind == GATEWAY || c.kind == LEAF {
                // The ingress account name will always be c.pa.account, but `acc` may
                // be different if we have an origin account header.
                if c.kind == LEAF {
                        ian = c.acc.GetName()
                } else {
                        ian = string(c.pa.account)
                }
                // The remote will have set the origin account header only if the
                // message changed account (think of service imports).
                oan = getHdrVal(MsgTraceOriginAccount)
                if oan == _EMPTY_ {
                        // For LEAF or ROUTER with pinned-account, we can use the c.acc.
                        if c.kind == LEAF || (c.kind == ROUTER && len(c.route.accName) > 0) {
                                acc = c.acc
                        } else {
                                // We will lookup account with c.pa.account (or ian).
                                oan = ian
                        }
                }
                // Unless we already got the account, we need to look it up.
                if acc == nil {
                        // We don't want to do account resolving here.
                        if acci, ok := c.srv.accounts.Load(oan); ok {
                                acc = acci.(*Account)
                                // Since we have looked-up the account, we don't need oan, so
                                // clear it in case it was set.
                                oan = _EMPTY_
                        } else {
                                // We still have to return a msgTrace object (if traceOnly is set)
                                // because if we don't, the message will end-up being delivered to
                                // applications, which may break them. We report the error in any case.
                                c.Errorf("Account %q was not found, won't be able to trace events", oan)
                                if !traceOnly {
                                        // We can bail, tracing will be disabled for this message.
                                        return nil
                                }
                        }
                }
                // Check the hop header
                hop = getHdrVal(MsgTraceHop)
        } else {
                acc = c.acc
                ian = acc.GetName()
        }
        // If external, we need to have the account's trace destination set,
        // otherwise, we are not enabling tracing.
        if external {
                var sampling int
                if acc != nil {
                        dest, sampling = acc.getTraceDestAndSampling()
                }
                if dest == _EMPTY_ {
                        // No account destination, no tracing for external trace headers.
                        return nil
                }
                // Check sampling, but only from origin server.
                if c.kind == CLIENT && !sample(sampling) {
                        // Need to disable tracing so that if the message is routed, it won't
                        // trigger a trace there.
                        c.msgBuf = c.setHeader(MsgTraceDest, MsgTraceDestDisabled, c.msgBuf)
                        return nil
                }
        }
        c.pa.trace = &msgTrace{
                srv:  c.srv,
                acc:  acc,
                oan:  oan,
                dest: dest,
                ct:   getCompressionType(getHdrVal(acceptEncodingHeader)),
                hop:  hop,
                event: &MsgTraceEvent{
                        Request: MsgTraceRequest{
                                Header:  headers,
                                MsgSize: c.pa.size,
                        },
                        Events: append(MsgTraceEvents(nil), &MsgTraceIngress{
                                MsgTraceBase: MsgTraceBase{
                                        Type:      MsgTraceIngressType,
                                        Timestamp: time.Now(),
                                },
                                Kind:    c.kind,
                                CID:     c.cid,
                                Name:    getConnName(c),
                                Account: ian,
                                Subject: string(c.pa.subject),
                        }),
                },
                tonly: traceOnly,
        }
        return c.pa.trace
}

func sample(sampling int) bool {
        // Option parsing should ensure that sampling is [1..100], but consider
        // any value outside of this range to be 100%.
        if sampling <= 0 || sampling >= 100 {
                return true
        }
        return rand.Int31n(100) <= int32(sampling)
}

// This function will return the header as a map (instead of http.Header because
// we want to preserve the header names' case) and a boolean that indicates if
// the headers have been lifted due to the presence of the external trace header
// only.
// Note that because of the traceParentHdr, the search is done in a case
// insensitive way. We used to rewrite it in lower case but no longer do since v2.14.
func genHeaderMapIfTraceHeadersPresent(hdr []byte) (map[string][]string, bool) {

        var (
                _keys               = [64][]byte{}
                _vals               = [64][]byte{}
                m                   map[string][]string
                traceDestHdrFound   bool
                traceParentHdrFound bool
        )
        // Skip the hdrLine
        if !bytes.HasPrefix(hdr, stringToBytes(hdrLine)) {
                return nil, false
        }

        keys := _keys[:0]
        vals := _vals[:0]

        for i := len(hdrLine); i < len(hdr); {
                // Search for key/val delimiter
                del := bytes.IndexByte(hdr[i:], ':')
                if del < 0 {
                        break
                }
                keyStart := i
                key := hdr[keyStart : keyStart+del]
                i += del + 1
                for i < len(hdr) && (hdr[i] == ' ' || hdr[i] == '\t') {
                        i++
                }
                valStart := i
                nl := bytes.Index(hdr[valStart:], crLFAsBytes)
                if nl < 0 {
                        break
                }
                valEnd := valStart + nl
                for valEnd > valStart && (hdr[valEnd-1] == ' ' || hdr[valEnd-1] == '\t') {
                        valEnd--
                }
                val := hdr[valStart:valEnd]
                if len(key) > 0 && len(val) > 0 {
                        vals = append(vals, val)

                        // We search for our special keys only if not already found.

                        // Check for the external trace header.
                        // Search needs to be case insensitive.
                        if !traceParentHdrFound && bytes.EqualFold(key, traceParentHdrAsBytes) {
                                // We will now check if the value has sampling or not.
                                // TODO(ik): Not sure if this header can have multiple values
                                // or not, and if so, what would be the rule to check for
                                // sampling. What is done here is to check them all until we
                                // found one with sampling.
                                tk := bytes.Split(val, dashAsBytes)
                                if len(tk) == 4 && len([]byte(tk[3])) == 2 {
                                        if hexVal, err := strconv.ParseInt(bytesToString(tk[3]), 16, 8); err == nil {
                                                if hexVal&0x1 == 0x1 {
                                                        traceParentHdrFound = true
                                                }
                                        }
                                }
                        } else if !traceDestHdrFound && bytes.Equal(key, traceDestHdrAsBytes) {
                                // This is the Nats-Trace-Dest header, check the value to see
                                // if it indicates that the trace was disabled.
                                if bytes.Equal(val, traceDestDisabledAsBytes) {
                                        return nil, false
                                }
                                traceDestHdrFound = true
                        }
                        // Add to the keys and preserve the key's case
                        keys = append(keys, key)
                }
                i += nl + 2
        }
        if !traceDestHdrFound && !traceParentHdrFound {
                return nil, false
        }
        m = make(map[string][]string, len(keys))
        for i, k := range keys {
                hname := string(k)
                m[hname] = append(m[hname], string(vals[i]))
        }
        return m, !traceDestHdrFound && traceParentHdrFound
}

// Special case where we create a trace event before parsing the message.
// This is for cases where the connection will be closed when detecting
// an error during early message processing (for instance max payload).
func (c *client) initAndSendIngressErrEvent(hdr []byte, dest string, ingressError error) {
        if ingressError == nil {
                return
        }
        ct := getAcceptEncoding(hdr)
        t := &msgTrace{
                srv:  c.srv,
                acc:  c.acc,
                dest: dest,
                ct:   ct,
                event: &MsgTraceEvent{
                        Request: MsgTraceRequest{MsgSize: c.pa.size},
                        Events: append(MsgTraceEvents(nil), &MsgTraceIngress{
                                MsgTraceBase: MsgTraceBase{
                                        Type:      MsgTraceIngressType,
                                        Timestamp: time.Now(),
                                },
                                Kind:  c.kind,
                                CID:   c.cid,
                                Name:  getConnName(c),
                                Error: ingressError.Error(),
                        }),
                },
        }
        t.sendEvent()
}

// Returns `true` if message tracing is enabled and we are tracing only,
// that is, we are not going to deliver the inbound message, returns
// `false` otherwise (no tracing, or tracing and message delivery).
func (t *msgTrace) traceOnly() bool {
        return t != nil && t.tonly
}

func (t *msgTrace) setOriginAccountHeaderIfNeeded(c *client, acc *Account, msg []byte) []byte {
        var oan string
        // If t.acc is set, only check that, not t.oan.
        if t.acc != nil {
                if t.acc != acc {
                        oan = t.acc.GetName()
                }
        } else if t.oan != acc.GetName() {
                oan = t.oan
        }
        if oan != _EMPTY_ {
                msg = c.setHeader(MsgTraceOriginAccount, oan, msg)
        }
        return msg
}

func (t *msgTrace) setHopHeader(c *client, msg []byte) []byte {
        e := t.event
        e.Hops++
        if len(t.hop) > 0 {
                t.nhop = fmt.Sprintf("%s.%d", t.hop, e.Hops)
        } else {
                t.nhop = fmt.Sprintf("%d", e.Hops)
        }
        return c.setHeader(MsgTraceHop, t.nhop, msg)
}

func (t *msgTrace) setIngressError(err string) {
        if i := t.event.Ingress(); i != nil {
                i.Error = err
        }
}

func (t *msgTrace) addSubjectMappingEvent(subj []byte) {
        if t == nil {
                return
        }
        t.event.Events = append(t.event.Events, &MsgTraceSubjectMapping{
                MsgTraceBase: MsgTraceBase{
                        Type:      MsgTraceSubjectMappingType,
                        Timestamp: time.Now(),
                },
                MappedTo: string(subj),
        })
}

func (t *msgTrace) addEgressEvent(dc *client, sub *subscription, err string) {
        if t == nil {
                return
        }
        e := &MsgTraceEgress{
                MsgTraceBase: MsgTraceBase{
                        Type:      MsgTraceEgressType,
                        Timestamp: time.Now(),
                },
                Kind:  dc.kind,
                CID:   dc.cid,
                Name:  getConnName(dc),
                Hop:   t.nhop,
                Error: err,
        }
        t.nhop = _EMPTY_
        // Specific to CLIENT connections...
        if dc.kind == CLIENT {
                // Set the subscription's subject and possibly queue name.
                e.Subscription = string(sub.subject)
                if len(sub.queue) > 0 {
                        e.Queue = string(sub.queue)
                }
        }
        if dc.kind == CLIENT || dc.kind == LEAF {
                if i := t.event.Ingress(); i != nil {
                        // If the Ingress' account is different from the destination's
                        // account, add the account name into the Egress trace event.
                        // This would happen with service imports.
                        if dcAccName := dc.acc.GetName(); dcAccName != i.Account {
                                e.Account = dcAccName
                        }
                }
        }
        t.event.Events = append(t.event.Events, e)
}

func (t *msgTrace) addStreamExportEvent(dc *client, to []byte) {
        if t == nil {
                return
        }
        dc.mu.Lock()
        accName := dc.acc.GetName()
        dc.mu.Unlock()
        t.event.Events = append(t.event.Events, &MsgTraceStreamExport{
                MsgTraceBase: MsgTraceBase{
                        Type:      MsgTraceStreamExportType,
                        Timestamp: time.Now(),
                },
                Account: accName,
                To:      string(to),
        })
}

func (t *msgTrace) addServiceImportEvent(accName, from, to string) {
        if t == nil {
                return
        }
        t.event.Events = append(t.event.Events, &MsgTraceServiceImport{
                MsgTraceBase: MsgTraceBase{
                        Type:      MsgTraceServiceImportType,
                        Timestamp: time.Now(),
                },
                Account: accName,
                From:    from,
                To:      to,
        })
}

func (t *msgTrace) addJetStreamEvent(streamName string) {
        if t == nil {
                return
        }
        t.js = &MsgTraceJetStream{
                MsgTraceBase: MsgTraceBase{
                        Type:      MsgTraceJetStreamType,
                        Timestamp: time.Now(),
                },
                Stream: streamName,
        }
        t.event.Events = append(t.event.Events, t.js)
}

func (t *msgTrace) updateJetStreamEvent(subject string, noInterest bool) {
        if t == nil {
                return
        }
        // JetStream event should have been created in addJetStreamEvent
        if t.js == nil {
                return
        }
        t.js.Subject = subject
        t.js.NoInterest = noInterest
        // Update the timestamp since this is more accurate than when it
        // was first added in addJetStreamEvent().
        t.js.Timestamp = time.Now()
}

func (t *msgTrace) sendEventFromJetStream(err error) {
        if t == nil {
                return
        }
        // JetStream event should have been created in addJetStreamEvent
        if t.js == nil {
                return
        }
        if err != nil {
                t.js.Error = err.Error()
        }
        t.sendEvent()
}

func (t *msgTrace) sendEvent() {
        if t == nil {
                return
        }
        if t.js != nil {
                ready := atomic.AddInt32(&t.ready, 1) == 2
                if !ready {
                        return
                }
        }
        t.srv.sendInternalAccountSysMsg(t.acc, t.dest, &t.event.Server, t.event, t.ct)
}

nats-io / nats-server / 25621091956

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous