comet.sh
Elixir ATProtocol ingestion and sync library.
refactor: simplify structure. Remove ConsumerGroup and make Consumer a simple behaviour
ovyerus.com
+3
-4
examples/record_consumer.ex
+3
-4
examples/record_consumer.ex
+1
-50
lib/consumer.ex
+1
-50
lib/consumer.ex
···
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
···
+12
-5
lib/drinkup.ex
+12
-5
lib/drinkup.ex
···
1
defmodule Drinkup do
2
use Supervisor
3
4
+
@type options() :: %{
5
+
required(:consumer) => module(),
6
+
optional(:host) => String.t(),
7
+
optional(:cursor) => pos_integer()
8
+
}
9
+
10
+
@spec start_link(options()) :: Supervisor.on_start()
11
+
def start_link(options) do
12
+
Supervisor.start_link(__MODULE__, options, name: __MODULE__)
13
end
14
15
+
def init(options) do
16
children = [
17
+
{Task.Supervisor, name: Drinkup.TaskSupervisor},
18
+
{Drinkup.Socket, options}
19
]
20
21
Supervisor.init(children, strategy: :one_for_one)
+16
lib/event.ex
+16
lib/event.ex
···
1
defmodule Drinkup.Event do
2
+
require Logger
3
alias Drinkup.Event
4
5
@type t() ::
···
22
def valid_seq?(last_seq, nil) when is_integer(last_seq), do: true
23
def valid_seq?(last_seq, seq) when is_integer(last_seq) and is_integer(seq), do: seq > last_seq
24
def valid_seq?(_last_seq, _seq), do: false
25
+
26
+
@spec dispatch(module(), t()) :: :ok
27
+
def dispatch(consumer, message) do
28
+
{:ok, _pid} =
29
+
Task.Supervisor.start_child(Drinkup.TaskSupervisor, fn ->
30
+
try do
31
+
consumer.handle_event(message)
32
+
rescue
33
+
e ->
34
+
Logger.error("Error in event handler: #{Exception.format(:error, e, __STACKTRACE__)}")
35
+
end
36
+
end)
37
+
38
+
:ok
39
+
end
40
end
+1
-1
lib/record_consumer.ex
+1
-1
lib/record_consumer.ex
+11
-13
lib/socket.ex
+11
-13
lib/socket.ex
···
4
"""
5
6
require Logger
7
-
alias Drinkup.{ConsumerGroup, Event}
8
9
@behaviour :gen_statem
10
@default_host "https://bsky.network"
···
15
@op_regular 1
16
@op_error -1
17
18
-
defstruct [:host, :seq, :conn, :stream]
19
20
@impl true
21
def callback_mode, do: [:state_functions, :state_enter]
···
30
}
31
end
32
33
-
def start_link(opts \\ [], statem_opts) do
34
-
opts = Keyword.validate!(opts, host: @default_host)
35
-
host = Keyword.get(opts, :host)
36
-
cursor = Keyword.get(opts, :cursor)
37
38
-
:gen_statem.start_link(__MODULE__, {host, cursor}, statem_opts)
39
end
40
41
@impl true
42
-
def init({host, cursor}) do
43
-
data = %__MODULE__{host: host, seq: cursor}
44
{:ok, :disconnected, data, [{:next_event, :internal, :connect}]}
45
end
46
···
54
{:next_state, :connecting_http, data}
55
end
56
57
-
def connecting_http(:enter, _from, data) do
58
Logger.debug("Connecting to http")
59
60
-
%{host: host, port: port} = URI.new!(data.host)
61
62
{:ok, conn} =
63
:gun.open(:binary.bin_to_list(host), port, %{
···
107
:keep_state_and_data
108
end
109
110
-
def connected(:info, {:gun_ws, conn, stream, {:binary, frame}}, data) do
111
# TODO: let clients specify a handler for raw* (*decoded) packets to support any atproto subscription
112
# Will also need support for JSON frames
113
with {:ok, header, next} <- CAR.DagCbor.decode(frame),
···
123
Logger.warning("Received unrecognised event from firehose: #{inspect({type, payload})}")
124
125
message ->
126
-
ConsumerGroup.dispatch(message)
127
end
128
129
{:keep_state, data}
···
4
"""
5
6
require Logger
7
+
alias Drinkup.Event
8
9
@behaviour :gen_statem
10
@default_host "https://bsky.network"
···
15
@op_regular 1
16
@op_error -1
17
18
+
defstruct [:options, :seq, :conn, :stream]
19
20
@impl true
21
def callback_mode, do: [:state_functions, :state_enter]
···
30
}
31
end
32
33
+
def start_link(%{consumer: _} = options, statem_opts) do
34
+
options = Map.merge(%{host: @default_host, cursor: nil}, options)
35
36
+
:gen_statem.start_link(__MODULE__, options, statem_opts)
37
end
38
39
@impl true
40
+
def init(%{cursor: seq} = options) do
41
+
data = %__MODULE__{seq: seq, options: options}
42
{:ok, :disconnected, data, [{:next_event, :internal, :connect}]}
43
end
44
···
52
{:next_state, :connecting_http, data}
53
end
54
55
+
def connecting_http(:enter, _from, %{options: options} = data) do
56
Logger.debug("Connecting to http")
57
58
+
%{host: host, port: port} = URI.new!(options.host)
59
60
{:ok, conn} =
61
:gun.open(:binary.bin_to_list(host), port, %{
···
105
:keep_state_and_data
106
end
107
108
+
def connected(:info, {:gun_ws, conn, stream, {:binary, frame}}, %{options: options} = data) do
109
# TODO: let clients specify a handler for raw* (*decoded) packets to support any atproto subscription
110
# Will also need support for JSON frames
111
with {:ok, header, next} <- CAR.DagCbor.decode(frame),
···
121
Logger.warning("Received unrecognised event from firehose: #{inspect({type, payload})}")
122
123
message ->
124
+
Event.dispatch(options.consumer, message)
125
end
126
127
{:keep_state, data}