src/lwes_channel.erl

   1 -module (lwes_channel).
   2
   3 -behaviour (gen_server).
   4
   5 -include_lib ("lwes.hrl").
   6 -include ("lwes_internal.hrl").
   7
   8 %% API
   9 -export ([ start_link/1,
  10            new/2,
  11            open/1,
  12            register_callback/4,
  13            send_to/2,
  14            close/1
  15          ]).
  16
  17 %% gen_server callbacks
  18 -export ([ init/1,
  19            handle_call/3,
  20            handle_cast/2,
  21            handle_info/2,
  22            terminate/2,
  23            code_change/3
  24          ]).
  25
  26 -record (state, {socket, channel, type, callback}).
  27 -record (callback, {function, format, state}).
  28
  29 %%====================================================================
  30 %% API functions
  31 %%====================================================================
  32 start_link (Channel) ->
  33   gen_server:start_link (?MODULE, [Channel], []).
  34
  35 new (Type, Config) ->
  36   #lwes_channel {
  37      type = Type,
  38      config = lwes_net_udp:new (Type, Config),
  39      ref = make_ref()
  40   }.
  41
  42 open (Channel) ->
  43   { ok, _Pid } = lwes_channel_manager:open_channel (Channel),
  44   { ok, Channel}.
  45
  46 register_callback (Channel, CallbackFunction, EventType, CallbackState) ->
  47   find_and_call ( Channel,
  48                   { register, CallbackFunction, EventType, CallbackState }).
  49
  50 send_to (Channel, Msg) ->
  51   find_and_call (Channel, { send, Msg }).
  52
  53 close (Channel) ->
  54   find_and_cast (Channel, stop).
  55
  56 %%====================================================================
  57 %% gen_server callbacks
  58 %%====================================================================
  59 init ([ Channel = #lwes_channel { type = Type, config = Config } ]) ->
  60   { ok, Socket } = lwes_net_udp:open (Type, Config),
  61   lwes_stats:initialize (lwes_net_udp:address(Config)),
  62   lwes_channel_manager:register_channel (Channel, self()),
  63   { ok, #state { socket = Socket,
  64                  channel = Channel,
  65                  type = Type
  66                }
  67   }.
  68
  69 handle_call ({ register, Function, Format, Accum },
  70              _From,
  71              State = #state {
  72                channel = #lwes_channel {type = listener }
  73              }) ->
  74   { reply,
  75     ok,
  76     State#state { callback = #callback { function = Function,
  77                                          format   = Format,
  78                                          state    = Accum } } };
  79
  80 handle_call ({ send, Packet },
  81              _From,
  82              State = #state {
  83                socket = Socket,
  84                channel = #lwes_channel { config = Config }
  85              }) ->
  86   Reply =
  87     case lwes_net_udp:send (Socket, Config, Packet) of
  88       ok ->
  89         lwes_stats:increment_sent(lwes_net_udp:address(Config)),
  90         ok;
  91       {error, Error} ->
  92         lwes_stats:increment_errors(lwes_net_udp:address(Config)),
  93         {error, Error}
  94     end,
  95   { reply, Reply, State };
  96 handle_call (_Request, _From, State) ->
  97   { reply, ok, State }.
  98
  99 handle_cast (stop, State) ->
 100   {stop, normal, State};
 101 handle_cast (_Request, State) ->
 102   { noreply, State }.
 103
 104 % skip if we don't have a handler
 105 handle_info ( {udp, _, _, _, _},
 106               State = #state {
 107                 type = listener,
 108                 channel = #lwes_channel { config = Config },
 109                 callback = undefined
 110               } ) ->
 111   lwes_stats:increment_received (lwes_net_udp:address(Config)),
 112   { noreply, State };
 113
 114 handle_info ( Packet = {udp, _, _, _, _},
 115               State = #state {
 116                 type = listener,
 117                 channel = #lwes_channel { config = Config },
 118                 callback = #callback { function = Function,
 119                                        format   = Format,
 120                                        state    = CbState }
 121               } ) ->
 122   lwes_stats:increment_received (lwes_net_udp:address(Config)),
 123   Event = lwes_event:from_udp_packet (Packet, Format),
 124   NewCbState = Function (Event, CbState),
 125   { noreply,
 126     State#state { callback = #callback { function = Function,
 127                                          format   = Format,
 128                                          state    = NewCbState }
 129                 }
 130   };
 131
 132 handle_info (_Request, State) ->
 133   {noreply, State}.
 134
 135 terminate (_Reason, #state {socket = Socket, channel = Channel}) ->
 136   lwes_net_udp:close (Socket),
 137   lwes_channel_manager:unregister_channel (Channel).
 138
 139 code_change (_OldVsn, State, _Extra) ->
 140   {ok, State}.
 141
 142 %%====================================================================
 143 %% Internal functions
 144 %%====================================================================
 145 find_and_call (Channel, Msg) ->
 146   case lwes_channel_manager:find_channel (Channel) of
 147     {error, not_open} ->
 148       {error, not_open};
 149     Pid ->
 150       gen_server:call ( Pid, Msg )
 151   end.
 152
 153 find_and_cast (Channel, Msg) ->
 154   case lwes_channel_manager:find_channel (Channel) of
 155     {error, not_open} ->
 156       {error, not_open};
 157     Pid ->
 158       gen_server:cast ( Pid, Msg )
 159   end.
 160
 161 %%====================================================================
 162 %% Test functions
 163 %%====================================================================
 164 -ifdef (TEST).
 165 -include_lib ("eunit/include/eunit.hrl").
 166
 167 -endif.