NATS Server Documentation

Access the following methods to retrieve real-time statistics about the server's connections and performance:

• NumClients(): Returns the number of registered clients.
• NumRoutes(): Returns the number of registered routes.
• NumRemotes(): Returns the number of registered remotes.
• NumLeafNodes(): Returns the number of leaf node connections.
• NumSubscriptions(): Returns the total number of active subscriptions.
• NumSlowConsumers(): Returns the total count of slow consumers.
• NumStalledClients(): Returns the total number of times clients have been stalled.
• NumStaleConnections(): Returns the number of stale connections.

For granular breakdowns, use the specific stats methods:

• NumSlowConsumersClients(), NumSlowConsumersRoutes(), NumSlowConsumersGateways(), NumSlowConsumersLeafs()
• NumStaleConnectionsClients(), NumStaleConnectionsRoutes(), NumStaleConnectionsGateways(), NumStaleConnectionsLeafs()

Sources: server/server.go

The consumer calculates the number of pending messages (numPending) based on the stream state and delivery policy. This is used to inform clients about how many messages are available for delivery.

Calculation Logic:

• The consumer checks the stream's LastSeq and compares it to the consumer's sseq (stream sequence).
• If sseq exceeds LastSeq, the pending count is reset to 0.
• For filtered consumers, the calculation uses NumPendingMulti with subject filters.
• For DeliverLastPerSubject policy, the calculation is adjusted to reflect the last message per subject.

Methods:

• streamNumPendingLocked(): Acquires the consumer lock and calls streamNumPending().
• streamNumPending(): Calculates the pending count based on delivery policy and filters.
• calculateNumPending(): Performs the actual calculation using the stream store.
• checkNumPending(): Validates the pending count against the stream state to prevent reporting more messages than exist.

Example Usage:

// Get the number of pending messages
pending, err := consumer.StreamNumPendingLocked()
if err != nil {
    // Handle error
}

Important Notes:

• The checkNumPending() method includes sanity checks to handle race conditions during stream deletion or message removal.
• If the consumer is not filtered, the calculation uses the stream's NumPending method with an empty filter.
• For filtered consumers, the calculation uses NumPendingMulti with the configured subject filters.

// Example: Calculate pending messages
func (o *consumer) streamNumPending() (uint64, error) {
    if o.mset == nil || o.mset.store == nil {
        o.npc, o.npf = 0, 0
        return 0, nil
    }
    npc, npf, err := o.calculateNumPending()
    if err != nil {
        return 0, err
    }
    o.npc, o.npf = int64(npc), npf
    return o.numPending(), nil
}

// Example: Check pending count sanity
func (o *consumer) checkNumPending() (uint64, error) {
    if o.mset != nil && o.mset.store != nil {
        var state StreamState
        o.mset.store.FastState(&state)
        npc := o.numPending()
        if o.sseq > state.LastSeq {
            o.npc = 0
        } else if npc > 0 {
            o.npc = int64(min(npc, state.Msgs, state.LastSeq-o.sseq+1))
        }
    }
    return o.numPending(), nil
}

Sources: server/consumer.go

1. Create stream in source account:

_, err := js.AddStream(&nats.StreamConfig{
    Name: "ORDERS",
    Subjects: []string{"foo"},
    Storage: nats.MemoryStorage,
})

2. Create consumer with delivery subject:

_, err = js.AddConsumer("ORDERS", &nats.ConsumerConfig{
    DeliverSubject: "deliver.ORDERS",
    AckPolicy: nats.AckExplicitPolicy,
})

3. Subscribe in importing account with remapped subject:

// Messages published to "foo" arrive as "bar.ORDERS" in importing account
sub, err := nc2.SubscribeSync("bar.ORDERS")

The to field in the import configuration remaps the subject from the source account to the importing account's namespace.

imports [
    { stream: { account: JS, subject: "deliver.ORDERS" }, to: "d.*" }
    { stream: { account: JS, subject: "foo.*" }, to: "bar.*" }
]

// Consumer receives messages with remapped subject
sub, err := nc2.SubscribeSync("bar.ORDERS")

Sources: server/jetstream_test.go