Standard library · Networking

net/nats

A NATS client with headers, request and reply, and JetStream streams, consumers and acknowledgements.

import net.nats

On this page

Headers
JetStream
Subjects and wildcards
Request and reply
Queue groups and a clean exit
Types
Functions

NATS is a small, fast message bus. net/nats is a client for it written in Salam, with no native library. Connect opens a Conn from a URL such as nats://127.0.0.1:4222, and ConnectWith adds an Options value (start from DefaultOptions), you publish to a subject, subscribe to one, or send a request and wait for the reply. Subjects are dotted names such as orders.created, and ValidSubject tells you whether a name is acceptable before you use it.

Headers

A Header carries key and value pairs next to the payload. NewHeader makes one, HeaderAdd appends a value, HeaderSet replaces the values of a key, HeaderGet and HeaderValues read them, and HeaderEncode and HeaderParse turn a header to and from its wire form.

net-nats-headers.salam

import net.nats

func main:
    mut h := nats.NewHeader()
    nats.HeaderAdd(h, "Trace", "a1")
    nats.HeaderAdd(h, "Trace", "b2")
    nats.HeaderSet(h, "Tenant", "acme")
    println nats.HeaderGet(h, "Tenant"), nats.HeaderCount(h), nats.HeaderHas(h, "Trace")
    println nats.ValidSubject("orders.created"), nats.ValidSubject("bad subject")
end

Outputacme 3 true true false

JetStream

JetStream keeps messages. AddStream creates a stream over some subjects, JSPublish publishes and returns a PubAck once the stream has stored the message, and AddConsumer with Fetch reads messages back in batches. After handling a message call Ack; Nak asks for a redelivery, InProgress extends the wait and Term gives up on the message. StreamInfo, PurgeStream and DeleteStream inspect and clean up.

Subjects and wildcards

A subscription may use two wildcards: * stands for exactly one word and > for one or more words at the end. So orders.* matches orders.created but not orders.created.eu, and orders.> matches both but not orders alone. SubjectMatches applies the same rule locally, which makes it handy in tests. ValidSubject rejects an empty name, spaces and control characters, a leading or trailing dot and an empty word between two dots. Neither call needs a server.

net-nats-guide-subjects.salam

import net.nats

func show(pattern: str, subject: str):
    println pattern, subject, nats.SubjectMatches(pattern, subject)
end

func main:
    show("orders.*", "orders.created")
    show("orders.*", "orders.created.eu")
    show("orders.>", "orders.created.eu")
    show("orders.>", "orders")
    show("*.created", "orders.created")
    show("orders.created", "orders.created")
    println nats.ValidSubject("orders..created"), nats.ValidSubject(".orders"), nats.ValidSubject("orders.")
    println nats.ValidSubject(""), nats.ValidSubject("a b"), nats.ValidSubject("orders.>")
end

Outputorders.* orders.created true orders.* orders.created.eu false orders.> orders.created.eu true orders.> orders false *.created orders.created true orders.created orders.created true false false false false false true

Request and reply

Request sends a message with a private reply subject and waits for one answer. It returns a Msg, so check ok before reading data. When nothing is subscribed to the subject the server says so and err is no responders; when somebody is subscribed but does not answer in time it is timeout. SubscribeReplier takes a function from a message to the reply text. Nothing runs in the background: handlers run while the connection is being pumped, which Request, NextMsg, Flush and Pump all do. To answer by hand, subscribe with Subscribe and call Respond with the incoming message.

net-nats-guide-request.salam

import net.nats

func main:
    mut nc := nats.Connect("nats://127.0.0.1:4222")
    if not nats.IsConnected(nc):
        println "cannot connect:", nats.LastError(nc)
        ret 1
    end
    _s := nats.SubscribeReplier(nc, "svc.shout", "", (m: nats.Msg): ret m.data + "!" end)

    r := nats.Request(nc, "svc.shout", "hello", 1000)
    if r.ok:
        println "reply:", r.data
    else:
        println "failed:", r.err
    end

    gone := nats.Request(nc, "svc.nobody", "x", 1000)
    println gone.ok, gone.err

    slow := nats.Request(nc, "svc.silent", "x", 150)
    println slow.ok, slow.err

    nats.Close(nc)
    ret 0
end

Queue groups and a clean exit

