%%%----------------------------------------------------------------------------- %% @doc Internal services provider for blockade. %% @private %% @end %%%----------------------------------------------------------------------------- -module(blockade_service). -include("blockade_header.hrl"). -export([rand_node/0, member_pids/3, send_messages/3, dispatch_event/4, queue_event/4, dispatch_queued/4, queue_prune/2, get_discard_opt/2, get_reset_opt/2, emit_priority/2, sync_priority/2, startup_prio_confr/1, cancel_ref/1]). %%------------------------------------------------------------------------------ %% Public functions. %%------------------------------------------------------------------------------ cancel_ref(Ref) when is_reference(Ref) -> erlang:cancel_timer(Ref), ok; cancel_ref(_) -> {error, no_ref}. sync_priority([], Default) -> Default; sync_priority(Priorities, _Default) -> [SelectedPrio | _] = most(Priorities), SelectedPrio. emit_priority(Man, Priority) -> gen_server:abcast(nodes(), Man, {priority_emit, Priority}), ok. startup_prio_confr(Opts) -> maps:get(priority, Opts, false) =/= false orelse length(nodes()) < 0. dispatch_queued([], _, _, Eq) -> lists:reverse(Eq); dispatch_queued([{Event, Payload, #{priority := Ep} = Opts} | Events], Man, Prio, Eq) when Ep >= Prio -> dispatch_event(Event, Payload, Man, Opts), dispatch_queued(Events, Man, Prio, Eq); dispatch_queued([Event | Events], Man, Prio, Eq) -> dispatch_queued(Events, Man, Prio, [Event | Eq]). queue_event(EventQueue, {_, _, #{priority := Ep}}, ManPrio, true) when Ep < ManPrio -> {event_discarded, EventQueue}; queue_event(EventQueue, {_, _, #{priority := Ep, discard_event := true}}, ManPrio, false) when Ep < ManPrio -> {event_discarded, EventQueue}; queue_event(EventQueue, {Event, Payload, Opts}, _ManPrio, _DiscardEvents) -> Neq = [{Event, Payload, Opts} | EventQueue], {event_queued, Neq}. dispatch_event(Event, Payload, Man, Opts) -> Mt = maps:get(members, Opts, global), Scope = ?PROCESS_NAME(Man, "pg"), Pids = blockade_service:member_pids(Scope, Event, Mt), blockade_service:send_messages(Pids, Event, Payload). send_messages([], _Event, _Payload) -> ok; send_messages([Pid | Pids], Event, Payload) -> Pid ! {Event, Payload}, send_messages(Pids, Event, Payload). rand_node() -> Nodes = erlang:nodes(), lists:nth( rand:uniform(length(Nodes)), Nodes). member_pids(Scope, Event, MemberType) when MemberType == local -> pg:get_local_members(Scope, Event); member_pids(Scope, Event, MemberType) when MemberType == global -> pg:get_members(Scope, Event); member_pids(_Scope, _Event, _MemberType) -> throw({error, invalid_members_option}). queue_prune(EventQueue, Priority) -> [Event || {_, _, #{priority := EventPrio}} = Event <- EventQueue, EventPrio >= Priority]. get_discard_opt(Opts, DefaultDiscard) -> case maps:get(discard_events, Opts, undefined) of undefined -> DefaultDiscard; DiscardEvents -> DiscardEvents end. get_reset_opt(Opts, DefaultRef) -> case maps:get(reset_after, Opts, undefined) of undefined -> DefaultRef; _ -> schedule_reset(Opts) end. %%------------------------------------------------------------------------------ %% Internal functions. %%------------------------------------------------------------------------------ schedule_reset(Opts) -> ResetAfter = maps:get(reset_after, Opts, undefined), case ResetAfter of undefined -> undefined; _ -> erlang:send_after(ResetAfter, self(), reset_priority) end. most(List) -> Lc = lists:foldl(fun(E1, E2) -> maps:put(E1, maps:get(E1, E2, 0) + 1, E2) end, #{}, List), Ls = lists:sort(fun({_, Ac}, {_, Bc}) -> Ac > Bc end, maps:to_list(Lc)), [Value || {Value, _Count} <- Ls].