Current section
Files
Jump to
Current section
Files
src/signal.erl
-module(signal).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch]).
-export([configure/1, with_subscriber/2, with_persistance_layer/2, with_pool_size_limit/2, aggregate/2, create/2, handle_command/2, get_state/1, get_id/1, get_current_pool_size/1, start/1]).
-export_type([emit_messages/3, event/1, persistance_interface/1, subscriber/2, consumer_message/2, emit_config/4, aggregate_config/3, emit_service/3, aggregate_state/3, aggregate_message/3, aggregate_update_context/3, pool_message/3, bus_message/1, store_messages/1]).
-opaque emit_messages(IGF, IGG, IGH) :: {get_pool,
gleam@erlang@process:subject(gleam@erlang@process:subject(pool_message(IGF, IGG, IGH)))} |
shutdown.
-type event(IGI) :: {event, integer(), binary(), binary(), IGI}.
-type persistance_interface(IGJ) :: {get_stored_events,
gleam@erlang@process:subject({ok, list(event(IGJ))} | {error, binary()}),
binary()} |
{is_identity_available,
gleam@erlang@process:subject({ok, boolean()} | {error, binary()}),
binary()} |
{store_events, list(event(IGJ))} |
shutdown_persistance_layer.
-type subscriber(IGK, IGL) :: {consumer,
gleam@erlang@process:subject(consumer_message(IGK, IGL))} |
{policy, gleam@otp@task:task(event(IGL))}.
-type consumer_message(IGM, IGN) :: {consume, event(IGN)} |
{get_consumer_state, gleam@erlang@process:subject(IGM)} |
shutdown_consumer.
-opaque emit_config(IGO, IGP, IGQ, IGR) :: {emit_config,
aggregate_config(IGO, IGQ, IGR),
gleam@option:option(gleam@erlang@process:subject(persistance_interface(IGR))),
list(subscriber(IGP, IGR)),
integer()}.
-type aggregate_config(IGS, IGT, IGU) :: {aggregate_config,
IGS,
fun((IGT, IGS) -> {ok, list(IGU)} | {error, binary()}),
fun((IGS, event(IGU)) -> IGS)}.
-type emit_service(IGV, IGW, IGX) :: {emit_service,
gleam@erlang@process:subject(pool_message(IGV, IGW, IGX)),
gleam@erlang@process:subject(bus_message(IGX)),
gleam@erlang@process:subject(store_messages(IGX))}.
-type aggregate_state(IGY, IGZ, IHA) :: {aggregate_state, integer(), IGY} |
{gleam_phantom, IGZ, IHA}.
-type aggregate_message(IHB, IHC, IHD) :: {state,
gleam@erlang@process:subject(IHB)} |
{identity, gleam@erlang@process:subject(binary())} |
{handle_command,
gleam@erlang@process:subject({ok, IHB} | {error, binary()}),
IHC} |
shutdown_aggregate |
{gleam_phantom, IHD}.
-type aggregate_update_context(IHE, IHF, IHG) :: {aggregate_update_context,
binary(),
gleam@erlang@process:subject(bus_message(IHG)),
fun((IHE, event(IHG)) -> IHE),
aggregate_state(IHE, IHF, IHG)}.
-type pool_message(IHH, IHI, IHJ) :: {create_aggregate,
gleam@erlang@process:subject({ok,
gleam@erlang@process:subject(aggregate_message(IHH, IHI, IHJ))} |
{error, binary()}),
binary()} |
{get_aggregate,
gleam@erlang@process:subject({ok,
gleam@erlang@process:subject(aggregate_message(IHH, IHI, IHJ))} |
{error, binary()}),
binary()} |
{pool_size, gleam@erlang@process:subject(integer())} |
shutdown_pool.
-type bus_message(IHK) :: {push_event, event(IHK)} | shutdown_bus.
-type store_messages(IHL) :: {store_event, event(IHL)} |
{get_events,
gleam@erlang@process:subject({ok, list(event(IHL))} | {error, binary()}),
binary()} |
{id_exists,
gleam@erlang@process:subject({ok, boolean()} | {error, binary()}),
binary()} |
{persistance_state, list(event(IHL)), boolean()} |
shutdown_store.
-spec configure(aggregate_config(IIW, IIX, IIY)) -> emit_config(IIW, any(), IIX, IIY).
configure(Agg) ->
{emit_config, Agg, none, [], 100}.
-spec with_subscriber(emit_config(IJH, IJI, IJJ, IJK), subscriber(IJI, IJK)) -> emit_config(IJH, IJI, IJJ, IJK).
with_subscriber(Config, Sub) ->
erlang:setelement(4, Config, [Sub | erlang:element(4, Config)]).
-spec with_persistance_layer(
emit_config(IJV, IJW, IJX, IJY),
gleam@erlang@process:subject(persistance_interface(IJY))
) -> emit_config(IJV, IJW, IJX, IJY).
with_persistance_layer(Config, Persist) ->
erlang:setelement(3, Config, {some, Persist}).
-spec with_pool_size_limit(emit_config(IKJ, IKK, IKL, IKM), integer()) -> emit_config(IKJ, IKK, IKL, IKM).
with_pool_size_limit(Config, Aggregates_in_memory) ->
erlang:setelement(5, Config, Aggregates_in_memory).
-spec aggregate(
gleam@erlang@process:subject(emit_messages(ILB, ILC, ILD)),
binary()
) -> {ok, gleam@erlang@process:subject(aggregate_message(ILB, ILC, ILD))} |
{error, binary()}.
aggregate(Signal, Id) ->
Pool = gleam@erlang@process:call(
Signal,
fun(Field@0) -> {get_pool, Field@0} end,
5
),
gleam@erlang@process:call(
Pool,
fun(_capture) -> {get_aggregate, _capture, Id} end,
5
).
-spec create(
gleam@erlang@process:subject(emit_messages(ILM, ILN, ILO)),
binary()
) -> {ok, gleam@erlang@process:subject(aggregate_message(ILM, ILN, ILO))} |
{error, binary()}.
create(Signal, Id) ->
Pool = gleam@erlang@process:call(
Signal,
fun(Field@0) -> {get_pool, Field@0} end,
5
),
gleam@erlang@process:call(
Pool,
fun(_capture) -> {create_aggregate, _capture, Id} end,
5
).
-spec handle_command(
gleam@erlang@process:subject(aggregate_message(ILX, ILY, any())),
ILY
) -> {ok, ILX} | {error, binary()}.
handle_command(Agg, Command) ->
gleam@erlang@process:call(
Agg,
fun(_capture) -> {handle_command, _capture, Command} end,
5
).
-spec get_state(
gleam@erlang@process:subject(aggregate_message(IMF, any(), any()))
) -> IMF.
get_state(Agg) ->
gleam@erlang@process:call(Agg, fun(_capture) -> {state, _capture} end, 5).
-spec get_id(
gleam@erlang@process:subject(aggregate_message(any(), any(), any()))
) -> binary().
get_id(Agg) ->
gleam@erlang@process:call(Agg, fun(_capture) -> {identity, _capture} end, 5).
-spec get_current_pool_size(
gleam@erlang@process:subject(emit_messages(any(), any(), any()))
) -> integer().
get_current_pool_size(Signal) ->
Pool = gleam@erlang@process:call(
Signal,
fun(_capture) -> {get_pool, _capture} end,
5
),
gleam@erlang@process:call(
Pool,
fun(_capture@1) -> {pool_size, _capture@1} end,
5
).
-spec emit_handler(emit_service(ING, INH, INI)) -> fun((emit_messages(ING, INH, INI), nil) -> gleam@otp@actor:next(any(), nil)).
emit_handler(Cfg) ->
fun(Message, _) -> case Message of
{get_pool, S} ->
gleam@erlang@process:send(S, erlang:element(2, Cfg)),
gleam@otp@actor:continue(nil);
shutdown ->
{stop, normal}
end end.
-spec hydrate_event(ION, aggregate_update_context(any(), any(), ION)) -> event(ION).
hydrate_event(Event, Ctx) ->
{event,
erlang:element(2, erlang:element(5, Ctx)) + 1,
erlang:element(2, Ctx),
gleam@string:inspect(Event),
Event}.
-spec apply_event(
event(IOU),
fun((IOW, event(IOU)) -> IOW),
aggregate_state(IOW, IOZ, IOU)
) -> aggregate_state(IOW, IOZ, IOU).
apply_event(Event, Handler, Agg) ->
{aggregate_state,
erlang:element(2, Agg) + 1,
Handler(erlang:element(3, Agg), Event)}.
-spec aggregate_init(list(event(INN)), aggregate_config(INQ, any(), INN)) -> fun(() -> gleam@otp@actor:init_result(aggregate_state(INQ, any(), INN), any())).
aggregate_init(Events, Cfg) ->
fun() ->
Aggregate = {aggregate_state, 0, erlang:element(2, Cfg)},
{ready,
begin
_pipe = Events,
_pipe@1 = gleam@list:unique(_pipe),
_pipe@2 = gleam@list:sort(
_pipe@1,
fun(E1, E2) ->
gleam@int:compare(
erlang:element(2, E1),
erlang:element(2, E2)
)
end
),
_pipe@3 = gleam@list:fold(
_pipe@2,
[],
fun(Dedup, Event) ->
case gleam@list:any(
Dedup,
fun(D) ->
erlang:element(2, D) =:= erlang:element(
2,
Event
)
end
) of
true ->
Dedup;
false ->
lists:append(Dedup, [Event])
end
end
),
gleam@list:fold(
_pipe@3,
Aggregate,
fun(Agg, E) ->
apply_event(E, erlang:element(4, Cfg), Agg)
end
)
end,
gleam_erlang_ffi:new_selector()}
end.
-spec send_event_to_bus(event(IPG), aggregate_update_context(any(), any(), IPG)) -> event(IPG).
send_event_to_bus(Event, Ctx) ->
gleam@erlang@process:send(erlang:element(3, Ctx), {push_event, Event}),
Event.
-spec update_aggregate(aggregate_update_context(IOG, IOH, IOI), IOI) -> aggregate_state(IOG, IOH, IOI).
update_aggregate(Ctx, Event) ->
_pipe = Event,
_pipe@1 = hydrate_event(_pipe, Ctx),
_pipe@2 = send_event_to_bus(_pipe@1, Ctx),
apply_event(_pipe@2, erlang:element(4, Ctx), erlang:element(5, Ctx)).
-spec aggregate_handler(
binary(),
fun((INX, INW) -> {ok, list(INY)} | {error, binary()}),
fun((INW, event(INY)) -> INW),
gleam@erlang@process:subject(bus_message(INY))
) -> fun((aggregate_message(INW, INX, INY), aggregate_state(INW, INX, INY)) -> gleam@otp@actor:next(any(), aggregate_state(INW, INX, INY))).
aggregate_handler(Id, Command_handler, Event_handler, Bus) ->
fun(Operation, Agg) -> case Operation of
{state, Client} ->
gleam@erlang@process:send(Client, erlang:element(3, Agg)),
gleam@otp@actor:continue(Agg);
{identity, Client@1} ->
gleam@erlang@process:send(Client@1, Id),
gleam@otp@actor:continue(Agg);
{handle_command, Client@2, Command} ->
case Command_handler(Command, erlang:element(3, Agg)) of
{ok, Events} ->
New_state = gleam@list:fold(
Events,
Agg,
fun(Agg@1, E) ->
update_aggregate(
{aggregate_update_context,
Id,
Bus,
Event_handler,
Agg@1},
E
)
end
),
gleam@erlang@process:send(
Client@2,
{ok, erlang:element(3, New_state)}
),
gleam@otp@actor:continue(New_state);
{error, Msg} ->
gleam@erlang@process:send(Client@2, {error, Msg}),
gleam@otp@actor:continue(Agg)
end;
shutdown_aggregate ->
{stop, normal}
end end.
-spec evict_aggregates_workflow(
integer(),
gleam@dict:dict(binary(), {gleam@erlang@process:subject(aggregate_message(IPX, IPY, IPZ)),
integer()})
) -> gleam@dict:dict(binary(), {gleam@erlang@process:subject(aggregate_message(IPX, IPY, IPZ)),
integer()}).
evict_aggregates_workflow(Max_size, State) ->
case maps:size(State) >= Max_size of
true ->
Eviction_list = begin
_pipe = maps:to_list(State),
_pipe@1 = gleam@list:filter(
_pipe,
fun(Agg) ->
{_, {_, Position}} = Agg,
Position >= Max_size
end
),
gleam@list:map(
_pipe@1,
fun(Agg@1) ->
{Id, {Agg@2, _}} = Agg@1,
gleam@erlang@process:send(Agg@2, shutdown_aggregate),
Id
end
)
end,
gleam@dict:drop(State, Eviction_list);
false ->
State
end.
-spec store_has_aggregate(
gleam@erlang@process:subject(store_messages(any())),
binary()
) -> boolean().
store_has_aggregate(Store, Key) ->
Response = gleam@erlang@process:call(
Store,
fun(_capture) -> {id_exists, _capture, Key} end,
5
),
case Response of
{error, _} ->
true;
{ok, false} ->
false;
{ok, true} ->
true
end.
-spec gather_aggregate_from_store(
fun((list(event(IRA))) -> {ok,
gleam@erlang@process:subject(aggregate_message(IRD, IRE, IRA))} |
{error, binary()}),
gleam@erlang@process:subject(store_messages(IRA)),
binary()
) -> {ok, gleam@erlang@process:subject(aggregate_message(IRD, IRE, IRA))} |
{error, binary()}.
gather_aggregate_from_store(Aggregate_initializer, Store, Id) ->
case gleam@erlang@process:call(
Store,
fun(_capture) -> {get_events, _capture, Id} end,
5
) of
{error, Msg} ->
{error, Msg};
{ok, Events} ->
Aggregate_initializer(Events)
end.
-spec gather_aggregate(
fun((list(event(IQJ))) -> {ok,
gleam@erlang@process:subject(aggregate_message(IQM, IQN, IQJ))} |
{error, binary()}),
gleam@erlang@process:subject(store_messages(IQJ)),
gleam@dict:dict(binary(), {gleam@erlang@process:subject(aggregate_message(IQM, IQN, IQJ)),
integer()}),
binary()
) -> {ok, gleam@erlang@process:subject(aggregate_message(IQM, IQN, IQJ))} |
{error, binary()}.
gather_aggregate(Aggregate_initializer, Store, Dict, Id) ->
case gleam@dict:get(Dict, Id) of
{ok, {Value, _}} ->
{ok, Value};
{error, _} ->
gather_aggregate_from_store(Aggregate_initializer, Store, Id)
end.
-spec start_aggregate(
aggregate_config(IRM, IRN, IRO),
gleam@erlang@process:subject(bus_message(IRO)),
binary(),
list(event(IRO))
) -> {ok, gleam@erlang@process:subject(aggregate_message(IRM, IRN, IRO))} |
{error, binary()}.
start_aggregate(Config, Bus, Id, Events) ->
case gleam@otp@actor:start_spec(
{spec,
aggregate_init(Events, Config),
5,
aggregate_handler(
Id,
erlang:element(3, Config),
erlang:element(4, Config),
Bus
)}
) of
{ok, Actor} ->
{ok, Actor};
_ ->
{error, <<"Could not initialize the actor: "/utf8, Id/binary>>}
end.
-spec pool_handler(
aggregate_config(IPO, IPP, IPQ),
gleam@erlang@process:subject(bus_message(IPQ)),
gleam@erlang@process:subject(store_messages(IPQ)),
integer()
) -> fun((pool_message(IPO, IPP, IPQ), gleam@dict:dict(binary(), {gleam@erlang@process:subject(aggregate_message(IPO, IPP, IPQ)),
integer()})) -> gleam@otp@actor:next(any(), gleam@dict:dict(binary(), {gleam@erlang@process:subject(aggregate_message(IPO, IPP, IPQ)),
integer()}))).
pool_handler(Config, Bus, Store, Max_size) ->
fun(Operation, State) -> case Operation of
{create_aggregate, Client, With_id} ->
Exists_in_store = store_has_aggregate(Store, With_id),
Exists_in_pool = gleam@dict:has_key(State, With_id),
case {Exists_in_store, Exists_in_pool} of
{false, false} ->
case start_aggregate(Config, Bus, With_id, []) of
{ok, Agg} ->
gleam@erlang@process:send(Client, {ok, Agg}),
gleam@otp@actor:continue(
gleam@dict:insert(
evict_aggregates_workflow(
Max_size,
State
),
With_id,
{Agg, maps:size(State) + 1}
)
);
Error ->
gleam@erlang@process:send(Client, Error),
gleam@otp@actor:continue(State)
end;
{_, _} ->
gleam@erlang@process:send(
Client,
{error,
<<<<"Aggregate ID must be unique, "/utf8,
With_id/binary>>/binary,
" is not!"/utf8>>}
),
gleam@otp@actor:continue(State)
end;
{get_aggregate, Client@1, Id} ->
case gather_aggregate(
fun(_capture) ->
start_aggregate(Config, Bus, Id, _capture)
end,
Store,
State,
Id
) of
{ok, Aggregate} ->
gleam@dict:insert(
State,
Id,
{Aggregate, maps:size(State) + 1}
),
gleam@erlang@process:send(Client@1, {ok, Aggregate});
{error, Msg} ->
gleam@erlang@process:send(Client@1, {error, Msg})
end,
gleam@otp@actor:continue(State);
{pool_size, S} ->
gleam@erlang@process:send(
S,
begin
_pipe = maps:to_list(State),
erlang:length(_pipe)
end
),
gleam@otp@actor:continue(State);
shutdown_pool ->
{stop, normal}
end end.
-spec notify_subscribers(event(ISD), list(subscriber(any(), ISD))) -> nil.
notify_subscribers(Event, Consumers) ->
case Consumers of
[] ->
nil;
[{consumer, S}] ->
gleam@erlang@process:send(S, {consume, Event});
[{policy, T}] ->
_ = gleam@otp@task:try_await(T, 5),
nil;
[{consumer, S@1} | Rest] ->
gleam@erlang@process:send(S@1, {consume, Event}),
notify_subscribers(Event, Rest);
[{policy, T@1} | Rest@1] ->
_ = gleam@otp@task:try_await(T@1, 5),
notify_subscribers(Event, Rest@1)
end.
-spec notify_store(
event(ISK),
gleam@erlang@process:subject(store_messages(ISK))
) -> nil.
notify_store(Event, Store) ->
gleam@erlang@process:send(Store, {store_event, Event}).
-spec bus_handler(
list(subscriber(any(), IRX)),
gleam@erlang@process:subject(store_messages(IRX))
) -> fun((bus_message(IRX), nil) -> gleam@otp@actor:next(any(), nil)).
bus_handler(Subscribers, Store) ->
fun(Message, _) -> case Message of
{push_event, Event} ->
notify_subscribers(Event, Subscribers),
notify_store(Event, Store),
gleam@otp@actor:continue(nil);
shutdown_bus ->
{stop, normal}
end end.
-spec store_handler(gleam@erlang@process:subject(persistance_interface(IST))) -> fun((store_messages(IST), {list(event(IST)),
list(event(IST))}) -> gleam@otp@actor:next(any(), {list(event(IST)),
list(event(IST))})).
store_handler(Persistor) ->
fun(Message, State) -> case Message of
{store_event, E} ->
{Wal, Processing} = State,
case gleam@list:is_empty(Processing) of
false ->
gleam@otp@actor:continue({[E | Wal], Processing});
true ->
gleam@erlang@process:send(
Persistor,
{store_events, [E | Wal]}
),
gleam@otp@actor:continue(
{[], lists:append(Wal, Processing)}
)
end;
{get_events, S, Id} ->
Events = gleam@erlang@process:call(
Persistor,
fun(_capture) -> {get_stored_events, _capture, Id} end,
5
),
gleam@erlang@process:send(S, Events),
gleam@otp@actor:continue(State);
{id_exists, S@1, Id@1} ->
Result = gleam@erlang@process:call(
Persistor,
fun(_capture@1) ->
{is_identity_available, _capture@1, Id@1}
end,
5
),
gleam@erlang@process:send(S@1, Result),
gleam@otp@actor:continue(State);
{persistance_state, Processed, _} ->
{Wal@1, Events_being_processed} = State,
{_, Not_yet_processed} = gleam@list:partition(
Events_being_processed,
fun(E@1) -> gleam@list:contains(Processed, E@1) end
),
gleam@otp@actor:continue({Wal@1, Not_yet_processed});
shutdown_store ->
{stop, normal}
end end.
-spec in_memory_persistance_handler(
persistance_interface(ISX),
list(event(ISX))
) -> gleam@otp@actor:next(any(), list(event(ISX))).
in_memory_persistance_handler(Message, State) ->
case Message of
{get_stored_events, S, Aggregate_id} ->
gleam@erlang@process:send(
S,
{ok,
gleam@list:filter(
State,
fun(E) -> erlang:element(3, E) =:= Aggregate_id end
)}
),
gleam@otp@actor:continue(State);
{is_identity_available, S@1, Aggregate_id@1} ->
case gleam@list:any(
State,
fun(E@1) -> erlang:element(3, E@1) =:= Aggregate_id@1 end
) of
true ->
gleam@erlang@process:send(S@1, {ok, true});
false ->
gleam@erlang@process:send(S@1, {ok, false})
end,
gleam@otp@actor:continue(State);
{store_events, Events} ->
gleam@otp@actor:continue(lists:append(State, Events));
shutdown_persistance_layer ->
{stop, normal}
end.
-spec set_up_store_handler(
gleam@option:option(gleam@erlang@process:subject(persistance_interface(ISO)))
) -> {ok,
fun((store_messages(ISO), {list(event(ISO)), list(event(ISO))}) -> gleam@otp@actor:next(any(), {list(event(ISO)),
list(event(ISO))}))} |
{error, binary()}.
set_up_store_handler(Persistor) ->
case Persistor of
{some, P} ->
{ok, store_handler(P)};
none ->
case gleam@otp@actor:start([], fun in_memory_persistance_handler/2) of
{ok, S} ->
{ok, store_handler(S)};
_ ->
{error,
<<"Failed to spin up the in memory persistance layer!"/utf8>>}
end
end.
-spec emit_init(emit_config(IMX, any(), IMZ, INA)) -> {ok,
emit_service(IMX, IMZ, INA)} |
{error, gleam@otp@actor:start_error()}.
emit_init(Config) ->
gleam@result:'try'(
gleam@result:replace_error(
set_up_store_handler(erlang:element(3, Config)),
init_timeout
),
fun(Store_handler) ->
gleam@result:'try'(
gleam@otp@actor:start({[], []}, Store_handler),
fun(Store) ->
gleam@result:'try'(
gleam@otp@actor:start(
nil,
bus_handler(erlang:element(4, Config), Store)
),
fun(Bus) ->
gleam@result:'try'(
gleam@otp@actor:start(
gleam@dict:new(),
pool_handler(
erlang:element(2, Config),
Bus,
Store,
erlang:element(5, Config)
)
),
fun(Pool) ->
{ok, {emit_service, Pool, Bus, Store}}
end
)
end
)
end
)
end
).
-spec start(emit_config(IKS, any(), IKU, IKV)) -> {ok,
gleam@erlang@process:subject(emit_messages(IKS, IKU, IKV))} |
{error, gleam@otp@actor:start_error()}.
start(Config) ->
gleam@result:'try'(
emit_init(Config),
fun(Service) -> gleam@otp@actor:start(nil, emit_handler(Service)) end
).