Skip to content

Commit 4e59e19

Browse files
ccamelccamel
authored
Merge pull request #61 from ccamel/feat/projection-management
✨ Feat/projection management
2 parents 911c43b + 4a597e2 commit 4e59e19

14 files changed

Lines changed: 446 additions & 9 deletions

‎README.md‎

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -82,6 +82,8 @@ io:format(" -> ~p~n", [Dispatch(withdraw, 10)]),
8282
io:format("[4] withdraw $1000 (should fail)~n", []),
8383
io:format(" -> ~p~n", [Dispatch(withdraw, 1000)]),
8484

85+
timer:sleep(500),
86+
8587
ok.
8688
```
8789
<!-- DEMO-END -->
@@ -179,13 +181,20 @@ es_projection:run_once(StoreContext, ProjectionModule, Options).
179181

180182
% Start a polling projection runner
181183
es_projection:start_link(StoreContext, ProjectionModule, Options).
184+
185+
% Start a managed polling projection runner under es_projection_sup
186+
es_projection:start(StoreContext, ProjectionModule, Options).
187+
188+
% Locate or stop a managed projection by ProjectionModule:name/0
189+
es_projection:lookup(ProjectionName).
190+
es_projection:stop(ProjectionName).
182191
```
183192

184193
The runner owns checkpointing. It loads the last processed global position for `ProjectionModule:name/0`, consumes events with `es_kernel_store:fold_all/4`, applies `event_filter/1` when present, calls `handle_event/2`, and stores the checkpoint after each processed position. Filtered events are checkpointed too, so a projection can keep moving through the global log.
185194

186195
By default, checkpoints are stored in ETS through `es_projection_checkpoint_ets`. Callers can provide another checkpoint backend with the `checkpoint_store` option. `start_position` defaults to `0`, and `poll_interval` defaults to `200` milliseconds.
187196

188-
The polling runner is fail-fast: any store, checkpoint, or projection handling error stops the process. If automatic recovery is needed, start projection runners under a supervisor with the desired restart strategy. Projection runners are started explicitly by callers; they are not currently part of the `es_kernel_sup` supervision tree.
197+
The polling runner is fail-fast: any store, checkpoint, or projection handling error stops the process. `es_projection:start/3` starts runners under the projection dynamic supervisor and tracks them by projection name through the manager. `start_link/3` remains available for callers that want to own the runner process directly.
189198

190199
#### Snapshot Store
191200

@@ -228,7 +237,6 @@ Setting `snapshot_interval => 0` (default) disables automatic snapshotting.
228237
#### Additional future features
229238

230239
- Add durable projection checkpoint backends for file or Mnesia stores if needed.
231-
- Add projection supervision/management helpers for named runners.
232240
- Support optional event subscriptions for real-time read-side updates.
233241
- Implement snapshot retention policies (e.g., keep only last N snapshots).
234242

