Overview

An Elixir Client for the RabbitMQ Streams Protocol.

Introduced in RabbiMQ 3.9, Streams is an alternative to Queues differing mainly by implementing “non-destructive consumer semantics”.

While this feature is avaiable when using the existing Queues, it shines when used with its dedicated protocol, allowing messages to be consumed extremelly fast.

This library implements a client for said protocol, managing the connection to the server and providing an API for producing, consuming and running available commands.

Installation

The package can be installed by adding rabbitmq_stream to your list of dependencies in mix.exs:

def deps do
  [
    {:rabbitmq_stream, "~> 0.4.0"},
    # ...
  ]
end

Consuming

First you define a connection

defmodule MyApp.MyConnection do
  use RabbitMQStream.Connection
end

You then can declare a consumer module by using the RabbitMQStream.Consumer:

defmodule MyApp.MyConsumer do
  use RabbitMQStream.Consumer,
    connection: MyApp.MyConnection,
    stream_name: "my_stream",
    initial_offset: :first

  @impl true
  def handle_message(_message) do
    :ok
  end
end

Producing

You can define a Producer with:

defmodule MyApp.MyProducer do
  use RabbitMQStream.Producer,
    stream_name: "stream-01",
    connection: MyApp.MyConnection
end

Then you can publish messages to the stream:

MyApp.MyProducer.publish("Hello World")

SuperStreams

A super stream is a logical stream made of individual, regular streams.

You can declare SuperStreams with:

:ok = MyApp.MyConnection.create_super_stream("my_super_stream", "route-A": ["stream-01", "stream-02"], "route-B": ["stream-03"])

And you can consume from it with:

defmodule MyApp.MySuperConsumer do
  use RabbitMQStream.SuperConsumer,
    initial_offset: :next,
    super_stream: "my_super_stream"

  @impl true
  def handle_message(_message) do
    # ...
    :ok
  end
end