A behaviour which provides the NATS JetStream Pull Consumer functionalities.

When a Consumer is pull-based, it means that the messages will be delivered when the server is asked for them.

Example

Declare a module which uses Gnat.Jetstream.PullConsumer and implements init/1 and handle_message/2 callbacks.

defmodule MyApp.PullConsumer do
  use Gnat.Jetstream.PullConsumer

  def start_link(arg) do
    Jetstream.PullConsumer.start_link(__MODULE__, arg)
  end

  @impl true
  def init(_arg) do
    {:ok, nil,
      connection_name: :gnat,
      stream_name: "TEST_STREAM",
      consumer_name: "TEST_CONSUMER"}
  end

  @impl true
  def handle_message(message, state) do
    # Do some processing with the message.
    {:ack, state}
  end
end

You can then place your Pull Consumer in a supervision tree. Remember that you need to have the Gnat.ConnectionSupervisor set up.

defmodule MyApp.Application do
  use Application

  @impl true
  def start(_type, _args) do
    children = [
      # Create NATS connection
      {Gnat.ConnectionSupervisor, ...},
      # Start NATS Jetstream Pull Consumer
      MyApp.PullConsumer,
    ]
    opts = [strategy: :one_for_one]
    Supervisor.start_link(children, opts)
  end
end

Connection Options

In order to establish consumer connection with NATS, you need to pass several connection options via keyword list in third element of a tuple returned from init/1 callback.

Following options must be provided. Omitting this options will cause the process to raise errors upon initialization:

• :connection_name - Gnat connection or Gnat.ConnectionSupervisor name/PID.

For durable consumers, provide:

• :stream_name - name of an existing stream the consumer will consume messages from.
• :consumer_name - name of an existing consumer pointing at the stream.

For ephemeral consumers, provide:

• :consumer - a Gnat.Jetstream.API.Consumer struct for creating an ephemeral consumer. The consumer struct must have durable_name: nil OR inactive_threshold set to ensure that the server will clean it up. The stream_name field must also be set.

You can also pass the optional ones:

• :connection_retry_timeout - a duration in milliseconds after which the PullConsumer which failed to establish NATS connection retries, defaults to 1000
• :connection_retries - a number of attempts the PullConsumer will make to establish the NATS connection. When this value is exceeded, the pull consumer stops with the :timeout reason, defaults to 10
• :inbox_prefix - allows the default _INBOX. prefix to be customized. Should end with a dot.
• :domain - use a JetStream domain, this is mostly used on leaf nodes.
• :batch_size - when set to a value greater than 1, enables batch mode. Messages are buffered and delivered to handle_message/2 in batches. Only the last message per batch is acknowledged, so the underlying consumer should use ack_policy: :all for correctness. This dramatically improves throughput for consumers that need to catch up on large backlogs. In batch mode, :nack and :term returns from handle_message/2 are treated as :ack since ack_policy: :all cannot selectively reject messages. Defaults to 1 (single-message mode).
• :request_expires - duration in nanoseconds that a long-poll pull request will linger on the server before the server replies with a 408 terminator and the consumer issues a fresh pull. Defaults to 5_000_000_000 (5 seconds).
• :idle_heartbeat - duration in nanoseconds at which the server is asked to emit 100-status idle heartbeat messages while a long-poll pull request is outstanding but no real messages are available. The PullConsumer also runs a local watchdog: if no traffic at all (data, status, or heartbeat) is observed within 2 * idle_heartbeat the consumer assumes the pull request was lost (e.g. dropped during a JetStream leadership change without killing the TCP connection) and forces a reconnect. Must be at most :request_expires / 2 — the server rejects pull requests that violate this. Defaults to half of :request_expires (2.5 seconds with default settings, watchdog fires at 5s).
• :heartbeat_check_interval - cadence in milliseconds at which the local watchdog checks for missed heartbeats. Independent of (and finer-grained than) the missed-heartbeat threshold itself. Defaults to 1_000 (1 second).

Telemetry

The PullConsumer emits the following telemetry events:

• [:gnat, :jetstream, :pull_consumer, :heartbeat_expired] — emitted when the local heartbeat watchdog observes that no inbound message has arrived within 2 * idle_heartbeat and the consumer is about to force a reconnect. Measurements: %{gap_ms, threshold_ms}. Metadata: %{module, stream_name, consumer_name, connection_name}.

