glomp

A STOMP client for talking to message brokers such as RabbitMQ, ActiveMQ and Artemis. Start a client, and it connects in the background and reconnects by itself if the connection drops. You can then send messages and subscribe to destinations.

let client =
  glomp.new(host: "localhost", port: 61_613)
  |> glomp.credentials(login: "guest", passcode: "guest")

let assert Ok(started) = glomp.start(client)
let assert Ok(Nil) = glomp.await_connected(client, within: 5000)
let assert Ok(Nil) =
  glomp.send(client, glomp.message(to: "/queue/jobs", body: <<"hi">>))

Types

AckMode

</>

How many messages the broker marks as done when you acknowledge one.

To acknowledge a message is to tell the broker you have handled it, so the broker does not send it again.

pub type AckMode {
  Cumulative
  Individual
}

Constructors

```
Cumulative
```
An acknowledgement covers this message and all earlier ones. This is STOMP’s client mode.
```
Individual
```
An acknowledgement covers only this message. This is STOMP’s client-individual mode.

AlreadySubscribed

</>

subscribe returns this when the client already has a subscription with this id.

pub type AlreadySubscribed {
  AlreadySubscribed(id: String)
}

Constructors

```
AlreadySubscribed(id: String)
```

Client

opaque </>

A client and its settings.

You get a Client from new. Change its settings with functions such as credentials, tls and with_subscription, then pass it to start or supervised:

let client =
  glomp.new(host: "localhost", port: 61_613)
  |> glomp.credentials(login: "guest", passcode: "guest")

let assert Ok(started) = glomp.start(client)
let assert Ok(Nil) = glomp.await_connected(client, within: 5000)

Every function that talks to the broker takes the same Client, such as send, subscribe, ack and stop. You can use it from any process.

A Client finds its running process by name, so it keeps working after a supervisor restarts the client. While the client is stopped, calls with it crash, because no process has that name.

Change the settings before you start the client. A running client keeps the settings it started with.

pub opaque type Client

Delivery

opaque </>

A token for acknowledging one message.

The handler of an acknowledged_subscription gets a Delivery with each message. To acknowledge the message later, return Defer from the handler, keep the Delivery, and pass it to ack or nack when the work is done:

