NATS

Synopsis

Creates a JetStream consumer that connects to NATS servers and processes messages from specified streams and subjects. Supports authentication, TLS encryption, and multiple workers with automatic message acknowledgment.

For details, see Appendix.

Schema

- id: <numeric>
  name: <string>
  description: <string>
  type: nats
  tags: <string[]>
  pipelines: <pipeline[]>
  status: <boolean>
  properties:
    address: <string>
    port: <numeric>
    username: <string>
    password: <string>
    auth_type: <string>
    stream: <string>
    consumer: <string>
    subject: <string>
    max_bytes: <numeric>
    timeout: <numeric>
    batch_size: <numeric>
    reuse: <boolean>
    workers: <numeric>
    tls:
      status: <boolean>
      cert_name: <string>
      key_name: <string>
      passphrase: <string>
      ca_name: <string>
      server_name: <string>
      min_tls_version: <string>
      max_tls_version: <string>
      insecure_skip_verify: <boolean>

Configuration

The following fields are used to define the device:

Device

Field	Required	Default	Description
id	Y	-	Unique numeric identifier
name	Y	-	Device name
description	N	-	Optional description
type	Y	-	Must be nats
tags	N	-	Optional tags
pipelines	N	-	Optional pre-processor pipelines
status	N	true	Enable/disable the device

Connection

Field	Required	Default	Description
address	N	"0.0.0.0"	NATS server address
port	Y	4222	NATS server port
username	N	-	Authentication username (or JWT for jwt auth, or NKey identity for nkey auth)
password	N	-	Authentication password (or JWT seed for jwt auth, or NKey seed for nkey auth)
auth_type	N	-	Authentication type: basic, jwt, nkey, or none. When omitted with credentials present, jwt is inferred if username contains a ., otherwise basic; with no credentials, none is used

Stream

Field	Required	Default	Description
stream	N	"vmetric"	JetStream stream name
consumer	N	"vmetric"	JetStream consumer name
subject	Y	-	Subject pattern to subscribe to
max_bytes	N	1048576	Maximum message size in bytes (1 MB default)
timeout	N	5	Fetch timeout in seconds (1-30)
batch_size	N	10000	Number of messages to fetch per request

note

The stream and consumer defaults (vmetric) must exist on the NATS server before the device can subscribe. Pre-create them via nats stream add and nats consumer add or override these fields to match an existing stream and consumer.

TLS

Field	Required	Default	Description
tls.status	N	false	Enable TLS encryption
tls.cert_name	Y*	cert.pem	Client certificate for mutual TLS
tls.key_name	Y*	key.pem	Client private key for mutual TLS
tls.passphrase	N	-	Passphrase for an encrypted private key
tls.ca_name	N	-	CA bundle used to verify the server certificate. When unset, the host trust store is used.
tls.server_name	N	-	SNI hostname override for the TLS handshake
tls.min_tls_version	N	tls1.2	Minimum accepted TLS version (tls1.0, tls1.1, tls1.2, tls1.3)
tls.max_tls_version	N	-	Maximum accepted TLS version. When unset, the highest mutually supported version is negotiated.
tls.insecure_skip_verify	N	false	Skip server certificate verification. Use only for testing.

* = Required when tls.status is true.

note

TLS material fields (cert_name, key_name, ca_name, client_ca_name) accept any of the following:

• File name — resolved relative to the service root directory. Nested paths such as certs/prod/server.pem are supported.
• Absolute path — honored only if it resolves inside the service root. Any path that escapes the root is refused.
• Inline PEM content — used verbatim when the value contains -----BEGIN.
• Environment variable — ${ENV_VAR}.
• Vault reference — $secret{id=...} or $secret{store=...,ref=...}.

Performance

Field	Required	Default	Description
reuse	N	true	Enable multi-worker mode
workers	N	4	Number of worker processes when reuse enabled (capped at the number of available CPU cores)

Examples

Basic

Creating a simple NATS consumer using the default stream and consumer...

- id: 1 name: basic_nats type: nats properties: address: "nats.example.com" port: 4222 subject: "logs.>"

Secure

Connecting with basic authentication and TLS encryption...

- id: 2 name: secure_nats type: nats properties: address: "nats.example.com" port: 4222 username: "consumer" password: "secret" auth_type: "basic" stream: "LOGS" consumer: "processor" subject: "logs.secure.>" tls: status: true cert_name: "nats.crt" key_name: "nats.key"

High-Volume

Tuning batch size and timeout for high-throughput consumption...

- id: 3 name: performant_nats type: nats properties: address: "nats.example.com" port: 4222 stream: "LOGS" consumer: "high-perf" subject: "logs.>" max_bytes: 4194304 timeout: 10 batch_size: 50000 reuse: true workers: 4

Subject Filters

Subject-based filtering with wildcards...

- id: 4 name: filtered_nats type: nats properties: address: "nats.example.com" port: 4222 stream: "LOGS" consumer: "filtered" subject: "logs.*.error" reuse: true workers: 2

tip

NATS subjects support * and > as wildcards for single and multiple tokens respectively.

Pipelines

Applying custom processing to messages...

- id: 5 name: pipeline_nats type: nats pipelines: - json_parser - field_extractor properties: address: "nats.example.com" port: 4222 stream: "LOGS" consumer: "processed" subject: "logs.raw.>"

note

Pipelines are processed sequentially, and can modify or drop messages before ingestion.