Packages
step_flow
0.0.18
1.9.0-rc2
1.9.0-rc1
1.9.0-rc0
1.8.2
1.8.1
1.8.1-rc8
1.8.1-rc7
1.8.1-rc6
1.8.1-rc5
1.8.1-rc4
1.8.1-rc3
1.8.1-rc2
1.8.1-rc1
1.8.1-rc0
1.8.0
1.8.0-rc3
1.8.0-rc2
1.8.0-rc1
1.8.0-rc0
1.7.3
1.7.3-rc4
1.7.3-rc3
1.7.3-rc2
1.7.3-rc1
1.7.3-rc0
1.7.2
1.7.2-rc4
1.7.2-rc3
1.7.2-rc2
1.7.2-rc1
1.7.2-rc0
1.7.1
1.7.0
1.7.0-rc1
1.7.0-rc0
1.6.1
1.6.1-rc1
1.6.1-rc0
1.6.0
1.6.0-rc9
1.6.0-rc8
1.6.0-rc7
1.6.0-rc6
1.6.0-rc5
1.6.0-rc4
1.6.0-rc3
1.6.0-rc20
1.6.0-rc2
1.6.0-rc19
1.6.0-rc18
1.6.0-rc17
1.6.0-rc16
1.6.0-rc15
1.6.0-rc14
1.6.0-rc13
1.6.0-rc12
1.6.0-rc11
1.6.0-rc10
1.6.0-rc1
1.5.0
1.5.0-rc1
1.4.2-rc2
1.4.2-rc1
1.4.1
1.4.1-rc1
1.4.0
1.4.0-rc4
1.4.0-rc3
1.4.0-rc2
1.4.0-rc1
1.3.1
1.3.0
1.3.0-rc
1.2.0
1.1.0
1.0.0
1.0.0-rc9
1.0.0-rc8
1.0.0-rc7
1.0.0-rc6
1.0.0-rc5
1.0.0-rc1
0.2.13
0.2.12
0.2.11
0.2.10
0.2.9
0.2.8
0.2.7
0.2.6
0.2.5
0.2.4
0.2.3
0.2.2
0.2.1
0.2.0
0.1.8
0.1.7
0.1.6
0.1.5
0.1.4
0.1.3
0.1.2
0.1.1
0.1.0
0.0.24
0.0.23
0.0.22
0.0.21
0.0.20
0.0.19
0.0.18
0.0.17
0.0.16
0.0.15
0.0.14
0.0.13
0.0.12
0.0.11
0.0.10
0.0.9
0.0.8
0.0.7
0.0.6
0.0.4
0.0.3
0.0.2
0.0.1
Step flow manager for Elixir applications
Current section
Files
Jump to
Current section
Files
lib/step_flow/amqp/common_consumer.ex
defmodule StepFlow.Amqp.CommonConsumer do
@moduledoc """
Definition of a Common Consumer of RabbitMQ queue.
To implement a consumer,
```elixir
defmodule MyModule do
use StepFlow.Amqp.CommonConsumer, %{
queue: "name_of_the_rabbit_mq_queue",
consumer: &MyModule.consume/4
}
def consume(channel, tag, redelivered, payload) do
...
Basic.ack(channel, tag)
end
end
```
"""
@doc false
defmacro __using__(opts) do
quote do
use GenServer
use AMQP
alias StepFlow.Amqp.Helpers
@doc false
def start_link do
GenServer.start_link(__MODULE__, :ok, name: __MODULE__)
end
@doc false
def init(:ok) do
rabbitmq_connect()
end
# Confirmation sent by the broker after registering this process as a consumer
def handle_info({:basic_consume_ok, %{consumer_tag: _consumer_tag}}, channel) do
{:noreply, channel}
end
# Sent by the broker when the consumer is unexpectedly cancelled (such as after a queue deletion)
def handle_info({:basic_cancel, %{consumer_tag: _consumer_tag}}, channel) do
{:stop, :normal, channel}
end
# Confirmation sent by the broker to the consumer process after a Basic.cancel
def handle_info({:basic_cancel_ok, %{consumer_tag: _consumer_tag}}, channel) do
{:noreply, channel}
end
def handle_info(
{:basic_deliver, payload, %{delivery_tag: tag, redelivered: redelivered}},
channel
) do
queue = unquote(opts).queue
data =
payload
|> Jason.decode!()
Logger.warn("#{__MODULE__}: receive message on queue: #{queue}")
spawn(fn -> unquote(opts).consumer.(channel, tag, redelivered, data) end)
{:noreply, channel}
end
def handle_info({:DOWN, _, :process, _pid, _reason}, _) do
{:ok, chan} = rabbitmq_connect()
# {:noreply, :ok}
end
def terminate(_reason, state) do
AMQP.Connection.close(state.connection)
end
defp rabbitmq_connect do
url = Helpers.get_amqp_connection_url()
case AMQP.Connection.open(url) do
{:ok, connection} ->
init_amqp_connection(connection)
{:error, message} ->
Logger.error(
"#{__MODULE__}: unable to connect to: #{url}, reason: #{inspect(message)}"
)
# Reconnection loop
:timer.sleep(10_000)
rabbitmq_connect()
end
end
defp init_amqp_connection(connection) do
Process.monitor(connection.pid)
{:ok, channel} = AMQP.Channel.open(connection)
queue = unquote(opts).queue
if queue == "worker_discovery" do
:ok = AMQP.Basic.qos(channel, prefetch_count: 1)
end
AMQP.Queue.declare(channel, "job_response_not_found", durable: true)
exchange =
AMQP.Exchange.topic(channel, "job_response",
durable: true,
arguments: [{"alternate-exchange", :longstr, "job_response_not_found"}]
)
exchange = AMQP.Exchange.fanout(channel, "job_response_delayed", durable: true)
{:ok, job_response_delayed_queue} =
AMQP.Queue.declare(channel, "job_response_delayed",
arguments: [
{"x-message-ttl", :short, 5000},
{"x-dead-letter-exchange", :longstr, ""}
]
)
AMQP.Queue.bind(channel, "job_response_delayed", "job_response_delayed", routing_key: "*")
AMQP.Queue.declare(channel, queue,
durable: true,
arguments: [
{"x-dead-letter-exchange", :longstr, "job_response_delayed"},
{"x-dead-letter-routing-key", :longstr, queue}
]
)
Logger.warn("#{__MODULE__}: bind #{queue}")
AMQP.Queue.bind(channel, queue, "job_response", routing_key: queue)
Logger.warn("#{__MODULE__}: connected to queue #{queue}")
{:ok, _consumer_tag} = AMQP.Basic.consume(channel, queue)
{:ok, channel}
end
end
end
end