%% Copyright (c) Meta Platforms, Inc. and affiliates. %% %% Licensed under the Apache License, Version 2.0 (the "License"); %% you may not use this file except in compliance with the License. %% You may obtain a copy of the License at %% %% http://www.apache.org/licenses/LICENSE-2.0 %% %% Unless required by applicable law or agreed to in writing, software %% distributed under the License is distributed on an "AS IS" BASIS, %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% See the License for the specific language governing permissions and %% limitations under the License. %% % @format -module(edb_events). -compile(warn_missing_spec_all). %% erlfmt:ignore % @fb-only -compile(warn_missing_spec_all). -moduledoc false. %% Public API -export([no_subscribers/0]). -export([subscribe/3, subscribe/4]). -export([unsubscribe/2, process_down/2]). -export([subscriber_pids/1, subscriptions/1]). -export([send_to/3, broadcast/2]). -export_type([event/0]). -export_type([subscription/0]). -export_type([subscribers/0]). -export_type([monitor_ref/0]). %% ------------------------------------------------------------------- %% Types %% ------------------------------------------------------------------- -type event() :: edb:event(). -opaque subscription() :: reference(). -opaque subscribers() :: #{ subscriptions := #{subscription() => {pid(), monitor_ref()}}, monitors := #{monitor_ref() => subscription()} }. -type monitor_ref() :: reference(). %% ------------------------------------------------------------------- %% Create and update %% ------------------------------------------------------------------- -spec no_subscribers() -> subscribers(). no_subscribers() -> #{subscriptions => #{}, monitors => #{}}. -spec subscribe(Pid, MonitorRef, Subscribers) -> {ok, {Subscription, Subscribers}} when Pid :: pid(), MonitorRef :: monitor_ref(), Subscription :: subscription(), Subscribers :: subscribers(). subscribe(Pid, MonitorRef, Subscribers0) -> Subscription = erlang:make_ref(), {ok, Subscribers1} = subscribe(Subscription, Pid, MonitorRef, Subscribers0), {ok, {Subscription, Subscribers1}}. -spec subscribe(Subscription, Pid, MonitorRef, Subscribers) -> {ok, Subscribers} when Subscription :: subscription(), Pid :: pid(), MonitorRef :: monitor_ref(), Subscribers :: subscribers(). subscribe(Subscription, _Pid, _MonitorRef, #{subscriptions := Subs}) when is_map_key(Subscription, Subs) -> error({duplicate_subscription, Subscription}); subscribe(Subscription, Pid, MonitorRef, #{subscriptions := Subs0, monitors := Mon0}) when is_pid(Pid), is_reference(MonitorRef), not is_map_key(MonitorRef, Mon0) -> Subs1 = Subs0#{Subscription => {Pid, MonitorRef}}, Mon1 = Mon0#{MonitorRef => Subscription}, {ok, #{subscriptions => Subs1, monitors => Mon1}}. -spec unsubscribe(Subscription, Subscribers) -> not_subscribed | {ok, {MonitorRef, Subscribers}} when Subscription :: subscription(), Subscribers :: subscribers(), MonitorRef :: monitor_ref(). unsubscribe(Subscription, #{subscriptions := Subs0, monitors := Mon0}) when is_reference(Subscription) -> case maps:take(Subscription, Subs0) of error -> not_subscribed; {{_Pid, MonitorRef}, Subs1} -> Mon1 = maps:remove(MonitorRef, Mon0), {ok, {MonitorRef, #{subscriptions => Subs1, monitors => Mon1}}} end. -spec process_down(MonitorRef, Subscribers) -> Subscribers when MonitorRef :: monitor_ref(), Subscribers :: subscribers(). process_down(MonitorRef, Subscribers0 = #{monitors := Mon0}) -> case Mon0 of #{MonitorRef := Subscription} -> {ok, {MonitorRef, Subscribers1}} = unsubscribe(Subscription, Subscribers0), Subscribers1; _ -> Subscribers0 end. %% ------------------------------------------------------------------- %% Query %% ------------------------------------------------------------------- -spec subscriber_pids(subscribers()) -> [pid()]. subscriber_pids(#{subscriptions := Subscriptions}) -> [Pid || _Subscription := {Pid, _MonitorRef} <- Subscriptions]. -spec subscriptions(subscribers()) -> [subscription()]. subscriptions(#{subscriptions := Subscriptions}) -> maps:keys(Subscriptions). %% ------------------------------------------------------------------- %% Send events %% ------------------------------------------------------------------- -spec send_to(Subscription, Event, Subscribers) -> ok | undefined when Subscription :: subscription(), Event :: event(), Subscribers :: subscribers(). send_to(Subscription, Event, #{subscriptions := Subscriptions}) when is_reference(Subscription) -> case Subscriptions of #{Subscription := {Pid, _MonitorRef}} -> Pid ! {edb_event, Subscription, Event}, ok; _ -> undefined end. -spec broadcast(Event, Subscribers) -> ok when Event :: event(), Subscribers :: subscribers(). broadcast(Event, #{subscriptions := Subscriptions}) -> [Pid ! {edb_event, Subscription, Event} || Subscription := {Pid, _MonitorRef} <- Subscriptions], ok.