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

build-options clj

(build-options
  {:nats.core/keys
     [auth-handler buffer-size client-side-limit-checks connection-listener
      connection-name connection-timeout credentials credentials-file-path
      data-port-type discard-messages-when-outgoing-queue-full? error-listener
      executor-service get-wait-time http-request-interceptor
      http-request-interceptors ignore-discovered-servers? inbox-prefix jwt
      jwt-file-path keystore-password keystore-path max-control-line
      max-messages-in-outgoing-queue max-pings-out max-reconnects nkey
      nkey-file-path no-echo? no-headers? no-no-responders? no-randomize?
      no-reconnect? no-resolve-hostnames? old-request-style? open-tls? pedantic?
      ping-interval reconnect-buffer-size reconnect-jitter reconnect-jitter-tls
      reconnect-wait report-no-responders? request-cleanup-interval secure?
      server-pool server-url server-urls ssl-context socket-so-linger
      socket-write-timeout statistics-collector utf8-subjects? tls-algorithm
      tls-first? token trace-connection? truststore-password truststore-path
      advanced-stats? use-dispatcher-with-executor? user-name password
      time-trace use-timeout-exception? verbose?]})

close clj

(close conn)

connect clj

(connect server-url-or-options
         &
         [{:keys [jet-stream-options key-value-options]}])

Connect to the NATS server. Optionally configure jet stream and key/value management, or use nats.stream/configure and nats.kv/configure respectively later.

server-url-or-options is either the NATS server URL as a string, or a map of the following keys:

• :nats.core/auth-handler a AuthHandler instance, see create-file-auth-handler and create-static-auth-handler.
• :nats.core/buffer-size
• :nats.core/client-side-limit-checks
• :nats.core/connection-listener A function that receives connection events. Will be passed the clj-nats connection and an event keyword.
• :nats.core/connection-name
• :nats.core/connection-timeout java.time.Duration or number of milliseconds
• :nats.core/credentials
• :nats.core/credentials-file-path
• :nats.core/data-port-type
• :nats.core/discard-messages-when-outgoing-queue-full?
• :nats.core/error-listener See create-error-listener
• :nats.core/executor-service A java.util.concurrent.ExecutorService instance
• :nats.core/get-wait-time A function that will be called with one argument, the number of connection attempts, and that returns a java.time.Duration dictating how long to wait before attempting another reconnect.
• :nats.core/http-request-interceptor See https://javadoc.io/static/io.nats/jnats/2.19.0/io/nats/client/Options.Builder.html#httpRequestInterceptor-java.util.function.Consumer-
• :nats.core/http-request-interceptors See https://javadoc.io/static/io.nats/jnats/2.19.0/io/nats/client/Options.Builder.html#httpRequestInterceptors-java.util.Collection-
• :nats.core/ignore-discovered-servers?
• :nats.core/inbox-prefix
• :nats.core/jwt
• :nats.core/jwt-file-path
• :nats.core/keystore-password A character array
• :nats.core/keystore-path
• :nats.core/max-control-line
• :nats.core/max-messages-in-outgoing-queue
• :nats.core/max-pings-out
• :nats.core/max-reconnects
• :nats.core/nkey
• :nats.core/nkey-file-path
• :nats.core/no-echo?
• :nats.core/no-headers?
• :nats.core/no-no-responders?
• :nats.core/no-randomize?
• :nats.core/no-reconnect?
• :nats.core/no-resolve-hostnames?
• :nats.core/old-request-style?
• :nats.core/open-tls?
• :nats.core/pedantic?
• :nats.core/ping-interval A java.time.Duration or a number of milliseconds
• :nats.core/reconnect-buffer-size
• :nats.core/reconnect-jitter java.time.Duration
• :nats.core/reconnect-jitter-tls ;; java.time.Duration
• :nats.core/reconnect-wait ;; java.time.Duration
• :nats.core/report-no-responders?
• :nats.core/request-cleanup-interval ;; java.time.Duration
• :nats.core/secure?
• :nats.core/server-pool See https://javadoc.io/static/io.nats/jnats/2.19.0/io/nats/client/ServerPool.html
• :nats.core/server-url
• :nats.core/server-urls
• :nats.core/ssl-context A javax.net.ssl.SSLContext
• :nats.core/socket-so-linger SO LINGER in seconds
• :nats.core/socket-write-timeout A java.time.Duration or a number of milliseconds
• :nats.core/statistics-collector See create-statistics-collector
• :nats.core/utf8-subjects?
• :nats.core/tls-algorithm
• :nats.core/tls-first?
• :nats.core/token A character array
• :nats.core/trace-connection?
• :nats.core/truststore-password A character array
• :nats.core/truststore-path
• :nats.core/advanced-stats?
• :nats.core/use-dispatcher-with-executor?
• :nats.core/user-name Character array or String. Must be same type as :password
• :nats.core/password Character array or String. Must be same type as :user-name
• :nats.core/time-trace
• :nats.core/use-timeout-exception?
• :nats.core/verbose?

