@spec ack(FireMQ.Message.t()) :: :ok | {:error, FireMQ.Error.t()}

Acknowledges a message. Does nothing for messages received with no_ack: true or already settled.

Like ack/1, but raises on error.

@spec bind_exchange(conn(), String.t(), String.t(), String.t(), keyword()) ::
  :ok | {:error, FireMQ.Error.t()}

Binds exchange destination to exchange source: messages that source routes with key also go to destination. Accepts the same options as bind_queue/5.

Like bind_exchange/5, but raises on error.

@spec bind_queue(conn(), String.t(), String.t(), String.t(), keyword()) ::
  :ok | {:error, FireMQ.Error.t()}

Binds queue to exchange with a routing key or pattern.

Options

• :args - binding arguments, such as "x-match" for headers exchanges.

Like bind_queue/5, but raises on error.

@spec close(conn()) :: :ok

Closes the connection gracefully. Returns :ok, also when the connection is already closed.

@spec connect(String.t(), keyword()) :: {:ok, pid()} | {:error, FireMQ.Error.t()}

Opens a connection.

The URL looks like fire://user:password@host:5673/vhost. User and password default to guest, the port to 5673 (fire://) or 5674 (fires://, TLS), and the vhost to /.

The connection belongs to the calling process and closes when that process exits. To run a connection under a supervisor, use FireMQ.Connection.start_link/1.

Options

• :heartbeat - heartbeat interval in seconds; 0 disables heartbeats. Defaults to 30.
• :connection_name - a name shown in the management UI.
• :connect_timeout - in milliseconds. Defaults to 10_000.
• :ssl - options for :ssl.connect/3 on fires:// URLs, for example ssl: [cacertfile: "ca.pem"].

Examples

{:ok, conn} = FireMQ.connect("fire://localhost")
{:error, %FireMQ.Error{kind: :server, code: 403}} =
  FireMQ.connect("fire://guest:wrong@localhost")

@spec connect!(String.t(), keyword()) :: pid()

Like connect/2, but returns the pid or raises.

@spec declare_exchange(conn(), String.t(), String.t() | atom(), keyword()) ::
  :ok | {:error, FireMQ.Error.t()}

Declares an exchange of type direct, fanout, topic or headers. The type can be a string or an atom.

Options

• :durable - survives a broker restart. Defaults to true.
• :auto_delete - deleted when its last binding is removed.
• :internal - cannot be published to directly.
• :passive - only check that the exchange exists.
• :args - exchange arguments, such as "alternate-exchange".

Like declare_exchange/4, but raises on error.

@spec declare_queue(conn(), String.t(), keyword()) ::
  {:ok, FireMQ.QueueInfo.t()} | {:error, FireMQ.Error.t()}

Declares a queue and returns its name and counts.

An empty name asks the broker to pick a unique one, which %FireMQ.QueueInfo{name: name} reports.

Options

• :durable - survives a broker restart. Defaults to true.
• :exclusive - used only by this connection and deleted when it closes.
• :auto_delete - deleted when its last subscription is removed.
• :passive - only check that the queue exists.
• :temporary - shorthand for durable: false, exclusive: true, auto_delete: true.
• :args - queue arguments with RabbitMQ's names, such as "x-message-ttl", "x-max-length", "x-overflow" or "x-dead-letter-exchange".

Examples

{:ok, %FireMQ.QueueInfo{name: "jobs"}} = FireMQ.declare_queue(conn, "jobs")
{:ok, %FireMQ.QueueInfo{name: "amq.gen-" <> _}} =
  FireMQ.declare_queue(conn, "", temporary: true)

Like declare_queue/3, but returns the FireMQ.QueueInfo or raises.

@spec delete_exchange(conn(), String.t(), keyword()) ::
  :ok | {:error, FireMQ.Error.t()}

Deletes an exchange.

Options

• :if_unused - fail with code 406 when the exchange has bindings.

Like delete_exchange/3, but raises on error.

@spec delete_queue(conn(), String.t(), keyword()) ::
  {:ok, non_neg_integer()} | {:error, FireMQ.Error.t()}

Deletes a queue and returns how many messages it held.

Options

• :if_unused - fail with code 406 when the queue has consumers.
• :if_empty - fail with code 406 when the queue has messages.

