diff --git a/changelog.d/+7fd21c4a.bugfix.md b/changelog.d/+7fd21c4a.bugfix.md new file mode 100644 index 000000000..286f6ba3e --- /dev/null +++ b/changelog.d/+7fd21c4a.bugfix.md @@ -0,0 +1 @@ +Fixed typing notifications not interrupting a sync long poll. Contributed by @eleboucher. diff --git a/src/api/client/sync/v5.rs b/src/api/client/sync/v5.rs index 3ea868947..7dbd5aa4c 100644 --- a/src/api/client/sync/v5.rs +++ b/src/api/client/sync/v5.rs @@ -63,7 +63,6 @@ type KnownRoomUpdates = BTreeMap>; struct SyncCollection { response: sync_events::v5::Response, known_room_updates: KnownRoomUpdates, - todo_rooms: TodoRooms, } /// `POST /_matrix/client/unstable/org.matrix.simplified_msc3575/sync` @@ -148,10 +147,6 @@ pub(crate) async fn sync_events_v5_route( } } - // Typing never wakes a sync loop, so collect it after the long poll settles. - collection.response.extensions.typing = - collect_typing_events(services, sender_user, &body, &collection.todo_rooms).await?; - commit_sync_collection( services, sender_user, @@ -290,7 +285,10 @@ async fn collect_sync_response( ) .await?; - Ok(SyncCollection { response, known_room_updates, todo_rooms }) + response.extensions.typing = + collect_typing_events(services, sender_user, body, &todo_rooms).await?; + + Ok(SyncCollection { response, known_room_updates }) } fn response_is_empty(response: &sync_events::v5::Response) -> bool { diff --git a/src/service/rooms/typing/mod.rs b/src/service/rooms/typing/mod.rs index a71a4c58f..6a3562c2c 100644 --- a/src/service/rooms/typing/mod.rs +++ b/src/service/rooms/typing/mod.rs @@ -1,7 +1,7 @@ use std::{collections::BTreeMap, sync::Arc}; use conduwuit::{ - Result, Server, debug_info, trace, + Result, Server, debug_info, utils::{self, IterStream}, }; use futures::StreamExt; @@ -10,9 +10,9 @@ use ruma::{ api::federation::transactions::edu::{Edu, TypingContent}, events::{SyncEphemeralRoomEvent, typing::TypingEventContent}, }; -use tokio::sync::{RwLock, broadcast}; +use tokio::sync::RwLock; -use crate::{Dep, globals, sending, sending::EduBuf, users}; +use crate::{Dep, globals, sending, sending::EduBuf, sync, users}; pub struct Service { server: Arc, @@ -21,12 +21,12 @@ pub struct Service { pub typing: RwLock>>, /// timestamp of the last change to typing users pub last_typing_update: RwLock>, - pub typing_update_sender: broadcast::Sender, } struct Services { globals: Dep, sending: Dep, + sync: Dep, users: Dep, } @@ -37,11 +37,11 @@ impl crate::Service for Service { services: Services { globals: args.depend::("globals"), sending: args.depend::("sending"), + sync: args.depend::("sync"), users: args.depend::("users"), }, typing: RwLock::new(BTreeMap::new()), last_typing_update: RwLock::new(BTreeMap::new()), - typing_update_sender: broadcast::channel(100).0, })) } @@ -71,9 +71,7 @@ impl Service { .await .insert(room_id.to_owned(), self.services.globals.next_count()?); - if self.typing_update_sender.send(room_id.to_owned()).is_err() { - trace!("receiver found what it was looking for and is no longer interested"); - } + self.services.sync.wake_all_joined(room_id).await; // update federation if self.services.globals.user_is_local(user_id) { @@ -99,9 +97,7 @@ impl Service { .await .insert(room_id.to_owned(), self.services.globals.next_count()?); - if self.typing_update_sender.send(room_id.to_owned()).is_err() { - trace!("receiver found what it was looking for and is no longer interested"); - } + self.services.sync.wake_all_joined(room_id).await; // update federation if self.services.globals.user_is_local(user_id) { @@ -111,15 +107,6 @@ impl Service { Ok(()) } - pub async fn wait_for_update(&self, room_id: &RoomId) { - let mut receiver = self.typing_update_sender.subscribe(); - while let Ok(next) = receiver.recv().await { - if next == room_id { - break; - } - } - } - /// Makes sure that typing events with old timestamps get removed. async fn typings_maintain(&self, room_id: &RoomId) -> Result<()> { let current_timestamp = utils::millis_since_unix_epoch(); @@ -152,9 +139,7 @@ impl Service { .await .insert(room_id.to_owned(), self.services.globals.next_count()?); - if self.typing_update_sender.send(room_id.to_owned()).is_err() { - trace!("receiver found what it was looking for and is no longer interested"); - } + self.services.sync.wake_all_joined(room_id).await; // update federation for user in &removable {