Packages
gen_amqp
5.0.3
7.0.0
6.0.0
5.0.6
5.0.5
5.0.4
5.0.3
5.0.2
5.0.1
5.0.0
4.5.1
4.5.0
4.4.1
4.4.0
4.3.2
4.3.1
4.3.0
4.2.1
4.2.0
4.1.0
4.0.1
4.0.0
3.6.3
3.6.2
3.6.1
3.6.0
3.5.1
3.5.0
3.4.0
3.3.1
3.3.0
3.2.1
3.2.0
3.1.0
3.0.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.4
2.0.2
2.0.0
1.0.0
0.7.1
0.7.0
0.6.6
0.6.5
0.6.0
GenAMQP is a set of utilities to make microservices using the worker pattern
Current section
Files
Jump to
Current section
Files
lib/gen_amqp/client.ex
defmodule GenAMQP.Client do
@moduledoc """
Client for consuming AMQP services
"""
alias GenAMQP.{Conn, Chan}
@spec call_with_conn(GenServer.name(), String.t(), String.t(), String.t(), Keyword.t()) :: any
def call_with_conn(conn_name, exchange, route, payload, opts \\ []) when is_binary(payload) do
max_time = Keyword.get(opts, :max_time, 5_000)
around_chan(conn_name, fn chan ->
{:ok, correlation_id} = Chan.request(chan, exchange, route, payload, self(), opts)
wait_response(correlation_id, max_time)
end)
end
@spec publish_with_conn(GenServer.name(), String.t(), String.t(), String.t(), Keyword.t()) ::
any
def publish_with_conn(conn_name, exchange, route, payload, opts \\ [])
when is_binary(payload) do
around_chan(conn_name, fn chan ->
Chan.publish(chan, exchange, route, payload, opts)
end)
end
defp around_chan(conn_name, execute) do
chan_name = UUID.uuid4()
{:ok, chan} = Conn.create_chan(conn_name, chan_name, store: false)
try do
execute.(chan)
after
:ok = Conn.close_chan(conn_name, chan)
else
resp -> resp
end
end
defp wait_response(correlation_id, max_time) do
receive do
{:basic_deliver, payload, %{correlation_id: ^correlation_id}} ->
payload
_ ->
wait_response(correlation_id, max_time)
after
max_time ->
{:error, :timeout}
end
end
end