diff --git a/crates/matrix-sdk-crypto/src/identities/manager.rs b/crates/matrix-sdk-crypto/src/identities/manager.rs index f4c1197be..dd09db28a 100644 --- a/crates/matrix-sdk-crypto/src/identities/manager.rs +++ b/crates/matrix-sdk-crypto/src/identities/manager.rs @@ -16,15 +16,10 @@ use std::{ collections::{BTreeMap, BTreeSet, HashSet}, ops::Deref, sync::Arc, - time::Duration, }; use futures_util::future::join_all; -use matrix_sdk_common::{ - executor::spawn, - locks::Mutex, - timeout::{timeout, ElapsedError}, -}; +use matrix_sdk_common::{executor::spawn, locks::Mutex}; use ruma::{ api::client::keys::get_keys::v3::Response as KeysQueryResponse, serde::Raw, DeviceId, OwnedDeviceId, OwnedServerName, OwnedTransactionId, OwnedUserId, ServerName, TransactionId, @@ -56,13 +51,6 @@ struct IdentityChange { private: Option, } -/// A listener that can notify if a `/keys/query` response has been received. -#[derive(Clone, Debug)] -pub(crate) struct KeysQueryListener { - inner: Arc, - store: Store, -} - /// Result type telling us if a `/keys/query` response was expected for a given /// user. #[derive(Clone, Copy, Debug, PartialEq, Eq)] @@ -74,59 +62,10 @@ pub(crate) enum UserKeyQueryResult { TimeoutExpired, } -impl KeysQueryListener { - pub(crate) fn new(store: Store) -> Self { - Self { inner: event_listener::Event::new().into(), store } - } - - /// Notify our listeners that we received a `/keys/query` response. - fn notify(&self) { - self.inner.notify(usize::MAX); - } - - /// Wait for a `/keys/query` response to be received if one is expected for - /// the given user. - /// - /// If the given timeout has elapsed the method will stop waiting and return - /// an error. - pub async fn wait_if_user_pending( - &self, - timeout: Duration, - user: &UserId, - ) -> Result { - let (users_for_key_query, _) = self.store.users_for_key_query().await.unwrap_or_default(); - - if users_for_key_query.contains(user) { - if let Err(e) = self.wait(timeout).await { - warn!( - user_id = ?user, - "The user has a pending `/key/query` request which did \ - not finish yet, some devices might be missing." - ); - - Err(e) - } else { - Ok(UserKeyQueryResult::WasPending) - } - } else { - Ok(UserKeyQueryResult::WasNotPending) - } - } - - /// Wait for a `/keys/query` response to be received. - /// - /// If the given timeout has elapsed the method will stop waiting and return - /// an error. - pub async fn wait(&self, duration: Duration) -> Result<(), ElapsedError> { - timeout(self.inner.listen(), duration).await - } -} - #[derive(Debug, Clone)] pub(crate) struct IdentityManager { user_id: Arc, device_id: Arc, - keys_query_listener: KeysQueryListener, failures: FailuresCache, store: Store, @@ -151,14 +90,12 @@ impl IdentityManager { const MAX_KEY_QUERY_USERS: usize = 250; pub fn new(user_id: Arc, device_id: Arc, store: Store) -> Self { - let keys_query_listener = KeysQueryListener::new(store.clone()); let keys_query_request_details = Mutex::new(KeysQueryRequestDetails::default()); IdentityManager { user_id, device_id, store, - keys_query_listener, failures: Default::default(), keys_query_request_details: keys_query_request_details.into(), } @@ -168,10 +105,6 @@ impl IdentityManager { &self.user_id } - pub fn listen_for_received_queries(&self) -> KeysQueryListener { - self.keys_query_listener.clone() - } - /// Receive a successful keys query response. /// /// Returns a list of devices newly discovered devices and devices that @@ -268,8 +201,6 @@ impl IdentityManager { "Finished handling of the keys/query response" ); - self.keys_query_listener.notify(); - Ok((devices, identities)) } @@ -1006,7 +937,7 @@ pub(crate) mod testing { #[cfg(test)] pub(crate) mod tests { - use std::{ops::Deref, time::Duration}; + use std::ops::Deref; use matrix_sdk_test::{async_test, response_from_file}; use ruma::{ @@ -1067,17 +998,11 @@ pub(crate) mod tests { let devices = manager.store.get_user_devices(other_user).await.unwrap(); assert_eq!(devices.devices().count(), 0); - let listener = manager.listen_for_received_queries(); - - let task = tokio::task::spawn(async move { listener.wait(Duration::from_secs(10)).await }); - manager .receive_keys_query_response(&TransactionId::new(), &other_key_query()) .await .unwrap(); - task.await.unwrap().unwrap(); - let devices = manager.store.get_user_devices(other_user).await.unwrap(); assert_eq!(devices.devices().count(), 1); diff --git a/crates/matrix-sdk-crypto/src/identities/mod.rs b/crates/matrix-sdk-crypto/src/identities/mod.rs index 1ea56f4fd..4f56dd43f 100644 --- a/crates/matrix-sdk-crypto/src/identities/mod.rs +++ b/crates/matrix-sdk-crypto/src/identities/mod.rs @@ -50,7 +50,7 @@ use std::sync::{ }; pub use device::{Device, LocalTrust, ReadOnlyDevice, UserDevices}; -pub(crate) use manager::{IdentityManager, KeysQueryListener, UserKeyQueryResult}; +pub(crate) use manager::{IdentityManager, UserKeyQueryResult}; use serde::{Deserialize, Deserializer, Serializer}; pub use user::{ OwnUserIdentity, ReadOnlyOwnUserIdentity, ReadOnlyUserIdentities, ReadOnlyUserIdentity, diff --git a/crates/matrix-sdk-crypto/src/machine.rs b/crates/matrix-sdk-crypto/src/machine.rs index 3731ac0b7..30dd32f7c 100644 --- a/crates/matrix-sdk-crypto/src/machine.rs +++ b/crates/matrix-sdk-crypto/src/machine.rs @@ -181,14 +181,11 @@ impl OlmMachine { let identity_manager = IdentityManager::new(user_id.clone(), device_id.clone(), store.clone()); - let event = identity_manager.listen_for_received_queries(); - let session_manager = SessionManager::new( account.clone(), users_for_key_claim, key_request_machine.clone(), store.clone(), - event, ); #[cfg(feature = "backups_v1")] diff --git a/crates/matrix-sdk-crypto/src/session_manager/sessions.rs b/crates/matrix-sdk-crypto/src/session_manager/sessions.rs index 5687ade05..8d655d4eb 100644 --- a/crates/matrix-sdk-crypto/src/session_manager/sessions.rs +++ b/crates/matrix-sdk-crypto/src/session_manager/sessions.rs @@ -34,7 +34,7 @@ use vodozemac::Curve25519PublicKey; use crate::{ error::OlmResult, gossiping::GossipMachine, - identities::{KeysQueryListener, UserKeyQueryResult}, + identities::UserKeyQueryResult, olm::Account, requests::{OutgoingRequest, ToDeviceRequest}, store::{Changes, Result as StoreResult, Store}, @@ -55,7 +55,6 @@ pub(crate) struct SessionManager { wedged_devices: Arc>>, key_request_machine: GossipMachine, outgoing_to_device_requests: Arc>, - keys_query_listener: KeysQueryListener, failures: FailuresCache, } @@ -69,7 +68,6 @@ impl SessionManager { users_for_key_claim: Arc>>, key_request_machine: GossipMachine, store: Store, - keys_query_listener: KeysQueryListener, ) -> Self { Self { account, @@ -78,7 +76,6 @@ impl SessionManager { users_for_key_claim, wedged_devices: Default::default(), outgoing_to_device_requests: Default::default(), - keys_query_listener, failures: Default::default(), } } @@ -419,7 +416,7 @@ mod tests { use super::SessionManager; use crate::{ gossiping::GossipMachine, - identities::{IdentityManager, KeysQueryListener, ReadOnlyDevice}, + identities::{IdentityManager, ReadOnlyDevice}, olm::{Account, PrivateCrossSigningIdentity, ReadOnlyAccount}, session_manager::GroupSessionCache, store::{IntoCryptoStore, MemoryStore, Store}, @@ -494,13 +491,7 @@ mod tests { users_for_key_claim.clone(), ); - SessionManager::new( - account, - users_for_key_claim, - key_request, - store.clone(), - KeysQueryListener::new(store), - ) + SessionManager::new(account, users_for_key_claim, key_request, store) } #[async_test]