nats.consumer — no.cjohansen/clj-nats 0.2025.02.21

ack clj

(ack conn message)

Acknowledge to the server that a message is successfully processed.

ack-in-progress clj

(ack-in-progress conn message)

Acknowledge to the server that processing is in progress. Use this if processing takes longer than your acknowledgement window. It's probably best to try to avoid needing to call this.

ack-policies clj

Available acknowledgment policies. One of:

• :nats.ack-policy/all
• :nats.ack-policy/explicit
• :nats.ack-policy/none

ack-term clj

(ack-term conn message)

Tell the server that processing this message was not successful, and that it should not be re-delivered.

create-consumer clj

(create-consumer conn config)

Create consumer. config is a map of:

• :nats.consumer/id - A keyword with the stream name for namespace and consumer name for name, e.g. :<stream>/<consumer>
• :nats.consumer/name
• :nats.consumer/stream-name
• :nats.consumer/ack-policy - See nats.consumer/ack-policies
• :nats.consumer/ack-wait - Number of milliseconds or a java.time.Duration
• :nats.consumer/deliver-policy - See nats.consumer/deliver-policies
• :nats.consumer/description
• :nats.consumer/durable? - Makes stream durable.
• :nats.consumer/mem-storage? - Forces consumer state to live in memory, instead of whatever the stream default is.
• :nats.consumer/filter-subjects
• :nats.consumer/headers-only?
• :nats.consumer/max-ack-pending - Maximum outstanding acks before consumers are paused
• :nats.consumer/max-batch - Maximum number of messages allowed in a single pull
• :nats.consumer/max-bytes - Maximum number of bytes allowed in a single pull
• :nats.consumer/max-deliver - Maximum number of times to deliver a message
• :nats.consumer/max-expires - Number of milliseconds or a java.time.Duration
• :nats.consumer/max-pull-waiting - Maximum number of outstanding pulls to accept
• :nats.consumer/metadata
• :nats.consumer/replicas - Number of replicas
• :nats.consumer/replay-policy - See nats.consumer/replay-policies
• :nats.consumer/sample-frequency - Percentage of requests to sample for monitoring purposes
• :nats.consumer/start-sequence
• :nats.consumer/sequence
• :nats.consumer/start-time - A java.time.Instant

Options for ephemeral consumers:

• :nats.consumer/inactive-threshold - Number of milliseconds or a java.time.Duration. Maximum idle time before consumer is removed.

Options for push consumers (not recommended)

• :nats.consumer/flow-control - Number of milliseconds or a java.time.Duration
• :nats.consumer/backoff
• :nats.consumer/deliver-group
• :nats.consumer/deliver-subject
• :nats.consumer/idle-heartbeat - Number of milliseconds or a java.time.Duration
• :nats.consumer/rate-limit

Requires a NATS 2.11 alpha server

• :nats.consumer/pause-until - A java.time.Instant

delete-consumer clj

(delete-consumer conn id)

(delete-consumer conn stream-name consumer-name)

deliver-policies clj

Available deliver policies. One of:

• :nats.deliver-policy/all
• :nats.deliver-policy/by-start-sequence
• :nats.deliver-policy/by-start-time
• :nats.deliver-policy/last
• :nats.deliver-policy/last-per-subject
• :nats.deliver-policy/new

get-consumer-info clj

(get-consumer-info conn id)

(get-consumer-info conn stream-name consumer-name)

get-consumer-names clj

(get-consumer-names conn stream-name)

get-consumers clj

(get-consumers conn stream-name)

nak clj

(nak conn message)

Tell the server that processing was not successful, and the message should be re-delivered later.

nak-with-delay clj

(nak-with-delay conn message duration)

Tell the server that processing was not successful, and the message should be re-delivered later and after at least the provided duration.

pull-message clj

(pull-message subscription timeout)

replay-policies clj

Available replay policies. One of:

• :nats.replay-policy/instant
• :nats.replay-policy/original

subscribe clj

(subscribe conn id)

(subscribe conn id opts)

(subscribe conn stream-name consumer-name opts)

Subscribe to messages on stream-name for consumer-name. opts is a map of:

• :batch-bytes
• :batch-size
• :bytes
• :threshold-pct

unsubscribe clj

(unsubscribe subscription)

update-consumer clj

(update-consumer conn config)

Update consumer. See create-consumer for keys in config.