diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/app.rs | 10 | ||||
| -rw-r--r-- | src/channel/app.rs | 129 | ||||
| -rw-r--r-- | src/channel/routes.rs | 5 | ||||
| -rw-r--r-- | src/channel/routes/test/on_send.rs | 11 | ||||
| -rw-r--r-- | src/events/app/broadcaster.rs (renamed from src/events/app.rs) | 8 | ||||
| -rw-r--r-- | src/events/app/events.rs | 138 | ||||
| -rw-r--r-- | src/events/app/mod.rs | 5 | ||||
| -rw-r--r-- | src/events/extract.rs (renamed from src/header.rs) | 0 | ||||
| -rw-r--r-- | src/events/mod.rs | 1 | ||||
| -rw-r--r-- | src/events/routes.rs | 12 | ||||
| -rw-r--r-- | src/events/routes/test.rs | 3 | ||||
| -rw-r--r-- | src/lib.rs | 1 | ||||
| -rw-r--r-- | src/test/fixtures/message.rs | 2 |
13 files changed, 175 insertions, 150 deletions
@@ -1,6 +1,10 @@ use sqlx::sqlite::SqlitePool; -use crate::{channel::app::Channels, events::app::Broadcaster, login::app::Logins}; +use crate::{ + channel::app::Channels, + events::app::{Broadcaster, Events}, + login::app::Logins, +}; #[derive(Clone)] pub struct App { @@ -20,6 +24,10 @@ impl App { Logins::new(&self.db) } + pub const fn events(&self) -> Events { + Events::new(&self.db, &self.broadcaster) + } + pub const fn channels(&self) -> Channels { Channels::new(&self.db, &self.broadcaster) } diff --git a/src/channel/app.rs b/src/channel/app.rs index 2da25d2..bb87734 100644 --- a/src/channel/app.rs +++ b/src/channel/app.rs @@ -1,22 +1,8 @@ -use chrono::TimeDelta; -use futures::{ - future, - stream::{self, StreamExt as _}, - Stream, -}; use sqlx::sqlite::SqlitePool; use crate::{ - clock::DateTime, - events::{ - app::Broadcaster, - repo::broadcast::{self, Provider as _}, - }, - repo::{ - channel::{self, Channel, Provider as _}, - error::NotFound as _, - login::Login, - }, + events::app::Broadcaster, + repo::channel::{Channel, Provider as _}, }; pub struct Channels<'a> { @@ -49,107 +35,6 @@ impl<'a> Channels<'a> { Ok(channels) } - - pub async fn send( - &self, - login: &Login, - channel: &channel::Id, - body: &str, - sent_at: &DateTime, - ) -> Result<broadcast::Message, EventsError> { - 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 events( - &self, - channel: &channel::Id, - subscribed_at: &DateTime, - resume_at: Option<broadcast::Sequence>, - ) -> Result<impl Stream<Item = broadcast::Message> + 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.listen(&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<broadcast::Sequence>, - ) -> impl for<'m> FnMut(&'m broadcast::Message) -> future::Ready<bool> { - 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<bool> { - let expire_at = expire_at.to_owned(); - move |msg| future::ready(msg.sent_at > expire_at) - } } #[derive(Debug, thiserror::Error)] @@ -177,13 +62,3 @@ pub enum InternalError { #[error(transparent)] DatabaseError(#[from] sqlx::Error), } - -#[derive(Debug, thiserror::Error)] -pub enum EventsError { - #[error("channel {0} not found")] - ChannelNotFound(channel::Id), - #[error(transparent)] - ResumeAtError(#[from] chrono::ParseError), - #[error(transparent)] - DatabaseError(#[from] sqlx::Error), -} diff --git a/src/channel/routes.rs b/src/channel/routes.rs index 674c876..bb6cde6 100644 --- a/src/channel/routes.rs +++ b/src/channel/routes.rs @@ -6,11 +6,12 @@ use axum::{ Router, }; -use super::app::{self, EventsError}; +use super::app; use crate::{ app::App, clock::RequestedAt, error::InternalError, + events::app::EventsError, repo::{ channel::{self, Channel}, login::Login, @@ -89,7 +90,7 @@ async fn on_send( login: Login, Json(request): Json<SendRequest>, ) -> Result<StatusCode, ErrorResponse> { - app.channels() + app.events() .send(&login, &channel, &request.message, &sent_at) .await // Could impl `From` here, but it's more code and this is used once. diff --git a/src/channel/routes/test/on_send.rs b/src/channel/routes/test/on_send.rs index eab7c32..6690374 100644 --- a/src/channel/routes/test/on_send.rs +++ b/src/channel/routes/test/on_send.rs @@ -5,7 +5,8 @@ use axum::{ use futures::stream::StreamExt; use crate::{ - channel::{app, routes}, + channel::routes, + events::app, repo::channel, test::fixtures::{self, future::Immediately as _}, }; @@ -42,8 +43,8 @@ async fn channel_exists() { let subscribed_at = fixtures::now(); let mut events = app - .channels() - .events(&channel.id, &subscribed_at, None) + .events() + .subscribe(&channel.id, &subscribed_at, None) .await .expect("subscribing to a valid channel"); @@ -99,8 +100,8 @@ async fn messages_in_order() { let subscribed_at = fixtures::now(); let events = app - .channels() - .events(&channel.id, &subscribed_at, None) + .events() + .subscribe(&channel.id, &subscribed_at, None) .await .expect("subscribing to a valid channel") .take(requests.len()); diff --git a/src/events/app.rs b/src/events/app/broadcaster.rs index 99e849e..6a1219a 100644 --- a/src/events/app.rs +++ b/src/events/app/broadcaster.rs @@ -6,8 +6,10 @@ use sqlx::sqlite::SqlitePool; use tokio::sync::broadcast::{channel, Sender}; use tokio_stream::wrappers::{errors::BroadcastStreamRecvError, BroadcastStream}; -use super::repo::broadcast; -use crate::repo::channel::{self, Provider as _}; +use crate::{ + events::repo::broadcast, + repo::channel::{self, Provider as _}, +}; // Clones will share the same senders collection. #[derive(Clone)] @@ -69,7 +71,7 @@ impl Broadcaster { // panic: if ``channel`` has not been previously registered, and was not // part of the initial set of channels. - pub fn listen( + pub fn subscribe( &self, channel: &channel::Id, ) -> impl Stream<Item = broadcast::Message> + std::fmt::Debug { diff --git a/src/events/app/events.rs b/src/events/app/events.rs new file mode 100644 index 0000000..a8814c9 --- /dev/null +++ b/src/events/app/events.rs @@ -0,0 +1,138 @@ +use chrono::TimeDelta; +use futures::{ + future, + stream::{self, StreamExt as _}, + Stream, +}; +use sqlx::sqlite::SqlitePool; + +use super::Broadcaster; +use crate::{ + clock::DateTime, + events::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<broadcast::Message, EventsError> { + 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<broadcast::Sequence>, + ) -> Result<impl Stream<Item = broadcast::Message> + 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<broadcast::Sequence>, + ) -> impl for<'m> FnMut(&'m broadcast::Message) -> future::Ready<bool> { + 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<bool> { + 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/mod.rs b/src/events/app/mod.rs new file mode 100644 index 0000000..03a7da2 --- /dev/null +++ b/src/events/app/mod.rs @@ -0,0 +1,5 @@ +mod broadcaster; +mod events; + +pub use self::broadcaster::Broadcaster; +pub use self::events::{Events, EventsError}; diff --git a/src/header.rs b/src/events/extract.rs index 683c1f9..683c1f9 100644 --- a/src/header.rs +++ b/src/events/extract.rs diff --git a/src/events/mod.rs b/src/events/mod.rs index f67ea04..e76d67c 100644 --- a/src/events/mod.rs +++ b/src/events/mod.rs @@ -1,4 +1,5 @@ pub mod app; +mod extract; pub mod repo; mod routes; diff --git a/src/events/routes.rs b/src/events/routes.rs index 7731680..5181370 100644 --- a/src/events/routes.rs +++ b/src/events/routes.rs @@ -16,13 +16,12 @@ use futures::{ stream::{self, Stream, StreamExt as _, TryStreamExt as _}, }; -use super::repo::broadcast; +use super::{extract::LastEventId, repo::broadcast}; use crate::{ app::App, - channel::app::EventsError, clock::RequestedAt, error::InternalError, - header::LastEventId, + events::app::EventsError, repo::{channel, login::Login}, }; @@ -67,8 +66,8 @@ async fn events( let resume_at = resume_at.get(&channel).copied(); let events = app - .channels() - .events(&channel, &now, resume_at) + .events() + .subscribe(&channel, &now, resume_at) .await? .map(ChannelEvent::wrap(channel)); @@ -122,9 +121,6 @@ impl IntoResponse for ErrorResponse { not_found @ EventsError::ChannelNotFound(_) => { (StatusCode::NOT_FOUND, not_found.to_string()).into_response() } - resume_at @ EventsError::ResumeAtError(_) => { - (StatusCode::BAD_REQUEST, resume_at.to_string()).into_response() - } other => InternalError::from(other).into_response(), } } diff --git a/src/events/routes/test.rs b/src/events/routes/test.rs index 131c751..d3f3fd6 100644 --- a/src/events/routes/test.rs +++ b/src/events/routes/test.rs @@ -6,8 +6,7 @@ use futures::{ }; use crate::{ - channel::app, - events::routes, + events::{app, routes}, repo::channel::{self}, test::fixtures::{self, future::Immediately as _}, }; @@ -4,7 +4,6 @@ pub mod cli; mod clock; mod error; mod events; -mod header; mod id; mod login; mod password; diff --git a/src/test/fixtures/message.rs b/src/test/fixtures/message.rs index 7fe3cb9..33feeae 100644 --- a/src/test/fixtures/message.rs +++ b/src/test/fixtures/message.rs @@ -15,7 +15,7 @@ pub async fn send( ) -> broadcast::Message { let body = propose(); - app.channels() + app.events() .send(login, &channel.id, &body, sent_at) .await .expect("should succeed if the channel exists") |