Packages
gen_amqp
5.0.2
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
alias GenAMQP.Chan
require Logger
def start_link(_) do
GenServer.start_link(__MODULE__, nil, [])
end
def init(_) do
{:ok, nil}
end
def handle_call({:do_work, data}, _from, state) do
work(data)
{:reply, nil, state}
end
defp work(%{
event: event,
exec_module: exec_module,
before_funcs: before_funcs,
after_funcs: after_funcs,
chan: chan,
payload: payload,
meta: meta
}) 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 = System.stacktrace()
Logger.error(inspect(st))
case create_error([e, st]) do
{:reply, resp} ->
{true, resp}
:noreply ->
{false, nil}
end
catch
kind, reason ->
Logger.error("STACKTRACE - EXIT")
st = System.stacktrace()
Logger.error(inspect(st))
case create_error([kind, reason, st]) do
{:reply, resp} ->
{true, resp}
:noreply ->
{false, nil}
end
end
resp = reduce_with_funcs(after_funcs, event, resp)
if reply? do
reply(chan, meta, resp)
end
end
defp reply(
_chan,
%{reply_to: :undefined, correlation_id: :undefined},
_resp
),
do: nil
defp reply(chan, %{reply_to: _, correlation_id: _} = meta, resp)
when is_binary(resp) do
Chan.response(chan, meta, resp)
end
defp reply(chan, %{reply_to: _, correlation_id: _} = meta, resp) do
Logger.error("message in wrong type #{inspect(resp)}")
Chan.response(chan, meta, create_error("message in wrong type"))
end
defp create_error(args) do
module = Application.get_env(:gen_amqp, :error_handler)
sol = apply(module, :handle, args)
IO.puts("SOL = #{inspect(sol)}")
sol
end
defp reduce_with_funcs(funcs, event, payload) do
Enum.reduce(funcs, payload, fn f, acc ->
f.(event, acc)
end)
end
end