NATS publisher

Publishes every incoming EventData record to a NATS subject (core NATS, at-most-once) using the CloudEvents NATS binding in binary content mode: attributes travel as ce-* headers, the payload is the raw message body.

Pins

Pin	Kind	Types	Description
Input	input	Viaducta.Plugins.Event.Types.EventData	Event to publish.
Continue	continue	same as input	Called with the unchanged input after the message was handed to the NATS connection.

Configuration

The configuration is shared with the NATS subscriber; fields marked subscriber 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 publish to (no wildcards).
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.
AllowSubjectOverride	boolean	false	true: a non-empty EventData.Channel is used as the subject instead of Subject.
DefaultSource	string (max 512)		CloudEvents source for events without one. Empty: viaducta://{flow}/{component}.
DefaultType	string (max 255)	com.viaducta.event	CloudEvents type for events without one.
QueueGroup	string (max 255)		Not used by the publisher (subscriber setting).
PendingCapacity	integer (min 1)	1024	Not used by the publisher (subscriber setting).
PendingFullMode	dropdown: 0 Wait, 1 DropNewest, 2 DropOldest, 3 DropWrite	2	Not used by the publisher (subscriber 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).
• Message built from the input EventData:
  • body = Data (null gives an empty body), reply subject = ReplyTo if set,
  • Headers → message headers, except keys that clash with the core ce-* attributes (those are dropped),
  • ce-id = Id (new random GUID if empty), ce-source = Source or the default source, ce-specversion = SpecVersion or 1.0, ce-type = Type or DefaultType, ce-time = Time in UTC (now if empty),
  • ce-datacontenttype and ce-subject only when DataContentType / Subject are set.
• The input record is not modified; defaults are applied only to the headers.
• A failed publish (payload too large, invalid subject or header, timeout while disconnected) raises a recoverable error for that record; the next record is tried again.

Notes

• No delivery confirmation: core NATS is fire-and-forget. Continue means the connection accepted the message, not that a subscriber received it. Messages with no subscriber listening are lost. JetStream is not supported.
• EventData.Subject is the CloudEvents subject, not the NATS subject. To route per record set Channel and enable AllowSubjectOverride.
• Set ReplyTo to have receivers answer on a reply subject; the publisher itself does not wait for replies.
• With InsecureSkipVerify set, CaCertificate is ignored.
• Keep secrets in expression settings, e.g. "Token": "=Secrets.NatsToken".

Example

{ "Url": "nats://nats.local:4222", "Subject": "orders.created", "DefaultType": "com.example.order.created", "AuthType": 2, "Token": "=Secrets.NatsToken" }

Http listener → Script (build EventData) → NATS publisher → Http response