NATS subscriber

Subscribes to a NATS subject (core NATS, at-most-once) and emits every received message as an EventData record, strictly in arrival order. CloudEvents attributes are read from ce-* headers (NATS binding, binary content mode). Use it as the starting point of a flow that reacts to NATS messages.

Pins

Pin	Kind	Types	Description
Start	start		Starts the subscription. Called by the system at flow start when unconnected. Further calls are ignored once subscribed.
Output	output	Viaducta.Plugins.Event.Types.EventData	Called once for every received message, one message at a time.

Configuration

The configuration is shared with the NATS publisher; fields marked publisher only have no effect here.

Field	Type	Default	Description
Url	string (max 1024)		Required. Server url, e.g. nats://host:4222. Several cluster urls comma separated.
Subject	string (max 255)		Required. Subject to subscribe on. Wildcards allowed (orders.*, orders.>).
ConnectTimeoutMs	integer (min 1)	5000	Connect timeout in milliseconds.
AuthType	dropdown: 0 None, 1 UserPassword, 2 Token, 3 NKey, 4 Creds	0	How the client authenticates.
Username	string (max 256)		Required when AuthType is 1.
Password	password (max 256)		Used when AuthType is 1.
Token	password		Required when AuthType is 2.
NKey	string		NKey public key. Required (with Seed) when AuthType is 3.
Seed	password		NKey seed (private key). Required when AuthType is 3.
Creds	text		Whole content of the .creds file (JWT and NKey seed). Required when AuthType is 4.
UseTls	boolean	false	true: TLS is required.
InsecureSkipVerify	boolean	false	Visible with UseTls. Skips server certificate verification. Development only.
CaCertificate	certificate name		Visible with UseTls. CA used to verify the server certificate (private CA). Empty: OS trust store.
ClientCertificate	certificate name		Visible with UseTls. Client certificate with private key for mutual TLS.
QueueGroup	string (max 255)		Queue group. Subscribers in the same group share (load balance) the messages. Empty: every subscriber receives every message.
PendingCapacity	integer (min 1)	1024	Max number of received messages buffered while the flow is processing.
PendingFullMode	dropdown: 0 Wait, 1 DropNewest, 2 DropOldest, 3 DropWrite	2	What happens when the buffer is full. 0 stops reading from the socket (the server may flag the client as a slow consumer); 1 / 2 drop the newest / oldest buffered message; 3 drops the incoming one.
AllowSubjectOverride	boolean	false	Not used by the subscriber (publisher setting).
DefaultSource	string (max 512)		Not used by the subscriber (publisher setting).
DefaultType	string (max 255)	com.viaducta.event	Not used by the subscriber (publisher setting).

Behaviour

• The connection is opened at flow start. Invalid configuration, a missing certificate or a failed first connect stops the flow with an unrecoverable error (no retry on the initial connect). After that, connection losses are reconnected automatically and logged as warnings. The plugin keeps the flow running (KeepAlive).
• On Start the plugin subscribes to Subject (with QueueGroup if set); a rejected subject or queue group is an unrecoverable error.
• Every message is mapped to EventData:
  • Data - message body (byte[]), Channel - subject the message arrived on, ReplyTo - reply subject,
  • ce-id, ce-source, ce-specversion, ce-type, ce-datacontenttype, ce-subject, ce-time headers → Id, Source, SpecVersion, Type, DataContentType, Subject, Time,
  • all other headers → Headers. For multi-valued headers the last value is used.
• Messages from non-CloudEvents publishers are accepted: missing attributes stay null (SpecVersion too), Time falls back to the receive time.
• Each message gets new metadata; CorrelationId is set to the event Id when present.
• The next message is read only after the flow finished processing the previous one. If processing fails, the error is logged and the message is skipped.

Notes

• Core NATS has no persistence or redelivery: messages published while the flow is stopped or disconnected, skipped after an error or dropped from a full buffer are lost. JetStream is not supported.
• Dropped messages are reported as aggregated warnings (at most every 5 seconds). Increase PendingCapacity or speed up the flow if they appear.
• If the server closes the subscription (for example a permission violation), an error is logged and no more messages arrive, but the flow keeps running.
• Run several flows with the same QueueGroup to scale out; ordering then holds only per instance.
• With InsecureSkipVerify set, CaCertificate is ignored.
• Keep secrets in expression settings, e.g. "Token": "=Secrets.NatsToken".
• Data is binary; convert it to text downstream before parsing JSON.

Example

{ "Url": "nats://nats.local:4222", "Subject": "orders.>", "QueueGroup": "order-workers", "AuthType": 2, "Token": "=Secrets.NatsToken" }

NATS subscriber (Output) → Script (decode Data) → Database writer