Packages
gen_amqp
4.0.0
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/worker.ex
defmodule GenAMQP.PoolWorker do
use GenServer
alias GenAMQP.Conn
require Logger
def start_link(_) do
GenServer.start_link(__MODULE__, nil, [])
end
def init(_) do
{:ok, nil}
end
def handle_cast(
{:incoming,
%{
event: event,
exec_module: exec_module,
before_funcs: before_funcs,
after_funcs: after_funcs,
conn_name: conn_name,
chan_name: chan_name,
payload: payload,
meta: meta
}},
state
) do
payload = reduce_with_funcs(before_funcs, event, payload)
{reply?, resp} =
try do
case apply(exec_module, :execute, [payload]) do
{:reply, resp} ->
{true, resp}
:noreply ->
{false, nil}
other ->
case apply(exec_module, :handle, [other]) do
{:reply, resp} ->
{true, resp}
:noreply ->
{false, nil}
end
end
rescue
e ->
Logger.error("STACKTRACE - RESCUE")
st = inspect(System.stacktrace())
Logger.error(st)
case create_error({inspect(e), st}) do
{:reply, resp} ->
{true, resp}
:noreply ->
{false, nil}
end
catch
:exit, reason ->
Logger.error("STACKTRACE - EXIT")
st = inspect(System.stacktrace())
Logger.error(st)
case create_error({reason, st}) do
{:reply, resp} ->
{true, resp}
:noreply ->
{false, nil}
end
end
resp = reduce_with_funcs(after_funcs, event, resp)
if reply? do
reply(conn_name, chan_name, meta, resp)
end
{:noreply, state}
end
defp reply(
_conn_name,
_chan_name,
%{reply_to: :undefined, correlation_id: :undefined},
_resp
),
do: nil
defp reply(conn_name, chan_name, %{reply_to: _, correlation_id: _} = meta, resp)
when is_binary(resp) do
Conn.response(conn_name, meta, resp, chan_name)
end
defp reply(conn_name, chan_name, %{reply_to: _, correlation_id: _} = meta, resp) do
Logger.error("message in wrong type #{inspect(resp)}")
Conn.response(conn_name, meta, create_error("message in wrong type"), chan_name)
end
defp create_error(msg) do
module = Application.get_env(:gen_amqp, :error_handler)
sol = apply(module, :handle, [msg])
IO.puts("SOL = #{inspect(sol)}")
apply(module, :handle, [msg])
end
defp reduce_with_funcs(funcs, event, payload) do
Enum.reduce(funcs, payload, fn f, acc ->
f.(event, acc)
end)
end
end