manhtranlinh
RPC with Broadway and RabbitMQ, the caller can auto start the receiver
Hi everyone,
I implemented the example from this blog RPC over RabbitMQ (with Elixir) – Andrea Leopardi.
My dev environment:
Elixir 1.14.2, RabbitMQ 3.12.0, Erlang 25.3.2.2, {:amqp, "~> 3.3"}, {:broadway, "~> 1.0"}, {:broadway_rabbitmq, "~> 0.7.0"}
I created: a caller module, a receiver module use Broadway and can start with Supervisor.
The situation is: if I did not start the receiver module, but run the caller program I still get the result as the receiver is started.
I don’t know if whether it is wrong with my dev environment or it is the auto start mechanism of Broadway, RabbitMQ and Supervisor.
If someone face this situation, please help to explain to me.
- caller_rpc_service.exs
defmodule CallerRPCService do
def wait_for_messages(_channel, correlation_id) do
receive do
{:basic_deliver, payload, %{correlation_id: ^correlation_id}} ->
IO.puts("Response Message: #{inspect(payload)}")
end
end
end
{:ok, connection} = AMQP.Connection.open()
{:ok, channel} = AMQP.Channel.open(connection)
{:ok, %{queue: queue_name}} = AMQP.Queue.declare(channel, "", exclusive: true)
AMQP.Basic.consume(channel, queue_name, nil)
{headers, message} = {[{"destination", "my_service"}], "Hello Broadway and RPC."}
correlation_id = :erlang.unique_integer() |> :erlang.integer_to_binary() |> Base.encode64()
AMQP.Exchange.declare(channel, "rpc", :headers,
arguments: [{"destination", :longstr, "my_service"}]
)
AMQP.Basic.publish(channel, "rpc", "", message,
reply_to: queue_name,
correlation_id: correlation_id,
headers: headers
)
CallerRPCService.wait_for_messages(channel, correlation_id)
- receiver_rpc_service.ex
defmodule MyService.RPCConsumer do
use Broadway
@producer BroadwayRabbitMQ.Producer
@producer_config [
queue: "my_service.rpcs",
declare: [durable: true],
bindings: [
{"rpc", arguments: [{"destination", :longstr, "my_service"}]}
],
metadata: [:reply_to, :correlation_id],
on_failure: :reject_and_requeue # mandatory option
]
def start_link(_args) do
options = [
name: RPCConsumerPipeline,
producer: [
module: {@producer, @producer_config}
],
processors: [
default: []
]
]
Broadway.start_link(__MODULE__, options)
end
def handle_message(_, %Broadway.Message{} = message, _context) do
IO.puts("Request message: #{inspect(message)}")
AMQP.Basic.publish(
message.metadata.amqp_channel,
"",
message.metadata.reply_to,
"We have got it. Ok!",
correlation_id: message.metadata.correlation_id
)
message
end
end
- application.ex
defmodule RabbimqTutorials.Application do
@moduledoc false
alias MyService.RPCConsumer
use Application
@impl true
def start(_type, _args) do
children = [
RPCConsumer
]
opts = [strategy: :one_for_one, name: RabbimqTutorials.Supervisor]
Supervisor.start_link(children, opts)
end
end
Popular in Questions
What is the idiomatic way of matching for not nil in Elixir?
E.g.,
First way:
defp halt_if_not_signed_in(conn, signed_in_account) when...
New
Hello!
Sorry for this astonishing simple question, but I’m really stuck. I try to set up the intellij-elixir plugin, but I don’t know ho...
New
Hi there,
I am working with Ecto-Postgresql and I need to call all of the records from a specific table but the table has 40,000 record...
New
I want to try my hand at web scraping. What tools/libraries do I need to use. I’m hoping to turn this into something professional so don’...
New
I wanted to check elixir version in phoenix because i found that my elixir is 1.5 but when i use Enum.chunk_by it said the function is un...
New
The Elixir Typespec docs show the following syntax for keyword lists in typespecs:
# ...
| [key: type] # keyword lis...
New
I have a list say
x = ["23gh", "56kh", "97mh"]
I would like to pass each element to Val in each iteration.
Say, in iteration 1 -------...
New
Forgive me if this is obvious, but how does one delete a database record WITHOUT selecting it first? https://hexdocs.pm/ecto/Ecto.Repo.h...
New
Sometimes I want to check if the input into a function is not a blank string.
My first approach:
defmodule Example do
def do_stuff(s...
New
To simplify some tasks at work, I wrote and published this package yesterday. It’s a simple macro that enables Access behaviour on struct...
New
Other popular topics
TL;DR: I’ve just released an implementation of Microsoft’s IDE-independent Language Server Protocol for Elixir. It adds language support ...
New
I have a relationship of love and hate with Elixir. Lots of things are just absolutely right, but there are some things that are kind of ...
New
I would like to know that is there any online source for learning Phoenix Framework for building E-Commerce Store? Any advantage on build...
New
What is the idiomatic way of matching for not nil in Elixir?
E.g.,
First way:
defp halt_if_not_signed_in(conn, signed_in_account) when...
New
Hello all!
I am typing this post from my new MacBook Pro with the M1 chip. I’m loving it so far, and will probably use it as my daily dr...
New
Hello!
Sorry for this astonishing simple question, but I’m really stuck. I try to set up the intellij-elixir plugin, but I don’t know ho...
New
Hi folks,
Few months ago I have announced the proof-of-concept of the library to manipulate the browsers DOM objects directly from Elixi...
New
Forgive me if this is obvious, but how does one delete a database record WITHOUT selecting it first? https://hexdocs.pm/ecto/Ecto.Repo.h...
New
ExUnit now has describe blocks which is a welcome addition coming from RSpec. In the docs, it states that nested hierarchies of describe ...
New
After calling mix ecto.create I get this error:
17:00:32.162 [error] GenServer #PID<0.412.0> terminating
** (Postgrex.Error) FATAL...
New