Subscriptions that share a queue name are one group, and the server delivers each message to just one of them, which is how several workers split a stream of jobs. Which member gets which message is up to the server, so count what was handled, not who handled it. NextMsg waits up to the given milliseconds and returns a message with ok false and err set to timeout when nothing arrives. AutoUnsubscribe removes a subscription after that many deliveries. Drain stops the subscriptions, flushes what is pending and closes the connection, so it is the right last call.

net-nats-guide-queue.salam

import net.nats

func main:
    mut nc := nats.Connect("nats://127.0.0.1:4222")
    if not nats.IsConnected(nc):
        println "cannot connect:", nats.LastError(nc)
        ret 1
    end
    a := nats.QueueSubscribe(nc, "jobs", "workers")
    b := nats.QueueSubscribe(nc, "jobs", "workers")
    _ok := nats.AutoUnsubscribe(nc, a, 10)
    repeat 6 in i:
        _sent := nats.Publish(nc, "jobs", "job " + i)
    end
    _flushed := nats.Flush(nc, 1000)

    mut handled := 0
    repeat 2 in k:
        s := k == 0 ? a : b
        mut m := nats.NextMsg(nc, s, 200)
        until m.ok:
            handled += 1
            m = nats.NextMsg(nc, s, 200)
        end
    end
    println "handled", handled, "subscriptions", nats.SubscriptionCount(nc)
    println "drained", nats.Drain(nc, 1000)
    ret 0
end

Types

`JSResult`

struct JSResult

The reply to a JetStream API request.

Name	Type	Description
`ok`	`bool = false`	The server accepted the request.
`err`	`str = ""`	The error text, empty on success.
`code`	`int = 0`	The JetStream error code, 0 when there is none.
`raw`	`str = ""`	The JSON reply as the server sent it.

net-nats-JetStream.salam

import net.nats

func main(): int:
    mut nc := nats.Connect("nats://127.0.0.1:4222")
    if not nats.IsConnected(nc):
        println "cannot connect:", nats.LastError(nc)
        ret 1
    end
    mut subjects := Vector {} as Vector<str>
    defer subjects.free()
    subjects.push("orders.*")
    made := nats.AddStream(nc, "ORDERS", subjects, 2000)
    println "stream:", made.ok, made.err
    ack := nats.JSPublish(nc, "orders.new", "order 1", 2000)
    println "stored at", ack.stream, ack.seq

    mut h := nats.NewHeader()
    nats.HeaderAdd(h, "Nats-Msg-Id", "order-2")
    again := nats.JSPublishHeaders(nc, "orders.new", h, "order 2", 2000)
    println again.ok, again.duplicate

    info := nats.StreamInfo(nc, "ORDERS", 2000)
    println info.messages, info.first_seq, info.last_seq, info.consumers
    raw := nats.JSRequest(nc, "STREAM.NAMES", "", 2000)
    println raw.ok

    nats.AddConsumer(nc, "ORDERS", "worker", "orders.*", 2000)
    batch := nats.Fetch(nc, "ORDERS", "worker", 10, 1000)
    each m in batch.msgs:
        println m.data, "number", nats.MsgSequence(m)
        if m.data == "order 1":
            nats.Ack(nc, m)
        else m.data == "order 2":
            nats.Nak(nc, m)
        else:
            nats.Term(nc, m)
        end
    end
    if batch.msgs.len() > 0:
        nats.InProgress(nc, batch.msgs.get(0))
    end

    nats.DeleteConsumer(nc, "ORDERS", "worker", 2000)
    nats.PurgeStream(nc, "ORDERS", 2000)
    nats.DeleteStream(nc, "ORDERS", 2000)
    nats.Close(nc)
    ret 0
end

`StreamState`

struct StreamState

The state of a stream, as StreamInfo returns it.

Name	Type	Description
`ok`	`bool = false`	The stream was found and read.
`err`	`str = ""`	The error text, empty on success.
`name`	`str = ""`	The name of the stream.
`messages`	`int = 0`	How many messages the stream holds.
`bytes`	`int = 0`	How many bytes the stream holds.
`first_seq`	`int = 0`	The sequence number of the oldest message.
`last_seq`	`int = 0`	The sequence number of the newest message.
`consumers`	`int = 0`	How many consumers the stream has.

The example for this item is shown with the ones above: net-nats-JetStream.salam

`PubAck`

struct PubAck

The acknowledgement of a message stored in a stream.

Name	Type	Description
`ok`	`bool = false`	The stream stored the message.
`err`	`str = ""`	The error text, empty on success.
`stream`	`str = ""`	The stream that stored the message.
`seq`	`int = 0`	The sequence number of the message in the stream.
`duplicate`	`bool = false`	The stream had this message already.

