kaa

-module(kaa_main_worker).

-behaviour(gen_server).

-include("kaa_core.hrl").

%% API
-export([start_link/0,
    start_link/1,
    get_worker/1,
    kaa_proto_in/2,
    stop_link/1]).

%% gen_server callbacks
-export([init/1,
    handle_call/3,
    handle_cast/2,
    handle_info/2,
    terminate/2,
    code_change/3]).

% the values can be override during initialization
-record(state, {jun_worker = undefined :: pid(),
    mon_ref = undefined :: reference(),
    storage = undefined :: atom(),
    key = <<"kaa">> :: binary()}).

start_link() ->
    start_link(default).

start_link(PyBin) ->
    {ok, Pid} = gen_server:start_link(?MODULE, [PyBin], []),
    % register process using syn
    Key = pid_to_list(Pid),
    ok = syn:register(Key, Pid),
    {ok, Key}.

get_worker(Key) ->
    Pid = syn:find_by_key(Key),
    gen_server:call(Pid, get_worker).

kaa_proto_in(Key, PBMsg) ->
    Pid = syn:find_by_key(Key),
    gen_server:call(Pid, {kaa_proto_in, PBMsg}, infinity).

stop_link(Key) ->
    Pid = syn:find_by_key(Key),
    gen_server:call(Pid, stop_link).

init([PyBin]) ->
    % start the py process and initializes its importing modules
    case jun_worker:start_link(PyBin) of
        {ok, JunPid} ->
            MonRef = erlang:monitor(process, JunPid),
            lager:info("initialized jun worker pid ~p", [JunPid]),
            Key = pid_to_list(self()),
            Storage = ets:new(?KAA_ENVIRONMENT(Key), [named_table, public]),
            {ok, #state{jun_worker = JunPid, mon_ref = MonRef, key = Key,
                storage = Storage}};
        Error      ->
            lager:error("cannot initializes jun worker due to ~p", [Error]),
            {stop, Error}
    end.

handle_call(get_worker, _From, State)            ->
    JunPid = State#state.jun_worker,
    Worker = kaa_proto:worker(JunPid),
    {reply, {ok, Worker}, State};

handle_call({kaa_proto_in, PBMsg}, _From, State) ->
    Result = kaa_proto:exec(PBMsg),    
    {reply, {ok, Result}, State};

handle_call(stop_link, _From, State) ->
    {stop, normal, ok, State};
 
handle_call(_Request, _From, State) ->    
    {reply, ok, State}.

handle_cast(_Request, State) ->
    {noreply, State}.

handle_info({'DOWN', MonRef, _Type, _Object, _Info}, State=#state{mon_ref = MonRef}) ->
    % process py pid is down, which one is the process to restart?
    {noreply, State};
handle_info(_Info, State) ->
    {noreply, State}.

terminate(_Reason, State) ->
    % terminate the jun worker process
    JunPid = State#state.jun_worker,
    ok = jun_worker:stop_link(JunPid),
    % also removes syn register & ets
    Key = State#state.key,
    true = ets:delete(?KAA_ENVIRONMENT(Key)),
    ok = syn:unregister(Key),
    ok.

code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

%% ===================================
%% Internal Funcionts
%% ===================================