connection-events clj

Available events:

• :nats.connection.event/closed
• :nats.connection.event/connected
• :nats.connection.event/disconnected
• :nats.connection.event/discovered-servers
• :nats.connection.event/lame-duck
• :nats.connection.event/reconnected
• :nats.connection.event/resubscribed

create-error-listener clj

(create-error-listener
  {:keys [error-occurred exception-occurred flow-control-processed
          heartbeat-alarm message-discarded pull-status-error
          pull-status-warning slow-consumer-detected socket-write-timeout
          supply-message unhandled-status]})

Create an io.nats.client.ErrorListener instance. Takes the following functions:

• :error-occurred (fn [conn error])
• :exception-occurred (fn [conn exception])
• :flow-control-processed (fn [conn subscription subject source])
• :heartbeat-alarm (fn [conn subscription last-stream-seq last-consumer-seq])
• :message-discarded (fn [conn message])
• :pull-status-error (fn [conn subscription status])
• :pull-status-warning (fn [conn subscription status])
• :slow-consumer-detected (fn [conn consumer])
• :socket-write-timeout (fn [conn])
• :supply-message (fn [label conn consumer sub pairs])
• :unhandled-status (fn [conn subscription status])

create-file-auth-handler clj

(create-file-auth-handler credentials-file-path)

(create-file-auth-handler jwt-file-path nkey-file-path)

Creates an io.nats.client.AuthHandler for a credentials file, or a jwt-file and a nkey-file. The result can be passed as :nats.core/auth-handler in nats.core/connect.

create-static-auth-handler clj

(create-static-auth-handler credentials)

(create-static-auth-handler jwt nkey)

Creates an io.nats.client.AuthHandler for a credential string, or a pair of JWT and nkey strings. The result can be passed as :nats.core/auth-handler in nats.core/connect.

create-statistics-collector clj

(create-statistics-collector
  {:keys [decrement-outstanding-requests increment-dropped-count
          increment-duplicate-replies-received increment-err-count
          increment-exception-count increment-flush-counter increment-in-bytes
          increment-in-msgs increment-ok-count increment-orphan-replies-received
          increment-out-bytes increment-out-msgs increment-outstanding-requests
          increment-ping-count increment-reconnects increment-replies-received
          increment-requests-sent register-read register-write
          set-advanced-tracking]})

Create a io.nats.client.StatisticsCollector instance. Takes a map of functions:

• :decrement-outstanding-requests (fn [])
• :increment-dropped-count (fn [])
• :increment-duplicate-replies-received (fn [])
• :increment-err-count (fn [])
• :increment-exception-count (fn [])
• :increment-flush-counter (fn [])
• :increment-in-bytes (fn [])
• :increment-in-msgs (fn [])
• :increment-ok-count (fn [])
• :increment-orphan-replies-received (fn [])
• :increment-out-bytes (fn [byte-count])
• :increment-out-msgs (fn [])
• :increment-outstanding-requests (fn [])
• :increment-ping-count (fn [])
• :increment-reconnects (fn [])
• :increment-replies-received (fn [])
• :increment-requests-sent (fn [])
• :register-read (fn [byte-count])
• :register-write (fn [byte-count])
• :set-advanced-tracking (fn [track-advance])

publish clj

(publish conn message)

Publish a message. Performs no publish acking; do not use for publishing to a JetStream subject, instead use nats.stream/publish.

message is a map of:

• :nats.message/subject - The subject to publish to
• :nats.message/data - The message data. Can be any Clojure value
• :nats.message/headers - An optional map of string keys to string (or collection of string) values to set as meta-data on the message.
• :nats.message/reply-to - An optional reply-to subject.

pull-message clj

(pull-message subscription timeout)

request clj

(request conn message)

(request conn message timeout)

Make a request and wait for the response. Returns a future that resolves with the response.

message is a map of:

• :nats.message/subject - The subject to publish to
• :nats.message/data - The message data. Can be any Clojure value
• :nats.message/headers - An optional map of string keys to string (or collection of string) values to set as meta-data on the message.

In request/response, :nats.message/reply-to is reserved for the server.

timeout is optional, and either a number of milliseconds to wait, or a java.time.Duration.

subscribe clj

(subscribe conn subject & [queue-name])

Subscribe to non-stream subject. For JetStream subjects, instead use nats.stream/subscribe. Pull messages with nats.core/pull-message.

unsubscribe clj

(unsubscribe subscription)