1 -module(beam_stats_consumer).
3 -include("include/beam_stats.hrl").
5 -behaviour(gen_server).
23 %% gen_server callbacks
34 queue:queue(beam_stats:t()).
36 %% ============================================================================
38 %% ============================================================================
40 -callback init(Options :: term()) ->
41 {ConsumptionInterval :: non_neg_integer(), State :: term()}.
43 -callback consume(queue(), State) ->
46 -callback terminate(State :: term()) ->
49 %% ============================================================================
51 %% ============================================================================
53 -define(FLUSH , flush).
54 -define(CONSUME_SYNC , consume_sync).
55 -define(CONSUME_ASYNC , consume_async).
58 { consumer_module :: module()
59 , consumer_state :: term()
60 , consumption_interval :: non_neg_integer()
61 , beam_stats_queue :: queue()
67 %% ============================================================================
69 %% ============================================================================
71 -spec add(module(), term()) ->
72 supervisor:startchild_ret().
73 add(ConsumerModule, ConsumerOptions) ->
74 beam_stats_sup_consumers:start_child(ConsumerModule, ConsumerOptions).
76 -spec consume_sync(pid(), beam_stats:t()) ->
78 consume_sync(PID, #beam_stats{}=BEAMStats) ->
79 {} = gen_server:call(PID, {?CONSUME_SYNC, BEAMStats}).
81 -spec consume_async(pid(), beam_stats:t()) ->
83 consume_async(PID, #beam_stats{}=BEAMStats) ->
84 ok = gen_server:cast(PID, {?CONSUME_ASYNC, BEAMStats}),
87 %% ============================================================================
89 %% ============================================================================
91 start_link(ConsumerModule, ConsumerOptions) ->
92 GenServerModule = ?MODULE,
94 InitArgs = [ConsumerModule, ConsumerOptions],
95 gen_server:start_link(GenServerModule, InitArgs, GenServerOpts).
97 %% ============================================================================
98 %% gen_server callbacks (unused)
99 %% ============================================================================
101 code_change(_OldVsn, State, _Extra) ->
104 %% ============================================================================
105 %% gen_server callbacks
106 %% ============================================================================
108 init([ConsumerModule, ConsumerOptions]) ->
109 {ConsumptionInterval, ConsumerState} = ConsumerModule:init(ConsumerOptions),
111 { consumer_module = ConsumerModule
112 , consumer_state = ConsumerState
113 , consumption_interval = ConsumptionInterval
114 , beam_stats_queue = queue:new()
116 ok = beam_stats_producer:subscribe(self()),
117 ok = schedule_first_flush(),
120 terminate(_Reason, _State) ->
121 ok = beam_stats_producer:unsubscribe(self()).
123 handle_call({?CONSUME_SYNC, #beam_stats{}=BEAMStats}, _, #state{}=State1) ->
124 State2 = consume_one(BEAMStats, State1),
127 handle_cast({?CONSUME_ASYNC, #beam_stats{}=BEAMStats}, #state{}=State1) ->
128 State2 = enqueue(BEAMStats, State1),
131 handle_info(?FLUSH, #state{consumption_interval=ConsumptionInterval}=State1) ->
132 State2 = consume_all_queued(State1),
133 ok = schedule_next_flush(ConsumptionInterval),
136 %% ============================================================================
138 %% ============================================================================
140 -spec consume_one(beam_stats:t(), state()) ->
142 consume_one(#beam_stats{}=BEAMStats, #state{}=State1) ->
143 Q = queue:in(BEAMStats, queue:new()),
146 -spec consume_all_queued(state()) ->
148 consume_all_queued(#state{beam_stats_queue=Q}=State1) ->
149 State2 = consume(Q, State1),
150 State2#state{beam_stats_queue = queue:new()}.
152 -spec consume(queue(), state()) ->
157 { consumer_module = ConsumerModule
158 , consumer_state = ConsumerState
161 State#state{consumer_state = ConsumerModule:consume(Q, ConsumerState)}.
163 -spec enqueue(beam_stats:t(), state()) ->
165 enqueue(#beam_stats{}=BEAMStats, #state{beam_stats_queue=Q}=State) ->
166 State#state{beam_stats_queue = queue:in(BEAMStats, Q)}.
168 -spec schedule_first_flush() ->
170 schedule_first_flush() ->
174 -spec schedule_next_flush(non_neg_integer()) ->
176 schedule_next_flush(Time) ->
177 _ = erlang:send_after(Time, self(), ?FLUSH),