Packages

ex_esdb

0.0.8-alpha
0.11.0 0.10.0 0.9.0 0.8.0 0.7.8 0.7.7 0.7.6 0.7.5 0.7.4 0.7.3 0.7.2 0.7.1 0.7.0 0.6.1 0.6.0 0.5.1 0.5.0 0.4.8 0.4.7 0.4.6 0.4.5 0.4.4 0.4.3 0.4.2 0.4.1 0.4.0 0.3.3 0.3.2 0.3.1 0.3.0 0.2.5 0.2.4 0.2.3 0.2.2 0.2.1 0.2.0 0.1.7 0.1.6 0.1.5 0.1.4 0.1.3 0.1.2 0.1.1 0.1.0 0.0.20 0.0.19 0.0.18 0.0.17 0.0.16 0.0.15 0.0.14-alpha 0.0.13-alpha 0.0.12-alpha 0.0.11-alpha 0.0.10-alpha 0.0.9-alpha 0.0.8-alpha 0.0.6-alpha 0.0.5-alpha 0.0.4-alpha 0.0.3-alpha 0.0.2-alfa 0.0.1-alfa

ExESDB is a reincarnation of rabbitmq/khepri, specialized for use as a BEAM-native event store.

HexDocs

Current section

Files

Jump to
Readme
63 Versions
8 Dependencies
Dependants
Files
Activity
ex_esdb src func_registrations.erl
Raw

src/func_registrations.erl 

-module(func_registrations).



-export([register_emitter/2]).



-spec reg_on_create(PubSub :: atom(), Store :: khepri:store()) -> ok | {error, term()}.

reg_on_create(PubSub, Store) ->

  khepri:put(Store,

             [procs, on_new_event],

             fun(Props) -> broadcast(PubSub, Store, Props) end).



-spec register_emitter(PubSub :: atom(), Store :: khepri:store()) -> ok | {error, term()}.

register_emitter(PubSub, Store) ->

  case reg_on_create(PubSub, Store) of

    ok ->

      NewEventFilter = khepri_evf:tree([Store, streams], #{on_actions => [create]}),

      khepri:register_trigger(Store, on_new_event, NewEventFilter, [procs, on_new_event]);

    {error, Reason} ->

      io:format("Khepri registration failed: ~p~n", [Reason]),

      {error, Reason}

  end.



-spec broadcast(PubSub :: atom(), Store :: khepri:store(), Props :: khepri:props()) -> ok.

broadcast(PubSubName, Topic, Message) ->

  Members = pg:get_members(PubSubName, Topic),

  lists:foreach(fun(Member) -> Member ! {message, Topic, Message} end, Members),

  ok.