ProtoRune. Jetstream
(proto_rune v0.5.2)
Copy Markdown
Consumer for the AT Protocol Jetstream, the filtered JSON re-encode of the repo firehose.
Where ProtoRune.Firehose delivers every commit on the network and
decodes every CAR block locally, Jetstream applies wantedCollections
and wantedDids filters server-side and ships plain JSON text frames:
a consumer that only cares about a handful of collections receives (and
pays to decode) only those events. Add it to your supervision tree:
children = [
{ProtoRune.Jetstream,
handler: MyConsumer, wanted_collections: ["place.quintal.feed.prosa"]}
]or start it directly with start_link/1.
Handlers
The required :handler option tells the consumer where events go:
- a pid, which receives
{:jetstream, %ProtoRune.Jetstream.Event{}}messages - a one-arity function, called with the event
- a
{module, function}tuple, called asfunction.(event)
Filtering and cursors
:wanted_collections- AT-URI collections to receive commit events for (repeatable server-side filter). Identity and account events are not collection-scoped and always flow through.:wanted_dids- restrict events to these repository DIDs.:cursor- atime_ustimestamp (microseconds) to resume from. Jetstream keeps a rolling playback window; resuming may re-deliver a handful of events around the cursor, so consumers must stay idempotent. When the connection drops, the consumer reconnects resuming from the last deliveredtime_us.cursor/1returns the current value, e.g. to persist it for a later restart.
Remaining options
:relay- the Jetstream instance base URL (default:"wss://jetstream2.us-east.bsky.network").:auto_reconnect- reconnect automatically on connection loss (default:true). Whenfalse, the process stops with reason{:jetstream_disconnected, reason}instead.:backoff_initial/:backoff_max- reconnect backoff bounds in milliseconds (default:1_000/30_000). The delay doubles after each failed attempt and resets once connected.:transport- theProtoRune.Firehose.Transportimplementation to use (default:ProtoRune.Firehose.Transport.Gun). Jetstream speaks the same WebSocket transport as the firehose, only with text frames.:transport_opts- options passed to the transport.:name- registers the process under the given name.
Summary
Functions
Returns a specification to start this module under a supervisor.
Returns the time_us of the last delivered event, or nil when no
event has been delivered yet.
Starts a Jetstream consumer process.
Types
@type handler() :: pid() | (ProtoRune.Jetstream.Event.t() -> term()) | {module(), atom()}
Where decoded events are delivered to.
Functions
Returns a specification to start this module under a supervisor.
See Supervisor.
@spec cursor(GenServer.server()) :: non_neg_integer() | nil
Returns the time_us of the last delivered event, or nil when no
event has been delivered yet.
Starts a Jetstream consumer process.
See the module documentation for the available options.