use super::*;
/// the relay event stream produced by [`Hydrant::subscribe_repos`].
pub struct RelayEventStream(mpsc::Receiver<Result<bytes::Bytes, RelayStreamError>>);
#[derive(Debug, Clone, thiserror::Error)]
pub enum RelayStreamError {
#[error("relay stream consumer too slow: {reason}")]
ConsumerTooSlow { reason: String },
#[error("cursor {cursor} is past the last seq, {head}")]
FutureCursor { cursor: u64, head: u64 },
}
impl RelayStreamError {
pub fn code(&self) -> &'static str {
match self {
Self::ConsumerTooSlow { .. } => "ConsumerTooSlow",
Self::FutureCursor { .. } => "FutureCursor",
}
}
}
impl futures::Stream for RelayEventStream {
type Item = Result<bytes::Bytes, RelayStreamError>;
fn poll_next(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<Option<Self::Item>> {
self.0.poll_recv(cx)
}
}
impl Hydrant {
/// subscribe to the relay's ordered `subscribeRepos` event stream.
///
/// returns a [`RelayEventStream`] that yields pre-encoded CBOR binary frames
/// ready to forward directly to ATProto clients via WebSocket.
///
/// - if `cursor` is `None`, streaming starts from the current head (live tail only).
/// - if `cursor` is `Some(seq)`, all persisted events from that seq onward are replayed first.
/// - if `cursor` is past the last seq, the stream yields [`RelayStreamError::FutureCursor`]
/// and ends.
///
/// a subscriber that falls behind the live broadcast catches up from the database. one
/// that stops reading for `HYDRANT_STREAM_SEND_TIMEOUT` receives
/// [`RelayStreamError::ConsumerTooSlow`] after the frames already queued, then the
/// stream ends.
///
/// the replay runs as fast as you read it. only `subscribeRepos` readers on the network are capped,
/// when `HYDRANT_THROTTLED_STREAMS` lists it.
pub fn subscribe_repos(&self, cursor: Option<u64>) -> RelayEventStream {
self.subscribe_repos_with(cursor, stream::StreamOptions::from_config(&self.config))
}
/// [`Self::subscribe_repos`] for a reader on the network, whose replay is capped.
pub(crate) fn subscribe_repos_network(&self, cursor: Option<u64>) -> RelayEventStream {
self.subscribe_repos_with(
cursor,
stream::StreamOptions::network(&self.config, crate::config::StreamKind::Relay),
)
}
fn subscribe_repos_with(
&self,
cursor: Option<u64>,
opts: stream::StreamOptions,
) -> RelayEventStream {
let (tx, rx) = mpsc::channel(stream::STREAM_CHANNEL_CAPACITY);
let state = self.state.clone();
let runtime = tokio::runtime::Handle::current();
std::thread::Builder::new()
.name("hydrant-relay-stream".into())
.spawn(move || {
let _g = runtime.enter();
relay_stream_thread(state, tx, cursor, opts);
})
.expect("failed to spawn relay stream thread");
RelayEventStream(rx)
}
}