glomp.acknowledged_subscription(
  id: "jobs",
  destination: "/queue/jobs",
  mode: glomp.Individual,
  handler: fn(message, delivery) {
    process.send(worker, #(message, delivery))
    glomp.Defer
  },
)

// Later, when the worker has finished
let assert Ok(Nil) = glomp.ack(client, delivery)

A Delivery only works on the connection its message arrived on. When that connection ends, ack and nack return Expired, and the broker sends the message again on the next connection.

pub opaque type Delivery

Disconnect

</>

The reason a connection ended, or why the client could not connect.

pub type Disconnect {
  Unreachable(
    ipv4: network.NetworkError,
    ipv6: option.Option(network.NetworkError),
  )
  ConnectorCrashed(reason: process.ExitReason)
  HandshakeTimedOut(waited_ms: Int)
  Refused(message: String, frame: frame.Frame)
  UnsupportedVersion(version: String)
  UnexpectedFrame(frame: frame.Frame)
  Malformed(error: frame.DecodeError)
  BrokerError(message: String, frame: frame.Frame)
  Closed
  SocketFailed(error: network.NetworkError)
  HeartBeatMissed(silent_ms: Int)
}

Constructors

Unreachable(
  ipv4: network.NetworkError,
  ipv6: option.Option(network.NetworkError),
)

Connecting over IPv4 failed, and so did IPv6. The client only tries IPv6 if there is time left.

```
ConnectorCrashed(reason: process.ExitReason)
```
The process that makes the connection crashed.
```
HandshakeTimedOut(waited_ms: Int)
```
The broker did not answer CONNECT within the connect timeout.
```
Refused(message: String, frame: frame.Frame)
```
The broker answered CONNECT with an ERROR frame. This usually means the credentials are wrong or another client is using the client id.
```
UnsupportedVersion(version: String)
```
The broker chose a STOMP version that glomp does not support. Glomp supports 1.1 and 1.2.
```
UnexpectedFrame(frame: frame.Frame)
```
The broker sent a frame that is not valid at that moment. For example, it answered CONNECT with a MESSAGE, or sent a second CONNECTED.
```
Malformed(error: frame.DecodeError)
```
The broker sent a frame that is invalid or too large.

BrokerError(message: String, frame: frame.Frame)

The broker sent an ERROR frame.

```
Closed
```
The broker closed the connection.

SocketFailed(error: network.NetworkError)

Reading or writing the socket failed.

```
HeartBeatMissed(silent_ms: Int)
```
Two heart-beat intervals passed without anything from the broker.

DropReason

</>

The reason the client ignored a frame from the broker.

pub type DropReason {
  MissingHeader(name: String)
  UnknownSubscription(id: String)
  UnknownReceipt(id: String)
}

Constructors

```
MissingHeader(name: String)
```
The frame is missing a header it needs.
```
UnknownSubscription(id: String)
```
A MESSAGE arrived for a subscription the client does not know. This usually happens just after you unsubscribe.
```
UnknownReceipt(id: String)
```
A RECEIPT arrived with an id the client does not know. This usually means it came after the wait ended with NoReceipt.

Error

</>

The reason a request such as send or ack failed.

pub type Error {
  NotConnected
  ConnectionLost(reason: Disconnect)
  Expired
  NoReceipt(waited_ms: Int)
}

Constructors

```
NotConnected
```
The client is not connected. It does not queue requests to send later, so the request was not sent.
```
ConnectionLost(reason: Disconnect)
```
The connection dropped while the frame was being written. The broker may or may not have received it.
```
Expired
```
The Delivery or Transaction came from an earlier connection, so the broker no longer knows it.
```
NoReceipt(waited_ms: Int)
```
The broker’s receipt did not arrive in time. The broker may still have done what you asked.

Event

</>

News about the connection, such as connecting or losing it. Your on_event function receives each one.

pub type Event {
  Connected(session: Session)
  Subscribed(id: String)
  Disconnected(reason: Disconnect, retry_in_ms: Int)
  Dropped(frame: frame.Frame, reason: DropReason)
}

Constructors

```
Connected(session: Session)
```
The client connected and asked the broker for every subscription.
```
Subscribed(id: String)
```
The broker confirmed this subscription.
```
Disconnected(reason: Disconnect, retry_in_ms: Int)
```
The connection ended, or an attempt to connect failed. The client tries again after retry_in_ms.
```
Dropped(frame: frame.Frame, reason: DropReason)
```
The client skipped a frame because it could not use it.

Message

</>

A message from the broker.

Your subscription’s handler gets each Message as it arrives. Read its fields directly, or use header to find one header. A Message is a plain value, so it stays valid after the connection ends.

pub type Message {
  Message(
    destination: String,
    id: String,
    subscription: String,
    headers: List(#(String, String)),
    body: BitArray,
  )
}

Constructors

```
Message(
  destination: String,
  id: String,
  subscription: String,
  headers: List(#(String, String)),
  body: BitArray,
)
```
Arguments

destination

The destination it was sent to.

id

The broker’s message-id.

subscription

The id of the subscription it arrived on.

headers

All headers, in the order the broker sent them.

body

The body exactly as the broker sent it.

NotSubscribed

</>

unsubscribe returns this when the client does not have a subscription with this id.

pub type NotSubscribed {
  NotSubscribed(id: String)
}

Constructors

```
NotSubscribed(id: String)
```

Outbound

opaque </>

A message to send.

Create one with message, change it with functions such as content_type, then pass it to send or send_confirmed:

let outbound =
  glomp.message(to: "/queue/jobs", body: <<"{}">>)
  |> glomp.content_type("application/json")

let assert Ok(Nil) = glomp.send(client, outbound)

An Outbound is a plain value, so you can send it more than once. One made with in_transaction only works while its transaction does.

pub opaque type Outbound

Session

</>

Details about a connection, which the client and broker agreed on when they connected.

You get a Session in the Connected event, which your on_event function receives. It describes that one connection. The next connection brings a new Session.

pub type Session {
  Session(
    version: frame.Version,
    server: option.Option(String),
    id: option.Option(String),
    heart_beat_send_ms: Int,
    heart_beat_receive_ms: Int,
    headers: List(#(String, String)),
  )
}

Constructors

```
Session(
  version: frame.Version,
  server: option.Option(String),
  id: option.Option(String),
  heart_beat_send_ms: Int,
  heart_beat_receive_ms: Int,
  headers: List(#(String, String)),
)
```
Arguments

server

The broker’s server header.

id

The broker’s session header.

heart_beat_send_ms

How often the client sends a heart-beat. If it is 0, the client does not send them.

heart_beat_receive_ms

How often the broker sends a heart-beat. If it is 0, the broker does not send them.

headers

All headers from the broker’s CONNECTED frame, which is its reply to CONNECT.

Stopped

</>

What happened to the connection when you called stop.

stop returns this once the client has exited. You can log it or ignore it, because the client is gone either way.

pub type Stopped {
  Graceful
  Unconfirmed
  Abrupt(reason: Disconnect)
  Idle
}

Constructors

```
Graceful
```
The broker confirmed the DISCONNECT.
```
Unconfirmed
```
The broker did not confirm the DISCONNECT within two seconds.
```
Abrupt(reason: Disconnect)
```
The connection failed before the broker could confirm.
```
Idle
```
The client was not connected.

Subscription

opaque </>

A subscription asks the broker to send you the messages from one destination, such as a queue. It has an id, the destination and the handler that receives each message.

Create one with subscription or acknowledged_subscription. Add it with with_subscription before the client starts, or with subscribe while it runs. The client subscribes again on every new connection, until you pass its id to unsubscribe.

pub opaque type Subscription

Transaction

opaque </>

A group of sends and acknowledgements that the broker applies all at once, when you commit.

You get a Transaction from begin. Add messages with in_transaction, and acknowledgements with ack_in and nack_in.

Then call commit to apply them, or abort to throw them away:

let assert Ok(transaction) = glomp.begin(client)

let outbound =
  glomp.message(to: "/queue/jobs", body: <<"hi">>)
  |> glomp.in_transaction(transaction)
let assert Ok(Nil) = glomp.send(client, outbound)

let assert Ok(Nil) = glomp.commit(client, transaction, within: 5000)

A Transaction only works on the connection that started it, because the broker aborts open transactions when that connection ends. After that, using it returns Expired.

pub opaque type Transaction

Verdict

</>

What your handler tells the broker about a message. The handler returns one of these for each message it gets.

pub type Verdict {
  Ack
  Nack
  Defer
}

Constructors

```
Ack
```
The message was handled. The client sends ACK to the broker.
```
Nack
```
The message was not handled. The client sends NACK, and the broker decides what to do with the message.
```
Defer
```
You will decide later. Keep the Delivery and call ack or nack when you’re done.

Values

abort

</>

pub fn abort(
  client: Client,
  transaction: Transaction,
) -> Result(Nil, Error)

Abort a transaction.

Errors: Expired, ConnectionLost.

ack

</>

pub fn ack(
  client: Client,
  delivery: Delivery,
) -> Result(Nil, Error)

Tell the broker you handled a deferred message, by sending ACK.

Errors: Expired, ConnectionLost.

ack_in

</>

pub fn ack_in(
  client: Client,
  delivery: Delivery,
  transaction: Transaction,
) -> Result(Nil, Error)

Like ack, but as part of a transaction.

acknowledged_subscription

</>

pub fn acknowledged_subscription(
  id id: String,
  destination destination: String,
  mode mode: AckMode,
  handler handler: fn(Message, Delivery) -> Verdict,
) -> Subscription

Create a subscription where you acknowledge each message.

The handler returns a Verdict. To decide later, return Defer and pass the Delivery to ack or nack.

After a reconnect, the broker sends again any message you had not acknowledged yet. This means a handler can see the same message twice.

await_connected

</>

pub fn await_connected(
  client: Client,
  within within_ms: Int,
) -> Result(Nil, Nil)

Wait up to within_ms for the client to connect.

Error means the time ran out. The client keeps trying to connect in the background.

begin

</>

pub fn begin(client: Client) -> Result(Transaction, Error)

Begin a transaction. The broker holds back what you add to it until you call commit. Add to it with in_transaction, ack_in and nack_in.

Errors: NotConnected, ConnectionLost.

client_id

</>

pub fn client_id(client: Client, id: String) -> Client

Set the client-id header. ActiveMQ and Artemis use it to keep durable subscriptions across reconnects. They allow only one connection per id at a time, so a second client with the same id is refused.

commit

</>

pub fn commit(
  client: Client,
  transaction: Transaction,
  within within_ms: Int,
) -> Result(Nil, Error)

Commit a transaction and wait for the broker’s receipt.

Errors: Expired, ConnectionLost, NoReceipt.

connect_header

</>

pub fn connect_header(
  client: Client,
  name name: String,
  value value: String,
) -> Client

Add a header to the CONNECT frame.

connect_timeout

</>

pub fn connect_timeout(
  client: Client,
  milliseconds: Int,
) -> Client

Set the time limit for one connection attempt, in milliseconds.

content_type

</>

pub fn content_type(
  outbound: Outbound,
  content_type: String,
) -> Outbound

Set the content-type header, such as application/json.

credentials

</>

pub fn credentials(
  client: Client,
  login login: String,
  passcode passcode: String,
) -> Client

Set the login and passcode headers.

Each STOMP header is one line of text, so a line feed ends it. Glomp sends the CONNECT frame exactly as written, because STOMP 1.0 brokers expect that. If login or passcode contains a line feed, the header ends there and the broker reads the rest as a separate header.

</>

pub fn header(
  message: Message,
  name: String,
) -> Result(String, Nil)

Get the first header called name.

heart_beat

</>

pub fn heart_beat(
  client: Client,
  send_ms send_ms: Int,
  receive_ms receive_ms: Int,
) -> Client

Set the heart-beat intervals, in milliseconds. A heart-beat is a small keep-alive that shows the connection is still working.

send_ms is how often the client can send one. receive_ms is how often it wants to receive one. Setting either to 0 turns that direction off.

For each direction, the client and broker use the larger of their two numbers. If either side says 0, that direction is off. You can see the agreed values in Session.

The client sends at half the agreed interval, so a slow network still delivers one in time. If two whole intervals pass with nothing from the broker, the client disconnects with HeartBeatMissed.

in_transaction

</>

pub fn in_transaction(
  outbound: Outbound,
  transaction: Transaction,
) -> Outbound

Send the message as part of a transaction. The broker only delivers it when you call commit.

limits

</>

pub fn limits(client: Client, limits: frame.Limits) -> Client

Set the largest frame the client will accept. This stops one huge frame from using up memory. If a larger frame arrives, the client closes the connection with Malformed.

message

</>

pub fn message(
  to destination: String,
  body body: BitArray,
) -> Outbound

Create a message for destination.

message_header

</>

pub fn message_header(
  outbound: Outbound,
  name name: String,
  value value: String,
) -> Outbound

Add a header to the SEND frame.

nack

</>

pub fn nack(
  client: Client,
  delivery: Delivery,
) -> Result(Nil, Error)

Tell the broker you did not handle a deferred message, by sending NACK.

Errors: Expired, ConnectionLost.

nack_in

</>

pub fn nack_in(
  client: Client,
  delivery: Delivery,
  transaction: Transaction,
) -> Result(Nil, Error)

Like nack, but as part of a transaction.

new

</>

pub fn new(host host: String, port port: Int) -> Client

Create a client for the broker at host and port.

Call new once per client, when your program starts. Each call creates an atom, which is a name the Erlang VM keeps forever. The VM can only hold a fixed number of atoms, so calling new over and over while your program runs will eventually crash the VM.

on_event

</>

pub fn on_event(
  client: Client,
  handler: fn(Event) -> Nil,
) -> Client

Set a function to receive Events. It runs in the handler process, in the same order as your message handlers.

reconnect

</>

pub fn reconnect(
  client: Client,
  first_ms first_ms: Int,
  longest_ms longest_ms: Int,
) -> Client

Set how long to wait before reconnecting, in milliseconds.

The wait starts at first_ms and doubles after each failure, up to longest_ms. This keeps a broker that is down from being flooded with attempts. Once a connection stays up for longest_ms, the wait goes back to first_ms.

send

</>

pub fn send(
  client: Client,
  outbound: Outbound,
) -> Result(Nil, Error)

Send a message. Ok means the message was written to the socket. It does not mean the broker has received it.

Errors: NotConnected, ConnectionLost, Expired.

send_confirmed

</>

pub fn send_confirmed(
  client: Client,
  outbound: Outbound,
  within within_ms: Int,
) -> Result(Nil, Error)

Send a message and wait for the broker’s receipt. A receipt is the broker’s reply saying it has received the frame.

Errors: the same as send, plus NoReceipt.

start

</>

pub fn start(
  client: Client,
) -> Result(actor.Started(Client), actor.StartError)

Start a client linked to the caller. The client stops when the caller exits.

This returns as soon as the client is running, and the client connects after that. To wait for the connection, use await_connected.

It fails if two subscriptions share an id, or if this client is already running.

stop

</>

pub fn stop(client: Client) -> Stopped

Stop a client started with start.

The client sends DISCONNECT and waits up to two seconds for the receipt, then it closes the socket and exits. Any handlers still running are killed.

A supervisor normally starts a stopped client again. The client disconnects and exits, its supervisor sees that it has exited, and the supervisor starts a new client that connects again.

To stop a supervised client for good, stop its supervisor. You can also give the client the Transient restart strategy. A supervisor only restarts a transient child after a crash, so the client stays stopped after stop:

glomp.supervised(client)
|> supervision.restart(supervision.Transient)

</>

pub fn subscribe(
  client: Client,
  subscription: Subscription,
) -> Result(Nil, AlreadySubscribed)

Add a subscription to a running client. If the client is connected, it subscribes straight away. It also subscribes on every later connection.

Ok means the client added it. The Subscribed event comes later, when the broker confirms it.

subscription

</>

pub fn subscription(
  id id: String,
  destination destination: String,
  handler handler: fn(Message) -> Nil,
) -> Subscription

Create a plain subscription. This is STOMP’s auto mode.

A plain subscription does not need to acknowledge messages. A message is only lost if the connection dies while you’re handling it. To keep those messages, use acknowledged_subscription.

Give each subscription in a client its own id. The broker puts the id on every message, so the client knows which handler to call. start fails if two subscriptions share an id.

subscription_header

</>

pub fn subscription_header(
  subscription: Subscription,
  name name: String,
  value value: String,
) -> Subscription

Add a header to the SUBSCRIBE frame, such as activemq.subscriptionName, selector or prefetch-count.

supervised

</>

pub fn supervised(
  client: Client,
) -> supervision.ChildSpecification(Client)

A child specification for a supervisor.

When the supervisor shuts it down, the client sends DISCONNECT first.

tls

</>

pub fn tls(client: Client, trust trust: network.Trust) -> Client

Connect over TLS. Pass the broker’s TLS port to new. Many brokers use port 61614 for this.

unsubscribe

</>

pub fn unsubscribe(
  client: Client,
  id: String,
) -> Result(Nil, NotSubscribed)

Remove a subscription. Messages the broker had already sent for it arrive as Dropped events.

virtual_host

</>

pub fn virtual_host(
  client: Client,
  virtual_host: String,
) -> Client

Set the host header, which picks a virtual host on the broker. It defaults to the broker’s host name. RabbitMQ calls this a vhost.

with_subscription

</>

pub fn with_subscription(
  client: Client,
  subscription: Subscription,
) -> Client

Add a subscription before the client starts.

glomp.new(host:, port:)
|> glomp.with_subscription(
  glomp.subscription(id: "jobs", destination: "/queue/jobs", handler:),
)

The client subscribes on every connection, in the order you added them.

To add one to a client that is already running, use subscribe.

without_content_length

</>

pub fn without_content_length(outbound: Outbound) -> Outbound

Leave out the content-length header. ActiveMQ treats a message with this header as bytes and a message without it as text.

A body that contains a NUL byte still gets the header. A NUL byte marks the end of a frame, so the broker needs the length to read past it.

Constructors

Constructors

Constructors

Constructors

Constructors

Constructors

Constructors

Arguments

Constructors

Constructors

Arguments

Constructors

Constructors