The example for this item is shown with the ones above: net-nats-JetStream.salam

`Batch`

struct Batch

The messages one Fetch call returned.

Name	Type	Description
`msgs`	`Vector`	The messages, possibly fewer than were asked for.
`err`	`str = ""`	The error text, empty on success.

The example for this item is shown with the ones above: net-nats-JetStream.salam

`KVEntry`

struct KVEntry

The result of reading a key from a key value bucket.

Name	Type	Description
`ok`	`bool = false`	The request itself succeeded, even when the key does not exist.
`found`	`bool = false`	The key exists and `value` holds its value.
`deleted`	`bool = false`	The last operation on the key removed it.
`key`	`str = ""`	The key that was read.
`value`	`str = ""`	The value of the key.
`revision`	`int = 0`	The stream sequence number of this revision of the key.
`err`	`str = ""`	The error text, such as `key not found`.

net-nats-KV.salam

import net.nats

func main(): int:
    mut nc := nats.Connect("nats://127.0.0.1:4222")
    if not nats.IsConnected(nc):
        println "cannot connect:", nats.LastError(nc)
        ret 1
    end
    made := nats.KVCreate(nc, "config", 5, 2000)
    println "bucket:", made.ok
    put := nats.KVPut(nc, "config", "theme", "dark", 2000)
    println "revision", put.seq
    got := nats.KVGet(nc, "config", "theme", 2000)
    if got.found:
        println got.key, got.value, got.revision
    end
    nats.KVDelete(nc, "config", "theme", 2000)
    gone := nats.KVGet(nc, "config", "theme", 2000)
    println gone.found, gone.deleted
    nats.KVDeleteBucket(nc, "config", 2000)
    nats.Close(nc)
    ret 0
end

`Header`

struct Header

Key and value pairs carried next to a payload. A key may appear several times.

Name	Type	Description
`keys`	`Vector`	The keys, in the order they were added.
`values`	`Vector`	The values, in the same order as the keys.

net-nats-Headers.salam

import net.nats
import str

func main:
    mut h := nats.NewHeader()
    nats.HeaderAdd(h, "Trace-Id", "abc")
    nats.HeaderAdd(h, "Tag", "one")
    nats.HeaderAdd(h, "Tag", "two")
    println nats.HeaderGet(h, "trace-id"), nats.HeaderValues(h, "TAG").len(), nats.HeaderHas(h, "nope")
    nats.HeaderSet(h, "Tag", "only")
    nats.HeaderDel(h, "Trace-Id")
    println nats.HeaderCount(h)
    println str.Replace(nats.HeaderEncode(h), "\r\n", "|")

    mut code := 0
    mut text := ""
    back := nats.HeaderParse("NATS/1.0 503 No Responders\r\nA: 1\r\n\r\n", code, text)
    println code, text, nats.HeaderGet(back, "a")
end

Outputabc 2 false 1 NATS/1.0|Tag: only|| 503 No Responders 1

`Msg`

struct Msg

One message, received or about to be published.

Name	Type	Description
`ok`	`bool = false`	A message arrived; false after a timeout or an error.
`err`	`str = ""`	The error text, such as `timeout`.
`subject`	`str = ""`	The subject of the message.
`reply`	`str = ""`	The subject to send a reply to, empty when none is expected.
`data`	`str = ""`	The payload.
`header`	`Header`	The headers of the message.
`sid`	`int = 0`	The id of the subscription that received it.
`status`	`int = 0`	The status code on a status message such as 503 for no responders, otherwise 0.
`status_text`	`str = ""`	The text next to the status code.

net-nats-Handlers.salam

import net.nats

mut seen := 0

func note(m: nats.Msg):
    seen += 1
    println "event", m.subject, m.data
end

func shout(m: nats.Msg): str:
    ret "HELLO " + m.data
end

func main(): int:
    mut nc := nats.Connect("nats://127.0.0.1:4222")
    if not nats.IsConnected(nc):
        println "cannot connect:", nats.LastError(nc)
        ret 1
    end
    nats.SubscribeHandler(nc, "events.>", "", (m: nats.Msg): note(m) end)
    nats.SubscribeReplier(nc, "greet", "", (m: nats.Msg): ret shout(m) end)
    nats.Publish(nc, "events.a.b", "started")

    mut fwd := nats.Msg {}
    fwd.subject = "events.copy"
    fwd.data = "forwarded"
    nats.PublishMsg(nc, fwd)

    repeat 20:
        nats.Pump(nc, 100)
    end
    println "handled", seen
    nats.Close(nc)
    ret 0
