Skip to content

Commit

Permalink
doc: Adding notes on using the lib on clustered enviroments
Browse files Browse the repository at this point in the history
  • Loading branch information
VictorGaiva committed Jul 14, 2024
1 parent c87a271 commit 7a8d5fe
Show file tree
Hide file tree
Showing 2 changed files with 56 additions and 1 deletion.
13 changes: 12 additions & 1 deletion lib/producer/producer.ex
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@ defmodule RabbitMQStream.Producer do
The RabbitMQStream.Producer accepts the following options:
* `:stream_name` - The name of the stream to publish to. Required.
* `:reference_name` - The string which is used by the server to prevent [Duplicate Message](https://blog.rabbitmq.com/posts/2021/07/rabbitmq-streams-message-deduplication/). Defaults to `__MODULE__.Producer`.
* `:reference_name` - The string which is used by the server to prevent [Duplicate Message](https://blog.rabbitmq.com/posts/2021/07/rabbitmq-streams-message-deduplication/). Defaults to `__MODULE__.Producer`. (If clustering in production, check notes and the end.)
* `:connection` - The identifier for a `RabbitMQStream.Connection`.
* `:serializer` - The module to use to decode the message. Defaults to `nil`, which means no encoding is done.
Expand Down Expand Up @@ -111,6 +111,17 @@ defmodule RabbitMQStream.Producer do
The default value for the `:serializer` is the module itself, unless a default is defined at a higher level of the
configuration. If there is a `encode!/1` callback defined, it is always used
# Notes on Clustering
Be aware that the sequence tracking for each `:reference_name` is global. Meaning the if you are running
your Elixir as a cluster of multiple nodes, and each having a process of a Producer with the same
`:reference_name`, you may encounter issues with message de-duplication, where messages are being
dropped because the sequence on each producer's state might not be up to date after another process
with the same `:reference_name` produced a message.
There might be cases where you would want this behaviour. If not, be sure to declare a unique
`:reference_name` for each process.
"""

defmacro __using__(opts) do
Expand Down
44 changes: 44 additions & 0 deletions test/consumer_test.exs
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,10 @@ defmodule RabbitMQStreamTest.Consumer do
use RabbitMQStream.Connection
end

defmodule SupervisedConnection2 do
use RabbitMQStream.Connection
end

defmodule SupervisorProducer do
use RabbitMQStream.Producer,
connection: SupervisedConnection
Expand Down Expand Up @@ -52,6 +56,9 @@ defmodule RabbitMQStreamTest.Consumer do
{:ok, _conn} = SupervisedConnection.start_link(host: "localhost", vhost: "/")
:ok = SupervisedConnection.connect()

{:ok, _conn} = SupervisedConnection2.start_link(host: "localhost", vhost: "/")
:ok = SupervisedConnection2.connect()

:ok
end

Expand Down Expand Up @@ -168,4 +175,41 @@ defmodule RabbitMQStreamTest.Consumer do

SupervisedConnection.delete_stream(@stream)
end

@stream "consumer-test-stream-04"
@reference_name "reference-04"
test "should not duplicate messages when the reference_name is used twice" do
SupervisedConnection.create_stream(@stream)

{:ok, _subscriber} =
Consumer.start_link(
initial_offset: :next,
stream_name: @stream,
private: self(),
offset_tracking: [count: [store_after: 1]]
)

{:ok, first} =
RabbitMQStream.Producer.start_link(
connection: SupervisedConnection,
stream_name: @stream,
reference_name: "#{@reference_name}"
)

{:ok, second} =
RabbitMQStream.Producer.start_link(
connection: SupervisedConnection2,
stream_name: @stream,
reference_name: "#{@reference_name}"
)

Process.sleep(1000)

:ok = RabbitMQStream.Producer.publish(first, Jason.encode!(%{message: "first"}))
Process.sleep(500)
:ok = RabbitMQStream.Producer.publish(second, Jason.encode!(%{message: "second"}))

assert_receive {:message, %{"message" => "first"}}, 500
refute_receive {:message, %{"message" => "second"}}, 500
end
end

0 comments on commit 7a8d5fe

Please sign in to comment.