src/beam_stats_consumer.erl

   1 -module(beam_stats_consumer).
   2
   3 -include("include/beam_stats.hrl").
   4
   5 -behaviour(gen_server).
   6
   7 -export_type(
   8     [ queue/0
   9     ]).
  10
  11 %% Public API
  12 -export(
  13     [ add/2
  14     , consume_sync/2
  15     , consume_async/2
  16     ]).
  17
  18 %% Internal API
  19 -export(
  20     [ start_link/2
  21     ]).
  22
  23 %% gen_server callbacks
  24 -export(
  25     [ init/1
  26     , handle_call/3
  27     , handle_cast/2
  28     , handle_info/2
  29     , terminate/2
  30     , code_change/3
  31     ]).
  32
  33 -type queue() ::
  34     queue:queue(beam_stats:t()).
  35
  36 %% ============================================================================
  37 %%  Consumer interface
  38 %% ============================================================================
  39
  40 -callback init(Options :: term()) ->
  41     {ConsumptionInterval :: non_neg_integer(), State :: term()}.
  42
  43 -callback consume(queue(), State) ->
  44     State.
  45
  46 -callback terminate(State :: term()) ->
  47     {}.
  48
  49 %% ============================================================================
  50 %% Internal data
  51 %% ============================================================================
  52
  53 -define(FLUSH         , flush).
  54 -define(CONSUME_SYNC  , consume_sync).
  55 -define(CONSUME_ASYNC , consume_async).
  56
  57 -record(state,
  58     { consumer_module      :: module()
  59     , consumer_state       :: term()
  60     , consumption_interval :: non_neg_integer()
  61     , beam_stats_queue     :: queue()
  62     }).
  63
  64 -type state() ::
  65     #state{}.
  66
  67 %% ============================================================================
  68 %%  Public API
  69 %% ============================================================================
  70
  71 -spec add(module(), term()) ->
  72     supervisor:startchild_ret().
  73 add(ConsumerModule, ConsumerOptions) ->
  74     beam_stats_sup_consumers:start_child(ConsumerModule, ConsumerOptions).
  75
  76 -spec consume_sync(pid(), beam_stats:t()) ->
  77     {}.
  78 consume_sync(PID, #beam_stats{}=BEAMStats) ->
  79     {} = gen_server:call(PID, {?CONSUME_SYNC, BEAMStats}).
  80
  81 -spec consume_async(pid(), beam_stats:t()) ->
  82     {}.
  83 consume_async(PID, #beam_stats{}=BEAMStats) ->
  84     ok = gen_server:cast(PID, {?CONSUME_ASYNC, BEAMStats}),
  85     {}.
  86
  87 %% ============================================================================
  88 %%  Internal API
  89 %% ============================================================================
  90
  91 start_link(ConsumerModule, ConsumerOptions) ->
  92     RegisteredName  = ConsumerModule,
  93     GenServerModule = ?MODULE,
  94     GenServerOpts   = [],
  95     InitArgs        = [ConsumerModule, ConsumerOptions],
  96     gen_server:start_link({local, RegisteredName}, GenServerModule, InitArgs, GenServerOpts).
  97
  98 %% ============================================================================
  99 %%  gen_server callbacks (unused)
 100 %% ============================================================================
 101
 102 code_change(_OldVsn, State, _Extra) ->
 103     {ok, State}.
 104
 105 %% ============================================================================
 106 %%  gen_server callbacks
 107 %% ============================================================================
 108
 109 init([ConsumerModule, ConsumerOptions]) ->
 110     {ConsumptionInterval, ConsumerState} = ConsumerModule:init(ConsumerOptions),
 111     State = #state
 112         { consumer_module      = ConsumerModule
 113         , consumer_state       = ConsumerState
 114         , consumption_interval = ConsumptionInterval
 115         , beam_stats_queue     = queue:new()
 116         },
 117     ok = beam_stats_producer:subscribe(self()),
 118     ok = schedule_first_flush(),
 119     {ok, State}.
 120
 121 terminate(_Reason, _State) ->
 122     ok = beam_stats_producer:unsubscribe(self()).
 123
 124 handle_call({?CONSUME_SYNC, #beam_stats{}=BEAMStats}, _, #state{}=State1) ->
 125     State2 = consume_one(BEAMStats, State1),
 126     {reply, {}, State2}.
 127
 128 handle_cast({?CONSUME_ASYNC, #beam_stats{}=BEAMStats}, #state{}=State1) ->
 129     State2 = enqueue(BEAMStats, State1),
 130     {noreply, State2}.
 131
 132 handle_info(?FLUSH, #state{consumption_interval=ConsumptionInterval}=State1) ->
 133     State2 = consume_all_queued(State1),
 134     ok = schedule_next_flush(ConsumptionInterval),
 135     {noreply, State2}.
 136
 137 %% ============================================================================
 138 %%  Internal
 139 %% ============================================================================
 140
 141 -spec consume_one(beam_stats:t(), state()) ->
 142     state().
 143 consume_one(#beam_stats{}=BEAMStats, #state{}=State1) ->
 144     Q = queue:in(BEAMStats, queue:new()),
 145     consume(Q, State1).
 146
 147 -spec consume_all_queued(state()) ->
 148     state().
 149 consume_all_queued(#state{beam_stats_queue=Q}=State1) ->
 150     State2 = consume(Q, State1),
 151     State2#state{beam_stats_queue = queue:new()}.
 152
 153 -spec consume(queue(), state()) ->
 154     state().
 155 consume(
 156     Q,
 157     #state
 158     { consumer_module  = ConsumerModule
 159     , consumer_state   = ConsumerState
 160     }=State
 161 ) ->
 162     State#state{consumer_state = ConsumerModule:consume(Q, ConsumerState)}.
 163
 164 -spec enqueue(beam_stats:t(), state()) ->
 165     state().
 166 enqueue(#beam_stats{}=BEAMStats, #state{beam_stats_queue=Q}=State) ->
 167     State#state{beam_stats_queue = queue:in(BEAMStats, Q)}.
 168
 169 -spec schedule_first_flush() ->
 170     ok.
 171 schedule_first_flush() ->
 172     _ = self() ! ?FLUSH,
 173     ok.
 174
 175 -spec schedule_next_flush(non_neg_integer()) ->
 176     ok.
 177 schedule_next_flush(Time) ->
 178     _ = erlang:send_after(Time, self(), ?FLUSH),
 179     ok.