end

`Subscription`

struct Subscription

A subscription handle returned by Subscribe.

Name	Type	Description
`ok`	`bool = false`	The subscription was created.
`sid`	`int = 0`	The subscription id.
`subject`	`str = ""`	The subject or pattern it listens on.
`queue`	`str = ""`	The queue group, empty when there is none.

net-nats-PubSub.salam

import net.nats

func main(): int:
    mut nc := nats.Connect("nats://127.0.0.1:4222")
    if not nats.IsConnected(nc):
        println "cannot connect:", nats.LastError(nc)
        ret 1
    end
    all := nats.Subscribe(nc, "orders.*")
    first := nats.QueueSubscribe(nc, "jobs", "workers")
    once := nats.Subscribe(nc, "ping")
    nats.AutoUnsubscribe(nc, once, 1)
    println "subscriptions:", nats.SubscriptionCount(nc)

    nats.Publish(nc, "orders.new", "order 1")
    nats.Publish(nc, "jobs", "job 1")
    nats.Publish(nc, "ping", "hello")
    nats.Flush(nc, 1000)
    println "waiting:", nats.Pending(nc, all), nats.Pending(nc, first)

    m := nats.NextMsg(nc, all, 1000)
    if m.ok:
        println m.subject, m.data
    end
    p := nats.NextMsg(nc, once, 1000)
    println p.data, nats.Delivered(nc, once), nats.IsValidSubscription(nc, once)

    nats.Unsubscribe(nc, all)
    nats.Close(nc)
    ret 0
end

`ServerInfo`

struct ServerInfo

What the server said about itself when the connection opened.

Name	Type	Description
`server_id`	`str = ""`	The unique id of the server.
`server_name`	`str = ""`	The configured name of the server.
`version`	`str = ""`	The server version.
`host`	`str = ""`	The host the server reports.
`port`	`int = 0`	The port the server reports.
`proto`	`int = 0`	The protocol version.
`client_id`	`int = 0`	The id the server gave this connection.
`max_payload`	`int = 1048576`	The largest payload the server accepts, in bytes.
`headers`	`bool = false`	The server supports message headers.
`tls_required`	`bool = false`	The server requires TLS.
`auth_required`	`bool = false`	The server requires authentication.
`jetstream`	`bool = false`	JetStream is enabled on the server.
`raw`	`str = ""`	The INFO line as the server sent it.

net-nats-Connect.salam

import net.nats

func main(): int:
    mut opts := nats.DefaultOptions()
    opts.name = "billing"
    opts.timeout_ms = 3000
    mut nc := nats.ConnectWith("nats://127.0.0.1:4222", opts)
    if not nats.IsConnected(nc):
        println "cannot connect:", nats.LastError(nc)
        ret 1
    end
    println "connected, largest payload", nats.MaxPayload(nc)
    nats.Close(nc)
    println nats.IsClosed(nc)
    ret 0
end

`Options`

struct Options

Connection settings. Start from DefaultOptions.

Name	Type	Description
`name`	`str = ""`	A name for this client, shown in server monitoring.
`user`	`str = ""`	The user name.
`password`	`str = ""`	The password.
`token`	`str = ""`	An authentication token, used instead of a user and password.
`no_echo`	`bool = false`	Do not deliver this connection's own messages back to it.
`verbose`	`bool = false`	Ask the server to acknowledge every protocol command.
`pedantic`	`bool = false`	Ask the server for stricter protocol checks.
`secure`	`bool = false`	Use TLS.
`insecure_skip_verify`	`bool = false`	Accept any TLS certificate. Use this only for testing.
`timeout_ms`	`int = 5000`	The time allowed for connecting, in milliseconds.
`ping_interval_ms`	`int = 120000`	How often to ping the server, in milliseconds.
`max_pings_out`	`int = 2`	How many unanswered pings mean the connection is dead.
`inbox_prefix`	`str = "_INBOX"`	The subject prefix of `NewInbox`.
`servers`	`Vector`	More server URLs to try when connecting or reconnecting.
`max_reconnects`	`int = 0`	How many times to reconnect after the connection drops; 0 turns reconnecting off.
`reconnect_wait_ms`	`int = 2000`	How long to wait between reconnect attempts, in milliseconds.
`reconnect_buffer`	`int = 8388608`	How many bytes of outgoing messages to keep while reconnecting.

