nats-pure Ruby Client

Groups allow you to aggregate endpoints under a common subject prefix. This enables building subject hierarchies.

Creating a Group: Use service.groups(name, options) or service.groups.add(name) to create a group. You can pass a queue: option to the group to apply it to all its endpoints.

Subject Hierarchy: When an endpoint is added to a group, its subject becomes group_name.endpoint_name. You can nest groups to create deeper hierarchies like parent.child.endpoint.

# Creating a nested hierarchy
numbers = service.groups.add("numbers")
aggregation = numbers.groups.add("agg")

aggregation.endpoints.add("avg") do |message|
  data = JSON.parse(message.data)
  avg = data.sum / data.size.to_f
  message.respond(avg.to_json)
end

# Requesting the nested endpoint
client.request("numbers.agg.avg", [3, 4, 5, 7].to_json)

To connect via WebSockets, add the websocket gem to your Gemfile and use the ws:// or wss:// protocol in your connection URL.

# Gemfile
gem 'websocket'

# Usage
nats = NATS.connect("wss://demo.nats.io:8443")

For NATS v2.0+ authentication, you can use JWT user credentials or bare NKEYS.

• User Credentials: Provide a path to a file containing the JWT and NKEYS using the user_credentials option.
• Bare NKEYS: Provide a path to a read-only file containing the nkey seed using the nkeys_seed option. This requires the nkeys gem to be installed.

# Using User Credentials
NATS.connect("tls://connect.ngs.global", user_credentials: "/path/to/creds")

# Using Bare NKEYS
NATS.connect("tls://connect.ngs.global", nkeys_seed: "path/to/seed.txt")

Install the pure Ruby NATS client using the standard Ruby gem command.

gem install nats-pure

JetStream allows for stream management and pull-based subscriptions. You can access JetStream via nc.jetstream. Common tasks include adding a stream and using pull_subscribe to fetch batches of messages with fetch.

require 'nats/client'

nc = NATS.connect("nats://demo.nats.io:4222")
js = nc.jetstream

# Add a stream
js.add_stream(name: "mystream", subjects: ["foo"])

# Pull subscription
psub = js.pull_subscribe("foo", "bar")

loop do
  begin
    msgs = psub.fetch(5)
    msgs.each do |msg|
      msg.ack
    end
  rescue NATS::IO::Timeout
    puts "Retry later..."
  end
end

Services manage internal subscriptions. You can manually stop a service and drain its subscriptions using service.stop.

Automatic Stopping: A service automatically stops if a NATS-related error occurs during work. You can use the on_stop callback to handle graceful shutdowns or log errors.

service.on_stop do |error|
  puts "Server stopped due to #{error.message}"
end

service.endpoints.add("error") do |message|
  raise NATS::IO::ServerError
end

To use TLS, pass a tls hash containing an OpenSSL::SSL::SSLContext object to the NATS.connect method. Ensure your server URL uses the tls:// protocol.

tls_context = OpenSSL::SSL::SSLContext.new
tls_context.ssl_version = :TLSv1_2

NATS.connect({
   servers: ['tls://127.0.0.1:4444'],
   reconnect: false,
   tls: {
     context: tls_context
   }
})

By default, the client learns about other servers in a cluster from the server it connects to. To prevent this, use the ignore_discovered_urls: true option during connection.

NATS.connect(servers: ['nats://127.0.0.1:4444'], ignore_discovered_urls: true)

When connecting to a NATS server, the client supports several authentication methods including User/Password, Auth Tokens, JWT, and NKey. The connection string (URI) determines the protocol (e.g., nats://, tls://, ws://, or wss://).

Authentication Types:

• User/Password: Provided via the URI (e.g., nats://user:pass@localhost).
• Auth Token: Provided via the @auth_token option.
• JWT/NKey: Uses callback mechanisms (@user_jwt_cb or @user_nkey_cb) and a signature callback (@signature_cb) to handle challenge-response authentication during the CONNECT phase.

When connecting to a NATS cluster, provide an options hash containing a servers array. You can also attach event handlers to monitor the connection lifecycle: on_error, on_reconnect, on_disconnect, and on_close.

cluster_opts = {
  servers: ["nats://127.0.0.1:4222", "nats://127.0.0.1:4223"],
  dont_randomize_servers: true,
  reconnect_time_wait: 0.5,
  max_reconnect_attempts: 2
}

nats = NATS.connect(cluster_opts)

nats.on_error { |e| puts "Error: #{e}" }
nats.on_reconnect { puts "Reconnected!" }
nats.on_disconnect { puts "Disconnected!" }
nats.on_close { puts "Connection closed" }

A NATS::Client instance moves through several states. You can check the current state using the .status reader or helper methods like .connected?, .disconnected?, etc.