rabbitmq-stream
rabbitmq-stream copied to clipboard
Elixir Client for RabbitMQ Streams Protocol.
RabbitMQStream
Elixir Client for RabbitMQ Streams Protocol.
Overview
RabbiMQ 3.9 introduced the Streams as an alternative to Queues, but differs mainly by implementing a "non-destructive consumer semantics". A consumer can read the messages starting at any offset, while receiving new messages.
While this feature is avaiable when using the existing Queues, it shines when used with its dedicated protocol, that allows messages to be consumed extremelly fast, in comparisson to Queues.
This library aims to be a Client for the Streams Protocol, managing connections and providing an idiomatic way of interacting with all the features avaiable for this functionallity.
Features
- Producing and Consuming from Streams
- Offset Tracking and Control Flow(Credits) helpers
- Stream Filtering
- Single Active Consumer
- Super Streams
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.1"},
# ...
]
end
Producing
RabbitMQ Streams protocol needs a static :reference_name
per producer. This is used to prevent message duplication. For this reason, each stream needs, for now, a static module to publish messages, which keeps track of its own publishing_id
.
You can define a Producer
module like this:
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")
Consuming
First you define a connection
defmodule MyApp.MyConnection do
use RabbitMQStream.Connection
end
You then can declare a consumer module with 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
Or you could manually consume from the stream with
{:ok, _subscription_id} = MyApp.MyConnection.subscribe("stream-01", self(), :next, 999)
The caller process will start receiving messages with the format {:deliver, %RabbitMQStream.Message.Types.DeliverData{} = deliver_data}
def handle_info({:deliver, %RabbitMQStream.Message.Types.DeliverData{} = deliver_data}, state) do
# do something with message
{:noreply, state}
end
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
Configuration
The configuration for the connection can be set in your config.exs
file:
config :rabbitmq_stream, MyApp.MyConnection,
username: "guest",
password: "guest"
# ...
end
You can configure a default Serializer module by passing it to the defaults configuration option
config :rabbitmq_stream, :defaults,
serializer: Jason
end
TLS
You can configure the RabbitmqStream to use TLS connections:
coonfig :rabbitmq_stream, :defaults,
connection: [
transport: :ssl,
ssl_opts: [
keyfile: "services/cert/client_box_key.pem",
certfile: "services/cert/client_box_certificate.pem",
cacertfile: "services/cert/ca_certificate.pem"
]
]
For more information, check the documentation.