glats

A NATS client for Gleam. This wraps Elixir’s client, Gnat.

Publish

import gleam/result
import glats

pub fn main() {
  use conn <- result.then(glats.connect("localhost", 4222, []))

  // Publish a single message to "some.subject".
  assert Ok(Nil) = glats.publish(conn, "some.subject", "hello world!")

  Ok(Nil)
}

Subscribe

import gleam/io
import gleam/result
import gleam/erlang/process
import glats

pub fn main() {
  use conn <- result.then(glats.connect("localhost", 4222, []))

  let subject = process.new_subject()

  // Subscribe to "some.subject".
  // Messages will be delivered to the erlang subject passed in.
  assert Ok(sid) = glats.subscribe(conn, subject, "some.subject")

  // Publish a single message to "some.subject".
  assert Ok(Nil) = glats.publish(conn, "some.subject", "hello world!")

  // Receive from erlang subject.
  assert Ok(glats.ReceivedMessage(
    conn: _conn, // Reference to the conn used
    sid: _sid,   // Subscription ID for the subscription
    message: glats.Message(
      subject: _subject,   // "some.subject"
      headers: _headers,   // empty map
      reply_to: _reply_to, // None
      body: _body,         // "hello world!"
    )
  )) = process.receive(subject, 1000)

  // Unsubscribe from the subscription.
  assert Ok(Nil) = glats.unsubscribe(conn, sid)

  Ok(Nil)
}

Request handler

import gleam/option.{None}
import gleam/result
import gleam/erlang/process
import glats
import glats/handler.{Reply, Request, Response}

pub fn main() {
  use conn <- result.then(glats.connect("localhost", 4222, []))

  // Start a request handler actor that will call `ping_pong_handler`
  // for every request received from NATS subject "do.ping".
  assert Ok(_actor) =
    handler.handle_request(conn, [], "do.ping", None, ping_pong_handler)

  process.sleep_forever()

  Ok(Nil)
}

pub fn ping_pong_handler(req: Request, state) {
  // Got message: Hello
  io.println("Got message: " <> req.body)

  // Reply with a message with the same headers and append to body.
  Reply(
    Response(
      headers: req.headers,
      reply_to: None,
      body: req.body <> " from glats!",
    ),
    state,
  )
}

Then in shell with natscli.

$ nats req do.ping 'Hello'
12:16:47 Sending request on "do.ping"
12:16:47 Received with rtt 427.64µs
Hello from glats!

Jetstream

Pull subscription

Pull consumers allow clients to request batches of messages on demand. For best performance set the batch size to as high as possible, but keep in mind that the client should be able to process (and ack) the entire batch in the AckWait time window of the consumer.

In the example below we request a batch of 50 messages with NoWait set. This instructs NATS to immediately return a response, event when no pending messages exist in the stream (in which case we get a message with status 404). That way we can exit the program when we’re caught up.

Normally you would request a batch with an expiry time (option Expires(Int)), in which case the request lasts for the time requested (and a message with status 408 is received when it expires).

This allows the client to control the flow of messages instead of NATS flooding it when there is high load as is possible with push consumers.

More info on pull consumers

All the logic in this example is handled for you in glats/jetstream/handler.{handle_pull_consumer}, see docs.

import gleam/io
import gleam/string
import gleam/result
import gleam/option.{Some}
import gleam/otp/actor.{InitFailed}
import gleam/erlang/process.{Abnormal}
import glats.{ReceivedMessage}
import glats/jetstream
import glats/jetstream/stream
import glats/jetstream/consumer.{
  AckExplicit, AckPolicy, Batch, BindStream, Description, InactiveThreshold,
  NoWait, Subscription, With,
}

const batch_size = 50

type State {
  State(sub: Subscription, remaining: Int)
}

pub fn main() {
  use conn <- result.then(glats.connect("localhost", 4222, []))

  // Create a stream
  let assert Ok(stream) =
    stream.create(conn, "mystream", ["orders.>", "items.>"], [])

  // Publish 3 messages to subjects contained in the stream
  let assert Ok(Nil) = glats.publish(conn, "orders.1", "order_data", [])
  let assert Ok(Nil) = glats.publish(conn, "orders.2", "order_data", [])
  let assert Ok(Nil) = glats.publish(conn, "items.1", "item_data", [])

  // Subscribe to subject in the stream using an ephemeral consumer
  let subject = process.new_subject()
  let assert Ok(sub) =
    consumer.subscribe(
      conn,
      subject,
      "orders.*",
      [
        // Bind to stream created above
        BindStream(stream.config.name),
        // Set description for the ephemeral consumer
        With(Description("An ephemeral consumer for subscription")),
        // Set ack policy for the consumer
        With(AckPolicy(AckExplicit)),
        // Sets the inactive threshold of the ephemeral consumer
        With(InactiveThreshold(60_000_000_000)),
      ],
    )
    |> io.debug

  // Start loop
  case request_and_loop(subject, State(sub: sub, remaining: 0)) {
    Ok(Nil) -> Ok(Nil)
    Error(err) -> {
      io.println("got error: " <> string.inspect(err))

      Error(InitFailed(Abnormal(string.inspect(err))))
    }
  }
}