‎apps/es_projection/src/es_projection.app.src‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,17 @@
11
{application, es_projection, [
22
{description, "Projection runtime for event sourcing implementation in Erlang"},
33
{vsn, "0.1.0"},
4-
{registered, []},
4+
{registered, [es_projection_root_sup, es_projection_sup, es_projection_mgr]},
55
{applications, [kernel, stdlib, es_kernel]},
6+
{mod, {es_projection_app, []}},
67
{modules, [
78
es_contract_projection,
89
es_contract_projection_checkpoint_store,
10+
es_projection_app,
911
es_projection_checkpoint_ets,
12+
es_projection_mgr,
13+
es_projection_root_sup,
14+
es_projection_sup,
1015
es_projection
1116
]},
1217
{licenses, ["BSD 3-Clause"]},

‎apps/es_projection/src/es_projection.erl‎

Lines changed: 47 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,9 @@ strategy when automatic recovery is desired.
1515

1616
-export([
1717
run_once/3,
18+
start/3,
1819
start_link/3,
20+
lookup/1,
1921
stop/1
2022
]).
2123

@@ -74,7 +76,25 @@ run_once(StoreContext, ProjectionModule, Options) ->
7476
end.
7577

7678
-doc """
77-
Start a polling projection runner.
79+
Start a managed polling projection runner.
80+
""".
81+
-spec start(StoreContext, ProjectionModule, Options) -> {ok, pid()} | {error, Reason} when
82+
StoreContext :: es_kernel_store:store_context(),
83+
ProjectionModule :: module(),
84+
Options :: options(),
85+
Reason :: term().
86+
start(StoreContext, ProjectionModule, Options) ->
87+
es_projection_mgr:start_projection(StoreContext, ProjectionModule, Options).
88+
89+
-doc """
90+
Return the pid of a managed projection runner.
91+
""".
92+
-spec lookup(atom()) -> {ok, pid()} | {error, not_found}.
93+
lookup(ProjectionName) ->
94+
es_projection_mgr:lookup(ProjectionName).
95+
96+
-doc """
97+
Start a polling projection runner linked to the caller.
7898
""".
7999
-spec start_link(StoreContext, ProjectionModule, Options) -> gen_server:start_ret() when
80100
StoreContext :: es_kernel_store:store_context(),
@@ -91,8 +111,24 @@ start_link(StoreContext, ProjectionModule, Options) ->
91111

92112
-doc """
93113
Stop a polling projection runner.
114+
115+
When passed an atom, the managed projection with that name is stopped. Other
116+
server references are stopped directly. If no projection manager is running,
117+
an atom is treated as a locally registered runner name.
94118
""".
95-
-spec stop(gen_server:server_ref()) -> ok.
119+
-spec stop(gen_server:server_ref()) -> ok | {error, not_found}.
120+
stop(ProjectionName) when is_atom(ProjectionName) ->
121+
case erlang:whereis(es_projection_mgr) of
122+
undefined ->
123+
stop_local_or_not_found(ProjectionName);
124+
_Pid ->
125+
case es_projection_mgr:stop_projection(ProjectionName) of
126+
ok ->
127+
ok;
128+
{error, not_found} ->
129+
stop_local_or_not_found(ProjectionName)
130+
end
131+
end;
96132
stop(ServerRef) ->
97133
gen_server:stop(ServerRef).
98134

@@ -174,6 +210,15 @@ init_runtime(StoreContext, ProjectionModule, Options) ->
174210
{error, Reason}
175211
end.
176212

213+
-spec stop_local_or_not_found(atom()) -> ok | {error, not_found}.
214+
stop_local_or_not_found(Name) ->
215+
case erlang:whereis(Name) of
216+
undefined ->
217+
{error, not_found};
218+
_Pid ->
219+
gen_server:stop(Name)
220+
end.
221+
177222
-spec ensure_checkpoint_store_started(module()) -> ok | {error, term()}.
178223
ensure_checkpoint_store_started(CheckpointStore) ->
179224
case code:ensure_loaded(CheckpointStore) of

‎apps/es_projection/src/es_projection_app.erl‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,19 @@
1+
-module(es_projection_app).
2+
3+
-moduledoc """
4+
OTP application callback for the projection runtime.
5+
""".
6+
7+
-behaviour(application).
8+
9+
-export([start/2, stop/1]).
10+
11+
-spec start(application:start_type(), term()) -> {ok, pid()}.
12+
start(_StartType, _StartArgs) ->
13+
es_projection_checkpoint_ets:start(),
14+
es_projection_root_sup:start_link().
15+
16+
-spec stop(term()) -> ok.
17+
stop(_State) ->
18+
ok = es_projection_checkpoint_ets:stop(),
19+
ok.

‎apps/es_projection/src/es_projection_mgr.erl‎

Lines changed: 139 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,139 @@
1+
-module(es_projection_mgr).
2+
3+
-moduledoc """
4+
Singleton manager for projection runner processes.
5+
""".
6+
7+
-behaviour(gen_server).
8+
9+
-export([
10+
start_link/0,
11+
stop/0,
12+
start_projection/3,
13+
stop_projection/1,
14+
lookup/1
15+
]).
16+
17+
-export([
18+
init/1,
19+
handle_call/3,
20+
handle_cast/2,
21+
handle_info/2,
22+
terminate/2,
23+
code_change/3
24+
]).
25+
26+
-record(state, {
27+
pids = #{} :: #{atom() => pid()},
28+
monitors = #{} :: #{reference() => atom()}
29+
}).
30+
31+
-opaque state() :: #state{}.
32+
-export_type([state/0]).
33+
34+
-spec start_link() -> gen_server:start_ret().
35+
start_link() ->
36+
gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
37+
38+
-spec stop() -> ok.
39+
stop() ->
40+
gen_server:stop(?MODULE).
41+
42+
-spec start_projection(StoreContext, ProjectionModule, Options) ->
43+
{ok, pid()} | {error, Reason}
44+
when
45+
StoreContext :: es_kernel_store:store_context(),
46+
ProjectionModule :: module(),
47+
Options :: es_projection:options(),
48+
Reason :: term().
49+
start_projection(StoreContext, ProjectionModule, Options) ->
50+
gen_server:call(?MODULE, {start_projection, StoreContext, ProjectionModule, Options}).
51+
52+
-spec stop_projection(atom()) -> ok | {error, not_found}.
53+
stop_projection(ProjectionName) ->
54+
gen_server:call(?MODULE, {stop_projection, ProjectionName}).
55+
56+
-spec lookup(atom()) -> {ok, pid()} | {error, not_found}.
57+
lookup(ProjectionName) ->
58+
gen_server:call(?MODULE, {lookup, ProjectionName}).
59+
60+
-spec init([]) -> {ok, state()}.
61+
init([]) ->
62+
{ok, #state{}}.
63+
64+
-spec handle_call(term(), {pid(), term()}, state()) ->
65+
{reply, term(), state()}.
66+
handle_call({start_projection, StoreContext, ProjectionModule, Options}, _From, State) ->
67+
ProjectionName = ProjectionModule:name(),
68+
case maps:find(ProjectionName, State#state.pids) of
69+
{ok, Pid} ->
70+
{reply, {ok, Pid}, State};
71+
error ->
72+
start_and_monitor(ProjectionName, StoreContext, ProjectionModule, Options, State)
73+
end;
74+
handle_call({stop_projection, ProjectionName}, _From, State) ->
75+
case maps:find(ProjectionName, State#state.pids) of
76+
{ok, Pid} ->
77+
ok = es_projection:stop(Pid),
78+
{reply, ok, remove_projection(ProjectionName, State)};
79+
error ->
80+
{reply, {error, not_found}, State}
81+
end;
82+
handle_call({lookup, ProjectionName}, _From, State) ->
83+
case maps:find(ProjectionName, State#state.pids) of
84+
{ok, Pid} ->
85+
{reply, {ok, Pid}, State};
86+
error ->
87+
{reply, {error, not_found}, State}
88+
end;
89+
handle_call(_Request, _From, State) ->
90+
{reply, {error, unknown_call}, State}.
91+
92+
-spec handle_cast(term(), state()) -> {noreply, state()}.
93+
handle_cast(_Request, State) ->
94+
{noreply, State}.
95+
96+
-spec handle_info(term(), state()) -> {noreply, state()}.
97+
handle_info({'DOWN', MonitorRef, process, _Pid, _Reason}, State) ->
98+
case maps:find(MonitorRef, State#state.monitors) of
99+
{ok, ProjectionName} ->
100+
{noreply, remove_projection(ProjectionName, State)};
101+
error ->
102+
{noreply, State}
103+
end;
104+
handle_info(_Info, State) ->
105+
{noreply, State}.
106+
107+
-spec terminate(term(), state()) -> ok.
108+
terminate(_Reason, _State) ->
109+
ok.
110+
111+
-spec code_change(term(), state(), term()) -> {ok, state()}.
112+
code_change(_OldVsn, State, _Extra) ->
113+
{ok, State}.
114+
115+
-spec start_and_monitor(
116+
atom(), es_kernel_store:store_context(), module(), es_projection:options(), state()
117+
) ->
118+
{reply, {ok, pid()} | {error, term()}, state()}.
119+
start_and_monitor(ProjectionName, StoreContext, ProjectionModule, Options, State) ->
120+
case es_projection_sup:start_projection(StoreContext, ProjectionModule, Options) of
121+
{ok, Pid} ->
122+
MonitorRef = erlang:monitor(process, Pid),
123+
NewState = State#state{
124+
pids = (State#state.pids)#{ProjectionName => Pid},
125+
monitors = (State#state.monitors)#{MonitorRef => ProjectionName}
126+
},
127+
{reply, {ok, Pid}, NewState};
128+
{error, Reason} ->
129+
{reply, {error, Reason}, State}
130+
end.
131+
132+
-spec remove_projection(atom(), state()) -> state().
133+
remove_projection(ProjectionName, State) ->
134+
Pids = maps:remove(ProjectionName, State#state.pids),
135+
Monitors = maps:filter(
136+
fun(_MonitorRef, Name) -> Name =/= ProjectionName end,
137+
State#state.monitors
138+
),
139+
State#state{pids = Pids, monitors = Monitors}.

‎apps/es_projection/src/es_projection_root_sup.erl‎

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,43 @@
1+
-module(es_projection_root_sup).
2+
3+
-moduledoc """
4+
Top-level supervisor for the projection application.
5+
""".
6+
7+
-behaviour(supervisor).
8+
9+
-export([start_link/0]).
10+
-export([init/1]).
11+
12+
-define(SERVER, ?MODULE).
13+
14+
-spec start_link() -> supervisor:startlink_ret().
15+
start_link() ->
16+
supervisor:start_link({local, ?SERVER}, ?MODULE, []).
17+
18+
-spec init([]) -> {ok, {supervisor:sup_flags(), [supervisor:child_spec()]}}.
19+
init([]) ->
20+
SupFlags = #{
21+
strategy => one_for_one,
22+
intensity => 10,
23+
period => 5
24+
},
25+
ChildSpecs = [
26+
#{
27+
id => es_projection_sup,
28+
start => {es_projection_sup, start_link, []},
29+
restart => permanent,
30+
shutdown => 5000,
31+
type => supervisor,
32+
modules => [es_projection_sup]
33+
},
34+
#{
35+
id => es_projection_mgr,
36+
start => {es_projection_mgr, start_link, []},
37+
restart => permanent,
38+
shutdown => 5000,
39+
type => worker,
40+
modules => [es_projection_mgr]
41+
}
42+
],
43+
{ok, {SupFlags, ChildSpecs}}.

‎apps/es_projection/src/es_projection_sup.erl‎

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,43 @@
1+
-module(es_projection_sup).
2+
3+
-moduledoc """
4+
Dynamic supervisor for projection runners.
5+
""".
6+
7+
-behaviour(supervisor).
8+
9+
-export([start_link/0, start_projection/3]).
10+
-export([init/1]).
11+
12+
-define(SERVER, ?MODULE).
13+
14+
-spec start_link() -> supervisor:startlink_ret().
15+
start_link() ->
16+
supervisor:start_link({local, ?SERVER}, ?MODULE, []).
17+
18+
-spec start_projection(StoreContext, ProjectionModule, Options) ->
19+
supervisor:startchild_ret()
20+
when
21+
StoreContext :: es_kernel_store:store_context(),
22+
ProjectionModule :: module(),
23+
Options :: es_projection:options().
24+
start_projection(StoreContext, ProjectionModule, Options) ->
25+
ProjectionName = ProjectionModule:name(),
26+
ChildSpec = #{
27+
id => {es_projection, ProjectionName},
28+
start => {es_projection, start_link, [StoreContext, ProjectionModule, Options]},
29+
restart => temporary,
30+
shutdown => 5000,
31+
type => worker,
32+
modules => [es_projection]
33+
},
34+
supervisor:start_child(?SERVER, ChildSpec).
35+
36+
-spec init([]) -> {ok, {supervisor:sup_flags(), [supervisor:child_spec()]}}.
37+
init([]) ->
38+
SupFlags = #{
39+
strategy => one_for_one,
40+
intensity => 10,
41+
period => 5
42+
},
43+
{ok, {SupFlags, []}}.

0 commit comments

Comments
 (0)