Packages
reckon_db
2.3.6
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_tracker_group.erl
%% @doc Tracker group management for reckon-db
%%
%% Uses pg (process groups) for managing tracker processes.
%% Trackers receive notifications about subscription lifecycle events.
%%
%% This module provides:
%% - Process group management for tracker processes
%% - Notification broadcasting for created/deleted/updated events
%%
%% Features that can be tracked:
%% - subscriptions: Subscription lifecycle
%% - streams: Stream lifecycle
%% - snapshots: Snapshot lifecycle
%%
%% @author rgfaber
-module(reckon_db_tracker_group).
-include("reckon_db.hrl").
-export([
join/3,
leave/3,
members/2,
group_key/2,
notify_created/3,
notify_deleted/3,
notify_updated/3
]).
%%====================================================================
%% Types
%%====================================================================
-type store_id() :: atom().
-type feature() :: subscriptions | streams | snapshots | atom().
%%====================================================================
%% API
%%====================================================================
%% @doc Generate the group key for a feature's trackers
-spec group_key(store_id(), feature()) -> integer().
group_key(StoreId, Feature) ->
erlang:phash2({StoreId, Feature, trackers}).
%% @doc Join one or more processes to the tracker group for a feature
-spec join(store_id(), feature(), pid() | [pid()]) -> ok.
join(StoreId, Feature, PidOrPids) ->
Group = group_key(StoreId, Feature),
ok = pg:join(?RECKON_DB_PG_SCOPE, Group, PidOrPids),
ok.
%% @doc Remove one or more processes from the tracker group
-spec leave(store_id(), feature(), pid() | [pid()]) -> ok.
leave(StoreId, Feature, PidOrPids) ->
Group = group_key(StoreId, Feature),
ok = pg:leave(?RECKON_DB_PG_SCOPE, Group, PidOrPids),
ok.
%% @doc Get all member processes tracking a feature
-spec members(store_id(), feature()) -> [pid()].
members(StoreId, Feature) ->
Group = group_key(StoreId, Feature),
pg:get_members(?RECKON_DB_PG_SCOPE, Group).
%% @doc Notify all trackers that a feature instance was created
-spec notify_created(store_id(), feature(), term()) -> ok.
notify_created(StoreId, Feature, Data) ->
Msg = created(Feature, Data),
Pids = members(StoreId, Feature),
lists:foreach(fun(Pid) -> Pid ! Msg end, Pids),
ok.
%% @doc Notify all trackers that a feature instance was deleted
-spec notify_deleted(store_id(), feature(), term()) -> ok.
notify_deleted(StoreId, Feature, Data) ->
Msg = deleted(Feature, Data),
Pids = members(StoreId, Feature),
lists:foreach(fun(Pid) -> Pid ! Msg end, Pids),
ok.
%% @doc Notify all trackers that a feature instance was updated
-spec notify_updated(store_id(), feature(), term()) -> ok.
notify_updated(StoreId, Feature, Data) ->
Msg = updated(Feature, Data),
Pids = members(StoreId, Feature),
lists:foreach(fun(Pid) -> Pid ! Msg end, Pids),
ok.
%%====================================================================
%% Internal functions
%%====================================================================
%% @private Create a feature_created message
-spec created(feature(), term()) -> {feature_created, feature(), term()}.
created(Feature, Data) ->
{feature_created, Feature, Data}.
%% @private Create a feature_deleted message
-spec deleted(feature(), term()) -> {feature_deleted, feature(), term()}.
deleted(Feature, Data) ->
{feature_deleted, Feature, Data}.
%% @private Create a feature_updated message
-spec updated(feature(), term()) -> {feature_updated, feature(), term()}.
updated(Feature, Data) ->
{feature_updated, Feature, Data}.