// Requests a batch of messages for the pull consumer and updates state
fn request_and_loop(subject, state: State) {
  case consumer.request_batch(state.sub, [Batch(batch_size), NoWait]) {
    Ok(Nil) -> loop(subject, State(..state, remaining: batch_size))
    Error(_) -> Error(Nil)
  }
}

// Receives messages
fn loop(subject, state: State) {
  case process.receive(subject, 2000) {
    // When request expires we get a message with status of `408`.
    Ok(ReceivedMessage(status: Some(408), ..)) -> {
      io.println("request expired, requesting more")

      request_and_loop(subject, state)
    }

    // When request is made with `NoWait` we get a message with status
    // of `404` when no new messages exist.
    Ok(ReceivedMessage(status: Some(404), ..)) -> {
      io.println("no new messages")

      Ok(Nil)
    }

    // We got a new message!
    Ok(ReceivedMessage(conn: conn, message: msg, ..)) -> {
      // Print message
      io.debug(msg)

      // Acknowledge message
      let assert Ok(_) = jetstream.ack(conn, msg)

      // Keep track of remaining messages in the request
      case state.remaining <= 1 {
        // Request more
        True -> request_and_loop(subject, state)
        // Decrement the counter
        False -> loop(subject, State(..state, remaining: state.remaining - 1))
      }
    }

    // An error!
    Error(err) -> Error(err)
  }
}

Push subscription

With push consumers messages are automatically published on a specified subject.

This example is considerably simpler than the pull consumer above but keep in mind that with push consumers the flow of messages is harder to control.

More info on push consumers

import gleam/io
import gleam/result
import gleam/erlang/process.{Subject}
import glats.{ReceivedMessage, SubscriptionMessage}
import glats/jetstream
import glats/jetstream/stream
import glats/jetstream/consumer.{
  AckExplicit, AckPolicy, BindStream, DeliverSubject, Description,
  InactiveThreshold, With,
}

pub fn main() {
  use conn <- result.then(glats.connect("localhost", 4222, []))

  // Create a stream
  let assert Ok(stream) =
    stream.create(conn, "mystream", ["orders.>", "items.>"], [])

  // Publish 3 messages to subjects contained in the stream
  let assert Ok(Nil) = glats.publish(conn, "orders.1", "order_data", [])
  let assert Ok(Nil) = glats.publish(conn, "orders.2", "order_data", [])
  let assert Ok(Nil) = glats.publish(conn, "items.1", "item_data", [])

  // Generate a random inbox topic for the push consumer's delivery topic
  let inbox = glats.new_inbox()

  // Subscribe to subject in the stream using an ephemeral consumer
  let subject = process.new_subject()
  let assert Ok(_sub) =
    consumer.subscribe(
      conn,
      subject,
      "orders.*",
      [
        // Bind to stream created above
        BindStream(stream.config.name),
        // Make it a push consumer
        With(DeliverSubject(inbox)),
        // Set description for the ephemeral consumer
        With(Description("An ephemeral consumer for subscription")),
        // Set ack policy for the consumer
        With(AckPolicy(AckExplicit)),
        // Sets the inactive threshold of the ephemeral consumer
        With(InactiveThreshold(60_000_000_000)),
      ],
    )
    |> io.debug

  // Start loop
  loop(subject)
}

fn loop(subject: Subject(SubscriptionMessage)) {
  case process.receive(subject, 2000) {
    // New message received
    Ok(ReceivedMessage(conn: conn, message: msg, ..)) -> {
      // Print message
      io.debug(msg)

      // Acknowledge message
      let assert Ok(Nil) = jetstream.ack(conn, msg)

      // Run loop again
      loop(subject)
    }

    // Error!
    Error(Nil) -> {
      io.println("no new message in 2 seconds")

      Ok(Nil)
    }
  }
}

Installation

gleam add glats

and its documentation can be found at https://hexdocs.pm/glats.