src/beam_stats_producer.erl

   1 -module(beam_stats_producer).
   2
   3 -behaviour(gen_server).
   4
   5 %% API
   6 -export(
   7     [ start_link/1
   8     ,   subscribe/1
   9     , unsubscribe/1
  10
  11     % For testing
  12     , sync_produce_consume/0
  13     ]).
  14
  15 %% gen_server callbacks
  16 -export(
  17     [ init/1
  18     , handle_call/3
  19     , handle_cast/2
  20     , handle_info/2
  21     , terminate/2
  22     , code_change/3
  23     ]).
  24
  25 %% ============================================================================
  26 %% Internal data
  27 %% ============================================================================
  28
  29 -define(PRODUCE_SYNC  , produce_sync).
  30 -define(PRODUCE_ASYNC , produce_async).
  31
  32 -record(state,
  33     { consumers = ordsets:new() :: ordsets:ordset(pid())
  34     , deltas_server :: beam_stats_delta:t()
  35     }).
  36
  37 -type state() ::
  38     #state{}.
  39
  40 %% ============================================================================
  41 %%  API
  42 %% ============================================================================
  43
  44 start_link(DeltasServer) ->
  45     gen_server:start_link({local, ?MODULE}, ?MODULE, DeltasServer, []).
  46
  47 -spec subscribe(pid()) ->
  48     ok.
  49 subscribe(PID) ->
  50     gen_server:cast(?MODULE, {subscribe, PID}).
  51
  52 -spec unsubscribe(pid()) ->
  53     ok.
  54 unsubscribe(PID) ->
  55     gen_server:cast(?MODULE, {unsubscribe, PID}).
  56
  57 -spec sync_produce_consume() ->
  58     {}.
  59 sync_produce_consume() ->
  60     {} = gen_server:call(?MODULE, ?PRODUCE_SYNC).
  61
  62 %% ============================================================================
  63 %%  gen_server callbacks (unused)
  64 %% ============================================================================
  65
  66 code_change(_OldVsn, State, _Extra) ->
  67     {ok, State}.
  68
  69 terminate(_Reason, _State) ->
  70     ok.
  71
  72 %% ============================================================================
  73 %%  gen_server callbacks
  74 %% ============================================================================
  75
  76 init(DeltasServer) ->
  77     ok = schedule_first_production(),
  78     Consumers = ordsets:new(),
  79     {ok, #state{consumers=Consumers, deltas_server=DeltasServer}}.
  80
  81 handle_cast({subscribe, PID}, #state{consumers=Consumers1}=State) ->
  82     Consumers2 = ordsets:add_element(PID, Consumers1),
  83     {noreply, State#state{consumers=Consumers2}};
  84
  85 handle_cast({unsubscribe, PID}, #state{consumers=Consumers1}=State) ->
  86     Consumers2 = ordsets:del_element(PID, Consumers1),
  87     {noreply, State#state{consumers=Consumers2}}.
  88
  89 handle_call(?PRODUCE_SYNC, _From, State1) ->
  90     State2 = produce_sync(State1),
  91     {reply, {}, State2}.
  92
  93 handle_info(?PRODUCE_ASYNC, #state{}=State1) ->
  94     State2 = produce_async(State1),
  95     ok = schedule_next_production(),
  96     {noreply, State2}.
  97
  98 %% ============================================================================
  99 %%  Private
 100 %% ============================================================================
 101
 102 -spec produce_sync(state()) ->
 103     state().
 104 produce_sync(#state{}=State) ->
 105     produce(State, fun beam_stats_consumer:consume_sync/2).
 106
 107 -spec produce_async(state()) ->
 108     state().
 109 produce_async(#state{}=State) ->
 110     produce(State, fun beam_stats_consumer:consume_async/2).
 111
 112 -spec produce(state(), fun((pid(), term()) -> ok)) ->
 113     state().
 114 produce(
 115     #state
 116     { consumers   = ConsumersSet
 117     , deltas_server = DeltasServer
 118     }=State,
 119     MsgSendFun
 120 ) ->
 121     Stats = beam_stats:collect(DeltasServer),
 122     ConsumersList = ordsets:to_list(ConsumersSet),
 123     Send = fun (Consumer) -> MsgSendFun(Consumer, Stats) end,
 124     ok = lists:foreach(Send, ConsumersList),
 125     State.
 126
 127 -spec schedule_first_production() ->
 128     ok.
 129 schedule_first_production() ->
 130     _ = self() ! ?PRODUCE_ASYNC,
 131     ok.
 132
 133 -spec schedule_next_production() ->
 134     ok.
 135 schedule_next_production() ->
 136     ProductionInterval = beam_stats_config:production_interval(),
 137     _ = erlang:send_after(ProductionInterval, self(), ?PRODUCE_ASYNC),
 138     ok.