comet.sh
Elixir ATProtocol ingestion and sync library.
feat: support running multiple instances
ovyerus.com
+1
-2
README.md
+1
-2
README.md
+26
examples/basic_consumer.ex
+26
examples/basic_consumer.ex
···
1
+
defmodule BasicConsumer do
2
+
@behaviour Drinkup.Consumer
3
+
4
+
def handle_event(%Drinkup.Event.Commit{} = event) do
5
+
IO.inspect(event, label: "Got commit event")
6
+
end
7
+
8
+
def handle_event(_), do: :noop
9
+
end
10
+
11
+
defmodule ExampleSupervisor do
12
+
use Supervisor
13
+
14
+
def start_link(arg \\ []) do
15
+
Supervisor.start_link(__MODULE__, arg, name: __MODULE__)
16
+
end
17
+
18
+
@impl true
19
+
def init(_) do
20
+
children = [
21
+
{Drinkup, %{consumer: BasicConsumer}}
22
+
]
23
+
24
+
Supervisor.init(children, strategy: :one_for_one)
25
+
end
26
+
end
+35
examples/multiple_consumers.ex
+35
examples/multiple_consumers.ex
···
1
+
defmodule PostDeleteConsumer do
2
+
use Drinkup.RecordConsumer, collections: ["app.bsky.feed.post"]
3
+
4
+
def handle_delete(record) do
5
+
IO.inspect(record, label: "update")
6
+
end
7
+
end
8
+
9
+
defmodule IdentityConsumer do
10
+
@behaviour Drinkup.Consumer
11
+
12
+
def handle_event(%Drinkup.Event.Identity{} = event) do
13
+
IO.inspect(event, label: "identity event")
14
+
end
15
+
16
+
def handle_event(_), do: :noop
17
+
end
18
+
19
+
defmodule ExampleSupervisor do
20
+
use Supervisor
21
+
22
+
def start_link(arg \\ []) do
23
+
Supervisor.start_link(__MODULE__, arg, name: __MODULE__)
24
+
end
25
+
26
+
@impl true
27
+
def init(_) do
28
+
children = [
29
+
{Drinkup, %{consumer: PostDeleteConsumer}},
30
+
{Drinkup, %{consumer: IdentityConsumer, name: :identities}}
31
+
]
32
+
33
+
Supervisor.init(children, strategy: :one_for_one)
34
+
end
35
+
end
+2
-2
examples/record_consumer.ex
+2
-2
examples/record_consumer.ex
···
17
17
defmodule ExampleSupervisor do
18
18
use Supervisor
19
19
20
-
def start_link(args \\ []) do
20
+
def start_link(arg \\ []) do
21
21
Supervisor.start_link(__MODULE__, arg, name: __MODULE__)
22
22
end
23
23
24
24
@impl true
25
25
def init(_) do
26
26
children = [
27
-
{Drinkup, %{module: ExampleRecordConsumer}}
27
+
{Drinkup, %{consumer: ExampleRecordConsumer}}
28
28
]
29
29
30
30
Supervisor.init(children, strategy: :one_for_one)
+8
lib/application.ex
+8
lib/application.ex
+23
-14
lib/drinkup.ex
+23
-14
lib/drinkup.ex
···
1
1
defmodule Drinkup do
2
2
use Supervisor
3
+
alias Drinkup.Options
3
4
4
-
@type options() :: %{
5
-
required(:consumer) => module(),
6
-
optional(:host) => String.t(),
7
-
optional(:cursor) => pos_integer()
8
-
}
5
+
@dialyzer nowarn_function: {:init, 1}
6
+
@impl true
7
+
def init({%Options{name: name} = drinkup_options, supervisor_options}) do
8
+
children = [
9
+
{Task.Supervisor, name: {:via, Registry, {Drinkup.Registry, {name, Tasks}}}},
10
+
{Drinkup.Socket, drinkup_options}
11
+
]
9
12
10
-
@spec start_link(options()) :: Supervisor.on_start()
11
-
def start_link(options) do
12
-
Supervisor.start_link(__MODULE__, options, name: __MODULE__)
13
+
Supervisor.start_link(
14
+
children,
15
+
supervisor_options ++ [name: {:via, Registry, {Drinkup.Registry, {name, Supervisor}}}]
16
+
)
13
17
end
14
18
15
-
def init(options) do
16
-
children = [
17
-
{Task.Supervisor, name: Drinkup.TaskSupervisor},
18
-
{Drinkup.Socket, options}
19
-
]
19
+
@spec child_spec(Options.options()) :: Supervisor.child_spec()
20
+
def child_spec(%{} = options), do: child_spec({options, [strategy: :one_for_one]})
20
21
21
-
Supervisor.init(children, strategy: :one_for_one)
22
+
@spec child_spec({Options.options(), Keyword.t()}) :: Supervisor.child_spec()
23
+
def child_spec({drinkup_options, supervisor_options}) do
24
+
%{
25
+
id: Map.get(drinkup_options, :name, __MODULE__),
26
+
start: {__MODULE__, :init, [{Options.from(drinkup_options), supervisor_options}]},
27
+
type: :supervisor,
28
+
restart: :permanent,
29
+
shutdown: 500
30
+
}
22
31
end
23
32
end
+6
-4
lib/event.ex
+6
-4
lib/event.ex
···
1
1
defmodule Drinkup.Event do
2
2
require Logger
3
-
alias Drinkup.Event
3
+
alias Drinkup.{Event, Options}
4
4
5
5
@type t() ::
6
6
Event.Commit.t()
···
23
23
def valid_seq?(last_seq, seq) when is_integer(last_seq) and is_integer(seq), do: seq > last_seq
24
24
def valid_seq?(_last_seq, _seq), do: false
25
25
26
-
@spec dispatch(module(), t()) :: :ok
27
-
def dispatch(consumer, message) do
26
+
@spec dispatch(t(), Options.t()) :: :ok
27
+
def dispatch(message, %Options{consumer: consumer, name: name}) do
28
+
supervisor_name = {:via, Registry, {Drinkup.Registry, {name, Tasks}}}
29
+
28
30
{:ok, _pid} =
29
-
Task.Supervisor.start_child(Drinkup.TaskSupervisor, fn ->
31
+
Task.Supervisor.start_child(supervisor_name, fn ->
30
32
try do
31
33
consumer.handle_event(message)
32
34
rescue
+22
lib/options.ex
+22
lib/options.ex
···
1
+
defmodule Drinkup.Options do
2
+
use TypedStruct
3
+
4
+
@default_host "https://bsky.network"
5
+
6
+
@type options() :: %{
7
+
required(:consumer) => module(),
8
+
optional(:name) => atom(),
9
+
optional(:host) => String.t(),
10
+
optional(:cursor) => pos_integer()
11
+
}
12
+
13
+
typedstruct do
14
+
field :consumer, module(), enforce: true
15
+
field :name, atom(), default: Drinkup
16
+
field :host, String.t(), default: @default_host
17
+
field :cursor, pos_integer() | nil
18
+
end
19
+
20
+
@spec from(options()) :: t()
21
+
def from(%{consumer: _} = options), do: struct(__MODULE__, options)
22
+
end
+3
-6
lib/socket.ex
+3
-6
lib/socket.ex
···
4
4
"""
5
5
6
6
require Logger
7
-
alias Drinkup.Event
7
+
alias Drinkup.{Event, Options}
8
8
9
9
@behaviour :gen_statem
10
-
@default_host "https://bsky.network"
11
10
@timeout :timer.seconds(5)
12
11
# TODO: `flow` determines messages in buffer. Determine ideal value?
13
12
@flow 10
···
30
29
}
31
30
end
32
31
33
-
def start_link(%{consumer: _} = options, statem_opts) do
34
-
options = Map.merge(%{host: @default_host, cursor: nil}, options)
35
-
32
+
def start_link(%Options{} = options, statem_opts) do
36
33
:gen_statem.start_link(__MODULE__, options, statem_opts)
37
34
end
38
35
···
121
118
Logger.warning("Received unrecognised event from firehose: #{inspect({type, payload})}")
122
119
123
120
message ->
124
-
Event.dispatch(options.consumer, message)
121
+
Event.dispatch(message, options)
125
122
end
126
123
127
124
{:keep_state, data}