from_nats

Consumes messages from a NATS JetStream subject.

from_nats subject:string, [url=secret, durable=string, count=int, tls=record,
          auth=record, _batch_size=int, _queue_capacity=int] { … }

Description

The from_nats operator consumes messages from a NATS JetStream subject. The operator passes each message payload as bytes to the subpipeline, then acknowledges the message after the subpipeline finishes successfully.

The NATS server must have a JetStream stream that captures the subject you consume from. The operator uses the default URL nats://localhost:4222 unless you provide url or configure plugins.nats.url.

subject: string

The NATS subject to consume from.

url = secret (optional)

The NATS server URL.

If the URL has no scheme, Tenzir uses nats:// by default or tls:// when TLS is enabled. Use nats://, tls://, ws://, or wss:// to select a specific transport.

durable = string (optional)

The durable consumer name to use for the JetStream subscription.

count = int (optional)

Exit successfully after consuming count messages.

auth = record (optional)

Authentication settings for the NATS connection. Each value can be a string or a secret.

Supported authentication records are:

• {token: secret("NATS_TOKEN")} for token authentication.
• {user: "alice", password: secret("NATS_PASSWORD")} for user/password authentication.
• {credentials: "/path/to/user.creds"} for NATS credentials files.
• {credentials: "/path/to/user.creds", seed: "/path/to/user.nk"} for credentials files with a separate seed file.
• {credentials_memory: secret("NATS_CREDS")} for credentials content stored in a secret.

tls = record (optional)

TLS configuration. Provide an empty record (tls={}) to enable TLS with defaults or set fields to customize it.

{
  skip_peer_verification: bool, // skip certificate verification.
  cacert: string,               // CA bundle to verify peers.
  certfile: string,             // client certificate to present.
  keyfile: string,              // private key for the client certificate.
  min_version: string,          // minimum TLS version (`"1.0"`, `"1.1"`, `"1.2"`, "1.3"`).
  ciphers: string,              // OpenSSL cipher list string.
  client_ca: string,            // CA to validate client certificates.
  require_client_cert,          // require clients to present a certificate.
}

The client_ca and require_client_cert options are only applied for operators that accept incoming client connections, and otherwise ignored.

Any value not specified in the record will either be picked up from the configuration or if not configured will not be used by the operator.

See the Node TLS Setup guide for more details.

NATS uses the standard Tenzir tls record. The nats.c library does not expose a minimum TLS version setting, so tls.min_version is accepted for record compatibility but ignored with a warning.

{ … }

The parsing subpipeline that receives each NATS message payload as bytes.

Inside the subpipeline, from_nats exposes the following metadata as dollar-variables:

Variable	Type	Description
$subject	string	The message subject.
$reply	string	The reply subject, or null when absent.
$headers	record	NATS headers as list<string> values.
$stream	string	The JetStream stream name, or null when absent.
$consumer	string	The JetStream consumer name, or null when absent.
$stream_sequence	uint64	The stream sequence number.
$consumer_sequence	uint64	The consumer sequence number.
$num_delivered	uint64	The message delivery count.
$num_pending	uint64	The number of pending messages for the consumer.
$timestamp	time	The JetStream message timestamp.

Examples

Consume JSON messages

from_nats "alerts" {
  read_json
}

Preserve NATS metadata

from_nats "alerts", durable="tenzir-alerts" {
  read_json
  nats_subject = $subject
  nats_stream = $stream
  nats_sequence = $stream_sequence
}

Connect with token authentication and TLS

from_nats "alerts",
  url="tls://nats.example.com:4222",
  auth={token: secret("NATS_TOKEN")},
  tls={}
{
  read_json
}

See Also

• to_nats
• Read from message brokers
• NATS