-module(bpe_proc). -author('Maxim Sokhatsky'). -include("bpe.hrl"). -include_lib("kvs/include/cursors.hrl"). -behaviour(gen_server). -export([start_link/1]). -export([init/1, handle_call/3, handle_cast/2, handle_continue/2, handle_info/2, terminate/2, code_change/3]). -export([debug/6, process_event/2, process_task/2, process_task/3]). start_link(Parameters) -> gen_server:start_link(?MODULE, Parameters, []). debug(Proc, Name, Targets, Target, Status, Reason) -> case application:get_env(bpe, debug, true) of true -> logger:notice("BPE: ~ts [~ts:~ts] ~p/~p ~p", [Proc#process.id, Name, Target, Status, Reason, Targets]); false -> skip end. process_messageEvent(#messageEvent{name=Name, payload=Payload}, #process{module=Module} = Proc) -> #result{type=T} = Res = Module:action({messageEvent,Name,Payload},Proc), debug(Proc, "messageEvent", "", Name, T, ""), case Res of #result{continue=C} = X when C =/= [] -> bpe:constructResult(X#result{opt={continue, C}}); X -> bpe:constructResult(X) end. process_event(Event, Proc) -> EventName = element(#messageEvent.id, Event), Targets = bpe_task:targets(EventName, Proc), Target0 = bpe_task:find_flow(EventName, Targets), Target1 = case Target0 of [] -> EventName; T -> T end, {Status, {Reason, Target}, ProcState} = bpe_event:handle_event(Event, Target1, Proc), bpe:add_trace(ProcState, [], Target), %It will be better always use EventName instead of Target #sched{pointer = Pointer, state = ScheduledFlows} = bpe:sched_head(ProcState#process.id), bpe:add_sched(ProcState, Pointer, ScheduledFlows), kvs:append(ProcState, "/bpe/proc"), debug(ProcState, EventName, Targets, Target, Status, Reason), case Status of stop -> {stop, Reason, Target, ProcState}; _ -> {Status, {Reason, Target}, ProcState} end. process_task(Stage, Proc) -> process_task(Stage, Proc, false). process_task(Stage, Proc, NoFlow) -> {_, Curr} = bpe:current_task(Proc), Task = bpe:step(Proc, Curr), Targets = case NoFlow of true -> noflow; _ -> bpe_task:targets(Curr, Proc) end, {Status, {Reason, Target}, ProcState} = case {Targets, Curr, Stage} of {noflow, _, _} -> {reply, {complete, Curr}, Proc}; {[], [], _} -> bpe_task:already_finished(Proc); {[], Curr, _} -> bpe_task:handle_task(Task, Curr, Curr, Proc); {[], _, _} -> bpe_task:denied_flow(Curr, Proc); {List, _, []} -> bpe_task:handle_task(Task, Curr, bpe_task:find_flow(Stage, List), Proc); {List, _, _} -> {reply, {complete, bpe_task:find_flow(Stage, List)}, Proc} end, case Status == stop orelse NoFlow == true of true -> {stop, Reason, Target, ProcState}; _ -> bpe:add_trace(ProcState, [], Target), debug(ProcState, Curr, Targets, Target, Status, Reason), {Status, {Reason, Target}, ProcState} end. convert_api_args(proc, [_ProcId]) -> {get}; convert_api_args(update, [_ProcId, State]) -> {set, State}; convert_api_args(assign, [_ProcId]) -> {ensure_mon}; convert_api_args(Fn, [_ | Args]) -> list_to_tuple([Fn | Args]). handleContinue({noreply, State}, Continue) -> {noreply, State, {continue, lists:flatten(Continue)}}; handleContinue({reply, Reply, State, {continue, C}}, Continue) -> {reply, Reply, State, {continue, lists:flatten(Continue) ++ C}}; handleContinue({reply, Reply, State, _}, Continue) -> {reply, Reply, State, {continue, lists:flatten(Continue)}}; handleContinue({noreply, State, {continue, C}}, Continue) -> {noreply, State, {continue, lists:flatten(Continue) ++ C}}; handleContinue({noreply, State, _}, Continue) -> {noreply, State, {continue, lists:flatten(Continue)}}; handleContinue({reply, Reply, State}, Continue) -> {reply, Reply, State, {continue, lists:flatten(Continue)}}; handleContinue({stop, _, State}, Continue) -> {noreply, State, {continue, lists:flatten(Continue) ++ [#continue{type=stop}]}}; handleContinue({stop, _, Reply, State}, Continue) -> {reply, Reply, State, {continue, lists:flatten(Continue) ++ [#continue{type=stop}]}}; handleContinue(X, _) -> X. % BPMN 2.0 Инфотех handle_call({mon_link, MID}, _, Proc) -> ProcNew = Proc#process{monitor = MID}, {reply, ProcNew, ProcNew}; handle_call({ensure_mon}, _, Proc) -> {Mon, ProcNew} = bpe:ensure_mon(Proc), {reply, Mon, ProcNew}; handle_call({get}, _, Proc) -> {reply, Proc, Proc}; handle_call({set, State}, _, Proc) -> {reply, Proc, State}; handle_call({persist, State}, _, #process{} = _Proc) -> kvs:append(State, "/bpe/proc"), {reply, State, State}; handle_call({next}, _, #process{} = Proc) -> try bpe:processFlow(Proc) catch _X:_Y:Z -> {reply, {error, 'next/1', Z}, Proc} end; handle_call({next, [#continue{} | _] = Continue}, _, #process{} = Proc) -> try handleContinue(bpe:processFlow(Proc), Continue) catch _X:_Y:Z -> {reply, {error, 'next/1', Z}, Proc, {continue, Continue}} end; handle_call({next, Stage}, _, Proc) -> try bpe:processFlow(Stage, Proc) catch _X:_Y:Z -> {reply, {error, 'next/2', Z}, Proc} end; handle_call({amend, Form}, _, Proc) -> try bpe:processFlow(bpe_env:append(env, Proc, Form)) catch _X:_Y:Z -> {reply, {error, 'amend/2', Z}, Proc} end; handle_call({discard, Form}, _, Proc) -> try bpe:processFlow(bpe_env:remove(env, Proc, Form)) catch _X:_Y:Z -> {reply, {error, 'discard/2', Z}, Proc} end; handle_call({event, Event}, _, Proc) -> try process_event(Event, Proc) catch _X:_Y:Z -> {reply, {error, 'event/2', Z}, Proc} end; handle_call({messageEvent, Event}, _, Proc) -> try process_messageEvent(Event, Proc) catch _X:_Y:Z -> {reply, {error, 'event/2', Z}, Proc} end; handle_call({messageEvent, Event, [#continue{} | _] = Continue}, _, Proc) -> try handleContinue(process_messageEvent(Event, Proc), Continue) catch _X:_Y:Z -> {reply, {error, 'event/2', Z}, Proc} end; % BPMN 1.0 ПриватБанк handle_call({complete}, _, Proc) -> try process_task([], Proc) catch _X:_Y:Z -> {reply, {error, 'complete/1', Z}, Proc} end; handle_call({complete, [#continue{} | _] = Continue}, _, Proc) -> try handleContinue(process_task([], Proc), Continue) catch _X:_Y:Z -> {reply, {error, 'complete/1', Z}, Proc, {continue, Continue}} end; handle_call({complete, Stage}, _, Proc) -> try process_task(Stage, Proc) catch _X:_Y:Z -> {reply, {error, 'complete/2', Z}, Proc} end; handle_call({modify, Form, append}, _, Proc) -> try process_task([], bpe_env:append(env, Proc, Form), true) catch _X:_Y:Z -> {reply, {error, 'append/2', Z}, Proc} end; handle_call({modify, Form, remove}, _, Proc) -> try process_task([], bpe_env:remove(env, Proc, Form), true) catch _X:_Y:Z -> {reply, {error, 'remove/2', Z}, Proc} end; handle_call({mon_link, MID, Continue}, _, Proc) -> ProcNew = Proc#process{monitor = MID}, {reply, ProcNew, ProcNew, {continue, Continue}}; handle_call({ensure_mon, Continue}, _, Proc) -> {Mon, ProcNew} = bpe:ensure_mon(Proc), {reply, Mon, ProcNew, {continue, Continue}}; handle_call({set, State, Continue}, _, Proc) -> {reply, Proc, State, {continue, Continue}}; handle_call({persist, State, Continue}, _, #process{} = _Proc) -> kvs:append(State, "/bpe/proc"), {reply, State, State, {continue, Continue}}; handle_call({next, Stage, Continue}, _, Proc) -> try handleContinue(bpe:processFlow(Stage, Proc), Continue) catch _X:_Y:Z -> {reply, {error, 'next/2', Z}, Proc, {continue, Continue}} end; handle_call({amend, Form, Continue}, _, Proc) -> try handleContinue(bpe:processFlow(bpe_env:append(env, Proc, Form)), Continue) catch _X:_Y:Z -> {reply, {error, 'amend/2', Z}, Proc, {continue, Continue}} end; handle_call({discard, Form, Continue}, _, Proc) -> try handleContinue(bpe:processFlow(bpe_env:remove(env, Proc, Form)), Continue) catch _X:_Y:Z -> {reply, {error, 'discard/2', Z}, Proc, {continue, Continue}} end; handle_call({event, Event, Continue}, _, Proc) -> try handleContinue(process_event(Event, Proc), Continue) catch _X:_Y:Z -> {reply, {error, 'event/2', Z}, Proc, {continue, Continue}} end; handle_call({complete, Stage, Continue}, _, Proc) -> try handleContinue(process_task(Stage, Proc), Continue) catch _X:_Y:Z -> {reply, {error, 'complete/2', Z}, Proc, {continue, Continue}} end; handle_call({modify, Form, append, Continue}, _, Proc) -> try handleContinue(process_task([], bpe_env:append(env, Proc, Form), true), Continue) catch _X:_Y:Z -> {reply, {error, 'append/2', Z}, Proc, {continue, Continue}} end; handle_call({modify, Form, remove, Continue}, _, Proc) -> try handleContinue(process_task([], bpe_env:remove(env, Proc, Form), true), Continue) catch _X:_Y:Z -> {reply, {error, 'remove/2', Z}, Proc, {continue, Continue}} end; handle_call(Command, _, Proc) -> {reply, {unknown, Command}, Proc}. handle_continue([#continue{type=spawn, module=Module, fn=Fn, args=Args} | T], #process{} = Proc) -> spawn(fun() -> apply(Module, Fn, Args) end), {noreply, Proc, {continue, T}}; handle_continue([#continue{type=bpe, fn=Fn, args=Args} | T], #process{} = Proc) -> Result = try handle_call(convert_api_args(Fn, Args), [], Proc) catch _X:_Y:Z -> {stop, {error, Z}, Proc} end, case Result of {noreply, State} -> {noreply, State, {continue, T}}; {reply, _, State, {continue, C}} -> {noreply, State, {continue, T ++ C}}; {reply, _, State, _} -> {noreply, State, {continue, T}}; {noreply, State, {continue, C}} -> {noreply, State, {continue, T ++ C}}; {noreply, State, _} -> {noreply, State, {continue, T}}; {reply, _, State} -> {noreply, State, {continue, T}}; {stop, _, State} -> {noreply, State, {continue, T ++ [#continue{type=stop}]}}; {stop, _, _, State} -> {noreply, State, {continue, T ++ [#continue{type=stop}]}}; X -> X end; handle_continue([#continue{type=stop}], #process{} = Proc) -> {stop, normal, Proc}; handle_continue([#continue{type=stop} = X | T], #process{} = Proc) -> case lists:member(#continue{type=stop}, T) of true -> {noreply, Proc, {continue, T}}; _ -> {noreply, Proc, {continue, T ++ [X]}} end; handle_continue([], #process{} = Proc) -> {noreply, Proc}. init(Process) -> Proc = bpe:load(Process#process.id, Process), logger:notice("BPE: ~ts spawned as ~p", [Proc#process.id, self()]), Till = bpe:till(calendar:local_time(), application:get_env(bpe, ttl, 24 * 60 * 60)), bpe:cache({process, Proc#process.id}, self(), Till), [bpe:reg({messageEvent, element(1, EventRec), Proc#process.id}) || EventRec <- bpe:events(Proc)], {ok, Proc#process{timer = erlang:send_after(rand:uniform(10000), self(), {timer, ping})}}. handle_cast(Msg, State) -> logger:notice("BPE: Unknown API async: ~p.", [Msg]), {stop, {error, {unknown_cast, Msg}}, State}. handle_info({timer, ping}, State = #process{timer = _Timer, id = _Id, events = _Events, notifications = _Pid}) -> case application:get_env(bpe, ping_discipline, bpe_ping) of undefined -> {noreply, State}; M -> M:ping(State) end; handle_info({'DOWN', _MonitorRef, _Type, _Object, _Info} = Msg, State = #process{id = Id}) -> logger:notice("BPE: Connection closed, shutting down " "session: ~p.", [Msg]), bpe:cache({process, Id}, undefined), {stop, normal, State}; handle_info(Info, State = #process{}) -> logger:notice("BPE: Unrecognized info: ~p", [Info]), {noreply, State}. terminate(Reason, #process{id = Id}) -> logger:notice("BPE: ~ts terminate Reason: ~p", [Id, Reason]), bpe:cache({process, Id}, undefined), ok. code_change(_OldVsn, State, _Extra) -> {ok, State}.