computeElement.erl

%%%----------------------------------------------------
%%% Copyright AnalyzERL 2011
%%% All rights reserved. No part of this computer programs(s) may be 
%%% used, reproduced,stored in any retrieval system, or transmitted,
%%% in any form or by any means, electronic, mechanical, photocopying,
%%% recording, or otherwise without prior written permission of 
%%% the authors
%%%--------------------------------------------------------------------- 

%%%--------------------------------------------------------------------- 
%%% Revision History
%%%--------------------------------------------------------------------- 
%%% Revision 1.0 Author: Souvik Ray (rsouvik@gmail.com)
%%%--------------------------------------------------------------------- 



-module(computeElement).
-compile(export_all).
-define(OUTPUTFILE, "output.txt").
-define(DEBUGFILE, "debug.txt").

% Computing Element 

% handler list will be read from configuration file
start(Name, Node, HandlerList) -> 
  {ok, Fdd} = file:open(?DEBUGFILE, [append]),
  io:fwrite(Fdd, "~p~n", ["Compute Element: Called on node"]),
  io:fwrite(Fdd, "~p~n", [Node]),
  io:fwrite(Fdd, "~p~n", [HandlerList]),
  file:close(Fdd),
  %parse HandlerList which is read from config file
  global:register_name(erlang:list_to_atom(Name), spawn(erlang:list_to_atom(Node), computeElement, init, [HandlerList])).

init(HandlerList) ->
  {ok, Fdd} = file:open(?DEBUGFILE, [append]),
  io:fwrite(Fdd, "~p", ["Compute Element: Initialized "]),
  file:close(Fdd),
  loop(initialize(HandlerList)).

initialize([]) -> [];
initialize([{Handler, InitData}|Rest]) ->
  [{Handler, Handler:init(InitData)}|initialize(Rest)].

call(Name, Msg) -> 
  %Name ! {request, self(), Msg},
    {ok, Fd} = file:open(?DEBUGFILE, [append]),
    io:fwrite(Fd, "~p", ["Inside call: "]),
    file:close(Fd),
  global:send(Name, {request, self(), Msg}),
  receive
    {reply, Reply} -> 
    {ok, Fdd} = file:open(?DEBUGFILE, [append]),
    io:fwrite(Fdd, "~p", ["Reply received from comp element: "]),
    io:fwrite(Fdd, "~p~n", [Reply]),
    file:close(Fdd),
    Reply 
   end.

reply(To, Msg) ->
  To ! {reply, Msg}.

loop(State) -> 
  receive
    {request, From, Msg} -> 
      {Reply, NewState} = handle_msg(Msg, State),
      reply(From, Reply),
      handle_output(NewState),
      loop(NewState);
    {stop, From} -> 
      reply(From, terminate(State))
  end.

%client fun
stop(Name) ->
  Name ! {stop, self()},
  receive {reply, Reply} -> Reply end.

terminate([]) -> [];
terminate([{Handler, Data}|Rest]) ->
  [{Handler, Handler:terminate(Data)}|terminate(Rest)].

%more client funs
add_handler(Name, Handler, InitData) ->
  call(Name, {add_handler, Handler, InitData}).

delete_handler(Name, Handler) ->
  call(Name, {delete_handler, Handler}).

get_data(Name, Handler) ->
  call(Name, {get_data, Handler}).

send_event(Name, Event) ->
  call(Name, {send_event, Event}).

send(Name, Event) -> 
  call(Name, Event).

handle_msg({add_handler, Handler, InitData}, LoopData) ->
  {ok, [{Handler, Handler:init(InitData)}|LoopData]};

handle_msg({delete_handler, Handler, InitData}, LoopData) ->
  case lists:keysearch(Handler, 1, LoopData) of
    false ->
      {{error, instance}, LoopData};
    {value, {Handler,Data}} ->
      Reply = {data, Handler:terminate(Data)},
      NewLoopData = lists:keydelete(Handler, 1, LoopData),
      {Reply, NewLoopData}
  end;

handle_msg({get_data, Handler}, LoopData) ->
  case lists:keysearch(Handler, 1, LoopData) of
    false               -> {{error, instance}, LoopData};
    {value, {Handler, Data}} -> {{data, Data}, LoopData}
  end;

handle_msg(Event, LoopData) ->
  {ok, event(Event, LoopData)}.

event(_Event, []) -> [];
event(Event, [{Handler, Data}|Rest]) ->
  [{Handler, Handler:handle_event(Event, Data)}|event(Event, Rest)].

%This can be controlled by timing specs in conf file
handle_output(LoopData) -> 
  {ok, generate_output(LoopData)}.

%Handler: aggregrator
%Data: {Event-key, Cnt}
generate_output([]) -> [];
generate_output([{Handler, Data}|Rest]) ->
  [{Handler, dispatchEvent(Handler:output(Data))}|generate_output(Rest)].

% EventKey is the routing key, EventData is the data generated by Handler output function
dispatchEvent(Data) ->
  case Data of 
    {EventKey, EventType, EventData} -> 
     Reply = networkManager:route(EventKey, EventType, EventData, self());
    %data sink
    %print to stdout or store to persistent store
    {EventData} ->  
       {ok, Fd} = file:open(?OUTPUTFILE, [append]),
       io:fwrite(Fd, "~p~n", [EventData]),
       file:close(Fd)
  end.