The example for this item is shown with the ones above: net-nats-Connect.salam

`Stats`

struct Stats

Traffic counters of a connection.

Name	Type	Description
`in_msgs`	`i64 = 0`	Messages received.
`out_msgs`	`i64 = 0`	Messages sent.
`in_bytes`	`i64 = 0`	Payload bytes received.
`out_bytes`	`i64 = 0`	Payload bytes sent.

`Conn`

struct Conn

A connection to a NATS server. It reads from the socket only when you call a function that waits, such as NextMsg or Pump.

Name	Type	Description
`ok`	`bool = false`	The connection was established.
`closed`	`bool = false`	The connection has been closed.
`draining`	`bool = false`	`Drain` is in progress.
`err`	`str = ""`	The last error text.
`url`	`str = ""`	The URL of the current server.
`info`	`ServerInfo`	What the server said about itself.
`opts`	`Options`	The settings of this connection.
`stats`	`Stats`	The traffic counters of this connection.

The example for this item is shown with the ones above: net-nats-Connect.salam

Functions

`JSRequest`

func JSRequest(nc &: Conn, api: str, body: str, timeout_ms: int): JSResult

Sends a request to the JetStream API, such as STREAM.INFO.orders, and returns the reply.

The example for this item is shown with the ones above: net-nats-JetStream.salam

`AddStream`

func AddStream(nc &: Conn, name: str, subjects: Vector<str>, timeout_ms: int): JSResult

Creates a stream over these subjects.

The example for this item is shown with the ones above: net-nats-JetStream.salam

`DeleteStream`

func DeleteStream(nc &: Conn, name: str, timeout_ms: int): JSResult

Deletes a stream and its messages.

The example for this item is shown with the ones above: net-nats-JetStream.salam

`PurgeStream`

func PurgeStream(nc &: Conn, name: str, timeout_ms: int): JSResult

Removes every message from a stream and keeps the stream.

The example for this item is shown with the ones above: net-nats-JetStream.salam

`StreamInfo`

func StreamInfo(nc &: Conn, name: str, timeout_ms: int): StreamState

Reads the state of a stream.

The example for this item is shown with the ones above: net-nats-JetStream.salam

`JSPublish`

func JSPublish(nc &: Conn, subject: str, data: str, timeout_ms: int): PubAck

Publishes a message and waits until a stream has stored it.

The example for this item is shown with the ones above: net-nats-JetStream.salam

`JSPublishHeaders`

func JSPublishHeaders(nc &: Conn, subject: str, h: Header, data: str, timeout_ms: int): PubAck

Like JSPublish, with headers.

The example for this item is shown with the ones above: net-nats-JetStream.salam

`AddConsumer`

func AddConsumer(nc &: Conn, stream: str, durable: str, filter: str, timeout_ms: int): JSResult

Creates a durable pull consumer on a stream. filter limits it to one subject and may be empty.

The example for this item is shown with the ones above: net-nats-JetStream.salam

`DeleteConsumer`

func DeleteConsumer(nc &: Conn, stream: str, durable: str, timeout_ms: int): JSResult

Deletes a consumer.

The example for this item is shown with the ones above: net-nats-JetStream.salam

`Fetch`

func Fetch(nc &: Conn, stream: str, durable: str, batch: int, timeout_ms: int): Batch

Asks a pull consumer for up to batch messages and waits until they arrive or the time runs out.

The example for this item is shown with the ones above: net-nats-JetStream.salam

`Ack`

func Ack(nc &: Conn, m: Msg): bool

Tells the server that the message has been handled.

The example for this item is shown with the ones above: net-nats-JetStream.salam

`Nak`

func Nak(nc &: Conn, m: Msg): bool

Asks the server to deliver the message again.

The example for this item is shown with the ones above: net-nats-JetStream.salam

`Term`

func Term(nc &: Conn, m: Msg): bool

Tells the server never to deliver the message again.

The example for this item is shown with the ones above: net-nats-JetStream.salam

`InProgress`

func InProgress(nc &: Conn, m: Msg): bool

Tells the server that work on the message continues, which extends its acknowledgement deadline.

The example for this item is shown with the ones above: net-nats-JetStream.salam

`MsgSequence`

func MsgSequence(m: Msg): int

The stream sequence number of a JetStream message, read from its reply subject; 0 for any other message.