Like delete_queue/3, but returns the message count or raises.

@spec get(conn(), String.t(), keyword()) ::
  {:ok, FireMQ.Message.t() | nil} | {:error, FireMQ.Error.t()}

Takes one message from a queue. Returns {:ok, nil} when the queue is empty.

Options

• :no_ack - treat the message as acknowledged once sent. Defaults to false, in which case settle it with ack/1 or nack/2.

Like get/3, but returns the message (or nil) or raises.

@spec nack(FireMQ.Message.t(), boolean()) :: :ok | {:error, FireMQ.Error.t()}

Rejects a message. With requeue set to true it is delivered again, with redelivered: true. Otherwise it is dead-lettered when the queue has a dead-letter exchange, and dropped if not.

Like nack/2, but raises on error.

@spec publish(conn(), String.t(), String.t(), iodata(), keyword()) ::
  :ok | {:error, FireMQ.Error.t()}

Publishes a message and waits for the broker's confirm.

Use exchange "" to send straight to the queue named by key. The body is a binary or iodata.

Options

• :persistent - write the message to disk. Defaults to true.
• :mandatory - fail with kind :unroutable when no queue receives the message.
• :confirm - wait for the broker's confirm. Defaults to true. When false, publish/5 returns once the message is written to the socket.
• :properties - a map or keyword list of message properties: :content_type, :content_encoding, :correlation_id, :reply_to, :expiration (milliseconds), :message_id, :timestamp (seconds since the Unix epoch, or a DateTime), :type, :user_id, :app_id and :priority (0 to 255).
• :headers - application headers, a map with string keys and any MessagePack value.

Errors

• kind :unroutable - a mandatory message reached no queue.
• kind :nacked - the broker rejected the message, for example because a queue with x-overflow: reject-publish is full.

Examples

:ok = FireMQ.publish(conn, "", "jobs", "hello")

:ok =
  FireMQ.publish(conn, "orders", "order.created", Jason.encode!(order),
    properties: [content_type: "application/json", message_id: "o-17"],
    headers: %{"trace" => "abc"}
  )

Like publish/5, but raises on error.

@spec purge_queue(conn(), String.t(), keyword()) ::
  {:ok, non_neg_integer()} | {:error, FireMQ.Error.t()}

Removes every ready message from a queue and returns how many were removed.

Like purge_queue/3, but returns the message count or raises.

@spec server_info(conn()) :: {:ok, map()} | {:error, FireMQ.Error.t()}

Returns what the broker reported in the handshake: :server (name and version), :conn_id, :heartbeat, :max_frame and :caps.

@spec subscribe(conn(), String.t(), keyword()) ::
  {:ok, sub_ref()} | {:error, FireMQ.Error.t()}

Starts consuming from a queue. Messages arrive in the calling process's mailbox as

{:firemq, sub_ref, %FireMQ.Message{}}

When the broker ends the subscription (for example because the queue was deleted) or the connection closes, the process receives {:firemq_cancelled, sub_ref, %FireMQ.Error{}}. The subscription ends when the subscribing process exits; messages it had not settled go back to the queue.

Acknowledge each message with ack/1 or nack/2. The broker sends at most :prefetch unacknowledged messages; the client grants more as you settle them, half the window at a time.

Options

• :prefetch - how many unacknowledged messages the broker may send at once. Defaults to 100.
• :no_ack - treat messages as acknowledged when sent. Faster, but messages are lost if the consumer crashes.
• :exclusive - fail with code 405 if the queue has other consumers, and block new ones.
• :tag - consumer tag shown in the management UI.
• :args - consumer arguments.

Like subscribe/3, but returns the subscription reference or raises.

@spec unbind_queue(conn(), String.t(), String.t(), String.t(), keyword()) ::
  :ok | {:error, FireMQ.Error.t()}

Removes a binding made with bind_queue/5. Takes the same options.

Like unbind_queue/5, but raises on error.

@spec unsubscribe(conn(), sub_ref(), keyword()) :: :ok | {:error, FireMQ.Error.t()}

Stops a subscription. Messages already received can still be acknowledged; messages that were on their way go back to the queue.

Like unsubscribe/3, but raises on error.