NATS::Client

How much outgoing data we will hold in memory while disconnected from the server. Once this is exceeded, publishing raises ReconnectBufferExceeded until the connection is re-established.

Connect to a single NATS server at the given URI

nats = NATS::Client.new(URI.parse("nats://nats.example.com"))

Connect to a NATS cluster at the given URIs

nats = NATS::Client.new([
  URI.parse("nats://nats-1.example.com"),
  URI.parse("nats://nats-2.example.com"),
  URI.parse("nats://nats-3.example.com"),
])

Close this NATS connection. This should be done explicitly before exiting the program so that the NATS server can remove any subscriptions that were associated with this client.

If the client is currently connected, it drains its subscriptions before closing. If it is disconnected, any data buffered while waiting to reconnect is discarded and the client stops trying to reconnect.

Flush the client's output buffer over the wire and wait for the server to acknowledge it. Raises NATS::Error if the server does not acknowledge within timeout or if the connection is lost while waiting. If the client is disconnected when this is called, the flush completes once the client reconnects and the buffered data has been acknowledged, if that happens within the timeout.

Flush the client's output buffer over the wire without waiting for the server to acknowledge it. This is a no-op while disconnected; the buffered data is sent as soon as the client reconnects.

Returns a NATS::JetStream::Client that uses this client's connection to the NATS server.

Returns a NATS::KV::Client that uses this client's connection to the NATS server.

Execute the given block whenever this client is disconnected from the NATS server.

nats = NATS::Client.new
nats.on_disconnect { Datadog.metrics.increment "nats.disconnect" }

Execute the given block whenever an exception is raised inside this NATS client.

nats = NATS::Client.new
nats.on_error { |error| Honeybadger.notify error }

Execute the given block when the NATS server this client is connected to enters lame-duck mode, meaning it is shutting down and will close this connection at some point within its lame-duck duration. The client will reconnect to another server automatically when that happens, but you can use this to stop taking on new work or to drain in-flight work first.

nats = NATS::Client.new
nats.on_lame_duck_mode { Log.info { "NATS server is shutting down" } }

Execute the given block whenever this client pings the server.

nats = NATS::Client.new
nats.on_ping { Datadog.metrics.increment "nats.ping" }

Execute the given block whenever this client receives a pong reply from the server.

nats = NATS::Client.new
nats.on_pong { Datadog.metrics.increment "nats.pong" }

Execute the given block whenever this client is reconnected to the NATS server.

nats = NATS::Client.new
nats.on_reconnect { Datadog.metrics.increment "nats.reconnect" }

Publish the given message body (either Bytes for binary data or String for text) on the given NATS subject, optionally supplying a reply_to subject (if expecting a reply or to notify the receiver where to send updates) and any headers.

# Send an empty message to a subject
nats.publish "hello"

# Serialize an object to a subject
nats.publish "orders.#{order.id}", order.to_json

# Tell a recipient where to send results. For example, to stream results
# to a given subject:
reply_subject = "replies.orders.list.customer.123"
orders = [] of Order
nats.subscribe reply_subject do |msg|
  case result = (Order | Complete).from_json(String.new(msg.body))
  in Order
    orders << result
  in Complete
    nats.unsubscribe reply_subject
  end
end
nats.publish "orders.list.customer.123", reply_to: reply_subject

# Publish a message to NATS JetStream with a message-deduplication header
# for idempotency:
nats.jetstream.subscribe consumer_subject, queue_group: "my-service" do |msg|
  # ...
end
nats.publish orders_subject, order.to_json, headers: NATS::Message::Headers{
  # Deduplicate using the equivalent of a cache key
  "Nats-Msg-Id" => "order-submitted-#{order.id}-#{order.updated_at.to_json}",
}

Publish the given message body (either Bytes for binary data or String for text) on the given NATS subject, optionally supplying a reply_to subject (if expecting a reply or to notify the receiver where to send updates) and any headers.

# Send an empty message to a subject
nats.publish "hello"

# Serialize an object to a subject
nats.publish "orders.#{order.id}", order.to_json

# Tell a recipient where to send results. For example, to stream results
# to a given subject:
reply_subject = "replies.orders.list.customer.123"
orders = [] of Order
nats.subscribe reply_subject do |msg|
  case result = (Order | Complete).from_json(String.new(msg.body))
  in Order
    orders << result
  in Complete
    nats.unsubscribe reply_subject
  end
end
nats.publish "orders.list.customer.123", reply_to: reply_subject

# Publish a message to NATS JetStream with a message-deduplication header
# for idempotency:
nats.jetstream.subscribe consumer_subject, queue_group: "my-service" do |msg|
  # ...
end
nats.publish orders_subject, order.to_json, headers: NATS::Message::Headers{
  # Deduplicate using the equivalent of a cache key
  "Nats-Msg-Id" => "order-submitted-#{order.id}-#{order.updated_at.to_json}",
}

Send the given body to the msg's reply_to subject, often used in a request/reply messaging model.

nats.subscribe "orders.*", queue_group: "orders-service" do |msg|
  _, id = msg.subject.split('.') # Similar to HTTP path routing

  if order = OrderQuery.new.find_by(id: id)
    nats.reply msg, {order: order}.to_json
  else
    nats.reply msg, {error: "No order with that id found"}.to_json
  end
end

Make a synchronous request to subscribers of the given subject, waiting up to timeout for a response from any of the subscribers. The first message to come back will be returned. If no messages comes back before the timeout elapses, nil is returned.

if order_response = nats.request("orders.info.#{order_id}")
  response << Order.from_json(String.new(order_response.body))
else
  response.status = :service_unavailable
end

Make an asynchronous request to subscribers of the given subject, not waiting for a response. The first message to come back will be passed to the block.

Send all messages and wait up to timeout for all of them to return. The method returns the responses for each request message in the original order. Requests that received no reply before timeout elapsed will have a nil response.

Make a synchronous request to subscribers of the given subject, waiting up to timeout for responses from any of the subscribers. The first max_replies messages to come back will be returned. If fewer replies are received before the timeout elapses, only those will be returned.

orders = nats.request_many("orders.info.#{order_id}", max_replies: 10).map do |response|
  Order.from_json(response.data_string)
end

Make a synchronous request to subscribers of the given subject, receiving as many messages as are sent until stall_timeout has passed without receiving any messages. If no replies are received before the stall_timeout elapses, the return value will be empty.

orders = nats.request("orders.info.#{order_id}", stall_timeout: 2.seconds).map do |response|
  Order.from_json(response.data_string)
end

Make a synchronous request to subscribers of the given subject, waiting up to timeout for responses from any of the subscribers and yielding each message to the given block to check for a sentinel message. The first message that returns true for the block will terminate the request and return all messages received up to that point.

orders = nats.request_many("orders.info.#{order_id}") do |response|
  # Our sentinel message will be empty
  response.data.empty?
end

NOTE: The sentinel message is not included in the array of messages returned by the method. It doesn't count as a reply. You should always send all of your replies and then send an additional sentinel message.

The current state of the connection

Subscribe to the given subject, optionally with a queue_group (so that each message is delivered to this application once instead of once for each instance of the application), executing the given block for each message.

require "nats"

nats = NATS::Client.new
nats.subscribe "orders.created" do |msg|
  order = Order.from_json(String.new(msg.body))

  # ...
end

Unsubscribe from the given subscription after the specified number of messages has been received.

Unsubscribe from the given subscription

nats = NATS::Client.new

new_orders = [] of NATS::Message
subscription = nats.subscribe "orders.created.*" do |msg|
  messages << msg
end

spawn do
  sleep 10.seconds
  nats.unsubscribe subscription
end