Elixir ATProtocol ingestion and sync library.
feat: add ConsumerGroup and dispatching via consumers
+135
-13
+58
lib/consumer.ex
+58
lib/consumer.ex
···
1
+
defmodule Drinkup.Consumer do
2
+
@moduledoc """
3
+
An unopinionated consumer of the Firehose. Will receive all events, not just commits.
4
+
"""
5
+
6
+
alias Drinkup.{ConsumerGroup, Event}
7
+
8
+
@callback handle_event(Event.t()) :: any()
9
+
10
+
defmacro __using__(_opts) do
11
+
quote location: :keep do
12
+
use GenServer
13
+
require Logger
14
+
15
+
@behaviour Drinkup.Consumer
16
+
17
+
def child_spec(opts) do
18
+
%{
19
+
id: __MODULE__,
20
+
start: {__MODULE__, :start_link, [opts]},
21
+
type: :worker,
22
+
restart: :permanent,
23
+
max_restarts: 0,
24
+
shutdown: 500
25
+
}
26
+
end
27
+
28
+
def start_link(opts) do
29
+
GenServer.start_link(__MODULE__, [], opts)
30
+
end
31
+
32
+
@impl GenServer
33
+
def init(_) do
34
+
ConsumerGroup.join()
35
+
{:ok, nil}
36
+
end
37
+
38
+
@impl GenServer
39
+
def handle_info({:event, event}, state) do
40
+
{:ok, _pid} =
41
+
Task.start(fn ->
42
+
try do
43
+
__MODULE__.handle_event(event)
44
+
rescue
45
+
e ->
46
+
Logger.error(
47
+
"Error in event handler: #{Exception.format(:error, e, __STACKTRACE__)}"
48
+
)
49
+
end
50
+
end)
51
+
52
+
{:noreply, state}
53
+
end
54
+
55
+
defoverridable GenServer
56
+
end
57
+
end
58
+
end
+39
lib/consumer_group.ex
+39
lib/consumer_group.ex
···
1
+
defmodule Drinkup.ConsumerGroup do
2
+
@moduledoc """
3
+
Register consumers and dispatch events to them.
4
+
"""
5
+
6
+
alias Drinkup.Event
7
+
8
+
@scope __MODULE__
9
+
@group :consumers
10
+
11
+
def start_link(_) do
12
+
:pg.start_link(@scope)
13
+
end
14
+
15
+
def child_spec(opts) do
16
+
%{
17
+
id: __MODULE__,
18
+
start: {__MODULE__, :start_link, [opts]},
19
+
type: :worker,
20
+
restart: :permanent,
21
+
shutdown: 500
22
+
}
23
+
end
24
+
25
+
@spec join() :: :ok
26
+
def join(), do: join(self())
27
+
28
+
@spec join(pid()) :: :ok
29
+
def join(pid), do: :pg.join(@scope, @group, pid)
30
+
31
+
@spec dispatch(Event.t()) :: :ok
32
+
def dispatch(event) do
33
+
@scope
34
+
|> :pg.get_members(@group)
35
+
|> Enum.each(&send(&1, {:event, event}))
36
+
end
37
+
38
+
# TODO: read `:pg` docs on what `monitor` is used fo
39
+
end
+14
lib/drinkup.ex
+14
lib/drinkup.ex
···
1
1
defmodule Drinkup do
2
+
use Supervisor
3
+
4
+
def start_link(arg \\ []) do
5
+
Supervisor.start_link(__MODULE__, arg, name: __MODULE__)
6
+
end
7
+
8
+
def init(_) do
9
+
children = [
10
+
Drinkup.ConsumerGroup,
11
+
Drinkup.Socket
12
+
]
13
+
14
+
Supervisor.init(children, strategy: :one_for_one)
15
+
end
2
16
end
+4
-2
lib/event.ex
+4
-2
lib/event.ex
···
1
1
defmodule Drinkup.Event do
2
2
alias Drinkup.Event
3
3
4
-
@spec from(String.t(), map()) ::
4
+
@type t() ::
5
5
Event.Commit.t()
6
6
| Event.Sync.t()
7
7
| Event.Identity.t()
8
8
| Event.Account.t()
9
9
| Event.Info.t()
10
-
| nil
10
+
11
+
@spec from(String.t(), map()) :: t() | nil
11
12
def from("#commit", payload), do: Event.Commit.from(payload)
12
13
def from("#sync", payload), do: Event.Sync.from(payload)
13
14
def from("#identity", payload), do: Event.Identity.from(payload)
···
17
18
18
19
@spec valid_seq?(integer() | nil, any()) :: boolean()
19
20
def valid_seq?(nil, seq) when is_integer(seq), do: true
21
+
def valid_seq?(last_seq, nil) when is_integer(last_seq), do: true
20
22
def valid_seq?(last_seq, seq) when is_integer(last_seq) and is_integer(seq), do: seq > last_seq
21
23
def valid_seq?(_last_seq, _seq), do: false
22
24
end
+20
-11
lib/socket.ex
+20
-11
lib/socket.ex
···
4
4
"""
5
5
6
6
require Logger
7
-
alias Drinkup.Event
7
+
alias Drinkup.{ConsumerGroup, Event}
8
8
9
9
@behaviour :gen_statem
10
10
@default_host "https://bsky.network"
···
20
20
@impl true
21
21
def callback_mode, do: [:state_functions, :state_enter]
22
22
23
-
def start_link(opts \\ []) do
23
+
def child_spec(opts) do
24
+
%{
25
+
id: __MODULE__,
26
+
start: {__MODULE__, :start_link, [opts, []]},
27
+
type: :worker,
28
+
restart: :permanent,
29
+
shutdown: 500
30
+
}
31
+
end
32
+
33
+
def start_link(opts \\ [], statem_opts) do
24
34
opts = Keyword.validate!(opts, host: @default_host)
25
35
host = Keyword.get(opts, :host)
26
36
cursor = Keyword.get(opts, :cursor)
27
37
28
-
:gen_statem.start_link(__MODULE__, {host, cursor}, [])
38
+
:gen_statem.start_link(__MODULE__, {host, cursor}, statem_opts)
29
39
end
30
40
31
41
@impl true
···
103
113
with {:ok, header, next} <- CAR.DagCbor.decode(frame),
104
114
{:ok, payload, _} <- CAR.DagCbor.decode(next),
105
115
{%{"op" => @op_regular, "t" => type}, _} <- {header, payload},
106
-
true <- type == "#info" || Event.valid_seq?(data.seq, payload["seq"]),
107
-
data <- %{data | seq: payload["seq"] || data.seq},
108
-
message <-
109
-
Event.from(type, payload) do
116
+
true <- Event.valid_seq?(data.seq, payload["seq"]) do
117
+
data = %{data | seq: payload["seq"] || data.seq}
118
+
message = Event.from(type, payload)
110
119
:ok = :gun.update_flow(conn, stream, @flow)
111
120
112
121
case message do
113
-
%Event.Commit{} = commit ->
114
-
IO.inspect(commit.ops, label: commit.repo)
122
+
nil ->
123
+
Logger.warning("Received unrecognised event from firehose: #{inspect({type, payload})}")
115
124
116
-
msg ->
117
-
IO.inspect(msg)
125
+
message ->
126
+
ConsumerGroup.dispatch(message)
118
127
end
119
128
120
129
{:keep_state, data}