char
/
matrix-rust-sdk
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Merge branch 'master' into invite-leave-sync

master
Devin R 2020-05-06 20:35:15 -04:00
parent 8afac39611 e9d6e28e13
commit 7b2ffd1f25
10 changed files with 810 additions and 546 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
matrix_sdk/Cargo.toml
View File

@ -32,8 +32,6 @@ matrix-sdk-crypto = { path = "../matrix_sdk_crypto", optional = true }
# Misc dependencies # Misc dependencies
thiserror = "1.0.16" thiserror = "1.0.16"
tracing = "0.1.13" tracing = "0.1.13"
atomic = "0.4.5"
dashmap = "3.11.1"
[dependencies.tracing-futures] [dependencies.tracing-futures]
version = "0.2.4" version = "0.2.4"
@ -45,12 +43,6 @@ version = "0.2.20"
default-features = false default-features = false
features = ["sync", "time", "fs"] features = ["sync", "time", "fs"]
[dependencies.sqlx]
version = "0.3.4"
optional = true
default-features = false
features = ["runtime-tokio", "sqlite"]
[dev-dependencies] [dev-dependencies]
tokio = { version = "0.2.20", features = ["rt-threaded", "macros"] } tokio = { version = "0.2.20", features = ["rt-threaded", "macros"] }
ruma-identifiers = { version = "0.16.1", features = ["rand"] } ruma-identifiers = { version = "0.16.1", features = ["rand"] }

361
matrix_sdk/src/async_client.rs
View File

