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.

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>
    secure:
      status: <boolean>
    tls:
      status: <boolean>
      cert_name: <string>
      key_name: <string>

Configuration

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
secure.status	N	false	When true, the password is decrypted with the service shared key before use

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*	-	TLS certificate file name
tls.key_name	Y*	-	TLS private key file name

* = Conditionally required when tls.status is true.

note

The TLS certificate and key files must be placed in the service root directory.

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.