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.

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)

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-reply 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.

Specs

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

Send a request and listen for a response synchronously

Following the nats request-reply 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

No Responders

If you send a request to a topic that has no registered listeners, it is sometimes convenient to find out right away, rather than waiting for a timeout to occur. In order to support this use-case, you can start your Gnat connection with the no_responders: true option and this function will return very quickly with an {:error, :no_responders} value. This behavior also works with request_multi/4

Specs

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

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

Specs

start_link(
  connection_settings(),
  keyword()
) :: GenServer.on_start()

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 use IPv6 addresses too
{:ok, gnat} = Gnat.start_link(%{host: '::1', port: 4222, tcp_opts: [:inet6, :binary]})

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)

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

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/reply 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)