24949216239

Committed 24 Apr 2026 08:34AM UTC coverage: 80.645% (-2.4%) from 83.05%

Build # 24949216239

Build Type

push

github

Committed by

web-flow

Commit Message

(2.14) [ADDED] `RemoteLeafOpts.IgnoreDiscoveredServers` option (#8067)

For a given leafnode remote, if this is set to true, this remote will
ignore any server leafnode URLs returned by the hub, allowing the user
to fully manage the servers this remote can connect to.

Resolves #8002

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>

Coverage Stats

74685 of 92610 relevant lines covered (80.64%)

632737.46 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 / 24949216239

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