pub(crate) mod cursor;
pub mod firehose;
pub mod firehose_stats;
#[cfg(feature = "indexer")]
pub mod indexer;
mod mailbox;
pub mod relay;
pub mod stream;
pub mod validation;
use url::Url;
use crate::ingest::mailbox::{ShardedMessage, ShardedReceiver, ShardedSender};
use crate::ingest::stream::SubscribeReposMessage;
#[derive(Debug)]
pub enum IngestMessage {
Firehose {
url: Url,
/// true when `relay` is a direct PDS connection (not an aggregating relay).
/// enables host authority enforcement in the worker.
is_pds: bool,
msg: SubscribeReposMessage<'static>,
ticket: cursor::Ticket,
},
}
#[derive(Clone, Debug)]
pub struct BufferTx {
inner: ShardedSender<IngestMessage>,
}
pub type BufferRx = ShardedReceiver<IngestMessage>;
impl BufferTx {
pub(crate) fn channel(num_shards: usize) -> (Self, Vec<BufferRx>) {
let (inner, rxs) = ShardedSender::channel(num_shards);
(Self { inner }, rxs)
}
pub async fn send(
&self,
msg: IngestMessage,
) -> Result<(), tokio::sync::mpsc::error::SendError<IngestMessage>> {
self.inner.send(msg).await
}
}
impl ShardedMessage for IngestMessage {
fn shard_idx(&self, num_shards: usize) -> usize {
let did = match self {
IngestMessage::Firehose { msg, .. } => msg.did(),
};
did.map(|did| (crate::util::hash(did) as usize) % num_shards)
.unwrap_or(0)
}
}