Dynamic Connection Options

It is possible that you have to determine some of the options dynamically depending on pull consumer's init argument. To do so, it is recommended to derive these options values from some init argument:

defmodule MyApp.PullConsumer do
  use Gnat.Jetstream.PullConsumer

  def start_link() do
    Gnat.Jetstream.PullConsumer.start_link(__MODULE__, %{counter: counter})
  end

  @impl true
  def init(%{counter: counter}) do
    {:ok, nil,
      connection_name: :gnat,
      stream_name: "TEST_STREAM_#{counter}",
      consumer_name: "TEST_CONSUMER_#{counter}"}
  end

  ...
end

Ephemeral Consumer Example

You can create ephemeral consumers by providing a :consumer struct with durable_name: nil. These are automatically created and cleaned up with the connection lifecycle:

defmodule MyApp.EphemeralPullConsumer do
  use Gnat.Jetstream.PullConsumer

  def start_link(arg) do
    Jetstream.PullConsumer.start_link(__MODULE__, arg)
  end

  @impl true
  def init(_arg) do
    consumer = %Gnat.Jetstream.API.Consumer{
      stream_name: "TEST_STREAM",
      durable_name: nil,  # Must be nil for ephemeral consumers
      filter_subject: "orders.*"
    }

    {:ok, nil,
      connection_name: :gnat,
      consumer: consumer}
  end

  @impl true
  def handle_message(message, state) do
    # Do some processing with the message.
    {:ack, state}
  end
end

Auto-Cleanup Durable Consumer Example

For scenarios where you want persistence across reconnections but automatic cleanup (e.g., Kubernetes pods), you can create durable consumers with inactive_threshold:

defmodule MyApp.AutoCleanupPullConsumer do
  use Gnat.Jetstream.PullConsumer

  def start_link(pod_name) do
    Jetstream.PullConsumer.start_link(__MODULE__, pod_name)
  end

  @impl true
  def init(pod_name) do
    consumer = %Gnat.Jetstream.API.Consumer{
      stream_name: "ORDERS_STREAM",
      durable_name: "orders-consumer-",  # Named after pod
      inactive_threshold: 300_000_000_000,  # 5 minutes in nanoseconds
      filter_subject: "orders.*"
    }

    {:ok, nil,
      connection_name: :gnat,
      consumer: consumer}
  end

  @impl true
  def handle_message(message, state) do
    # Process order message with state persistence
    {:ack, state}
  end
end

How to supervise

A PullConsumer is most commonly started under a supervision tree. When we invoke use Gnat.Jetstream.PullConsumer, it automatically defines a child_spec/1 function that allows us to start the pull consumer directly under a supervisor. To start a pull consumer under a supervisor with an initial argument of :example, one may do:

children = [
  {MyPullConsumer, :example}
]
Supervisor.start_link(children, strategy: :one_for_all)

While one could also simply pass the MyPullConsumer as a child to the supervisor, such as:

children = [
  MyPullConsumer # Same as {MyPullConsumer, []}
]
Supervisor.start_link(children, strategy: :one_for_all)

A common approach is to use a keyword list, which allows setting init argument and server options, for example:

def start_link(opts) do
  {initial_state, opts} = Keyword.pop(opts, :initial_state, nil)
  Gnat.Jetstream.PullConsumer.start_link(__MODULE__, initial_state, opts)
end

and then you can use MyPullConsumer, {MyPullConsumer, name: :my_consumer} or even {MyPullConsumer, initial_state: :example, name: :my_consumer} as a child specification.

use Gnat.Jetstream.PullConsumer also accepts a list of options which configures the child specification and therefore how it runs under a supervisor. The generated child_spec/1 can be customized with the following options:

• :id - the child specification identifier, defaults to the current module
• :restart - when the child should be restarted, defaults to :permanent
• :shutdown - how to shut down the child, either immediately or by giving it time to shut down

For example:

use Gnat.Jetstream.PullConsumer, restart: :transient, shutdown: 10_000

See the "Child specification" section in the Supervisor module for more detailed information. The @doc annotation immediately preceding use Jetstream.PullConsumer will be attached to the generated child_spec/1 function.

Name registration

A pull consumer is bound to the same name registration rules as GenServers. Read more about it in the GenServer documentation.