firemq-elixir

Elixir client for FireMQ, a RabbitMQ-compatible message broker. It speaks the native FireMQ protocol over TCP or TLS and is written in pure Elixir.

Defaults favour safety: queues and exchanges are durable, messages are persistent, and publish returns only after the broker confirms the message.

Install

Add :firemq to your dependencies in mix.exs. It needs Elixir 1.15 or later.

def deps do
  [
    {:firemq, "~> 0.1"}
  ]
end

Quickstart

Start a broker, for example with firemq --dev, then:

{:ok, conn} = FireMQ.connect("fire://guest:guest@localhost:5673")
{:ok, _info} = FireMQ.declare_queue(conn, "jobs")

# Returns once the broker has taken responsibility for the message.
:ok = FireMQ.publish(conn, "", "jobs", "hello")

{:ok, sub} = FireMQ.subscribe(conn, "jobs")

receive do
  {:firemq, ^sub, message} ->
    IO.puts(message.body)
    :ok = FireMQ.ack(message)
end

:ok = FireMQ.close(conn)

The examples/ directory has the classic tutorials: hello_world, work_queues, publish_subscribe, routing, topics and rpc. Run one with FIREMQ_URL=fire://localhost mix run examples/topics.exs.

Connect

{:ok, conn} = FireMQ.connect("fire://user:password@host:5673/vhost", heartbeat: 30)

The URL schemes are fire:// (TCP, port 5673 by default) and fires:// (TLS, port 5674). User and password default to guest. The vhost is the URL path without its leading /, percent-decoded; an empty path means vhost /.

Options:

For fires://, the client verifies the broker's certificate against the system CA store. Pass your own CA with ssl: [cacertfile: "ca.pem"]. For local development only, you can skip verification with ssl: [verify: :verify_none].

A connection opened with FireMQ.connect/2 belongs to the calling process: it closes when that process exits. To keep a connection running independently, put it under a supervisor (see Supervision).

API

Every function takes the connection first and returns :ok, {:ok, value} or {:error, %FireMQ.Error{}}. Each one also has a bang variant, such as publish!/5, that returns the value or raises FireMQ.Error. Functions that wait for the broker accept a :timeout option in milliseconds, 30_000 by default.

Topology

{:ok, %FireMQ.QueueInfo{name: "jobs", messages: 0, consumers: 0}} =
  FireMQ.declare_queue(conn, "jobs", args: %{"x-max-length" => 10_000})

# An empty name lets the broker pick one. A temporary queue is exclusive to this
# connection and deleted when it closes.
{:ok, %FireMQ.QueueInfo{name: reply_queue}} = FireMQ.declare_queue(conn, "", temporary: true)

:ok = FireMQ.declare_exchange(conn, "orders", :topic)
:ok = FireMQ.bind_queue(conn, "jobs", "orders", "order.*")
:ok = FireMQ.unbind_queue(conn, "jobs", "orders", "order.*")
:ok = FireMQ.bind_exchange(conn, "audit", "orders", "#")

{:ok, purged} = FireMQ.purge_queue(conn, "jobs")
{:ok, deleted} = FireMQ.delete_queue(conn, "jobs", if_empty: true)
:ok = FireMQ.delete_exchange(conn, "orders")

declare_queue/3 options: :durable (default true), :exclusive, :auto_delete, :passive, :temporary and :args. declare_exchange/4 options: :durable (default true), :auto_delete, :internal, :passive and :args. Arguments use RabbitMQ's names, such as x-message-ttl, x-dead-letter-exchange or x-overflow.

Publish

:ok =
  FireMQ.publish(conn, "orders", "order.created", ~s({"id":42}),
    mandatory: true,
    properties: [content_type: "application/json", message_id: "o-42"],
    headers: %{"trace" => "abc", "attempt" => 1}
  )

Use exchange "" to send straight to the queue named by the routing key.

Consume

subscribe/3 delivers messages to the calling process as {:firemq, sub_ref, %FireMQ.Message{}}:

{:ok, sub} = FireMQ.subscribe(conn, "jobs", prefetch: 10)

receive do
  {:firemq, ^sub, %FireMQ.Message{body: body} = message} ->
    case process(body) do
      :ok -> FireMQ.ack(message)
      :retry -> FireMQ.nack(message, true)
      :fail -> FireMQ.nack(message, false)
    end

  {:firemq_cancelled, ^sub, %FireMQ.Error{} = reason} ->
    # The broker ended the subscription (for example, the queue was deleted)
    # or the connection closed.
    {:stopped, reason}
end

:ok = FireMQ.unsubscribe(conn, sub)

A message has body, exchange, routing_key, properties (a map with atom keys), headers (a map with string keys), redelivered and delivery_tag.

The broker sends at most :prefetch unacknowledged messages (100 by default). The client grants more credit as you settle messages, half the window at a time. With no_ack: true, messages count as acknowledged when sent and the client grants credit as they arrive.

nack(message, true) requeues the message, which comes back with redelivered: true. nack(message, false) dead-letters it when the queue has a dead-letter exchange, and drops it otherwise.

When the subscribing process exits, the client cancels the subscription and returns the messages that process had not settled to the queue.

To take a single message, use get/3. It returns {:ok, nil} when the queue is empty:

{:ok, message} = FireMQ.get(conn, "jobs")
if message, do: FireMQ.ack(message)

Supervision

FireMQ.Connection is a GenServer with a child_spec/1, so you can start it under a supervisor and reach it by name:

children = [
  {FireMQ.Connection, url: "fire://guest:guest@localhost:5673", name: MyApp.FireMQ}
]

Supervisor.start_link(children, strategy: :one_for_one)

FireMQ.publish(MyApp.FireMQ, "", "jobs", "hello")

FireMQ.Connection.start_link/1 takes the same options as FireMQ.connect/2, plus :url and :name. If the broker cannot be reached or refuses the login, start_link/1 returns {:error, %FireMQ.Error{}}.

When the broker closes the connection or the socket fails, the connection process exits with {:shutdown, %FireMQ.Error{}} and the supervisor starts a new one. The client does not reconnect on its own, and subscriptions do not survive a restart: subscribers receive {:firemq_cancelled, sub_ref, error} and must subscribe again.

Errors

Failures return {:error, %FireMQ.Error{kind: kind, code: code, message: message}}.

case FireMQ.publish(conn, "orders", "order.unknown", "x", mandatory: true) do
  :ok -> :ok
  {:error, %FireMQ.Error{kind: :unroutable}} -> Logger.warning("nobody is listening")
  {:error, %FireMQ.Error{} = error} -> raise error
end

A server error on one request, such as a 406 when you redeclare a queue with other properties, does not close the connection.

Development

The unit tests need no broker:

mix test

The integration tests and examples need a running broker:

firemq --ephemeral &
FIREMQ_URL=fire://guest:guest@localhost:5673 mix test --only integration

The conformance driver in conformance/driver.exs runs the FireMQ SDK conformance suite. Compile first, so that Mix prints nothing on the driver's standard output:

mix compile
firemq-conformance run --driver "mix run conformance/driver.exs"

License

Apache-2.0. See LICENSE.