The example for this item is shown with the ones above: net-nats-JetStream.salam

`AddPushConsumer`

func AddPushConsumer(nc &: Conn, stream: str, durable: str, filter: str, deliver: str, queue: str, timeout_ms: int): JSResult

Creates a durable push consumer that delivers to the subject deliver; subscribe to that subject to receive the messages.

net-nats-Push.salam

import net.nats

func main(): int:
    mut nc := nats.Connect("nats://127.0.0.1:4222")
    if not nats.IsConnected(nc):
        println "cannot connect:", nats.LastError(nc)
        ret 1
    end
    sub := nats.Subscribe(nc, "deliver.orders")
    made := nats.AddPushConsumer(nc, "ORDERS", "pusher", "orders.*", "deliver.orders", "", 2000)
    if not made.ok:
        println "consumer:", made.err
        ret 1
    end
    until true:
        m := nats.NextPush(nc, sub, 2000)
        if not m.ok:
            break
        end
        println m.data, nats.MsgSequence(m)
        nats.Ack(nc, m)
    end
    nats.Close(nc)
    ret 0
end

`NextPush`

func NextPush(nc &: Conn, s: Subscription, timeout_ms: int): Msg

The next message of a push subscription. It answers idle heartbeats by itself and never returns them.

The example for this item is shown with the ones above: net-nats-Push.salam

`KVCreate`

func KVCreate(nc &: Conn, bucket: str, history: int, timeout_ms: int): JSResult

Creates a key value bucket that keeps history revisions of each key.

The example for this item is shown with the ones above: net-nats-KV.salam

`KVDeleteBucket`

func KVDeleteBucket(nc &: Conn, bucket: str, timeout_ms: int): JSResult

Deletes a bucket and its keys.

The example for this item is shown with the ones above: net-nats-KV.salam

`KVPut`

func KVPut(nc &: Conn, bucket: str, key: str, value: str, timeout_ms: int): PubAck

Writes a key and returns the acknowledgement; its seq is the new revision.

The example for this item is shown with the ones above: net-nats-KV.salam

`KVGet`

func KVGet(nc &: Conn, bucket: str, key: str, timeout_ms: int): KVEntry

Reads the newest revision of a key.

The example for this item is shown with the ones above: net-nats-KV.salam

`KVDelete`

func KVDelete(nc &: Conn, bucket: str, key: str, timeout_ms: int): PubAck

Deletes a key by writing a delete marker, so its history stays until the bucket limit removes it.

The example for this item is shown with the ones above: net-nats-KV.salam

`DefaultOptions`

func DefaultOptions(): Options

The default settings, ready to be changed.

The example for this item is shown with the ones above: net-nats-Connect.salam

`NewHeader`

func NewHeader(): Header

Creates an empty header.

The example for this item is shown with the ones above: net-nats-Headers.salam

`HeaderAdd`

func HeaderAdd(h &: Header, key: str, value: str)

Adds a value to a key and keeps the values it already has.

The example for this item is shown with the ones above: net-nats-Headers.salam

`HeaderDel`

func HeaderDel(h &: Header, key: str)

Removes a key and all its values.

The example for this item is shown with the ones above: net-nats-Headers.salam

`HeaderSet`

func HeaderSet(h &: Header, key: str, value: str)

Replaces the values of a key with this one.

The example for this item is shown with the ones above: net-nats-Headers.salam

`HeaderGet`

func HeaderGet(h: Header, key: str): str

The first value of a key, or an empty text.

The example for this item is shown with the ones above: net-nats-Headers.salam

`HeaderValues`

func HeaderValues(h: Header, key: str): Vector<str>

All the values of a key.

The example for this item is shown with the ones above: net-nats-Headers.salam

`HeaderHas`

func HeaderHas(h: Header, key: str): bool

True when the key exists.

The example for this item is shown with the ones above: net-nats-Headers.salam

`HeaderCount`

func HeaderCount(h: Header): int

The number of key and value pairs.

The example for this item is shown with the ones above: net-nats-Headers.salam

`HeaderEncode`

func HeaderEncode(h: Header): str

The header in its wire form.

The example for this item is shown with the ones above: net-nats-Headers.salam

`HeaderParse`

func HeaderParse(block: str, status &: int, status_text &: str): Header

Reads a header from its wire form, and sets the status code and text when the block has them.

The example for this item is shown with the ones above: net-nats-Headers.salam

`NewInbox`

func NewInbox(nc: Conn): str

