Gnat (gnat v1.5.2) View Source

Link to this section Summary

Types

  • connection_timeout - limits how long it can take to establish a connection to a server
  • host - The location of the NATS server
  • port - The port the NATS server is listening on
  • ssl_opts - Options for connecting over SSL
  • tcp_opts - Options for connecting over TCP
  • tls - If the server should use a TLS connection
  • inbox_prefix - Prefix to use for the message inbox of this connection
t()

Functions

get the number of active subscriptions

Returns a specification to start this module under a supervisor.

ping(pid) deprecated

Ping the NATS server

Send a request and listen for a response synchronously

Send a request and listen for multiple responses synchronously

Get information about the NATS server the connection is for

Starts a connection to a nats broker

Gracefully shuts down a connection

Unsubscribe from a topic

Link to this section Types

Specs

connection_settings() :: %{
  optional(:connection_timeout) => non_neg_integer(),
  optional(:host) => binary(),
  optional(:inbox_prefix) => binary(),
  optional(:port) => non_neg_integer(),
  optional(:ssl_opts) => list(),
  optional(:tcp_opts) => list(),
  optional(:tls) => boolean()
}
  • connection_timeout - limits how long it can take to establish a connection to a server
  • host - The location of the NATS server
  • port - The port the NATS server is listening on
  • ssl_opts - Options for connecting over SSL
  • tcp_opts - Options for connecting over TCP
  • tls - If the server should use a TLS connection
  • inbox_prefix - Prefix to use for the message inbox of this connection

Specs

headers() :: [{binary(), iodata()}]

Specs

message() :: %{
  gnat: t(),
  topic: String.t(),
  body: String.t(),
  sid: non_neg_integer(),
  reply_to: String.t(),
  headers: headers()
}

Specs

sent_message() :: {:msg, message()}

Specs

server_info() :: %{
  :client_id => non_neg_integer(),
  :client_ip => binary(),
  optional(:ip) => binary(),
  optional(:cluster) => binary(),
  optional(:cluster_dynamic) => boolean(),
  optional(:connect_urls) => [binary()],
  optional(:ws_connect_urls) => [binary()],
  optional(:git_commit) => binary(),
  :go => binary(),
  :headers => boolean(),
  :host => binary(),
  optional(:jetstream) => binary(),
  :max_payload => integer(),
  :port => non_neg_integer(),
  :proto => integer(),
  :server_id => binary(),
  :server_name => binary(),
  :version => binary(),
  optional(:ldm) => boolean(),
  optional(:tls_verify) => boolean(),
  optional(:tls_available) => boolean(),
  optional(:tls_required) => boolean(),
  optional(:auth_required) => boolean()
}

Info Protocol

  • client_id - An optional unsigned integer (64 bits) representing the internal client identifier in the server. This can be used to filter client connections in monitoring, correlate with error logs, etc...
  • client_ip - The IP address the client is connecting from
  • cluster - The name of the cluster if any
  • cluster_dynamic - If the cluster is dynamic
  • connect_urls - An optional list of server urls that a client can connect to.
  • ws_connect_urls - An optional list of server urls that a websocket client can connect to.
  • git_commit - The git commit associated with this NATS version
  • go - The version of golang the NATS server was built with
  • headers - If messages can have headers in them
  • host - The IP address used to start the NATS server, by default this will be 0.0.0.0 and can be configured with -client_advertise host:port
  • jetstream - If the server is using JetStream features
  • max_payload - Maximum payload size, in bytes, that the server will accept from the client
  • port - The port number the NATS server is configured to listen on
  • proto - An integer indicating the protocol version of the server. The server version 1.2.0 sets this to 1 to indicate that it supports the "Echo" feature.
  • server_id - The unique identifier of the NATS server
  • server_name - A name for the server
  • version - The version of the NATS server
  • ldm - If the server supports Lame Duck Mode notifications, and the current server has transitioned to lame duck, ldm will be set to true.
  • auth_required - If this is set, then the client should try to authenticate upon connect.
  • tls_required - If this is set, then the client must perform the TLS/1.2 handshake. Note, this used to be ssl_required and has been updated along with the protocol from SSL to TLS.
  • tls_verify - If this is set, the client must provide a valid certificate during the TLS handshake.
  • tls_available - If the server can use TLS

Specs

t() :: GenServer.server()

Link to this section Functions

Link to this function

active_subscriptions(pid)

View Source

Specs

active_subscriptions(t()) :: {:ok, non_neg_integer()}

get the number of active subscriptions

Returns a specification to start this module under a supervisor.

See Supervisor.

This function is deprecated. Pinging is handled internally by the connection, this functionality will be removed.

Ping the NATS server

This correlates to the PING command in the NATS protocol. If the NATS server responds with a PONG message this function will return :ok

{:ok, gnat} = Gnat.start_link()
:ok = Gnat.ping(gnat)
Link to this function

pub(pid, topic, message, opts \\ [])

View Source

Specs

pub(t(), String.t(), binary(), keyword()) :: :ok

Publish a message

{:ok, gnat} = Gnat.start_link()
:ok = Gnat.pub(gnat, "characters", "Ron Swanson")

