%%--------------------------------------------------------------------
%%
%% pipeline_bridge: this server spawns a pipeline and feeds 
%% data into it.
%%
%%--------------------------------------------------------------------
-module(pipeline_bridge).
-behaviour(gen_server).

-include("qa.hrl").

%% Server state
-record(state, { port }).

-export([start_link/0, pipeline/1, list_pipelines/0, reset/0]).

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

start_link() ->
    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).

pipeline({ Pipeline, Data }) ->
    gen_server:call(?MODULE, {pipeline_request, { Pipeline, Data }}).

list_pipelines() ->
    gen_server:call(?MODULE, {ls}).

reset() ->
    gen_server:call(?MODULE, {reset}).

init([]) ->
    process_flag(trap_exit, true),
    Port = open_pbridge(),
    {ok, #state{port = Port }}.

open_pbridge() ->
    DrvPath = os:getenv("RGEAGGPATH") ++ "/lib/erl_drv/pbridge",
    bobutils:local_open_port(DrvPath).

handle_call(Call, _From, #state{port = Port } = State) ->
    case Call of
        { pipeline_request, Msg } ->
            { reply, call_port(Port, [ pipeline, Msg ]), State};
        { Cmd } ->
            { reply, call_port(Port, [ Cmd ]), State}
    end.

handle_cast(_Msg, N) -> { noreply, N + 1 }.

handle_info({'EXIT', Port, Reason}, #state{port = Port} = State) ->
    {stop, {port_terminated, Reason}, State};

handle_info(_Info, State) ->
    {noreply, State}.

terminate(_Reason, _State) -> 
    ok.

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

%%These are all the port management functions

call_port(Port, [ pipeline, { PipelineName, Data } ])  ->
    MD5_chksum = bobutils:md5_hex(term_to_binary(Data)),
    Row = #pdata_note{
        status=noError, 
        time_ingress=erlang:now(), 
        time_egress=0,
        size_ingress=size(term_to_binary(Data)),
        size_egress=0,
        pipeline=PipelineName,
        chksum=MD5_chksum
    },
    case send_msg(Port, lists:append([ pipeline, PipelineName], Data)) of
        { error, _ } ->
            Row1 = Row#pdata_note{time_egress = erlang:now(), 
                                    size_egress=0,
                                    status=error},
            ReturnValue = error;
        Resp ->
            Row1 = Row#pdata_note{time_egress = erlang:now(),
                                    size_egress = size(Resp)},
            ReturnValue = Resp
    end,
    RpcResp = rpc:call(list_to_atom("collection@" ++ net_adm:localhost()), 
        collection_manager, save_pdata_note, [Row1]),
    case RpcResp of
        { badrpc, Reason } ->
            error_logger:error_msg("save pdata failed: ~p~n", [ Reason ]);
        _ ->
            ok
    end,
    ReturnValue;

call_port(Port, [stop]) ->
    erlang:port_close(Port);

call_port(Port, Msg) ->
    send_msg(Port, Msg).
    
send_msg(Port, Msg) ->
    case (catch erlang:term_to_binary(Msg)) of
        Cmd when binary(Cmd) ->
            case (catch erlang:port_command(Port, Cmd)) of
                {'EXIT', Reason} ->
                    error_logger:error_msg("caught exception ~p~n", [Reason]),
                    { error, "port_command threw exception" };
                _ ->
                    get_response(Port)
            end;
        _ ->
            { error, "bad term_to_binary" }
    end.

get_response(Port) ->
   receive
        {Port, { data, Data }} ->
            case (catch erlang:binary_to_term(Data)) of
                {'EXIT', {badarg, _}} ->
                    error_logger:error_msg("badarg from binary_to_term~n"),
                    { error, "badarg" };
                Val ->
                    Val
            end;
        {Port, {exit_status, Status}} ->
            error_logger:error_msg("Port terminated with sig: ~p~n", [Status]),
            call_port(Port, [reset]);
        {'EXIT', _Port, Reason} ->
            error_logger:error_msg("died at exit in receive ~p~n", [Reason]),
            call_port(Port, [reset]),
            { error, "port died" };
        Msg ->
            error_logger:error_msg("caught einval ~p~n", [Msg]),
            { error, "something weird happened" }
    end.