A new unique subject for receiving replies.

net-nats-Request.salam

import net.nats

func main(): int:
    mut nc := nats.Connect("nats://127.0.0.1:4222")
    if not nats.IsConnected(nc):
        println "cannot connect:", nats.LastError(nc)
        ret 1
    end
    r := nats.Request(nc, "time.now", "", 1000)
    if r.ok:
        println "answer:", r.data
    else:
        println "no answer:", r.err
    end

    mut h := nats.NewHeader()
    nats.HeaderAdd(h, "Trace-Id", "abc-123")
    r2 := nats.RequestMsg(nc, "svc.echo", h, "ping", 1000)
    println r2.ok, r2.data

    inbox := nats.NewInbox(nc)
    nats.PublishRequest(nc, "svc.echo", inbox, "fire")
    reply := nats.Subscribe(nc, inbox)
    println nats.NextMsg(nc, reply, 500).ok

    println "drained:", nats.Drain(nc, 1000)
    ret 0
end

`ValidSubject`

func ValidSubject(subject: str): bool

True when the text can be used as a subject to publish on.

net-nats-Subjects.salam

import net.nats

func main:
    println nats.ValidSubject("orders.eu.new"), nats.ValidSubject("orders..new"), nats.ValidSubject("a b")
    println nats.SubjectMatches("orders.*", "orders.new"), nats.SubjectMatches("orders.*", "orders.eu.new")
    println nats.SubjectMatches("orders.>", "orders.eu.new"), nats.SubjectMatches("orders.>", "orders")
end

Outputtrue false false true false true false

`SubjectMatches`

func SubjectMatches(pattern: str, subject: str): bool

True when the subject matches the pattern. * stands for one token and > for one or more at the end.

The example for this item is shown with the ones above: net-nats-Subjects.salam

`Pump`

func Pump(nc &: Conn, timeout_ms: int): int

Reads from the socket for up to timeout_ms, handles what arrived and returns how many messages it handled. Handlers run from here, so call it in a loop when you use them.

The example for this item is shown with the ones above: net-nats-Handlers.salam

`Connect`

func Connect(url: str): Conn

Connects to a server with the default settings.

The example for this item is shown with the ones above: net-nats-PubSub.salam

`ConnectWith`

func ConnectWith(url: str, opts: Options): Conn

Connects to a server with these settings.

The example for this item is shown with the ones above: net-nats-Connect.salam

`IsConnected`

func IsConnected(nc: Conn): bool

True while the connection is usable.

The example for this item is shown with the ones above: net-nats-Connect.salam

`IsClosed`

func IsClosed(nc: Conn): bool

True after the connection has been closed.

The example for this item is shown with the ones above: net-nats-Connect.salam

`LastError`

func LastError(nc: Conn): str

The last error text.

The example for this item is shown with the ones above: net-nats-Connect.salam

`Close`

func Close(nc &: Conn)

Closes the connection at once. Use Drain to finish outstanding work first.

The example for this item is shown with the ones above: net-nats-Connect.salam

`Publish`

func Publish(nc &: Conn, subject: str, data: str): bool

Publishes a message and returns false when it could not be written.

The example for this item is shown with the ones above: net-nats-PubSub.salam

`PublishRequest`

func PublishRequest(nc &: Conn, subject: str, reply: str, data: str): bool

Publishes a message and names the subject the reply should go to.

The example for this item is shown with the ones above: net-nats-Request.salam

`PublishHeaders`

func PublishHeaders(nc &: Conn, subject: str, h: Header, data: str): bool

Publishes a message with headers.

`PublishMsg`

func PublishMsg(nc &: Conn, m: Msg): bool

Publishes a Msg with its subject, reply, headers and payload.

The example for this item is shown with the ones above: net-nats-Handlers.salam

`Respond`

func Respond(nc &: Conn, m: Msg, data: str): bool

Sends a reply to the reply subject of a message; false when it has none.

`Subscribe`

func Subscribe(nc &: Conn, subject: str): Subscription

Subscribes to a subject or pattern. Read the messages with NextMsg.

The example for this item is shown with the ones above: net-nats-PubSub.salam

`QueueSubscribe`

func QueueSubscribe(nc &: Conn, subject: str, queue: str): Subscription

Subscribes as a member of a queue group, so each message goes to one member only.

The example for this item is shown with the ones above: net-nats-PubSub.salam

`SubscribeHandler`

func SubscribeHandler(nc &: Conn, subject: str, queue: str, handler: func(Msg)): Subscription