@ -17,7 +17,6 @@
use std::collections::BTreeMap; use std::collections::BTreeMap;
use std::collections::HashMap; use std::collections::HashMap;
use std::convert::{TryFrom, TryInto}; use std::convert::{TryFrom, TryInto};
use std::ops::Deref;
use std::result::Result as StdResult; use std::result::Result as StdResult;
use std::sync::Arc; use std::sync::Arc;
use std::time::{Duration, Instant}; use std::time::{Duration, Instant};
@ -50,21 +49,21 @@ use crate::models::Room;
use crate::session::Session; use crate::session::Session;
use crate::state::StateStore; use crate::state::StateStore;
use crate::VERSION; use crate::VERSION;
use crate::{Error, EventEmitter, Result, RoomStateType}; use crate::{Error, EventEmitter, Result};
const DEFAULT_SYNC_TIMEOUT: Duration = Duration::from_secs(30); const DEFAULT_SYNC_TIMEOUT: Duration = Duration::from_secs(30);
#[derive(Clone)]
/// An async/await enabled Matrix client. /// An async/await enabled Matrix client.
/// ///
/// All of the state is held in an `Arc` so the `AsyncClient` can be cloned freely. /// All of the state is held in an `Arc` so the `AsyncClient` can be cloned freely.
#[derive(Clone)]
pub struct AsyncClient { pub struct AsyncClient {
/// The URL of the homeserver to connect to. /// The URL of the homeserver to connect to.
homeserver: Url, homeserver: Url,
/// The underlying HTTP client. /// The underlying HTTP client.
http_client: reqwest::Client, http_client: reqwest::Client,
/// User session data. /// User session data.
pub(crate) base_client: Arc<RwLock<BaseClient>>, pub(crate) base_client: BaseClient,
} }
impl std::fmt::Debug for AsyncClient { impl std::fmt::Debug for AsyncClient {
@ -286,24 +285,22 @@ impl AsyncClient {
let http_client = http_client.default_headers(headers).build()?; let http_client = http_client.default_headers(headers).build()?;
let mut base_client = BaseClient::new(session)?; let base_client = if let Some(store) = config.state_store {
BaseClient::new_with_state_store(session, store)?
if let Some(store) = config.state_store { } else {
base_client.state_store = Some(store); BaseClient::new(session)?
}; };
Ok(Self { Ok(Self {
homeserver, homeserver,
http_client, http_client,
base_client: Arc::new(RwLock::new(base_client)), base_client,
}) })
} }
/// Is the client logged in. /// Is the client logged in.
pub async fn logged_in(&self) -> bool { pub async fn logged_in(&self) -> bool {
// TODO turn this into a atomic bool so this method doesn't need to be self.base_client.logged_in().await
// async.
self.base_client.read().await.logged_in()
} }
/// The Homeserver of the client. /// The Homeserver of the client.
@ -315,7 +312,7 @@ impl AsyncClient {
/// ///
/// The methods of `EventEmitter` are called when the respective `RoomEvents` occur. /// The methods of `EventEmitter` are called when the respective `RoomEvents` occur.
pub async fn add_event_emitter(&mut self, emitter: Box<dyn EventEmitter>) { pub async fn add_event_emitter(&mut self, emitter: Box<dyn EventEmitter>) {
self.base_client.write().await.event_emitter = Some(emitter); self.base_client.add_event_emitter(emitter).await;
} }
/// Returns an `Option` of the room name from a `RoomId`. /// Returns an `Option` of the room name from a `RoomId`.
@ -323,41 +320,67 @@ impl AsyncClient {
/// This is a human readable room name. /// This is a human readable room name.
pub async fn get_room_name(&self, room_id: &RoomId) -> Option<String> { pub async fn get_room_name(&self, room_id: &RoomId) -> Option<String> {
// TODO do we want to use the `RoomStateType` enum here or should we have // TODO do we want to use the `RoomStateType` enum here or should we have
// 3 seperate `room_name` methods. The other option is to remove this and have // 3 separate `room_name` methods. The other option is to remove this and have
// the user get a `Room` and use `Room::calculate_name` method? // the user get a `Room` and use `Room::calculate_name` method?
self.base_client self.base_client.calculate_room_name(room_id).await
.read()
.await
.calculate_room_name(room_id)
.await
} }
/// Returns a `Vec` of the room names this client knows about. /// Returns a `Vec` of the room names this client knows about.
/// ///
/// This is a human readable list of room names. /// This is a human readable list of room names.
pub async fn get_room_names(&self) -> Vec<String> { pub async fn get_room_names(&self) -> Vec<String> {
self.base_client.read().await.calculate_room_names().await // TODO same as get_room_name
self.base_client.calculate_room_names().await
} }
/// Returns the joined rooms this client knows about. /// Returns the joined rooms this client knows about.
/// ///
/// A `HashMap` of room id to `matrix::models::Room` /// A `HashMap` of room id to `matrix::models::Room`
pub async fn get_joined_rooms(&self) -> HashMap<RoomId, Arc<tokio::sync::RwLock<Room>>> { pub fn joined_rooms(&self) -> Arc<RwLock<HashMap<RoomId, Arc<tokio::sync::RwLock<Room>>>>> {
self.base_client.read().await.joined_rooms.clone() self.base_client.joined_rooms()
} }
/// Returns the invited rooms this client knows about. /// Returns the invited rooms this client knows about.
/// ///
/// A `HashMap` of room id to `matrix::models::Room` /// A `HashMap` of room id to `matrix::models::Room`
pub async fn get_invited_rooms(&self) -> HashMap<RoomId, Arc<tokio::sync::RwLock<Room>>> { pub async fn invited_rooms(
self.base_client.read().await.invited_rooms.clone() &self,
) -> Arc<RwLock<HashMap<RoomId, Arc<tokio::sync::RwLock<Room>>>>> {
self.base_client.invited_rooms()
} }
/// Returns the left rooms this client knows about. /// Returns the left rooms this client knows about.
/// ///
/// A `HashMap` of room id to `matrix::models::Room` /// A `HashMap` of room id to `matrix::models::Room`
pub async fn get_left_rooms(&self) -> HashMap<RoomId, Arc<tokio::sync::RwLock<Room>>> { pub async fn left_rooms(&self) -> Arc<RwLock<HashMap<RoomId, Arc<tokio::sync::RwLock<Room>>>>> {
self.base_client.read().await.lefted_rooms.clone() self.base_client.left_rooms()
}
/// Get a joined room with the given room id.
///
/// # Arguments
///
/// `room_id` - The unique id of the room that should be fetched.
pub async fn get_joined_room(&self, room_id: &RoomId) -> Option<Arc<RwLock<Room>>> {
self.base_client.get_joined_room(room_id).await
}
/// Get an invited room with the given room id.
///
/// # Arguments
///
/// `room_id` - The unique id of the room that should be fetched.
pub async fn get_invited_room(&self, room_id: &RoomId) -> Option<Arc<RwLock<Room>>> {
self.base_client.get_invited_room(room_id).await
}
/// Get a left room with the given room id.
///
/// # Arguments
///
/// `room_id` - The unique id of the room that should be fetched.
pub async fn get_left_room(&self, room_id: &RoomId) -> Option<Arc<RwLock<Room>>> {
self.base_client.get_left_room(room_id).await
} }
/// This allows `AsyncClient` to manually sync state with the provided `StateStore`. /// This allows `AsyncClient` to manually sync state with the provided `StateStore`.
@ -384,7 +407,7 @@ impl AsyncClient {
/// # }); /// # });
/// ``` /// ```
pub async fn sync_with_state_store(&self) -> Result<bool> { pub async fn sync_with_state_store(&self) -> Result<bool> {
self.base_client.write().await.sync_with_state_store().await self.base_client.sync_with_state_store().await
} }
/// Login to the server. /// Login to the server.
@ -419,9 +442,7 @@ impl AsyncClient {
}; };
let response = self.send(request).await?; let response = self.send(request).await?;
let mut client = self.base_client.write().await; self.base_client.receive_login_response(&response).await?;
client.receive_login_response(&response).await?;
Ok(response) Ok(response)
} }
@ -642,7 +663,7 @@ impl AsyncClient {
pub async fn sync(&self, mut sync_settings: SyncSettings) -> Result<sync_events::Response> { pub async fn sync(&self, mut sync_settings: SyncSettings) -> Result<sync_events::Response> {
{ {
// if the client has been synced from the state store don't sync again // if the client has been synced from the state store don't sync again
if !self.base_client.read().await.is_state_store_synced() { if !self.base_client.is_state_store_synced() {
// this will bail out returning false if the store has not been set up // this will bail out returning false if the store has not been set up
if let Ok(synced) = self.sync_with_state_store().await { if let Ok(synced) = self.sync_with_state_store().await {
if synced { if synced {
@ -663,236 +684,13 @@ impl AsyncClient {
let mut response = self.send(request).await?; let mut response = self.send(request).await?;
// when events change state updated signals to state store to update database self.base_client
let mut updated = self.iter_joined_rooms(&mut response).await?; .receive_sync_response(&mut response)
.await?;
if self.iter_invited_rooms(&response).await? {
updated = true;
}
if self.iter_left_rooms(&mut response).await? {
updated = true;
}
let mut client = self.base_client.write().await;
client.receive_sync_response(&mut response, updated).await?;
Ok(response) Ok(response)
} }
async fn iter_joined_rooms(&self, response: &mut sync_events::Response) -> Result<bool> {
let mut updated = false;
for (room_id, joined_room) in &mut response.rooms.join {
let matrix_room = {
let mut client = self.base_client.write().await;
for event in &joined_room.state.events {
if let Ok(e) = event.deserialize() {
if client.receive_joined_state_event(&room_id, &e).await {
updated = true;
}
}
}
client.get_or_create_joined_room(&room_id).clone()
};
// RoomSummary contains information for calculating room name
matrix_room
.write()
.await
.set_room_summary(&joined_room.summary);
// re looping is not ideal here
for event in &mut joined_room.state.events {
if let Ok(e) = event.deserialize() {
let client = self.base_client.read().await;
client
.emit_state_event(&room_id, &e, RoomStateType::Joined)
.await;
}
}
for mut event in &mut joined_room.timeline.events {
let decrypted_event = {
let mut client = self.base_client.write().await;
let (decrypt_ev, timeline_update) = client
.receive_joined_timeline_event(room_id, &mut event)
.await;
if timeline_update {
updated = true;
};
decrypt_ev
};
if let Some(e) = decrypted_event {
*event = e;
}
if let Ok(e) = event.deserialize() {
let client = self.base_client.read().await;
client
.emit_timeline_event(&room_id, &e, RoomStateType::Joined)
.await;
}
}
// look at AccountData to further cut down users by collecting ignored users
if let Some(account_data) = &joined_room.account_data {
for account_data in &account_data.events {
{
if let Ok(e) = account_data.deserialize() {
let mut client = self.base_client.write().await;
if client.receive_account_data_event(&room_id, &e).await {
updated = true;
}
client
.emit_account_data_event(room_id, &e, RoomStateType::Joined)
.await;
}
}
}
}
// After the room has been created and state/timeline events accounted for we use the room_id of the newly created
// room to add any presence events that relate to a user in the current room. This is not super
// efficient but we need a room_id so we would loop through now or later.
for presence in &mut response.presence.events {
{
if let Ok(e) = presence.deserialize() {
let mut client = self.base_client.write().await;
if client.receive_presence_event(&room_id, &e).await {
updated = true;
}
client
.emit_presence_event(&room_id, &e, RoomStateType::Joined)
.await;
}
}
}
for ephemeral in &mut joined_room.ephemeral.events {
{
if let Ok(e) = ephemeral.deserialize() {
let mut client = self.base_client.write().await;
if client.receive_ephemeral_event(&room_id, &e).await {
updated = true;
}
client
.emit_ephemeral_event(&room_id, &e, RoomStateType::Joined)
.await;
}
}
}
if updated {
if let Some(store) = self.base_client.read().await.state_store.as_ref() {
store
.store_room_state(matrix_room.read().await.deref())
.await?;
}
}
}
Ok(updated)
}
async fn iter_left_rooms(&self, response: &mut sync_events::Response) -> Result<bool> {
let mut updated = false;
for (room_id, left_room) in &mut response.rooms.leave {
let matrix_room = {
let mut client = self.base_client.write().await;
for event in &left_room.state.events {
if let Ok(e) = event.deserialize() {
if client.receive_left_state_event(&room_id, &e).await {
updated = true;
}
}
}
client.get_or_create_left_room(&room_id).clone()
};
for event in &mut left_room.state.events {
if let Ok(e) = event.deserialize() {
let client = self.base_client.read().await;
client
.emit_state_event(&room_id, &e, RoomStateType::Left)
.await;
}
}
for mut event in &mut left_room.timeline.events {
let decrypted_event = {
let mut client = self.base_client.write().await;
let (decrypt_ev, timeline_update) = client
.receive_left_timeline_event(room_id, &mut event)
.await;
if timeline_update {
updated = true;
};
decrypt_ev
};
if let Some(e) = decrypted_event {
*event = e;
}
if let Ok(e) = event.deserialize() {
let client = self.base_client.read().await;
client
.emit_timeline_event(&room_id, &e, RoomStateType::Left)
.await;
}
}
if updated {
if let Some(store) = self.base_client.read().await.state_store.as_ref() {
store
.store_room_state(matrix_room.read().await.deref())
.await?;
}
}
}
Ok(updated)
}
async fn iter_invited_rooms(&self, response: &sync_events::Response) -> Result<bool> {
let mut updated = false;
for (room_id, invited_room) in &response.rooms.invite {
let matrix_room = {
let mut client = self.base_client.write().await;
for event in &invited_room.invite_state.events {
if let Ok(e) = event.deserialize() {
if client.receive_invite_state_event(&room_id, &e).await {
updated = true;
}
}
}
client.get_or_create_left_room(&room_id).clone()
};
for event in &invited_room.invite_state.events {
if let Ok(e) = event.deserialize() {
let client = self.base_client.read().await;
client
.emit_stripped_state_event(&room_id, &e, RoomStateType::Invited)
.await;
}
}
if updated {
if let Some(store) = self.base_client.read().await.state_store.as_ref() {
store
.store_room_state(matrix_room.read().await.deref())
.await?;
}
}
}
Ok(updated)
}
/// Repeatedly call sync to synchronize the client state with the server. /// Repeatedly call sync to synchronize the client state with the server.
/// ///
/// # Arguments /// # Arguments
@ -975,7 +773,7 @@ impl AsyncClient {
#[cfg(feature = "encryption")] #[cfg(feature = "encryption")]
{ {
if self.base_client.read().await.should_upload_keys().await { if self.base_client.should_upload_keys().await {
let response = self.keys_upload().await; let response = self.keys_upload().await;
if let Err(e) = response { if let Err(e) = response {
@ -983,7 +781,7 @@ impl AsyncClient {
} }
} }
if self.base_client.read().await.should_query_keys().await { if self.base_client.should_query_keys().await {
let response = self.keys_query().await; let response = self.keys_query().await;
if let Err(e) = response { if let Err(e) = response {
@ -1048,9 +846,9 @@ impl AsyncClient {
}; };
let request_builder = if Request::METADATA.requires_authentication { let request_builder = if Request::METADATA.requires_authentication {
let client = self.base_client.read().await; let session = self.base_client.session().read().await;
if let Some(ref session) = client.session { if let Some(session) = session.as_ref() {
request_builder.bearer_auth(&session.access_token) request_builder.bearer_auth(&session.access_token)
} else { } else {
return Err(Error::AuthenticationRequired); return Err(Error::AuthenticationRequired);
@ -1134,8 +932,7 @@ impl AsyncClient {
#[cfg(feature = "encryption")] #[cfg(feature = "encryption")]
{ {
let encrypted = { let encrypted = {
let client = self.base_client.read().await; let room = self.base_client.get_joined_room(room_id).await;
let room = client.joined_rooms.get(room_id);
match room { match room {
Some(r) => r.read().await.is_encrypted(), Some(r) => r.read().await.is_encrypted(),
@ -1145,40 +942,24 @@ impl AsyncClient {
if encrypted { if encrypted {
let missing_sessions = { let missing_sessions = {
let client = self.base_client.read().await; let room = self.base_client.get_joined_room(room_id).await;
let room = client.joined_rooms.get(room_id);
let room = room.as_ref().unwrap().read().await; let room = room.as_ref().unwrap().read().await;
let users = room.members.keys(); let users = room.members.keys();
self.base_client self.base_client.get_missing_sessions(users).await?
.read()
.await
.get_missing_sessions(users)
.await?
}; };
if !missing_sessions.is_empty() { if !missing_sessions.is_empty() {
self.claim_one_time_keys(missing_sessions).await?; self.claim_one_time_keys(missing_sessions).await?;
} }
if self if self.base_client.should_share_group_session(room_id).await {
.base_client
.read()
.await
.should_share_group_session(room_id)
.await
{
// TODO we need to make sure that only one such request is // TODO we need to make sure that only one such request is
// in flight per room at a time. // in flight per room at a time.
self.share_group_session(room_id).await?; self.share_group_session(room_id).await?;
} }
raw_content = serde_json::value::to_raw_value( raw_content = serde_json::value::to_raw_value(
&self &self.base_client.encrypt(room_id, content).await?,
.base_client
.read()
.await
.encrypt(room_id, content)
.await?,
)?; )?;
event_type = EventType::RoomEncrypted; event_type = EventType::RoomEncrypted;
} }
@ -1219,8 +1000,6 @@ impl AsyncClient {
let response = self.send(request).await?; let response = self.send(request).await?;
self.base_client self.base_client
.read()
.await
.receive_keys_claim_response(&response) .receive_keys_claim_response(&response)
.await?; .await?;
Ok(response) Ok(response)
@ -1242,8 +1021,6 @@ impl AsyncClient {
async fn share_group_session(&self, room_id: &RoomId) -> Result<()> { async fn share_group_session(&self, room_id: &RoomId) -> Result<()> {
let mut requests = self let mut requests = self
.base_client .base_client
.read()
.await
.share_group_session(room_id) .share_group_session(room_id)
.await .await
.expect("Keys don't need to be uploaded"); .expect("Keys don't need to be uploaded");
@ -1270,8 +1047,6 @@ impl AsyncClient {
async fn keys_upload(&self) -> Result<upload_keys::Response> { async fn keys_upload(&self) -> Result<upload_keys::Response> {
let (device_keys, one_time_keys) = self let (device_keys, one_time_keys) = self
.base_client .base_client
.read()
.await
.keys_for_upload() .keys_for_upload()
.await .await
.expect("Keys don't need to be uploaded"); .expect("Keys don't need to be uploaded");
@ -1289,8 +1064,6 @@ impl AsyncClient {
let response = self.send(request).await?; let response = self.send(request).await?;
self.base_client self.base_client
.read()
.await
.receive_keys_upload_response(&response) .receive_keys_upload_response(&response)
.await?; .await?;
Ok(response) Ok(response)
@ -1299,7 +1072,7 @@ impl AsyncClient {
/// Get the current, if any, sync token of the client. /// Get the current, if any, sync token of the client.
/// This will be None if the client didn't sync at least once. /// This will be None if the client didn't sync at least once.
pub async fn sync_token(&self) -> Option<String> { pub async fn sync_token(&self) -> Option<String> {
self.base_client.read().await.sync_token.clone() self.base_client.sync_token().await
} }
/// Query the server for users device keys. /// Query the server for users device keys.
@ -1313,8 +1086,6 @@ impl AsyncClient {
async fn keys_query(&self) -> Result<get_keys::Response> { async fn keys_query(&self) -> Result<get_keys::Response> {
let mut users_for_query = self let mut users_for_query = self
.base_client .base_client
.read()
.await
.users_for_key_query() .users_for_key_query()
.await .await
.expect("Keys don't need to be uploaded"); .expect("Keys don't need to be uploaded");
@ -1338,8 +1109,6 @@ impl AsyncClient {
let response = self.send(request).await?; let response = self.send(request).await?;
self.base_client self.base_client
.read()
.await
.receive_keys_query_response(&response) .receive_keys_query_response(&response)
.await?; .await?;

888
matrix_sdk/src/base_client.rs
View File

File diff suppressed because it is too large Load Diff

4
matrix_sdk/src/event_emitter/mod.rs
View File

@ -266,7 +266,7 @@ mod test {
self.0.lock().await.push("account ignore".to_string()) self.0.lock().await.push("account ignore".to_string())
} }
async fn on_account_push_rules(&self, _: RoomState, _: &PushRulesEvent) { async fn on_account_push_rules(&self, _: RoomState, _: &PushRulesEvent) {
self.0.lock().await.push("".to_string()) self.0.lock().await.push("account push rules".to_string())
} }
async fn on_account_data_fully_read(&self, _: RoomState, _: &FullyReadEvent) { async fn on_account_data_fully_read(&self, _: RoomState, _: &FullyReadEvent) {
self.0.lock().await.push("account read".to_string()) self.0.lock().await.push("account read".to_string())
@ -317,6 +317,7 @@ mod test {
assert_eq!( assert_eq!(
v.as_slice(), v.as_slice(),
[ [
"state rules",
"state member", "state member",
"state aliases", "state aliases",
"state power", "state power",
@ -396,6 +397,7 @@ mod test {
assert_eq!( assert_eq!(
v.as_slice(), v.as_slice(),
[ [
"state rules",
"state member", "state member",
"state aliases", "state aliases",
"state power", "state power",

3
matrix_sdk/src/models/room.rs
View File

@ -507,7 +507,8 @@ mod test {
let _response = client.sync(sync_settings).await.unwrap(); let _response = client.sync(sync_settings).await.unwrap();
let rooms = &client.base_client.read().await.joined_rooms; let rooms_lock = &client.base_client.joined_rooms();
let rooms = rooms_lock.read().await;
let room = &rooms let room = &rooms
.get(&RoomId::try_from("!SVkFJHzfwvuaIEawgC:localhost").unwrap()) .get(&RoomId::try_from("!SVkFJHzfwvuaIEawgC:localhost").unwrap())
.unwrap() .unwrap()

11
matrix_sdk/src/state/mod.rs
View File

@ -13,13 +13,12 @@
// See the License for the specific language governing permissions and // See the License for the specific language governing permissions and
// limitations under the License. // limitations under the License.
use serde::{Deserialize, Serialize};
use std::collections::HashMap; use std::collections::HashMap;
pub mod state_store; pub mod state_store;
pub use state_store::JsonStore; pub use state_store::JsonStore;
use serde::{Deserialize, Serialize};
use crate::base_client::{Client as BaseClient, Token}; use crate::base_client::{Client as BaseClient, Token};
use crate::events::push_rules::Ruleset; use crate::events::push_rules::Ruleset;
use crate::identifiers::{RoomId, UserId}; use crate::identifiers::{RoomId, UserId};
@ -48,7 +47,7 @@ impl PartialEq for ClientState {
} }
impl ClientState { impl ClientState {
pub fn from_base_client(client: &BaseClient) -> ClientState { pub async fn from_base_client(client: &BaseClient) -> ClientState {
let BaseClient { let BaseClient {
sync_token, sync_token,
ignored_users, ignored_users,
@ -56,9 +55,9 @@ impl ClientState {
.. ..
} = client; } = client;
Self { Self {
sync_token: sync_token.clone(), sync_token: sync_token.read().await.clone(),
ignored_users: ignored_users.clone(), ignored_users: ignored_users.read().await.clone(),
push_ruleset: push_ruleset.clone(), push_ruleset: push_ruleset.read().await.clone(),
} }
} }
} }

8
matrix_sdk/src/state/state_store.rs
View File

@ -287,16 +287,16 @@ mod test {
AsyncClient::new_with_config(homeserver, Some(session.clone()), config).unwrap(); AsyncClient::new_with_config(homeserver, Some(session.clone()), config).unwrap();
client.sync(sync_settings).await.unwrap(); client.sync(sync_settings).await.unwrap();
let base_client = client.base_client.read().await; let base_client = &client.base_client;
// assert the synced client and the logged in client are equal // assert the synced client and the logged in client are equal
assert_eq!(base_client.session, Some(session)); assert_eq!(*base_client.session().read().await, Some(session));
assert_eq!( assert_eq!(
base_client.sync_token, base_client.sync_token().await,
Some("s526_47314_0_7_1_1_1_11444_1".to_string()) Some("s526_47314_0_7_1_1_1_11444_1".to_string())
); );
assert_eq!( assert_eq!(
base_client.ignored_users, *base_client.ignored_users.read().await,
vec![UserId::try_from("@someone:example.org").unwrap()] vec![UserId::try_from("@someone:example.org").unwrap()]
); );
} }

6
matrix_sdk/src/test_builder.rs
View File

@ -343,13 +343,11 @@ impl ClientTestRunner {
} }
async fn stream_client_events(&mut self) { async fn stream_client_events(&mut self) {
let mut cli = self let cli = &self
.client .client
.as_ref() .as_ref()
.expect("`AsyncClient` must be set use `ClientTestRunner::set_client`") .expect("`AsyncClient` must be set use `ClientTestRunner::set_client`")
.base_client .base_client;
.write()
.await;
let room_id = &self.room_user_id.0; let room_id = &self.room_user_id.0;

65
matrix_sdk_crypto/src/machine.rs
View File

@ -984,7 +984,7 @@ impl OlmMachine {
.map_err(|_| EventError::UnsupportedOlmType)?; .map_err(|_| EventError::UnsupportedOlmType)?;
// Decrypt the OlmMessage and get a Ruma event out of it. // Decrypt the OlmMessage and get a Ruma event out of it.
let (mut decrypted_event, signing_key) = self let (decrypted_event, signing_key) = self
.decrypt_olm_message(&event.sender, &content.sender_key, message) .decrypt_olm_message(&event.sender, &content.sender_key, message)
.await?; .await?;
@ -992,14 +992,23 @@ impl OlmMachine {
// Handle the decrypted event, e.g. fetch out Megolm sessions out of // Handle the decrypted event, e.g. fetch out Megolm sessions out of
// the event. // the event.
self.handle_decrypted_to_device_event( if let Some(event) = self
&content.sender_key, .handle_decrypted_to_device_event(
&signing_key, &content.sender_key,
&mut decrypted_event, &signing_key,
) &decrypted_event,
.await?; )
.await?
Ok(decrypted_event) {
// Some events may have sensitive data e.g. private keys, while we
// wan't to notify our users that a private key was received we
// don't want them to be able to do silly things with it. Handling
// events modifies them and returns a modified one, so replace it
// here if we get one.
Ok(event)
} else {
Ok(decrypted_event)
}
} else { } else {
warn!("Olm event doesn't contain a ciphertext for our key"); warn!("Olm event doesn't contain a ciphertext for our key");
Err(EventError::MissingCiphertext.into()) Err(EventError::MissingCiphertext.into())
@ -1012,7 +1021,7 @@ impl OlmMachine {
sender_key: &str, sender_key: &str,
signing_key: &str, signing_key: &str,
event: &mut ToDeviceRoomKey, event: &mut ToDeviceRoomKey,
) -> OlmResult<()> { ) -> OlmResult<Option<EventJson<ToDeviceEvent>>> {
match event.content.algorithm { match event.content.algorithm {
Algorithm::MegolmV1AesSha2 => { Algorithm::MegolmV1AesSha2 => {
let session_key = GroupSessionKey(mem::take(&mut event.content.session_key)); let session_key = GroupSessionKey(mem::take(&mut event.content.session_key));
@ -1024,14 +1033,24 @@ impl OlmMachine {
session_key, session_key,
)?; )?;
let _ = self.store.save_inbound_group_session(session).await?; let _ = self.store.save_inbound_group_session(session).await?;
Ok(()) // TODO ideally we would rewrap the event again just like so
// let event = EventJson::from(ToDeviceEvent::RoomKey(event.clone()));
// This saidly lacks a type once it's serialized again, fix
// this in Ruma.
let mut json = serde_json::to_value(event.clone())?;
json.as_object_mut()
.unwrap()
.insert("type".to_owned(), Value::String("m.room_key".to_owned()));
let event = serde_json::from_value::<EventJson<ToDeviceEvent>>(json)?;
Ok(Some(event))
} }
_ => { _ => {
warn!( warn!(
"Received room key with unsupported key algorithm {}", "Received room key with unsupported key algorithm {}",
event.content.algorithm event.content.algorithm
); );
Ok(()) Ok(None)
} }
} }
} }
@ -1330,25 +1349,26 @@ impl OlmMachine {
&mut self, &mut self,
sender_key: &str, sender_key: &str,
signing_key: &str, signing_key: &str,
event: &mut EventJson<ToDeviceEvent>, event: &EventJson<ToDeviceEvent>,
) -> OlmResult<()> { ) -> OlmResult<Option<EventJson<ToDeviceEvent>>> {
let event = if let Ok(e) = event.deserialize() { let event = if let Ok(e) = event.deserialize() {
e e
} else { } else {
warn!("Decrypted to-device event failed to be parsed correctly"); warn!("Decrypted to-device event failed to be parsed correctly");
return Ok(()); return Ok(None);
}; };
match event { match event {
ToDeviceEvent::RoomKey(mut e) => { ToDeviceEvent::RoomKey(mut e) => {
self.add_room_key(sender_key, signing_key, &mut e).await Ok(self.add_room_key(sender_key, signing_key, &mut e).await?)
} }
ToDeviceEvent::ForwardedRoomKey(e) => { ToDeviceEvent::ForwardedRoomKey(e) => {
self.add_forwarded_room_key(sender_key, signing_key, &e) self.add_forwarded_room_key(sender_key, signing_key, &e)?;
Ok(None)
} }
_ => { _ => {
warn!("Received a unexpected encrypted to-device event"); warn!("Received a unexpected encrypted to-device event");
Ok(()) Ok(None)
} }
} }
} }
@ -1657,7 +1677,7 @@ mod test {
let mut bob_keys = BTreeMap::new(); let mut bob_keys = BTreeMap::new();
let one_time_key = one_time_keys.iter().nth(0).unwrap(); let one_time_key = one_time_keys.iter().next().unwrap();
let mut keys = BTreeMap::new(); let mut keys = BTreeMap::new();
keys.insert(one_time_key.0.clone(), one_time_key.1.clone()); keys.insert(one_time_key.0.clone(), one_time_key.1.clone());
bob_keys.insert(bob.device_id.clone(), keys); bob_keys.insert(bob.device_id.clone(), keys);
@ -1820,7 +1840,7 @@ mod test {
let identity_keys = machine.account.identity_keys(); let identity_keys = machine.account.identity_keys();
let ed25519_key = identity_keys.ed25519(); let ed25519_key = identity_keys.ed25519();
let mut one_time_key = one_time_keys.values_mut().nth(0).unwrap(); let mut one_time_key = one_time_keys.values_mut().next().unwrap();
let ret = machine.verify_json( let ret = machine.verify_json(
&machine.user_id, &machine.user_id,
@ -1848,7 +1868,7 @@ mod test {
&machine.user_id, &machine.user_id,
&machine.device_id, &machine.device_id,
ed25519_key, ed25519_key,
&mut json!(&mut one_time_keys.as_mut().unwrap().values_mut().nth(0)), &mut json!(&mut one_time_keys.as_mut().unwrap().values_mut().next()),
); );
assert!(ret.is_ok()); assert!(ret.is_ok());
@ -1923,7 +1943,7 @@ mod test {
let mut bob_keys = BTreeMap::new(); let mut bob_keys = BTreeMap::new();
let one_time_key = one_time_keys.iter().nth(0).unwrap(); let one_time_key = one_time_keys.iter().next().unwrap();
let mut keys = BTreeMap::new(); let mut keys = BTreeMap::new();
keys.insert(one_time_key.0.clone(), one_time_key.1.clone()); keys.insert(one_time_key.0.clone(), one_time_key.1.clone());
bob_keys.insert(bob_machine.device_id.clone(), keys); bob_keys.insert(bob_machine.device_id.clone(), keys);
@ -2011,6 +2031,7 @@ mod test {
if let AnyToDeviceEvent::RoomKey(e) = event.deserialize().unwrap() { if let AnyToDeviceEvent::RoomKey(e) = event.deserialize().unwrap() {
assert_eq!(e.sender, alice.user_id); assert_eq!(e.sender, alice.user_id);
assert!(e.content.session_key.is_empty())
} else { } else {
panic!("Event had the wrong type"); panic!("Event had the wrong type");
} }

2
matrix_sdk_types/Cargo.toml
View File

@ -13,6 +13,6 @@ version = "0.1.0"
[dependencies] [dependencies]
js_int = "0.1.5" js_int = "0.1.5"
ruma-api = "0.16.0" ruma-api = "0.16.0"
ruma-client-api = { git = "https://github.com/matrix-org/ruma-client-api" } ruma-client-api = "0.8.0"
ruma-events = "0.21.0" ruma-events = "0.21.0"
ruma-identifiers = "0.16.1" ruma-identifiers = "0.16.1"