-module(vent_handler_worker). -behaviour(gen_server). %% API -export([start_link/1, process/2]). %% gen_server callbacks -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). -include("vent_internal.hrl"). -include_lib("amqp_client/include/amqp_client.hrl"). -type handler_response() :: ok | {requeue, term()} | {requeue, number(), term()} | {drop, term()}. -type message() :: #amqp_msg{}. -type timed_message() :: #{msg => message(), processing_start => monotonic_tstamp()}. -record(state, {handler :: module(), handler_state :: term()}). -type state() :: #state{}. %%%=================================================================== %%% API %%%=================================================================== -spec start_link(term()) -> gen_server_startlink_ret(). start_link(Args) -> gen_server:start_link(?MODULE, Args, []). -spec process(pid(), Msg :: timed_message()) -> handler_response(). process(Pid, Msg) -> gen_server:call(Pid, {process, Msg}). %%%=================================================================== %%% gen_server callbacks %%%=================================================================== -spec init([module()]) -> {ok, state()}. init([Handler]) -> {ok, HandlerState} = vent_handler:init(Handler), {ok, #state{ handler = Handler, handler_state = HandlerState }}. -spec handle_call(any(), any(), state()) -> {reply, ok, state()}. handle_call({process, Msg}, _From, State) -> {Reply, State1} = call_handler(Msg, State), {reply, Reply, State1}; handle_call(_Request, _From, State) -> Reply = ok, {reply, Reply, State}. -spec handle_cast(any(), state()) -> {noreply, state()}. handle_cast(_Msg, State) -> {noreply, State}. -spec handle_info(any(), state()) -> {noreply, state()} | {stop, any(), state()}. handle_info(_Info, State) -> {noreply, State}. -spec terminate(any(), any()) -> ok. terminate(_Reason, #state{ handler = Handler, handler_state = HState}) -> catch vent_handler:terminate(Handler, HState), ok. -spec code_change(any(), state(), any()) -> {ok, state()}. code_change(_OldVsn, State, _Extra) -> {ok, State}. %%%=================================================================== %%% Internal functions %%%=================================================================== -spec call_handler(timed_message(), state()) -> {handler_response(), state()}. call_handler(#{msg := {#'basic.deliver'{}, #amqp_msg{payload = Payload}}}, #state{handler = Handler, handler_state = HState} = State) -> DecodedPayload = jsone:decode(Payload), case vent_handler:handle(Handler, DecodedPayload, HState) of {ok, HState1} -> {ok, State#state{handler_state = HState1}}; {requeue, Reason, HState1} -> lager:error("Handler asked to requeue message: ~p~n", [Reason]), {{requeue, Reason}, State#state{handler_state = HState1}}; {requeue, Timeout, Reason, HState1} when Timeout > 0 -> {{requeue, Timeout, Reason}, State#state{handler_state = HState1}}; {drop, Reason, HState1} -> {{drop, Reason}, State#state{handler_state = HState1}} end.