Packages
reckon_db
2.2.2
5.11.0
5.10.4
5.10.3
5.10.1
5.10.0
5.9.1
5.9.0
5.8.3
5.8.2
5.8.1
5.8.0
5.7.0
5.6.1
5.6.0
5.5.5
5.5.4
5.5.3
5.5.2
5.5.1
5.5.0
5.4.0
5.2.2
5.2.1
5.2.0
5.1.0
5.0.0
4.0.0
3.1.2
3.1.1
3.0.0
2.3.7
2.3.6
2.3.5
2.3.4
2.3.3
2.3.2
2.3.1
2.3.0
2.2.2
2.2.0
2.1.4
2.1.3
2.1.2
2.1.1
2.1.0
2.0.0
1.7.7
1.7.6
1.7.5
1.7.4
1.7.3
1.7.2
1.7.1
1.7.0
1.6.3
1.6.2
1.6.1
1.6.0
1.5.1
1.5.0
1.4.5
1.4.4
1.4.3
1.4.2
1.4.1
1.4.0
1.3.3
1.3.2
1.3.1
1.3.0
1.2.7
1.2.6
1.2.5
1.2.4
1.2.3
1.2.2
1.2.1
1.2.0
1.1.1
1.1.0
1.0.3
1.0.2
1.0.1
1.0.0
BEAM-native Event Store built on Khepri/Ra with Raft consensus. Event sourcing, persistent subscriptions, snapshots, and automatic cluster formation via UDP multicast discovery. Ships embedded Rust NIFs for 3-15x acceleration of crypto, hashing, compression, aggregation, filter matching, and grap...
Current section
Files
Jump to
Current section
Files
src/reckon_db_emitter_pool.erl
%% @doc Emitter pool supervisor for reckon-db
%%
%% Supervises a pool of emitter workers for a single subscription.
%% Each subscription can have multiple emitter workers for load distribution.
%%
%% @author rgfaber
-module(reckon_db_emitter_pool).
-behaviour(supervisor).
-include("reckon_db.hrl").
-include("reckon_db_telemetry.hrl").
%% API
-export([
start_link/2,
start_emitter/2,
stop_emitter/2,
update_emitter/2,
stop/2,
name/2
]).
%% Supervisor callbacks
-export([init/1]).
%%====================================================================
%% API
%%====================================================================
%% @doc Generate the name for an emitter pool
-spec name(atom(), binary()) -> atom().
name(StoreId, SubscriptionKey) ->
reckon_db_naming:emitter_pool_name(StoreId, SubscriptionKey).
%% @doc Start the emitter pool supervisor
-spec start_link(atom(), subscription()) -> {ok, pid()} | {error, term()}.
start_link(StoreId, #subscription{id = SubId} = Subscription) ->
Name = name(StoreId, SubId),
supervisor:start_link({local, Name}, ?MODULE, {StoreId, Subscription}).
%% @doc Start an emitter pool for a subscription
-spec start_emitter(atom(), subscription()) -> {ok, pid()} | {error, term()}.
start_emitter(StoreId, #subscription{id = SubId} = Subscription) ->
Name = name(StoreId, SubId),
case whereis(Name) of
undefined ->
%% Start via emitter supervisor
reckon_db_emitter_sup:start_emitter_pool(StoreId, Subscription);
Pid ->
{error, {already_started, Pid}}
end.
%% @doc Stop an emitter pool for a subscription
-spec stop_emitter(atom(), subscription()) -> ok | {error, term()}.
stop_emitter(StoreId, #subscription{id = SubId}) ->
Name = name(StoreId, SubId),
case whereis(Name) of
undefined ->
ok;
_Pid ->
reckon_db_emitter_sup:stop_emitter_pool(StoreId, SubId)
end.
%% @doc Update an emitter pool configuration
-spec update_emitter(atom(), subscription()) -> ok | {error, term()}.
update_emitter(StoreId, #subscription{} = Subscription) ->
case stop_emitter(StoreId, Subscription) of
ok ->
restart_emitter(StoreId, Subscription);
Error ->
Error
end.
restart_emitter(StoreId, Subscription) ->
case start_emitter(StoreId, Subscription) of
{ok, _Pid} -> ok;
Error -> Error
end.
%% @doc Stop an emitter pool by key (deprecated, use stop_emitter/2)
-spec stop(atom(), binary()) -> ok | {error, term()}.
stop(StoreId, SubscriptionKey) ->
Name = name(StoreId, SubscriptionKey),
case whereis(Name) of
undefined -> ok;
_Pid ->
reckon_db_emitter_sup:stop_emitter_pool(StoreId, SubscriptionKey)
end.
%%====================================================================
%% Supervisor callbacks
%%====================================================================
init({StoreId, #subscription{id = SubId, subscriber_pid = Subscriber,
pool_size = PoolSize} = _Subscription}) ->
%% Get or create emitter names
EmitterNames = case reckon_db_emitter_group:retrieve_emitters(StoreId, SubId) of
[] ->
%% Not yet persisted, create them
reckon_db_emitter_group:persist_emitters(StoreId, SubId, PoolSize);
Names ->
Names
end,
%% Create child specs for each emitter
Children = [
reckon_db_emitter:child_spec(StoreId, SubId, Subscriber, EmitterName)
|| EmitterName <- EmitterNames
],
%% Emit telemetry
telemetry:execute(
?EMITTER_POOL_CREATED,
#{system_time => erlang:system_time(millisecond)},
#{store_id => StoreId, subscription_id => SubId, pool_size => PoolSize}
),
logger:info("Starting emitter pool: store=~p, subscription=~s, pool_size=~p",
[StoreId, SubId, PoolSize]),
SupFlags = #{
strategy => one_for_one,
intensity => 10,
period => 60
},
{ok, {SupFlags, Children}}.