Packages
ex_esdb
0.0.19
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.
Current section
Files
Jump to
Current section
Files
lib/ex_esdb/emitters.ex
defmodule ExESDB.Emitters do
@moduledoc """
As part of the ExESDB.System, ExESDB.Emitters is responsible for managing the
lifetime of the Emitter processes.
"""
alias ExESDB.Topics, as: Topics
defp build_filter(:by_stream, selector), do: :streams_filters.by_stream(selector)
defp build_filter(:by_event_type, selector), do: :streams_filters.by_event_type(selector)
defp build_filter(:by_event_pattern, selector), do: :streams_filters.by_event_pattern(selector)
defp build_filter(:by_event_payload, selector), do: :streams_filters.by_event_payload(selector)
defp partition(term) do
partitions = System.schedulers_online()
:erlang.phash2(term, (:rand.uniform(partitions) + 1) * 10)
end
def start_emitter(
store,
%{
type: type,
subscription_name: subscription_name,
selector: selector,
subscriber: subscriber
},
pool_size \\ 3
) do
filter =
type
|> build_filter(selector)
sub_topic = Topics.sub_topic(type, subscription_name, selector)
args = {store, sub_topic, subscriber, pool_size, filter}
DynamicSupervisor.start_child(
{:via, PartitionSupervisor, {ExESDB.EmitterPools, partition(args)}},
{ExESDB.EmitterPool, args}
)
end
end