From 69ca4f2ec12635eb5ecbcb96c0b0f9ce5656897c Mon Sep 17 00:00:00 2001 From: Owen Jacobson Date: Wed, 25 Sep 2024 02:23:28 -0400 Subject: More reorganizing. --- src/events/app.rs | 140 +++++++++++++++++++++++++++++++++++++++++ src/events/app/broadcaster.rs | 116 ---------------------------------- src/events/app/events.rs | 141 ------------------------------------------ src/events/app/mod.rs | 5 -- src/events/broadcaster.rs | 116 ++++++++++++++++++++++++++++++++++ src/events/mod.rs | 1 + 6 files changed, 257 insertions(+), 262 deletions(-) create mode 100644 src/events/app.rs delete mode 100644 src/events/app/broadcaster.rs delete mode 100644 src/events/app/events.rs delete mode 100644 src/events/app/mod.rs create mode 100644 src/events/broadcaster.rs (limited to 'src/events') diff --git a/src/events/app.rs b/src/events/app.rs new file mode 100644 index 0000000..7229551 --- /dev/null +++ b/src/events/app.rs @@ -0,0 +1,140 @@ +use chrono::TimeDelta; +use futures::{ + future, + stream::{self, StreamExt as _}, + Stream, +}; +use sqlx::sqlite::SqlitePool; + +use super::{ + broadcaster::Broadcaster, + repo::broadcast::{self, Provider as _}, +}; +use crate::{ + clock::DateTime, + repo::{ + channel::{self, Provider as _}, + error::NotFound as _, + login::Login, + }, +}; + +pub struct Events<'a> { + db: &'a SqlitePool, + broadcaster: &'a Broadcaster, +} + +impl<'a> Events<'a> { + pub const fn new(db: &'a SqlitePool, broadcaster: &'a Broadcaster) -> Self { + Self { db, broadcaster } + } + + pub async fn send( + &self, + login: &Login, + channel: &channel::Id, + body: &str, + sent_at: &DateTime, + ) -> Result { + let mut tx = self.db.begin().await?; + let channel = tx + .channels() + .by_id(channel) + .await + .not_found(|| EventsError::ChannelNotFound(channel.clone()))?; + let message = tx + .broadcast() + .create(login, &channel, body, sent_at) + .await?; + tx.commit().await?; + + self.broadcaster.broadcast(&channel.id, &message); + Ok(message) + } + + pub async fn subscribe( + &self, + channel: &channel::Id, + subscribed_at: &DateTime, + resume_at: Option, + ) -> Result + std::fmt::Debug, EventsError> { + // Somewhat arbitrarily, expire after 90 days. + let expire_at = subscribed_at.to_owned() - TimeDelta::days(90); + + let mut tx = self.db.begin().await?; + let channel = tx + .channels() + .by_id(channel) + .await + .not_found(|| EventsError::ChannelNotFound(channel.clone()))?; + + // Subscribe before retrieving, to catch messages broadcast while we're + // querying the DB. We'll prune out duplicates later. + let live_messages = self.broadcaster.subscribe(&channel.id); + + tx.broadcast().expire(&expire_at).await?; + let stored_messages = tx.broadcast().replay(&channel, resume_at).await?; + tx.commit().await?; + + let resume_broadcast_at = stored_messages + .last() + .map(|message| message.sequence) + .or(resume_at); + + // This should always be the case, up to integer rollover, primarily + // because every message in stored_messages has a sequence not less + // than `resume_at`, or `resume_at` is None. We use the last message + // (if any) to decide when to resume the `live_messages` stream. + // + // It probably simplifies to assert!(resume_at <= resume_broadcast_at), but + // this form captures more of the reasoning. + assert!( + (resume_at.is_none() && resume_broadcast_at.is_none()) + || (stored_messages.is_empty() && resume_at == resume_broadcast_at) + || resume_at < resume_broadcast_at + ); + + // no skip_expired or resume transforms for stored_messages, as it's + // constructed not to contain messages meeting either criterion. + // + // * skip_expired is redundant with the `tx.broadcasts().expire(…)` call; + // * resume is redundant with the resume_at argument to + // `tx.broadcasts().replay(…)`. + let stored_messages = stream::iter(stored_messages); + let live_messages = live_messages + // Sure, it's temporally improbable that we'll ever skip a message + // that's 90 days old, but there's no reason not to be thorough. + .filter(Self::skip_expired(&expire_at)) + // Filtering on the broadcast resume point filters out messages + // before resume_at, and filters out messages duplicated from + // stored_messages. + .filter(Self::resume(resume_broadcast_at)); + + Ok(stored_messages.chain(live_messages)) + } + + fn resume( + resume_at: Option, + ) -> impl for<'m> FnMut(&'m broadcast::Message) -> future::Ready { + move |msg| { + future::ready(match resume_at { + None => true, + Some(resume_at) => msg.sequence > resume_at, + }) + } + } + fn skip_expired( + expire_at: &DateTime, + ) -> impl for<'m> FnMut(&'m broadcast::Message) -> future::Ready { + let expire_at = expire_at.to_owned(); + move |msg| future::ready(msg.sent_at > expire_at) + } +} + +#[derive(Debug, thiserror::Error)] +pub enum EventsError { + #[error("channel {0} not found")] + ChannelNotFound(channel::Id), + #[error(transparent)] + DatabaseError(#[from] sqlx::Error), +} diff --git a/src/events/app/broadcaster.rs b/src/events/app/broadcaster.rs deleted file mode 100644 index 6a1219a..0000000 --- a/src/events/app/broadcaster.rs +++ /dev/null @@ -1,116 +0,0 @@ -use std::collections::{hash_map::Entry, HashMap}; -use std::sync::{Arc, Mutex, MutexGuard}; - -use futures::{future, stream::StreamExt as _, Stream}; -use sqlx::sqlite::SqlitePool; -use tokio::sync::broadcast::{channel, Sender}; -use tokio_stream::wrappers::{errors::BroadcastStreamRecvError, BroadcastStream}; - -use crate::{ - events::repo::broadcast, - repo::channel::{self, Provider as _}, -}; - -// Clones will share the same senders collection. -#[derive(Clone)] -pub struct Broadcaster { - // The use of std::sync::Mutex, and not tokio::sync::Mutex, follows Tokio's - // own advice: . Methods that - // lock it must be sync. - senders: Arc>>>, -} - -impl Broadcaster { - pub async fn from_database(db: &SqlitePool) -> Result { - let mut tx = db.begin().await?; - let channels = tx.channels().all().await?; - tx.commit().await?; - - let channels = channels.iter().map(|c| &c.id); - let broadcaster = Self::new(channels); - Ok(broadcaster) - } - - fn new<'i>(channels: impl IntoIterator) -> Self { - let senders: HashMap<_, _> = channels - .into_iter() - .cloned() - .map(|id| (id, Self::make_sender())) - .collect(); - - Self { - senders: Arc::new(Mutex::new(senders)), - } - } - - // panic: if ``channel`` is already registered. - pub fn register_channel(&self, channel: &channel::Id) { - match self.senders().entry(channel.clone()) { - // This ever happening indicates a serious logic error. - Entry::Occupied(_) => panic!("duplicate channel registration for channel {channel}"), - Entry::Vacant(entry) => { - entry.insert(Self::make_sender()); - } - } - } - - // panic: if ``channel`` has not been previously registered, and was not - // part of the initial set of channels. - pub fn broadcast(&self, channel: &channel::Id, message: &broadcast::Message) { - let tx = self.sender(channel); - - // Per the Tokio docs, the returned error is only used to indicate that - // there are no receivers. In this use case, that's fine; a lack of - // listening consumers (chat clients) when a message is sent isn't an - // error. - // - // The successful return value, which includes the number of active - // receivers, also isn't that interesting to us. - let _ = tx.send(message.clone()); - } - - // panic: if ``channel`` has not been previously registered, and was not - // part of the initial set of channels. - pub fn subscribe( - &self, - channel: &channel::Id, - ) -> impl Stream + std::fmt::Debug { - let rx = self.sender(channel).subscribe(); - - BroadcastStream::from(rx) - .take_while(|r| { - future::ready(match r { - Ok(_) => true, - // Stop the stream here. This will disconnect SSE clients - // (see `routes.rs`), who will then resume from - // `Last-Event-ID`, allowing them to catch up by reading - // the skipped messages from the database. - Err(BroadcastStreamRecvError::Lagged(_)) => false, - }) - }) - .map(|r| { - // Since the previous transform stops at the first error, this - // should always hold. - // - // See also . - r.expect("after filtering, only `Ok` messages should remain") - }) - } - - // panic: if ``channel`` has not been previously registered, and was not - // part of the initial set of channels. - fn sender(&self, channel: &channel::Id) -> Sender { - self.senders()[channel].clone() - } - - fn senders(&self) -> MutexGuard>> { - self.senders.lock().unwrap() // propagate panics when mutex is poisoned - } - - fn make_sender() -> Sender { - // Queue depth of 16 chosen entirely arbitrarily. Don't read too much - // into it. - let (tx, _) = channel(16); - tx - } -} diff --git a/src/events/app/events.rs b/src/events/app/events.rs deleted file mode 100644 index 8b76994..0000000 --- a/src/events/app/events.rs +++ /dev/null @@ -1,141 +0,0 @@ -use chrono::TimeDelta; -use futures::{ - future, - stream::{self, StreamExt as _}, - Stream, -}; -use sqlx::sqlite::SqlitePool; - -use super::Broadcaster; -use crate::{ - clock::DateTime, - events::{ - app::EventsError, - repo::broadcast::{self, Provider as _}, - }, - repo::{ - channel::{self, Provider as _}, - error::NotFound as _, - login::Login, - }, -}; - -pub struct Events<'a> { - db: &'a SqlitePool, - broadcaster: &'a Broadcaster, -} - -impl<'a> Events<'a> { - pub const fn new(db: &'a SqlitePool, broadcaster: &'a Broadcaster) -> Self { - Self { db, broadcaster } - } - - pub async fn send( - &self, - login: &Login, - channel: &channel::Id, - body: &str, - sent_at: &DateTime, - ) -> Result { - let mut tx = self.db.begin().await?; - let channel = tx - .channels() - .by_id(channel) - .await - .not_found(|| EventsError::ChannelNotFound(channel.clone()))?; - let message = tx - .broadcast() - .create(login, &channel, body, sent_at) - .await?; - tx.commit().await?; - - self.broadcaster.broadcast(&channel.id, &message); - Ok(message) - } - - pub async fn subscribe( - &self, - channel: &channel::Id, - subscribed_at: &DateTime, - resume_at: Option, - ) -> Result + std::fmt::Debug, Error> { - // Somewhat arbitrarily, expire after 90 days. - let expire_at = subscribed_at.to_owned() - TimeDelta::days(90); - - let mut tx = self.db.begin().await?; - let channel = tx - .channels() - .by_id(channel) - .await - .not_found(|| EventsError::ChannelNotFound(channel.clone()))?; - - // Subscribe before retrieving, to catch messages broadcast while we're - // querying the DB. We'll prune out duplicates later. - let live_messages = self.broadcaster.subscribe(&channel.id); - - tx.broadcast().expire(&expire_at).await?; - let stored_messages = tx.broadcast().replay(&channel, resume_at).await?; - tx.commit().await?; - - let resume_broadcast_at = stored_messages - .last() - .map(|message| message.sequence) - .or(resume_at); - - // This should always be the case, up to integer rollover, primarily - // because every message in stored_messages has a sequence not less - // than `resume_at`, or `resume_at` is None. We use the last message - // (if any) to decide when to resume the `live_messages` stream. - // - // It probably simplifies to assert!(resume_at <= resume_broadcast_at), but - // this form captures more of the reasoning. - assert!( - (resume_at.is_none() && resume_broadcast_at.is_none()) - || (stored_messages.is_empty() && resume_at == resume_broadcast_at) - || resume_at < resume_broadcast_at - ); - - // no skip_expired or resume transforms for stored_messages, as it's - // constructed not to contain messages meeting either criterion. - // - // * skip_expired is redundant with the `tx.broadcasts().expire(…)` call; - // * resume is redundant with the resume_at argument to - // `tx.broadcasts().replay(…)`. - let stored_messages = stream::iter(stored_messages); - let live_messages = live_messages - // Sure, it's temporally improbable that we'll ever skip a message - // that's 90 days old, but there's no reason not to be thorough. - .filter(Self::skip_expired(&expire_at)) - // Filtering on the broadcast resume point filters out messages - // before resume_at, and filters out messages duplicated from - // stored_messages. - .filter(Self::resume(resume_broadcast_at)); - - Ok(stored_messages.chain(live_messages)) - } - - fn resume( - resume_at: Option, - ) -> impl for<'m> FnMut(&'m broadcast::Message) -> future::Ready { - move |msg| { - future::ready(match resume_at { - None => true, - Some(resume_at) => msg.sequence > resume_at, - }) - } - } - fn skip_expired( - expire_at: &DateTime, - ) -> impl for<'m> FnMut(&'m broadcast::Message) -> future::Ready { - let expire_at = expire_at.to_owned(); - move |msg| future::ready(msg.sent_at > expire_at) - } -} - -#[derive(Debug, thiserror::Error)] -pub enum Error { - #[error("channel {0} not found")] - ChannelNotFound(channel::Id), - #[error(transparent)] - DatabaseError(#[from] sqlx::Error), -} diff --git a/src/events/app/mod.rs b/src/events/app/mod.rs deleted file mode 100644 index f364e00..0000000 --- a/src/events/app/mod.rs +++ /dev/null @@ -1,5 +0,0 @@ -mod broadcaster; -mod events; - -pub use self::broadcaster::Broadcaster; -pub use self::events::{Error as EventsError, Events}; diff --git a/src/events/broadcaster.rs b/src/events/broadcaster.rs new file mode 100644 index 0000000..6a1219a --- /dev/null +++ b/src/events/broadcaster.rs @@ -0,0 +1,116 @@ +use std::collections::{hash_map::Entry, HashMap}; +use std::sync::{Arc, Mutex, MutexGuard}; + +use futures::{future, stream::StreamExt as _, Stream}; +use sqlx::sqlite::SqlitePool; +use tokio::sync::broadcast::{channel, Sender}; +use tokio_stream::wrappers::{errors::BroadcastStreamRecvError, BroadcastStream}; + +use crate::{ + events::repo::broadcast, + repo::channel::{self, Provider as _}, +}; + +// Clones will share the same senders collection. +#[derive(Clone)] +pub struct Broadcaster { + // The use of std::sync::Mutex, and not tokio::sync::Mutex, follows Tokio's + // own advice: . Methods that + // lock it must be sync. + senders: Arc>>>, +} + +impl Broadcaster { + pub async fn from_database(db: &SqlitePool) -> Result { + let mut tx = db.begin().await?; + let channels = tx.channels().all().await?; + tx.commit().await?; + + let channels = channels.iter().map(|c| &c.id); + let broadcaster = Self::new(channels); + Ok(broadcaster) + } + + fn new<'i>(channels: impl IntoIterator) -> Self { + let senders: HashMap<_, _> = channels + .into_iter() + .cloned() + .map(|id| (id, Self::make_sender())) + .collect(); + + Self { + senders: Arc::new(Mutex::new(senders)), + } + } + + // panic: if ``channel`` is already registered. + pub fn register_channel(&self, channel: &channel::Id) { + match self.senders().entry(channel.clone()) { + // This ever happening indicates a serious logic error. + Entry::Occupied(_) => panic!("duplicate channel registration for channel {channel}"), + Entry::Vacant(entry) => { + entry.insert(Self::make_sender()); + } + } + } + + // panic: if ``channel`` has not been previously registered, and was not + // part of the initial set of channels. + pub fn broadcast(&self, channel: &channel::Id, message: &broadcast::Message) { + let tx = self.sender(channel); + + // Per the Tokio docs, the returned error is only used to indicate that + // there are no receivers. In this use case, that's fine; a lack of + // listening consumers (chat clients) when a message is sent isn't an + // error. + // + // The successful return value, which includes the number of active + // receivers, also isn't that interesting to us. + let _ = tx.send(message.clone()); + } + + // panic: if ``channel`` has not been previously registered, and was not + // part of the initial set of channels. + pub fn subscribe( + &self, + channel: &channel::Id, + ) -> impl Stream + std::fmt::Debug { + let rx = self.sender(channel).subscribe(); + + BroadcastStream::from(rx) + .take_while(|r| { + future::ready(match r { + Ok(_) => true, + // Stop the stream here. This will disconnect SSE clients + // (see `routes.rs`), who will then resume from + // `Last-Event-ID`, allowing them to catch up by reading + // the skipped messages from the database. + Err(BroadcastStreamRecvError::Lagged(_)) => false, + }) + }) + .map(|r| { + // Since the previous transform stops at the first error, this + // should always hold. + // + // See also . + r.expect("after filtering, only `Ok` messages should remain") + }) + } + + // panic: if ``channel`` has not been previously registered, and was not + // part of the initial set of channels. + fn sender(&self, channel: &channel::Id) -> Sender { + self.senders()[channel].clone() + } + + fn senders(&self) -> MutexGuard>> { + self.senders.lock().unwrap() // propagate panics when mutex is poisoned + } + + fn make_sender() -> Sender { + // Queue depth of 16 chosen entirely arbitrarily. Don't read too much + // into it. + let (tx, _) = channel(16); + tx + } +} diff --git a/src/events/mod.rs b/src/events/mod.rs index e76d67c..b9f3f5b 100644 --- a/src/events/mod.rs +++ b/src/events/mod.rs @@ -1,4 +1,5 @@ pub mod app; +pub mod broadcaster; mod extract; pub mod repo; mod routes; -- cgit v1.2.3