src/beam_stats_producer.erl

   1 -module(beam_stats_producer).
   2
   3 -include("beam_stats.hrl").
   4
   5 -behaviour(gen_server).
   6
   7 %% API
   8 -export(
   9     [ start_link/0
  10     ,   subscribe/1
  11     , unsubscribe/1
  12     ]).
  13
  14 %% gen_server callbacks
  15 -export(
  16     [ init/1
  17     , handle_call/3
  18     , handle_cast/2
  19     , handle_info/2
  20     , terminate/2
  21     , code_change/3
  22     ]).
  23
  24 %% ============================================================================
  25 %% Internal data
  26 %% ============================================================================
  27
  28 -define(SIGNAL_PRODUCTION , beam_stats_production_signal).
  29
  30 -record(state,
  31     { consumers = ordsets:new() :: ordsets:ordset(pid())
  32     }).
  33
  34 %% ============================================================================
  35 %%  API
  36 %% ============================================================================
  37
  38 start_link() ->
  39     gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
  40
  41 -spec subscribe(pid()) ->
  42     ok.
  43 subscribe(PID) ->
  44     gen_server:cast(?MODULE, {subscribe, PID}).
  45
  46 -spec unsubscribe(pid()) ->
  47     ok.
  48 unsubscribe(PID) ->
  49     gen_server:cast(?MODULE, {unsubscribe, PID}).
  50
  51 %% ============================================================================
  52 %%  gen_server callbacks (unused)
  53 %% ============================================================================
  54
  55 handle_call(_Request, _From, State) ->
  56     ?METHOD_SHOULD_NOT_BE_USED(handle_call, State).
  57
  58 code_change(_OldVsn, State, _Extra) ->
  59     {ok, State}.
  60
  61 terminate(_Reason, _State) ->
  62     ok.
  63
  64 %% ============================================================================
  65 %%  gen_server callbacks
  66 %% ============================================================================
  67
  68 init([]) ->
  69     ok = schedule_first_production(),
  70     Consumers = ordsets:new(),
  71     {ok, #state{consumers=Consumers}}.
  72
  73 handle_cast({subscribe, PID}, #state{consumers=Consumers1}=State) ->
  74     Consumers2 = ordsets:add_element(PID, Consumers1),
  75     {noreply, State#state{consumers=Consumers2}};
  76
  77 handle_cast({unsubscribe, PID}, #state{consumers=Consumers1}=State) ->
  78     Consumers2 = ordsets:del_element(PID, Consumers1),
  79     {noreply, State#state{consumers=Consumers2}}.
  80
  81 handle_info(?SIGNAL_PRODUCTION, #state{consumers=ConsumersSet}=State) ->
  82     ConsumersList = ordsets:to_list(ConsumersSet),
  83     ok = collect_and_push_to_consumers(ConsumersList),
  84     ok = schedule_next_production(),
  85     {noreply, State}.
  86
  87 %% ============================================================================
  88 %%  Private
  89 %% ============================================================================
  90
  91 -spec schedule_first_production() ->
  92     ok.
  93 schedule_first_production() ->
  94     _ = self() ! ?SIGNAL_PRODUCTION,
  95     ok.
  96
  97 -spec schedule_next_production() ->
  98     ok.
  99 schedule_next_production() ->
 100     ProductionInterval = beam_stats_config:production_interval(),
 101     _ = erlang:send_after(ProductionInterval, self(), ?SIGNAL_PRODUCTION),
 102     ok.
 103
 104 -spec collect_and_push_to_consumers([pid()]) ->
 105     ok.
 106 collect_and_push_to_consumers(Consumers) ->
 107     BEAMStats = beam_stats:collect(),
 108     Push = fun (Consumer) -> gen_server:cast(Consumer, BEAMStats) end,
 109     lists:foreach(Push, Consumers).