diff options
Diffstat (limited to 'src/channel')
| -rw-r--r-- | src/channel/app.rs | 111 | ||||
| -rw-r--r-- | src/channel/mod.rs | 1 | ||||
| -rw-r--r-- | src/channel/repo/broadcast.rs | 121 | ||||
| -rw-r--r-- | src/channel/repo/mod.rs | 1 |
4 files changed, 4 insertions, 230 deletions
diff --git a/src/channel/app.rs b/src/channel/app.rs index 8ae0c3c..f9a75d7 100644 --- a/src/channel/app.rs +++ b/src/channel/app.rs @@ -1,6 +1,3 @@ -use std::collections::{hash_map::Entry, HashMap}; -use std::sync::{Arc, Mutex, MutexGuard}; - use chrono::TimeDelta; use futures::{ future, @@ -8,12 +5,13 @@ use futures::{ Stream, }; use sqlx::sqlite::SqlitePool; -use tokio::sync::broadcast::{channel, Sender}; -use tokio_stream::wrappers::{errors::BroadcastStreamRecvError, BroadcastStream}; -use super::repo::broadcast::{self, Provider as _}; use crate::{ clock::DateTime, + events::{ + app::Broadcaster, + repo::broadcast::{self, Provider as _}, + }, repo::{ channel::{self, Channel, Provider as _}, error::NotFound as _, @@ -158,104 +156,3 @@ pub enum EventsError { #[error(transparent)] DatabaseError(#[from] sqlx::Error), } - -// 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: <https://tokio.rs/tokio/tutorial/shared-state>. Methods that - // lock it must be sync. - senders: Arc<Mutex<HashMap<channel::Id, Sender<broadcast::Message>>>>, -} - -impl Broadcaster { - pub async fn from_database(db: &SqlitePool) -> Result<Self, sqlx::Error> { - 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<Item = &'i channel::Id>) -> 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); - } - - // panic: if ``channel`` has not been previously registered, and was not - // part of the initial set of channels. - pub fn listen(&self, channel: &channel::Id) -> impl Stream<Item = broadcast::Message> { - 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 <https://users.rust-lang.org/t/taking-from-stream-while-ok/48854>. - 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<broadcast::Message> { - self.senders()[channel].clone() - } - - fn senders(&self) -> MutexGuard<HashMap<channel::Id, Sender<broadcast::Message>>> { - self.senders.lock().unwrap() // propagate panics when mutex is poisoned - } - - fn make_sender() -> Sender<broadcast::Message> { - // Queue depth of 16 chosen entirely arbitrarily. Don't read too much - // into it. - let (tx, _) = channel(16); - tx - } -} diff --git a/src/channel/mod.rs b/src/channel/mod.rs index f67ea04..9f79dbb 100644 --- a/src/channel/mod.rs +++ b/src/channel/mod.rs @@ -1,5 +1,4 @@ pub mod app; -pub mod repo; mod routes; pub use self::routes::router; diff --git a/src/channel/repo/broadcast.rs b/src/channel/repo/broadcast.rs deleted file mode 100644 index 182203a..0000000 --- a/src/channel/repo/broadcast.rs +++ /dev/null @@ -1,121 +0,0 @@ -use sqlx::{sqlite::Sqlite, SqliteConnection, Transaction}; - -use crate::{ - clock::DateTime, - repo::{ - channel::Channel, - login::{self, Login}, - message, - }, -}; - -pub trait Provider { - fn broadcast(&mut self) -> Broadcast; -} - -impl<'c> Provider for Transaction<'c, Sqlite> { - fn broadcast(&mut self) -> Broadcast { - Broadcast(self) - } -} - -pub struct Broadcast<'t>(&'t mut SqliteConnection); - -#[derive(Clone, Debug, serde::Serialize)] -pub struct Message { - pub id: message::Id, - pub sender: Login, - pub body: String, - pub sent_at: DateTime, -} - -impl<'c> Broadcast<'c> { - pub async fn create( - &mut self, - sender: &Login, - channel: &Channel, - body: &str, - sent_at: &DateTime, - ) -> Result<Message, sqlx::Error> { - let id = message::Id::generate(); - - let message = sqlx::query!( - r#" - insert into message - (id, sender, channel, body, sent_at) - values ($1, $2, $3, $4, $5) - returning - id as "id: message::Id", - sender as "sender: login::Id", - body, - sent_at as "sent_at: DateTime" - "#, - id, - sender.id, - channel.id, - body, - sent_at, - ) - .map(|row| Message { - id: row.id, - sender: sender.clone(), - body: row.body, - sent_at: row.sent_at, - }) - .fetch_one(&mut *self.0) - .await?; - - Ok(message) - } - - pub async fn expire(&mut self, expire_at: &DateTime) -> Result<(), sqlx::Error> { - sqlx::query!( - r#" - delete from message - where sent_at < $1 - "#, - expire_at, - ) - .execute(&mut *self.0) - .await?; - - Ok(()) - } - - pub async fn replay( - &mut self, - channel: &Channel, - resume_at: Option<&DateTime>, - ) -> Result<Vec<Message>, sqlx::Error> { - let messages = sqlx::query!( - r#" - select - message.id as "id: message::Id", - login.id as "sender_id: login::Id", - login.name as sender_name, - message.body, - message.sent_at as "sent_at: DateTime" - from message - join login on message.sender = login.id - where channel = $1 - and coalesce(sent_at > $2, true) - order by sent_at asc - "#, - channel.id, - resume_at, - ) - .map(|row| Message { - id: row.id, - sender: Login { - id: row.sender_id, - name: row.sender_name, - }, - body: row.body, - sent_at: row.sent_at, - }) - .fetch_all(&mut *self.0) - .await?; - - Ok(messages) - } -} diff --git a/src/channel/repo/mod.rs b/src/channel/repo/mod.rs deleted file mode 100644 index 2ed3062..0000000 --- a/src/channel/repo/mod.rs +++ /dev/null @@ -1 +0,0 @@ -pub mod broadcast; |