Merge branch 'lock_requests'

matrix_sdk/src/client.rs

@@ -88,16 +88,15 @@ use matrix_sdk_common::{
 };
 
 #[cfg(feature = "encryption")]
-use matrix_sdk_common::{
+use matrix_sdk_common::api::r0::{
-    api::r0::{
+    keys::{get_keys, upload_keys, upload_signing_keys::Request as UploadSigningKeysRequest},
-        keys::{get_keys, upload_keys, upload_signing_keys::Request as UploadSigningKeysRequest},
+    to_device::send_event_to_device::{
-        to_device::send_event_to_device::{
+        Request as RumaToDeviceRequest, Response as ToDeviceResponse,
-            Request as RumaToDeviceRequest, Response as ToDeviceResponse,
-        },
     },
-    locks::Mutex,
 };
 
+use matrix_sdk_common::locks::Mutex;
+
 use crate::{
     error::HttpError,
     http_client::{client_with_config, HttpClient, HttpSend},
@@ -134,10 +133,12 @@ pub struct Client {
     /// Locks making sure we only have one group session sharing request in
     /// flight per room.
     #[cfg(feature = "encryption")]
-    pub(crate) group_session_locks: DashMap<RoomId, Arc<Mutex<()>>>,
+    pub(crate) group_session_locks: Arc<DashMap<RoomId, Arc<Mutex<()>>>>,
     #[cfg(feature = "encryption")]
     /// Lock making sure we're only doing one key claim request at a time.
     key_claim_lock: Arc<Mutex<()>>,
+    pub(crate) members_request_locks: Arc<DashMap<RoomId, Arc<Mutex<()>>>>,
+    pub(crate) typing_notice_times: Arc<DashMap<RoomId, Instant>>,
 }
 
 #[cfg(not(tarpaulin_include))]
@@ -390,9 +391,11 @@ impl Client {
             http_client,
             base_client,
             #[cfg(feature = "encryption")]
-            group_session_locks: DashMap::new(),
+            group_session_locks: Arc::new(DashMap::new()),
             #[cfg(feature = "encryption")]
             key_claim_lock: Arc::new(Mutex::new(())),
+            members_request_locks: Arc::new(DashMap::new()),
+            typing_notice_times: Arc::new(DashMap::new()),
         })
     }
 
@@ -1668,7 +1671,6 @@ impl Client {
     /// # use std::{path::PathBuf, time::Duration};
     /// # use matrix_sdk::{
     /// #     Client, SyncSettings,
-    /// #     api::r0::typing::create_typing_event::Typing,
     /// #     identifiers::room_id,
     /// # };
     /// # use futures::executor::block_on;
@@ -1746,7 +1748,6 @@ impl Client {
     /// # use std::{path::PathBuf, time::Duration};
     /// # use matrix_sdk::{
     /// #     Client, SyncSettings,
-    /// #     api::r0::typing::create_typing_event::Typing,
     /// #     identifiers::room_id,
     /// # };
     /// # use futures::executor::block_on;
@@ -1801,7 +1802,7 @@ mod test {
         api::r0::{
             account::register::Request as RegistrationRequest,
             directory::get_public_rooms_filtered::Request as PublicRoomsFilterRequest,
-            membership::Invite3pid, typing::create_typing_event::Typing, uiaa::AuthData,
+            membership::Invite3pid, uiaa::AuthData,
         },
         assign,
         directory::Filter,
@@ -2450,9 +2451,7 @@ mod test {
             .get_joined_room(&room_id!("!SVkFJHzfwvuaIEawgC:localhost"))
             .unwrap();
 
-        room.typing_notice(Typing::Yes(std::time::Duration::from_secs(1)))
+        room.typing_notice(true).await.unwrap();
-            .await
-            .unwrap();
     }
 
     #[tokio::test]

matrix_sdk/src/room/common.rs

@@ -2,7 +2,9 @@ use matrix_sdk_common::api::r0::{
     membership::{get_member_events, join_room_by_id, leave_room},
     message::get_message_events,
 };
-use std::ops::Deref;
+use matrix_sdk_common::locks::Mutex;
+
+use std::{ops::Deref, sync::Arc};
 
 use crate::{Client, Result, Room, RoomMember};
 
@@ -96,14 +98,34 @@ impl Common {
     }
 
     pub(crate) async fn request_members(&self) -> Result<()> {
-        // TODO: don't send a request if a request is being sent
+        #[allow(clippy::map_clone)]
-        let request = get_member_events::Request::new(self.inner.room_id());
+        if let Some(mutex) = self
-        let response = self.client.send(request, None).await?;
+            .client
+            .members_request_locks
+            .get(self.inner.room_id())
+            .map(|m| m.clone())
+        {
+            mutex.lock().await;
+        } else {
+            let mutex = Arc::new(Mutex::new(()));
+            self.client
+                .members_request_locks
+                .insert(self.inner.room_id().clone(), mutex.clone());
 
-        self.client
+            let _guard = mutex.lock().await;
-            .base_client
+
-            .receive_members(self.inner.room_id(), &response)
+            let request = get_member_events::Request::new(self.inner.room_id());
-            .await?;
+            let response = self.client.send(request, None).await?;
+
+            self.client
+                .base_client
+                .receive_members(self.inner.room_id(), &response)
+                .await?;
+
+            self.client
+                .members_request_locks
+                .remove(self.inner.room_id());
+        }
 
         Ok(())
     }

matrix_sdk/src/room/joined.rs

@@ -27,6 +27,7 @@ use matrix_sdk_common::{
         AnyMessageEventContent, AnyStateEventContent,
     },
     identifiers::{EventId, UserId},
+    instant::{Duration, Instant},
     uuid::Uuid,
 };
 
@@ -41,6 +42,9 @@ use matrix_sdk_base::crypto::AttachmentEncryptor;
 #[cfg(feature = "encryption")]
 use tracing::instrument;
 
+const TYPING_NOTICE_TIMEOUT: Duration = Duration::from_secs(4);
+const TYPING_NOTICE_RESEND_TIMEOUT: Duration = Duration::from_secs(3);
+
 /// A room in the joined state.
 ///
 /// The `JoinedRoom` contains all methodes specific to a `Room` with type `RoomType::Joined`.
@@ -138,11 +142,16 @@ impl Joined {
         Ok(())
     }
 
-    /// Send a request to notify this room of a user typing.
+    /// Activate typing notice for this room.
+    ///
+    /// The typing notice remains active for 4s. It can be deactivate at any point by setting
+    /// typing to `false`. If this method is called while the typing notice is active nothing will happen.
+    /// This method can be called on every key stroke, since it will do nothing while typing is
+    /// active.
     ///
     /// # Arguments
     ///
-    /// * `typing` - Whether the user is typing, and how long.
+    /// * `typing` - Whether the user is typing or has stopped typing.
     ///
     /// # Examples
     ///
@@ -164,20 +173,46 @@ impl Joined {
     /// #    .unwrap();
     ///
     /// room
-    ///     .typing_notice(Typing::Yes(Duration::from_secs(4)))
+    ///     .typing_notice(true)
     ///     .await
     ///     .expect("Can't get devices from server");
     /// # });
     /// ```
-    pub async fn typing_notice(&self, typing: impl Into<Typing>) -> Result<()> {
+    pub async fn typing_notice(&self, typing: bool) -> Result<()> {
-        // TODO: don't send a request if a typing notice is being sent or is already active
+        // Only send a request to the homeserver if the old timeout has elapsed or the typing
-        let request = TypingRequest::new(
+        // notice changed state within the TYPING_NOTICE_TIMEOUT
-            self.inner.own_user_id(),
+        let send =
-            self.inner.room_id(),
+            if let Some(typing_time) = self.client.typing_notice_times.get(self.inner.room_id()) {
-            typing.into(),
+                if typing_time.elapsed() > TYPING_NOTICE_RESEND_TIMEOUT {
-        );
+                    // We always reactivate the typing notice if typing is true or we may need to
+                    // deactivate it if it's currently active if typing is false
+                    typing || typing_time.elapsed() <= TYPING_NOTICE_TIMEOUT
+                } else {
+                    // Only send a request when we need to deactivate typing
+                    !typing
+                }
+            } else {
+                // Typing notice is currently deactivated, therefore, send a request only when it's
+                // about to be activated
+                typing
+            };
+
+        if send {
+            let typing = if typing {
+                self.client
+                    .typing_notice_times
+                    .insert(self.inner.room_id().clone(), Instant::now());
+                Typing::Yes(TYPING_NOTICE_TIMEOUT)
+            } else {
+                self.client.typing_notice_times.remove(self.inner.room_id());
+                Typing::No
+            };
+
+            let request =
+                TypingRequest::new(self.inner.own_user_id(), self.inner.room_id(), typing);
+            self.client.send(request, None).await?;
+        }
 
-        self.client.send(request, None).await?;
         Ok(())
     }
 
@@ -539,7 +574,7 @@ impl Joined {
     ///    avatar_url: Some(avatar_url.to_string()),
     ///    membership: MembershipState::Join,
     ///    is_direct: None,
-    ///    displayname: None,    
+    ///    displayname: None,
     ///    third_party_invite: None,
     /// };
     /// # let room = client