Subscribes and calls this function for each message from inside Pump. queue may be empty.

The example for this item is shown with the ones above: net-nats-Handlers.salam

`SubscribeReplier`

func SubscribeReplier(nc &: Conn, subject: str, queue: str, handler: func(Msg): str): Subscription

Like SubscribeHandler, and the text the function returns is sent back as the reply.

The example for this item is shown with the ones above: net-nats-Handlers.salam

`Unsubscribe`

func Unsubscribe(nc &: Conn, s: Subscription): bool

Ends a subscription.

The example for this item is shown with the ones above: net-nats-PubSub.salam

`AutoUnsubscribe`

func AutoUnsubscribe(nc &: Conn, s: Subscription, max: int): bool

Ends a subscription by itself after max messages have been delivered.

The example for this item is shown with the ones above: net-nats-PubSub.salam

`IsValidSubscription`

func IsValidSubscription(nc: Conn, s: Subscription): bool

True while the subscription is active.

The example for this item is shown with the ones above: net-nats-PubSub.salam

`Delivered`

func Delivered(nc: Conn, s: Subscription): int

How many messages this subscription has received.

The example for this item is shown with the ones above: net-nats-PubSub.salam

`Pending`

func Pending(nc: Conn, s: Subscription): int

How many received messages wait to be read with NextMsg.

The example for this item is shown with the ones above: net-nats-PubSub.salam

`NextMsg`

func NextMsg(nc &: Conn, s: Subscription, timeout_ms: int): Msg

Waits for the next message of a subscription. On a timeout ok is false and err is timeout.

The example for this item is shown with the ones above: net-nats-PubSub.salam

`Flush`

func Flush(nc &: Conn, timeout_ms: int): bool

Waits until the server has processed everything sent so far.

The example for this item is shown with the ones above: net-nats-PubSub.salam

`Request`

func Request(nc &: Conn, subject: str, data: str, timeout_ms: int): Msg

Publishes a request and waits for one reply.

The example for this item is shown with the ones above: net-nats-Request.salam

`RequestMsg`

func RequestMsg(nc &: Conn, subject: str, h: Header, data: str, timeout_ms: int): Msg

Like Request, with headers.

The example for this item is shown with the ones above: net-nats-Request.salam

`Drain`

func Drain(nc &: Conn, timeout_ms: int): bool

Ends every subscription, waits for the server to process what was sent, and closes the connection.

The example for this item is shown with the ones above: net-nats-Request.salam

`MaxPayload`

func MaxPayload(nc: Conn): int

The largest payload the server accepts, in bytes.

The example for this item is shown with the ones above: net-nats-Connect.salam

`SubscriptionCount`

func SubscriptionCount(nc: Conn): int

The number of active subscriptions.

The example for this item is shown with the ones above: net-nats-PubSub.salam

`IsReconnecting`

func IsReconnecting(nc: Conn): bool

True while the connection is reconnecting.

net-nats-Reconnect.salam

import net.nats

func main(): int:
    mut opts := nats.DefaultOptions()
    opts.servers.push("nats://10.0.0.2:4222")
    opts.servers.push("nats://10.0.0.3:4222")
    opts.max_reconnects = 20
    opts.reconnect_wait_ms = 500
    opts.reconnect_buffer = 1048576
    mut nc := nats.ConnectWith("nats://10.0.0.1:4222", opts)
    if not nats.IsConnected(nc):
        println "cannot connect:", nats.LastError(nc)
        ret 1
    end
    sub := nats.Subscribe(nc, "metrics.>")
    until not nats.IsClosed(nc):
        m := nats.NextMsg(nc, sub, 1000)
        if m.ok:
            println m.subject, m.data
        end
        if nats.IsReconnecting(nc):
            println "link is down, buffered", nats.BufferedBytes(nc), "bytes"
            nats.Publish(nc, "audit", "waiting for the server")
        end
        if nats.IsClosed(nc):
            break
        end
    end
    println "reconnected", nats.Reconnects(nc), "times"
    nats.Close(nc)
    ret 0
end

`Reconnects`

func Reconnects(nc: Conn): int

How many times the connection has reconnected.

The example for this item is shown with the ones above: net-nats-Reconnect.salam

`BufferedBytes`

func BufferedBytes(nc: Conn): int

How many bytes of messages wait to be sent after a reconnect.

The example for this item is shown with the ones above: net-nats-Reconnect.salam