---
id: event--nats-subscriber
title: "NATS subscriber"
summary: "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."
audience: user
keywords: "start, output, url, subject, connecttimeoutms, authtype, username, password, token, nkey, seed, creds, usetls, insecureskipverify, cacertificate, clientcertificate, queuegroup, pendingcapacity, pendingfullmode, allowsubjectoverride, defaultsource, defaulttype"
---
# 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

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

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