Reduce the size of Client

… by moving all of its fields into an inner reference-counted type
rather than using reference counting for all fields individually.
Some fields are still refcounted individually and thus go through an
extra layer of indirection, but the size and clone simplification should
still make this change worthwhile.

Author: jplatte

crates/matrix-sdk/src/client.rs

@@ -108,27 +108,31 @@ pub enum LoopCtrl {
 /// All of the state is held in an `Arc` so the `Client` can be cloned freely.
 #[derive(Clone)]
 pub struct Client {
+    pub(crate) inner: Arc<ClientInner>,
+}
+
+pub(crate) struct ClientInner {
     /// The URL of the homeserver to connect to.
     homeserver: Arc<RwLock<Url>>,
     /// The underlying HTTP client.
     http_client: HttpClient,
     /// User session data.
-    pub(crate) base_client: BaseClient,
+    base_client: BaseClient,
     /// Locks making sure we only have one group session sharing request in
     /// flight per room.
     #[cfg(feature = "encryption")]
-    pub(crate) group_session_locks: Arc<DashMap<RoomId, Arc<Mutex<()>>>>,
+    pub(crate) group_session_locks: DashMap<RoomId, Arc<Mutex<()>>>,
     #[cfg(feature = "encryption")]
     /// Lock making sure we're only doing one key claim request at a time.
-    pub(crate) key_claim_lock: Arc<Mutex<()>>,
-    pub(crate) members_request_locks: Arc<DashMap<RoomId, Arc<Mutex<()>>>>,
-    pub(crate) typing_notice_times: Arc<DashMap<RoomId, Instant>>,
+    pub(crate) key_claim_lock: Mutex<()>,
+    pub(crate) members_request_locks: DashMap<RoomId, Arc<Mutex<()>>>,
+    pub(crate) typing_notice_times: DashMap<RoomId, Instant>,
     /// Event handlers. See `register_event_handler`.
-    event_handlers: Arc<RwLock<EventHandlerMap>>,
+    event_handlers: RwLock<EventHandlerMap>,
     /// Custom event handler context. See `register_event_handler_context`.
-    event_handler_data: Arc<StdRwLock<AnyMap>>,
+    event_handler_data: StdRwLock<AnyMap>,
     /// Notification handlers. See `register_notification_handler`.
-    notification_handlers: Arc<RwLock<Vec<NotificationHandlerFn>>>,
+    notification_handlers: RwLock<Vec<NotificationHandlerFn>>,
     /// Whether the client should operate in application service style mode.
     /// This is low-level functionality. For an high-level API check the
     /// `matrix_sdk_appservice` crate.
@@ -141,7 +145,7 @@ pub struct Client {
     /// synchronization, e.g. if we send out a request to create a room, we can
     /// wait for the sync to get the data to fetch a room object from the state
     /// store.
-    pub(crate) sync_beat: Arc<event_listener::Event>,
+    pub(crate) sync_beat: event_listener::Event,
 }
 
 #[cfg(not(tarpaulin_include))]
@@ -185,7 +189,7 @@ impl Client {
         let http_client =
             HttpClient::new(client, homeserver.clone(), session, config.request_config);
 
-        Ok(Self {
+        let inner = Arc::new(ClientInner {
             homeserver,
             http_client,
             base_client,
@@ -200,8 +204,10 @@ impl Client {
             notification_handlers: Default::default(),
             appservice_mode: config.appservice_mode,
             use_discovery_response: config.use_discovery_response,
-            sync_beat: event_listener::Event::new().into(),
-        })
+            sync_beat: event_listener::Event::new(),
+        });
+
+        Ok(Self { inner })
     }
 
     /// Create a new [`Client`] using homeserver auto discovery.
@@ -267,6 +273,24 @@ impl Client {
         Ok(client)
     }
 
+    pub(crate) fn base_client(&self) -> &BaseClient {
+        &self.inner.base_client
+    }
+
+    #[cfg(feature = "encryption")]
+    pub(crate) async fn olm_machine(&self) -> Option<matrix_sdk_base::crypto::OlmMachine> {
+        self.base_client().olm_machine().await
+    }
+
+    #[cfg(feature = "encryption")]
+    pub(crate) async fn mark_request_as_sent(
+        &self,
+        request_id: &matrix_sdk_base::uuid::Uuid,
+        response: impl Into<matrix_sdk_base::crypto::IncomingResponse<'_>>,
+    ) -> Result<(), matrix_sdk_base::Error> {
+        self.base_client().mark_request_as_sent(request_id, response).await
+    }
+
     fn homeserver_from_user_id(user_id: &UserId) -> Result<Url> {
         let homeserver = format!("https://{}", user_id.server_name());
         #[allow(unused_mut)]
@@ -289,7 +313,7 @@ impl Client {
     ///
     /// * `homeserver_url` - The new URL to use.
     pub async fn set_homeserver(&self, homeserver_url: Url) {
-        let mut homeserver = self.homeserver.write().await;
+        let mut homeserver = self.inner.homeserver.write().await;
         *homeserver = homeserver_url;
     }
 
@@ -323,23 +347,23 @@ impl Client {
 
     /// Is the client logged in.
     pub async fn logged_in(&self) -> bool {
-        self.base_client.logged_in().await
+        self.inner.base_client.logged_in().await
     }
 
     /// The Homeserver of the client.
     pub async fn homeserver(&self) -> Url {
-        self.homeserver.read().await.clone()
+        self.inner.homeserver.read().await.clone()
     }
 
     /// Get the user id of the current owner of the client.
     pub async fn user_id(&self) -> Option<UserId> {
-        let session = self.base_client.session().read().await;
+        let session = self.inner.base_client.session().read().await;
         session.as_ref().cloned().map(|s| s.user_id)
     }
 
     /// Get the device id that identifies the current session.
     pub async fn device_id(&self) -> Option<DeviceIdBox> {
-        let session = self.base_client.session().read().await;
+        let session = self.inner.base_client.session().read().await;
         session.as_ref().map(|s| s.device_id.clone())
     }
 
@@ -350,7 +374,7 @@ impl Client {
     /// Can be used with [`Client::restore_login`] to restore a previously
     /// logged in session.
     pub async fn session(&self) -> Option<Session> {
-        self.base_client.session().read().await.clone()
+        self.inner.base_client.session().read().await.clone()
     }
 
     /// Fetches the display name of the owner of the client.
@@ -468,7 +492,7 @@ impl Client {
 
     /// Get a reference to the store.
     pub fn store(&self) -> &Store {
-        self.base_client.store()
+        self.inner.base_client.store()
     }
 
     /// Sets the mxc avatar url of the client's owner. The avatar gets unset if
@@ -610,34 +634,39 @@ impl Client {
         <H::Future as Future>::Output: EventHandlerResult,
     {
         let event_type = H::ID.1;
-        self.event_handlers.write().await.entry(H::ID).or_default().push(Box::new(move |data| {
-            let maybe_fut = serde_json::from_str(data.raw.get())
-                .map(|ev| handler.clone().handle_event(ev, data));
-
-            Box::pin(async move {
-                match maybe_fut {
-                    Ok(Some(fut)) => {
-                        fut.await.print_error(event_type);
-                    }
-                    Ok(None) => {
-                        error!("Event handler for {} has an invalid context argument", event_type);
-                    }
-                    Err(e) => {
-                        warn!(
-                            "Failed to deserialize `{}` event, skipping event handler.\n\
+        self.inner.event_handlers.write().await.entry(H::ID).or_default().push(Box::new(
+            move |data| {
+                let maybe_fut = serde_json::from_str(data.raw.get())
+                    .map(|ev| handler.clone().handle_event(ev, data));
+
+                Box::pin(async move {
+                    match maybe_fut {
+                        Ok(Some(fut)) => {
+                            fut.await.print_error(event_type);
+                        }
+                        Ok(None) => {
+                            error!(
+                                "Event handler for {} has an invalid context argument",
+                                event_type
+                            );
+                        }
+                        Err(e) => {
+                            warn!(
+                                "Failed to deserialize `{}` event, skipping event handler.\n\
                                  Deserialization error: {}",
-                            event_type, e,
-                        );
+                                event_type, e,
+                            );
+                        }
                     }
-                }
-            })
-        }));
+                })
+            },
+        ));
 
         self
     }
 
     pub(crate) async fn event_handlers(&self) -> RwLockReadGuard<'_, EventHandlerMap> {
-        self.event_handlers.read().await
+        self.inner.event_handlers.read().await
     }
 
     /// Add an arbitrary value for use as event handler context.
@@ -681,15 +710,15 @@ impl Client {
     where
         T: Clone + Send + Sync + 'static,
     {
-        self.event_handler_data.write().unwrap().insert(ctx);
+        self.inner.event_handler_data.write().unwrap().insert(ctx);
         self
     }
 
     pub(crate) fn event_handler_context<T>(&self) -> Option<T>
     where
         T: Clone + Send + Sync + 'static,
     {
-        let map = self.event_handler_data.read().unwrap();
+        let map = self.inner.event_handler_data.read().unwrap();
         map.get::<T>().cloned()
     }
 
@@ -703,7 +732,7 @@ impl Client {
         H: Fn(Notification, room::Room, Client) -> Fut + Send + Sync + 'static,
         Fut: Future<Output = ()> + Send + 'static,
     {
-        self.notification_handlers.write().await.push(Box::new(
+        self.inner.notification_handlers.write().await.push(Box::new(
             move |notification, room, client| Box::pin((handler)(notification, room, client)),
         ));
 
@@ -713,7 +742,7 @@ impl Client {
     pub(crate) async fn notification_handlers(
         &self,
     ) -> RwLockReadGuard<'_, Vec<NotificationHandlerFn>> {
-        self.notification_handlers.read().await
+        self.inner.notification_handlers.read().await
     }
 
     /// Get all the rooms the client knows about.
@@ -1183,15 +1212,15 @@ impl Client {
     ///
     /// * `response` - A successful login response.
     async fn receive_login_response(&self, response: &login::Response) -> Result<()> {
-        if self.use_discovery_response {
+        if self.inner.use_discovery_response {
             if let Some(well_known) = &response.well_known {
                 if let Ok(homeserver) = Url::parse(&well_known.homeserver.base_url) {
                     self.set_homeserver(homeserver).await;
                 }
             }
         }
 
-        self.base_client.receive_login_response(response).await?;
+        self.inner.base_client.receive_login_response(response).await?;
 
         Ok(())
     }
@@ -1254,7 +1283,7 @@ impl Client {
     ///
     /// [`login`]: #method.login
     pub async fn restore_login(&self, session: Session) -> Result<()> {
-        Ok(self.base_client.restore_login(session).await?)
+        Ok(self.inner.base_client.restore_login(session).await?)
     }
 
     /// Register a user to the server.
@@ -1301,8 +1330,8 @@ impl Client {
         let homeserver = self.homeserver().await;
         info!("Registering to {}", homeserver);
 
-        let config = if self.appservice_mode {
-            Some(self.http_client.request_config.force_auth())
+        let config = if self.inner.appservice_mode {
+            Some(self.inner.http_client.request_config.force_auth())
         } else {
             None
         };
@@ -1367,14 +1396,14 @@ impl Client {
         filter_name: &str,
         definition: FilterDefinition<'_>,
     ) -> Result<String> {
-        if let Some(filter) = self.base_client.get_filter(filter_name).await? {
+        if let Some(filter) = self.inner.base_client.get_filter(filter_name).await? {
             Ok(filter)
         } else {
             let user_id = self.user_id().await.ok_or(Error::AuthenticationRequired)?;
             let request = FilterUploadRequest::new(&user_id, definition);
             let response = self.send(request, None).await?;
 
-            self.base_client.receive_filter_upload(filter_name, &response).await?;
+            self.inner.base_client.receive_filter_upload(filter_name, &response).await?;
 
             Ok(response.filter_id)
         }
@@ -1586,8 +1615,8 @@ impl Client {
             content_type: Some(content_type.essence_str()),
         });
 
-        let request_config = self.http_client.request_config.timeout(timeout);
-        Ok(self.http_client.upload(request, Some(request_config)).await?)
+        let request_config = self.inner.http_client.request_config.timeout(timeout);
+        Ok(self.inner.http_client.upload(request, Some(request_config)).await?)
     }
 
     /// Send an arbitrary request to the server, without updating client state.
@@ -1639,7 +1668,7 @@ impl Client {
         Request: OutgoingRequest + Debug,
         HttpError: From<FromHttpResponseError<Request::EndpointError>>,
     {
-        Ok(self.http_client.send(request, config).await?)
+        Ok(self.inner.http_client.send(request, config).await?)
     }
 
     /// Get information of all our own devices.
@@ -1832,9 +1861,9 @@ impl Client {
             timeout: sync_settings.timeout,
         });
 
-        let request_config = self.http_client.request_config.timeout(
+        let request_config = self.inner.http_client.request_config.timeout(
             sync_settings.timeout.unwrap_or_else(|| Duration::from_secs(0))
-                + self.http_client.request_config.timeout,
+                + self.inner.http_client.request_config.timeout,
         );
 
         let response = self.send(request, Some(request_config)).await?;
@@ -1845,7 +1874,7 @@ impl Client {
             error!(error =? e, "Error while sending outgoing E2EE requests");
         };
 
-        self.sync_beat.notify(usize::MAX);
+        self.inner.sync_beat.notify(usize::MAX);
 
         Ok(response)
     }
@@ -2050,7 +2079,7 @@ impl Client {
     /// Get the current, if any, sync token of the client.
     /// This will be None if the client didn't sync at least once.
     pub async fn sync_token(&self) -> Option<String> {
-        self.base_client.sync_token().await
+        self.inner.base_client.sync_token().await
     }
 
     /// Get a media file's content.
@@ -2069,7 +2098,7 @@ impl Client {
         use_cache: bool,
     ) -> Result<Vec<u8>> {
         let content = if use_cache {
-            self.base_client.store().get_media_content(request).await?
+            self.inner.base_client.store().get_media_content(request).await?
         } else {
             None
         };
@@ -2113,7 +2142,7 @@ impl Client {
             };
 
             if use_cache {
-                self.base_client.store().add_media_content(request, content.clone()).await?;
+                self.inner.base_client.store().add_media_content(request, content.clone()).await?;
             }
 
             Ok(content)
@@ -2126,7 +2155,7 @@ impl Client {
     ///
     /// * `request` - The `MediaRequest` of the content.
     pub async fn remove_media_content(&self, request: &MediaRequest) -> Result<()> {
-        Ok(self.base_client.store().remove_media_content(request).await?)
+        Ok(self.inner.base_client.store().remove_media_content(request).await?)
     }
 
     /// Delete all the media content corresponding to the given
@@ -2136,7 +2165,7 @@ impl Client {
     ///
     /// * `uri` - The `MxcUri` of the files.
     pub async fn remove_media_content_for_uri(&self, uri: &MxcUri) -> Result<()> {
-        Ok(self.base_client.store().remove_media_content_for_uri(uri).await?)
+        Ok(self.inner.base_client.store().remove_media_content_for_uri(uri).await?)
     }
 
     /// Get the file of the given media event content.
@@ -2655,7 +2684,7 @@ pub(crate) mod test {
             .add_state_event(EventsJson::PowerLevels)
             .build_sync_response();
 
-        client.base_client.receive_sync_response(response).await.unwrap();
+        client.inner.base_client.receive_sync_response(response).await.unwrap();
         let room_id = room_id!("!SVkFJHzfwvuaIEawgC:localhost");
 
         assert_eq!(client.homeserver().await, Url::parse(&mockito::server_url()).unwrap());