> ## Documentation Index
> Fetch the complete documentation index at: https://cosmo-docs.wundergraph.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Stream And Consumer Configuration

> Configuring a stream and consumer for a NATS provider.

<Note>
  Your NATS server must be configured to allow jetstreams before initiation. Please see [the documentation](https://docs.nats.io/nats-concepts/jetstream) for more details.
</Note>

By configuring a stream and consumer, messages that are sent while a subscription is inactive can be consumed at a later time, e.g., when the subscription is restarted. This can help to prevent the loss of messages.

## Event-Driven Graph schema configuration

The `@edfs__natsSubscribe` directive defines an optional `streamConfiguration` argument. Providing this input object allows the configuration of a stream and consumer on a NATS connection.

Note that if the `streamConfiguration` argument is undefined, the connection will be interpreted not to use a stream/consumer. If defined, all input object fields are required:

| Input name                | Type    | Value                                                                               |
| ------------------------- | ------- | ----------------------------------------------------------------------------------- |
| consumerInactiveThreshold | Int!    | The inactive threshold of the consumer in seconds. The default value is 30 seconds. |
| consumerName              | String! | The name of the consumer.                                                           |
| streamName                | String! | The name of the stream.                                                             |

## NATS configuration

### Stream

The stream name that is supplied to the `streamName` input must already be configured by your NATS server. If the relevant stream for a stream-reliant subscription cannot be found, an error will be returned.

Please consult the following documentation on creating a stream (and ensure your NATS server has been configured to allow jetstreams):

[https://docs.nats.io/nats-concepts/jetstream/streams](https://docs.nats.io/nats-concepts/jetstream/streams)

Typically, this can be achieved using the [NATS cli](https://docs.nats.io/using-nats/nats-tools/nats_cli) and following the prompts:

```bash theme={"system"}
nats str add stream-name
```

### Consumer

As long as the stream exists, when a subscription is started, the router will attempt to identify an existing consumer. This identification is the name (supplied by the `consumerName` input), followed by a hash of the subjects, and the host name and listen address of the router.

If a consumer is not found, the router will create a new durable consumer with the same name.

By default, consumers will automatically be deleted after 30 seconds of inactivity. Inactivity is classed as the lack of a running subscription. You can change this value by specifying the value for the `consumerInactiveThreshold` input in the stream configuration. If you set the `consumerInactiveThreshold` to `0` (or any negative value), the consumer will never be deleted.

If you wish for your consumer to have a timeout threshold, please configure accordingly using the following documentation:

[https://docs.nats.io/nats-concepts/jetstream/consumers](https://docs.nats.io/nats-concepts/jetstream/consumers)

[https://docs.nats.io/nats-concepts/jetstream/consumers/example\_configuration](https://docs.nats.io/nats-concepts/jetstream/consumers/example_configuration)

Typically, this can be achieved using the [NATS cli](https://docs.nats.io/using-nats/nats-tools/nats_cli) and following the prompts:

```bash theme={"system"}
nats consumer add stream-name consumer-name
```

## Example

In the example below, the NATS provider `my-nats` has also defined a stream configuration. The `streamName` input has been set to "myStream", and the `consumerName` input has been set to `myConsumer`.

```js theme={"system"}
directive @edfs__natsSubscribe(
  subjects: [String!]!,
  providerId: String! = "default",
  streamConfiguration: edfs__NatsStreamConfiguration
) on FIELD_DEFINITION

input edfs__NatsStreamConfiguration {
    consumerInactiveThreshold: Int! = 30
    consumerName: String!
    streamName: String!
}

type Subscription {
    userUpdated(id: Int!): User! @edfs__natsSubscribe(
        subjects: ["user.{{ args.id }}"],
        providerId: "my-nats",
        streamConfiguration: { consumerName: "myConsumer", streamName: "myStream" },
    )
}

type User @key(fields: "id", resolvable: false) {
    id: Int! @external
}
```