If you want to provide a reply address to receive a response you can pass it as an option. See request-response pattern.

{:ok, gnat} = Gnat.start_link()
:ok = Gnat.pub(gnat, "characters", "Star Lord", reply_to: "me")

If you want to publish a message with headers you can pass the :headers key in the opts like this.

{:ok, gnat} = Gnat.start_link()
:ok = Gnat.pub(gnat, "listen", "Yo", headers: [{"foo", "bar"}])

Headers must be passed as a t:headers() value (a list of tuples). Sending and parsing headers has more overhead than typical nats messages (see the Nats 2.2 release notes for details), so only use them when they are really valuable.

Link to this function

request(pid, topic, body, opts \\ [])

View Source

Specs

request(t(), String.t(), binary(), keyword()) ::
  {:ok, message()} | {:error, :timeout}

Send a request and listen for a response synchronously

Following the nats request-response pattern this function generates a one-time topic to receive replies and then sends a message to the provided topic.

Supported options:

  • receive_timeout: an integer number of milliseconds to wait for a response. Defaults to 60_000
  • headers: a set of headers you want to send with the request (see Gnat.pub/4)
{:ok, gnat} = Gnat.start_link()
case Gnat.request(gnat, "i_can_haz_cheezburger", "plZZZZ?!?!?") do
  {:ok, %{body: delicious_cheezburger}} -> :yum
  {:error, :timeout} -> :sad_cat
end
Link to this function

request_multi(pid, topic, body, opts \\ [])

View Source

Specs

request_multi(t(), String.t(), binary(), keyword()) :: {:ok, [message()]}

Send a request and listen for multiple responses synchronously

This function makes it easy to do a scatter-gather operation where you wait for a limited time and optionally a maximum number of replies.

Supported options:

  • receive_timeout: an integer number of milliseconds to wait for responses. Defaults to 60_000
  • max_messages: an integer number of messages to listen for. Defaults to -1 meaning unlimited
  • headers: a set of headers you want to send with the request (see Gnat.pub/4)
{:ok, gnat} = Gnat.start_link()
{:ok, responses} = Gnat.request_multi(gnat, "i_can_haz_fries", "plZZZZZ!?!?", max_messages: 5)
Enum.count(responses) #=> 5

Specs

server_info(t()) :: server_info()

Get information about the NATS server the connection is for

Link to this function

start_link(connection_settings \\ %{}, opts \\ [])

View Source

Specs

Starts a connection to a nats broker

{:ok, gnat} = Gnat.start_link(%{host: '127.0.0.1', port: 4222})
# if the server requires TLS you can start a connection with:
{:ok, gnat} = Gnat.start_link(%{host: '127.0.0.1', port: 4222, tls: true})
# if the server requires TLS and a client certificate you can start a connection with:
{:ok, gnat} = Gnat.start_link(%{tls: true, ssl_opts: [certfile: "client-cert.pem", keyfile: "client-key.pem"]})
# you can customize default "_INBOX." inbox prefix with:
{:ok, gnat} = Gnat.start_link(%{host: '127.0.0.1', port: 4222, inbox_prefix: "my_prefix._INBOX."})

You can also pass arbitrary SSL or TCP options in the tcp_opts and ssl_opts keys. If you pass custom TCP options please include :binary. Gnat uses binary matching to parse messages.

The final opts argument will be passed to the GenServer.start_link call so you can pass things like [name: :gnat_connection].

Specs

stop(t()) :: :ok

Gracefully shuts down a connection

{:ok, gnat} = Gnat.start_link()
:ok = Gnat.stop(gnat)
Link to this function

sub(pid, subscriber, topic, opts \\ [])

View Source

Specs

sub(t(), pid(), String.t(), keyword()) ::
  {:ok, non_neg_integer()} | {:ok, String.t()} | {:error, String.t()}

Subscribe to a topic

Supported options:

  • queue_group: a string that identifies which queue group you want to join

By default each subscriber will receive a copy of every message on the topic. When a queue_group is supplied messages will be spread among the subscribers in the same group. (see nats queueing)

The subscribed process will begin receiving messages with a structure of sent_message/0

{:ok, gnat} = Gnat.start_link()
{:ok, subscription} = Gnat.sub(gnat, self(), "topic")
receive do
  {:msg, %{topic: "topic", body: body}} ->
    IO.puts "Received: #{body}"
end
Link to this function

unsub(pid, sid, opts \\ [])

View Source

Specs

unsub(t(), non_neg_integer() | String.t(), keyword()) :: :ok

Unsubscribe from a topic

Supported options:

  • max_messages: number of messages to be received before automatically unsubscribed

This correlates to the UNSUB command in the nats protocol. By default the unsubscribe is affected immediately, but an optional max_messages value can be provided which will allow max_messages to be received before affecting the unsubscribe. This is especially useful for request response patterns.

{:ok, gnat} = Gnat.start_link()
{:ok, subscription} = Gnat.sub(gnat, self(), "my_inbox")
:ok = Gnat.unsub(gnat, subscription)
# OR
:ok = Gnat.unsub(gnat, subscription, max_messages: 2)