Merge branch 'master' into state-store

.travis.yml

@@ -5,8 +5,6 @@ addons:
     packages:
       - libssl-dev
 
-cache: cargo
-
 jobs:
   include:
     - os: linux

examples/command_bot.rs

@@ -0,0 +1,118 @@
+use std::sync::Arc;
+use std::{env, process::exit};
+
+use matrix_sdk::{
+    self,
+    events::room::message::{MessageEvent, MessageEventContent, TextMessageEventContent},
+    AsyncClient, AsyncClientConfig, EventEmitter, Room, SyncSettings,
+};
+use tokio::sync::RwLock;
+use url::Url;
+
+struct CommandBot {
+    /// This clone of the `AsyncClient` will send requests to the server,
+    /// while the other keeps us in sync with the server using `sync_forever`.
+    client: AsyncClient,
+}
+
+impl CommandBot {
+    pub fn new(client: AsyncClient) -> Self {
+        Self { client }
+    }
+}
+
+#[async_trait::async_trait]
+impl EventEmitter for CommandBot {
+    async fn on_room_message(&self, room: Arc<RwLock<Room>>, event: &MessageEvent) {
+        let msg_body = if let MessageEvent {
+            content: MessageEventContent::Text(TextMessageEventContent { body: msg_body, .. }),
+            ..
+        } = event
+        {
+            msg_body.clone()
+        } else {
+            String::new()
+        };
+
+        if msg_body.contains("!party") {
+            let content = MessageEventContent::Text(TextMessageEventContent {
+                body: "🎉🎊🥳 let's PARTY!! 🥳🎊🎉".to_string(),
+                format: None,
+                formatted_body: None,
+                relates_to: None,
+            });
+            // we clone here to hold the lock for as little time as possible.
+            let room_id = room.read().await.room_id.clone();
+
+            println!("sending");
+
+            self.client
+                // send our message to the room we found the "!party" command in
+                // the last parameter is an optional Uuid which we don't care about.
+                .room_send(&room_id, content, None)
+                .await
+                .unwrap();
+
+            println!("message sent");
+        }
+    }
+}
+
+async fn login_and_sync(
+    homeserver_url: String,
+    username: String,
+    password: String,
+) -> Result<(), matrix_sdk::Error> {
+    let client_config = AsyncClientConfig::new()
+        .proxy("http://localhost:8080")?
+        .disable_ssl_verification();
+
+    let homeserver_url = Url::parse(&homeserver_url)?;
+    // create a new AsyncClient with the given homeserver url and config
+    let mut client = AsyncClient::new_with_config(homeserver_url, None, client_config).unwrap();
+
+    client
+        .login(
+            username.clone(),
+            password,
+            None,
+            Some("command bot".to_string()),
+        )
+        .await?;
+
+    println!("logged in as {}", username);
+
+    // initial sync to set up state and so our bot doesn't respond to old messages
+    client.sync(SyncSettings::default()).await.unwrap();
+    // add our CommandBot to be notified of incoming messages, we do this after the initial
+    // sync to avoid responding to messages before the bot was running.
+    client
+        .add_event_emitter(Box::new(CommandBot::new(client.clone())))
+        .await;
+
+    // since we called sync before we `sync_forever` we must pass that sync token to
+    // `sync_forever`
+    let settings = SyncSettings::default().token(client.sync_token().await.unwrap());
+    // this keeps state from the server streaming in to CommandBot via the EventEmitter trait
+    client.sync_forever(settings, |_| async {}).await;
+
+    Ok(())
+}
+
+#[tokio::main]
+async fn main() -> Result<(), matrix_sdk::Error> {
+    let (homeserver_url, username, password) =
+        match (env::args().nth(1), env::args().nth(2), env::args().nth(3)) {
+            (Some(a), Some(b), Some(c)) => (a, b, c),
+            _ => {
+                eprintln!(
+                    "Usage: {} <homeserver_url> <username> <password>",
+                    env::args().next().unwrap()
+                );
+                exit(1)
+            }
+        };
+
+    login_and_sync(homeserver_url, username, password).await?;
+    Ok(())
+}

examples/login.rs

@@ -1,4 +1,3 @@
-use std::ops::Deref;
 use std::sync::Arc;
 use std::{env, process::exit};
 use url::Url;
@@ -8,26 +7,31 @@ use matrix_sdk::{
     events::room::message::{MessageEvent, MessageEventContent, TextMessageEventContent},
     AsyncClient, AsyncClientConfig, EventEmitter, Room, SyncSettings,
 };
-use tokio::sync::Mutex;
+use tokio::sync::RwLock;
 
 struct EventCallback;
 
 #[async_trait::async_trait]
 impl EventEmitter for EventCallback {
-    async fn on_room_message(&mut self, room: Arc<Mutex<Room>>, event: Arc<Mutex<MessageEvent>>) {
+    async fn on_room_message(&self, room: Arc<RwLock<Room>>, event: &MessageEvent) {
         if let MessageEvent {
             content: MessageEventContent::Text(TextMessageEventContent { body: msg_body, .. }),
             sender,
             ..
-        } = event.lock().await.deref()
+        } = event
         {
-            let rooms = room.lock().await;
+            let name = {
-            let member = rooms.members.get(&sender).unwrap();
+                // any reads should be held for the shortest time possible to
-            println!(
+                // avoid dead locks
-                "{}: {}",
+                let room = room.read().await;
-                member.display_name.as_ref().unwrap_or(&sender.to_string()),
+                let member = room.members.get(&sender).unwrap();
-                msg_body
+                member
-            );
+                    .display_name
+                    .as_ref()
+                    .map(ToString::to_string)
+                    .unwrap_or(sender.to_string())
+            };
+            println!("{}: {}", name, msg_body);
         }
     }
 }
@@ -43,9 +47,7 @@ async fn login(
     let homeserver_url = Url::parse(&homeserver_url)?;
     let mut client = AsyncClient::new_with_config(homeserver_url, None, client_config).unwrap();
 
-    client
+    client.add_event_emitter(Box::new(EventCallback)).await;
-        .add_event_emitter(Arc::new(Mutex::new(Box::new(EventCallback))))
-        .await;
 
     client
         .login(username, password, None, Some("rust-sdk".to_string()))

src/async_client.rs

@@ -37,10 +37,10 @@ use ruma_api::{Endpoint, Outgoing};
 use ruma_events::room::message::MessageEventContent;
 use ruma_events::EventResult;
 pub use ruma_events::EventType;
-use ruma_identifiers::RoomId;
+use ruma_identifiers::{RoomId, RoomIdOrAliasId, UserId};
 
 #[cfg(feature = "encryption")]
-use ruma_identifiers::{DeviceId, UserId};
+use ruma_identifiers::DeviceId;
 
 use crate::api;
 use crate::base_client::Client as BaseClient;
@@ -53,6 +53,8 @@ const DEFAULT_SYNC_TIMEOUT: Duration = Duration::from_secs(30);
 
 #[derive(Clone)]
 /// An async/await enabled Matrix client.
+///
+/// All of the state is held in an `Arc` so the `AsyncClient` can be cloned freely.
 pub struct AsyncClient {
     /// The URL of the homeserver to connect to.
     homeserver: Url,
@@ -181,7 +183,17 @@ impl SyncSettings {
 use api::r0::client_exchange::send_event_to_device;
 #[cfg(feature = "encryption")]
 use api::r0::keys::{claim_keys, get_keys, upload_keys, KeyAlgorithm};
+use api::r0::membership::join_room_by_id;
+use api::r0::membership::join_room_by_id_or_alias;
+use api::r0::membership::kick_user;
+use api::r0::membership::leave_room;
+use api::r0::membership::{
+    invite_user::{self, InvitationRecipient},
+    Invite3pid,
+};
 use api::r0::message::create_message_event;
+use api::r0::message::get_message_events;
+use api::r0::room::create_room;
 use api::r0::session::login;
 use api::r0::sync::sync_events;
 
@@ -263,10 +275,7 @@ impl AsyncClient {
     /// Add `EventEmitter` to `AsyncClient`.
     ///
     /// The methods of `EventEmitter` are called when the respective `RoomEvents` occur.
-    pub async fn add_event_emitter(
+    pub async fn add_event_emitter(&mut self, emitter: Box<dyn EventEmitter>) {
-        &mut self,
-        emitter: Arc<tokio::sync::Mutex<Box<dyn EventEmitter>>>,
-    ) {
         self.base_client.write().await.event_emitter = Some(emitter);
     }
 
@@ -291,7 +300,7 @@ impl AsyncClient {
     /// Returns the rooms this client knows about.
     ///
     /// A `HashMap` of room id to `matrix::models::Room`
-    pub async fn get_rooms(&self) -> HashMap<RoomId, Arc<tokio::sync::Mutex<Room>>> {
+    pub async fn get_rooms(&self) -> HashMap<RoomId, Arc<tokio::sync::RwLock<Room>>> {
         self.base_client.read().await.joined_rooms.clone()
     }
 
@@ -309,7 +318,7 @@ impl AsyncClient {
     ///     only if the client also holds the encryption keys for this device.
     #[instrument(skip(password))]
     pub async fn login<S: Into<String> + std::fmt::Debug>(
-        &mut self,
+        &self,
         user: S,
         password: S,
         device_id: Option<S>,
@@ -333,16 +342,214 @@ impl AsyncClient {
         Ok(response)
     }
 
+    /// Join a room by `RoomId`.
+    ///
+    /// Returns a `join_room_by_id::Response` consisting of the
+    /// joined rooms `RoomId`.
+    ///
+    /// # Arguments
+    ///
+    /// * room_id - The `RoomId` of the room to be joined.
+    pub async fn join_room_by_id(&mut self, room_id: &RoomId) -> Result<join_room_by_id::Response> {
+        let request = join_room_by_id::Request {
+            room_id: room_id.clone(),
+            third_party_signed: None,
+        };
+        self.send(request).await
+    }
+
+    /// Join a room by `RoomId`.
+    ///
+    /// Returns a `join_room_by_id_or_alias::Response` consisting of the
+    /// joined rooms `RoomId`.
+    ///
+    /// # Arguments
+    ///
+    /// * alias - The `RoomId` or `RoomAliasId` of the room to be joined.
+    /// An alias looks like this `#name:example.com`
+    pub async fn join_room_by_id_or_alias(
+        &self,
+        alias: &RoomIdOrAliasId,
+    ) -> Result<join_room_by_id_or_alias::Response> {
+        let request = join_room_by_id_or_alias::Request {
+            room_id_or_alias: alias.clone(),
+            third_party_signed: None,
+        };
+        self.send(request).await
+    }
+
+    /// Kick a user out of the specified room.
+    ///
+    /// Returns a `kick_user::Response`, an empty response.
+    ///
+    /// # Arguments
+    ///
+    /// * room_id - The `RoomId` of the room the user should be kicked out of.
+    ///
+    /// * user_id - The `UserId` of the user that should be kicked out of the room.
+    ///
+    /// * reason - Optional reason why the room member is being kicked out.
+    pub async fn kick_user(
+        &self,
+        room_id: &RoomId,
+        user_id: &UserId,
+        reason: Option<String>,
+    ) -> Result<kick_user::Response> {
+        let request = kick_user::Request {
+            reason,
+            room_id: room_id.clone(),
+            user_id: user_id.clone(),
+        };
+        self.send(request).await
+    }
+
+    /// Leave the specified room.
+    ///
+    /// Returns a `leave_room::Response`, an empty response.
+    ///
+    /// # Arguments
+    ///
+    /// * room_id - The `RoomId` of the room to leave.
+    ///
+    pub async fn leave_room(&self, room_id: &RoomId) -> Result<leave_room::Response> {
+        let request = leave_room::Request {
+            room_id: room_id.clone(),
+        };
+        self.send(request).await
+    }
+
+    /// Invite the specified user by `UserId` to the given room.
+    ///
+    /// Returns a `invite_user::Response`, an empty response.
+    ///
+    /// # Arguments
+    ///
+    /// * room_id - The `RoomId` of the room to invite the specified user to.
+    ///
+    /// * user_id - The `UserId` of the user to invite to the room.
+    pub async fn invite_user_by_id(
+        &self,
+        room_id: &RoomId,
+        user_id: &UserId,
+    ) -> Result<invite_user::Response> {
+        let request = invite_user::Request {
+            room_id: room_id.clone(),
+            recipient: InvitationRecipient::UserId {
+                user_id: user_id.clone(),
+            },
+        };
+        self.send(request).await
+    }
+
+    /// Invite the specified user by third party id to the given room.
+    ///
+    /// Returns a `invite_user::Response`, an empty response.
+    ///
+    /// # Arguments
+    ///
+    /// * room_id - The `RoomId` of the room to invite the specified user to.
+    ///
+    /// * invite_id - A third party id of a user to invite to the room.
+    pub async fn invite_user_by_3pid(
+        &self,
+        room_id: &RoomId,
+        invite_id: &Invite3pid,
+    ) -> Result<invite_user::Response> {
+        let request = invite_user::Request {
+            room_id: room_id.clone(),
+            recipient: InvitationRecipient::ThirdPartyId(invite_id.clone()),
+        };
+        self.send(request).await
+    }
+
+    /// Create a room using the `RoomBuilder` and send the request.
+    ///
+    /// Sends a request to `/_matrix/client/r0/createRoom`, returns a `create_room::Response`,
+    /// this is an empty response.
+    ///
+    /// # Arguments
+    ///
+    /// * room - The easiest way to create this request is using the `RoomBuilder`.
+    ///
+    /// # Examples
+    /// ```no_run
+    /// use matrix_sdk::{AsyncClient, RoomBuilder};
+    /// # use matrix_sdk::api::r0::room::Visibility;
+    /// # use url::Url;
+    ///
+    /// # let homeserver = Url::parse("http://example.com").unwrap();
+    /// let mut builder = RoomBuilder::default();
+    /// builder.creation_content(false)
+    ///     .initial_state(vec![])
+    ///     .visibility(Visibility::Public)
+    ///     .name("name")
+    ///     .room_version("v1.0");
+    ///
+    /// let mut cli = AsyncClient::new(homeserver, None).unwrap();
+    /// # use futures::executor::block_on;
+    /// # block_on(async {
+    /// assert!(cli.create_room(builder).await.is_ok());
+    /// # });
+    /// ```
+    pub async fn create_room<R: Into<create_room::Request>>(
+        &self,
+        room: R,
+    ) -> Result<create_room::Response> {
+        let request = room.into();
+        self.send(request).await
+    }
+
+    /// Get messages starting at a specific sync point using the
+    /// `MessagesRequestBuilder`s `from` field as a starting point.
+    ///
+    /// Sends a request to `/_matrix/client/r0/rooms/{room_id}/messages` and
+    /// returns a `get_message_events::IncomingResponse` that contains chunks
+    /// of `RoomEvents`.
+    ///
+    /// # Arguments
+    ///
+    /// * request - The easiest way to create a `Request` is using the
+    /// `MessagesRequestBuilder`.
+    ///
+    /// # Examples
+    /// ```no_run
+    /// # use std::convert::TryFrom;
+    /// use matrix_sdk::{AsyncClient, MessagesRequestBuilder};
+    /// # use matrix_sdk::identifiers::RoomId;
+    /// # use matrix_sdk::api::r0::filter::RoomEventFilter;
+    /// # use matrix_sdk::api::r0::message::get_message_events::Direction;
+    /// # use url::Url;
+    /// # use js_int::UInt;
+    ///
+    /// # let homeserver = Url::parse("http://example.com").unwrap();
+    /// let mut builder = MessagesRequestBuilder::new();
+    /// builder.room_id(RoomId::try_from("!roomid:example.com").unwrap())
+    ///     .from("t47429-4392820_219380_26003_2265".to_string())
+    ///     .to("t4357353_219380_26003_2265".to_string())
+    ///     .direction(Direction::Backward)
+    ///     .limit(UInt::new(10).unwrap());
+    ///
+    /// let mut cli = AsyncClient::new(homeserver, None).unwrap();
+    /// # use futures::executor::block_on;
+    /// # block_on(async {
+    /// assert!(cli.room_messages(builder).await.is_ok());
+    /// # });
+    /// ```
+    pub async fn room_messages<R: Into<get_message_events::Request>>(
+        &self,
+        request: R,
+    ) -> Result<get_message_events::IncomingResponse> {
+        let req = request.into();
+        self.send(req).await
+    }
+
     /// Synchronize the client's state with the latest state on the server.
     ///
     /// # Arguments
     ///
     /// * `sync_settings` - Settings for the sync call.
     #[instrument]
-    pub async fn sync(
+    pub async fn sync(&self, sync_settings: SyncSettings) -> Result<sync_events::IncomingResponse> {
-        &mut self,
-        sync_settings: SyncSettings,
-    ) -> Result<sync_events::IncomingResponse> {
         let request = sync_events::Request {
             filter: None,
             since: sync_settings.token,
@@ -354,9 +561,8 @@ impl AsyncClient {
         let mut response = self.send(request).await?;
 
         for (room_id, room) in &mut response.rooms.join {
-            let mut client = self.base_client.write().await;
-
             let _matrix_room = {
+                let mut client = self.base_client.write().await;
                 for event in &room.state.events {
                     if let EventResult::Ok(e) = event {
                         client.receive_joined_state_event(&room_id, &e).await;
@@ -369,12 +575,14 @@ impl AsyncClient {
             // re looping is not ideal here
             for event in &mut room.state.events {
                 if let EventResult::Ok(e) = event {
+                    let client = self.base_client.read().await;
                     client.emit_state_event(room_id, e).await;
                 }
             }
 
             for mut event in &mut room.timeline.events {
                 let decrypted_event = {
+                    let mut client = self.base_client.write().await;
                     client
                         .receive_joined_timeline_event(room_id, &mut event)
                         .await
@@ -385,6 +593,7 @@ impl AsyncClient {
                 }
 
                 if let EventResult::Ok(e) = event {
+                    let client = self.base_client.read().await;
                     client.emit_timeline_event(room_id, e).await;
                 }
             }
@@ -393,8 +602,8 @@ impl AsyncClient {
             for account_data in &mut room.account_data.events {
                 {
                     if let EventResult::Ok(e) = account_data {
+                        let mut client = self.base_client.write().await;
                         client.receive_account_data_event(&room_id, e).await;
-
                         client.emit_account_data_event(room_id, e).await;
                     }
                 }
@@ -406,6 +615,7 @@ impl AsyncClient {
             for presence in &mut response.presence.events {
                 {
                     if let EventResult::Ok(e) = presence {
+                        let mut client = self.base_client.write().await;
                         client.receive_presence_event(&room_id, e).await;
 
                         client.emit_presence_event(room_id, e).await;
@@ -416,6 +626,7 @@ impl AsyncClient {
             for ephemeral in &mut room.ephemeral.events {
                 {
                     if let EventResult::Ok(e) = ephemeral {
+                        let mut client = self.base_client.write().await;
                         client.receive_ephemeral_event(&room_id, e).await;
 
                         client.emit_ephemeral_event(room_id, e).await;
@@ -487,7 +698,7 @@ impl AsyncClient {
     /// ```
     #[instrument(skip(callback))]
     pub async fn sync_forever<C>(
-        &mut self,
+        &self,
         sync_settings: SyncSettings,
         callback: impl Fn(sync_events::IncomingResponse) -> C + Send,
     ) where
@@ -599,7 +810,6 @@ impl AsyncClient {
         } else {
             request_builder
         };
-
         let mut response = request_builder.send().await?;
 
         trace!("Got response: {:?}", response);
@@ -635,6 +845,10 @@ impl AsyncClient {
     ///
     /// * `content` - The content of the message event.
     ///
+    /// * `txn_id` - A unique `Uuid` that can be attached to a `MessageEvent` held
+    /// in it's unsigned field as `transaction_id`. If not given one is created for the
+    /// message.
+    ///
     /// # Example
     /// ```no_run
     /// # use matrix_sdk::Room;
@@ -649,6 +863,7 @@ impl AsyncClient {
     /// # let homeserver = Url::parse("http://localhost:8080").unwrap();
     /// # let mut client = AsyncClient::new(homeserver, None).unwrap();
     /// # let room_id = RoomId::try_from("!test:localhost").unwrap();
+    /// use uuid::Uuid;
     ///
     /// let content = MessageEventContent::Text(TextMessageEventContent {
     ///     body: "Hello world".to_owned(),
@@ -656,14 +871,15 @@ impl AsyncClient {
     ///     formatted_body: None,
     ///     relates_to: None,
     /// });
-    ///
+    /// let txn_id = Uuid::new_v4();
-    /// client.room_send(&room_id, content).await.unwrap();
+    /// client.room_send(&room_id, content, Some(txn_id)).await.unwrap();
     /// })
     /// ```
     pub async fn room_send(
-        &mut self,
+        &self,
         room_id: &RoomId,
         #[allow(unused_mut)] mut content: MessageEventContent,
+        txn_id: Option<Uuid>,
     ) -> Result<create_message_event::Response> {
         #[allow(unused_mut)]
         let mut event_type = EventType::RoomMessage;
@@ -675,7 +891,7 @@ impl AsyncClient {
                 let room = client.joined_rooms.get(room_id);
 
                 match room {
-                    Some(r) => r.lock().await.is_encrypted(),
+                    Some(r) => r.read().await.is_encrypted(),
                     None => false,
                 }
             };
@@ -684,7 +900,7 @@ impl AsyncClient {
                 let missing_sessions = {
                     let client = self.base_client.read().await;
                     let room = client.joined_rooms.get(room_id);
-                    let room = room.as_ref().unwrap().lock().await;
+                    let room = room.as_ref().unwrap().read().await;
                     let users = room.members.keys();
                     self.base_client
                         .read()
@@ -722,7 +938,7 @@ impl AsyncClient {
         let request = create_message_event::Request {
             room_id: room_id.clone(),
             event_type,
-            txn_id: Uuid::new_v4().to_string(),
+            txn_id: txn_id.unwrap_or_else(Uuid::new_v4).to_string(),
             data: content,
         };
 

src/base_client.rs

@@ -36,7 +36,9 @@ use crate::models::Room;
 use crate::session::Session;
 use crate::EventEmitter;
 
+#[cfg(feature = "encryption")]
 use tokio::sync::Mutex;
+use tokio::sync::RwLock;
 
 #[cfg(feature = "encryption")]
 use crate::crypto::{OlmMachine, OneTimeKeys};
@@ -65,14 +67,14 @@ pub struct Client {
     /// The current sync token that should be used for the next sync call.
     pub sync_token: Option<Token>,
     /// A map of the rooms our user is joined in.
-    pub joined_rooms: HashMap<RoomId, Arc<Mutex<Room>>>,
+    pub joined_rooms: HashMap<RoomId, Arc<RwLock<Room>>>,
     /// A list of ignored users.
     pub ignored_users: Vec<UserId>,
     /// The push ruleset for the logged in user.
     pub push_ruleset: Option<Ruleset>,
     /// Any implementor of EventEmitter will act as the callbacks for various
     /// events.
-    pub event_emitter: Option<Arc<Mutex<Box<dyn EventEmitter>>>>,
+    pub event_emitter: Option<Box<dyn EventEmitter>>,
 
     #[cfg(feature = "encryption")]
     olm: Arc<Mutex<Option<OlmMachine>>>,
@@ -125,10 +127,7 @@ impl Client {
     /// Add `EventEmitter` to `Client`.
     ///
     /// The methods of `EventEmitter` are called when the respective `RoomEvents` occur.
-    pub async fn add_event_emitter(
+    pub async fn add_event_emitter(&mut self, emitter: Box<dyn EventEmitter>) {
-        &mut self,
-        emitter: Arc<tokio::sync::Mutex<Box<dyn EventEmitter>>>,
-    ) {
         self.event_emitter = Some(emitter);
     }
 
@@ -160,7 +159,7 @@ impl Client {
 
     pub(crate) async fn calculate_room_name(&self, room_id: &RoomId) -> Option<String> {
         if let Some(room) = self.joined_rooms.get(room_id) {
-            let room = room.lock().await;
+            let room = room.read().await;
             Some(room.room_name.calculate_name(room_id, &room.members))
         } else {
             None
@@ -170,17 +169,17 @@ impl Client {
     pub(crate) async fn calculate_room_names(&self) -> Vec<String> {
         let mut res = Vec::new();
         for (id, room) in &self.joined_rooms {
-            let room = room.lock().await;
+            let room = room.read().await;
             res.push(room.room_name.calculate_name(id, &room.members))
         }
         res
     }
 
-    pub(crate) fn get_or_create_room(&mut self, room_id: &RoomId) -> &mut Arc<Mutex<Room>> {
+    pub(crate) fn get_or_create_room(&mut self, room_id: &RoomId) -> &mut Arc<RwLock<Room>> {
         #[allow(clippy::or_fun_call)]
         self.joined_rooms
             .entry(room_id.clone())
-            .or_insert(Arc::new(Mutex::new(Room::new(
+            .or_insert(Arc::new(RwLock::new(Room::new(
                 room_id,
                 &self
                     .session
@@ -190,7 +189,7 @@ impl Client {
             ))))
     }
 
-    pub(crate) fn get_room(&self, room_id: &RoomId) -> Option<&Arc<Mutex<Room>>> {
+    pub(crate) fn get_room(&self, room_id: &RoomId) -> Option<&Arc<RwLock<Room>>> {
         self.joined_rooms.get(room_id)
     }
 
@@ -259,7 +258,7 @@ impl Client {
                     }
                 }
 
-                let mut room = self.get_or_create_room(&room_id).lock().await;
+                let mut room = self.get_or_create_room(&room_id).write().await;
                 room.receive_timeline_event(e);
                 decrypted_event
             }
@@ -282,7 +281,7 @@ impl Client {
         room_id: &RoomId,
         event: &StateEvent,
     ) -> bool {
-        let mut room = self.get_or_create_room(room_id).lock().await;
+        let mut room = self.get_or_create_room(room_id).write().await;
         room.receive_state_event(event)
     }
 
@@ -303,7 +302,7 @@ impl Client {
     ) -> bool {
         // this should be the room that was just created in the `Client::sync` loop.
         if let Some(room) = self.get_room(room_id) {
-            let mut room = room.lock().await;
+            let mut room = room.write().await;
             room.receive_presence_event(event)
         } else {
             false
@@ -376,7 +375,7 @@ impl Client {
                 // part where we already iterate through the rooms to avoid yet
                 // another room loop.
                 for room in self.joined_rooms.values() {
-                    let room = room.lock().await;
+                    let room = room.write().await;
                     if !room.is_encrypted() {
                         continue;
                     }
@@ -459,7 +458,7 @@ impl Client {
 
         match &mut *olm {
             Some(o) => {
-                let room = room.lock().await;
+                let room = room.write().await;
                 let members = room.members.keys();
                 Ok(o.share_group_session(room_id, members).await?)
             }
@@ -573,37 +572,26 @@ impl Client {
         Ok(())
     }
 
-    pub(crate) async fn emit_timeline_event(&mut self, room_id: &RoomId, event: &mut RoomEvent) {
+    pub(crate) async fn emit_timeline_event(&self, room_id: &RoomId, event: &RoomEvent) {
         match event {
             RoomEvent::RoomMember(mem) => {
                 if let Some(ee) = &self.event_emitter {
                     if let Some(room) = self.get_room(&room_id) {
-                        ee.lock()
+                        ee.on_room_member(Arc::clone(&room), &mem).await;
-                            .await
-                            .on_room_member(Arc::clone(&room), Arc::new(Mutex::new(mem.clone())))
-                            .await;
                     }
                 }
             }
             RoomEvent::RoomName(name) => {
                 if let Some(ee) = &self.event_emitter {
                     if let Some(room) = self.get_room(&room_id) {
-                        ee.lock()
+                        ee.on_room_name(Arc::clone(&room), &name).await;
-                            .await
-                            .on_room_name(Arc::clone(&room), Arc::new(Mutex::new(name.clone())))
-                            .await;
                     }
                 }
             }
             RoomEvent::RoomCanonicalAlias(canonical) => {
                 if let Some(ee) = &self.event_emitter {
                     if let Some(room) = self.get_room(&room_id) {
-                        ee.lock()
+                        ee.on_room_canonical_alias(Arc::clone(&room), &canonical)
-                            .await
-                            .on_room_canonical_alias(
-                                Arc::clone(&room),
-                                Arc::new(Mutex::new(canonical.clone())),
-                            )
                             .await;
                     }
                 }
@@ -611,45 +599,28 @@ impl Client {
             RoomEvent::RoomAliases(aliases) => {
                 if let Some(ee) = &self.event_emitter {
                     if let Some(room) = self.get_room(&room_id) {
-                        ee.lock()
+                        ee.on_room_aliases(Arc::clone(&room), &aliases).await;
-                            .await
-                            .on_room_aliases(
-                                Arc::clone(&room),
-                                Arc::new(Mutex::new(aliases.clone())),
-                            )
-                            .await;
                     }
                 }
             }
             RoomEvent::RoomAvatar(avatar) => {
                 if let Some(ee) = &self.event_emitter {
                     if let Some(room) = self.get_room(&room_id) {
-                        ee.lock()
+                        ee.on_room_avatar(Arc::clone(&room), &avatar).await;
-                            .await
-                            .on_room_avatar(Arc::clone(&room), Arc::new(Mutex::new(avatar.clone())))
-                            .await;
                     }
                 }
             }
             RoomEvent::RoomMessage(msg) => {
                 if let Some(ee) = &self.event_emitter {
                     if let Some(room) = self.get_room(&room_id) {
-                        ee.lock()
+                        ee.on_room_message(Arc::clone(&room), &msg).await;
-                            .await
-                            .on_room_message(Arc::clone(&room), Arc::new(Mutex::new(msg.clone())))
-                            .await;
                     }
                 }
             }
             RoomEvent::RoomMessageFeedback(msg_feedback) => {
                 if let Some(ee) = &self.event_emitter {
                     if let Some(room) = self.get_room(&room_id) {
-                        ee.lock()
+                        ee.on_room_message_feedback(Arc::clone(&room), &msg_feedback)
-                            .await
-                            .on_room_message_feedback(
-                                Arc::clone(&room),
-                                Arc::new(Mutex::new(msg_feedback.clone())),
-                            )
                             .await;
                     }
                 }
@@ -657,26 +628,14 @@ impl Client {
             RoomEvent::RoomRedaction(redaction) => {
                 if let Some(ee) = &self.event_emitter {
                     if let Some(room) = self.get_room(&room_id) {
-                        ee.lock()
+                        ee.on_room_redaction(Arc::clone(&room), &redaction).await;
-                            .await
-                            .on_room_redaction(
-                                Arc::clone(&room),
-                                Arc::new(Mutex::new(redaction.clone())),
-                            )
-                            .await;
                     }
                 }
             }
             RoomEvent::RoomPowerLevels(power) => {
                 if let Some(ee) = &self.event_emitter {
                     if let Some(room) = self.get_room(&room_id) {
-                        ee.lock()
+                        ee.on_room_power_levels(Arc::clone(&room), &power).await;
-                            .await
-                            .on_room_power_levels(
-                                Arc::clone(&room),
-                                Arc::new(Mutex::new(power.clone())),
-                            )
-                            .await;
                     }
                 }
             }
@@ -684,40 +643,26 @@ impl Client {
         }
     }
 
-    pub(crate) async fn emit_state_event(&mut self, room_id: &RoomId, event: &mut StateEvent) {
+    pub(crate) async fn emit_state_event(&self, room_id: &RoomId, event: &StateEvent) {
         match event {
             StateEvent::RoomMember(member) => {
                 if let Some(ee) = &self.event_emitter {
                     if let Some(room) = self.get_room(&room_id) {
-                        ee.lock()
+                        ee.on_state_member(Arc::clone(&room), &member).await;
-                            .await
-                            .on_state_member(
-                                Arc::clone(&room),
-                                Arc::new(Mutex::new(member.clone())),
-                            )
-                            .await;
                     }
                 }
             }
             StateEvent::RoomName(name) => {
                 if let Some(ee) = &self.event_emitter {
                     if let Some(room) = self.get_room(&room_id) {
-                        ee.lock()
+                        ee.on_state_name(Arc::clone(&room), &name).await;
-                            .await
-                            .on_state_name(Arc::clone(&room), Arc::new(Mutex::new(name.clone())))
-                            .await;
                     }
                 }
             }
             StateEvent::RoomCanonicalAlias(canonical) => {
                 if let Some(ee) = &self.event_emitter {
                     if let Some(room) = self.get_room(&room_id) {
-                        ee.lock()
+                        ee.on_state_canonical_alias(Arc::clone(&room), &canonical)
-                            .await
-                            .on_state_canonical_alias(
-                                Arc::clone(&room),
-                                Arc::new(Mutex::new(canonical.clone())),
-                            )
                             .await;
                     }
                 }
@@ -725,52 +670,28 @@ impl Client {
             StateEvent::RoomAliases(aliases) => {
                 if let Some(ee) = &self.event_emitter {
                     if let Some(room) = self.get_room(&room_id) {
-                        ee.lock()
+                        ee.on_state_aliases(Arc::clone(&room), &aliases).await;
-                            .await
-                            .on_state_aliases(
-                                Arc::clone(&room),
-                                Arc::new(Mutex::new(aliases.clone())),
-                            )
-                            .await;
                     }
                 }
             }
             StateEvent::RoomAvatar(avatar) => {
                 if let Some(ee) = &self.event_emitter {
                     if let Some(room) = self.get_room(&room_id) {
-                        ee.lock()
+                        ee.on_state_avatar(Arc::clone(&room), &avatar).await;
-                            .await
-                            .on_state_avatar(
-                                Arc::clone(&room),
-                                Arc::new(Mutex::new(avatar.clone())),
-                            )
-                            .await;
                     }
                 }
             }
             StateEvent::RoomPowerLevels(power) => {
                 if let Some(ee) = &self.event_emitter {
                     if let Some(room) = self.get_room(&room_id) {
-                        ee.lock()
+                        ee.on_state_power_levels(Arc::clone(&room), &power).await;
-                            .await
-                            .on_state_power_levels(
-                                Arc::clone(&room),
-                                Arc::new(Mutex::new(power.clone())),
-                            )
-                            .await;
                     }
                 }
             }
             StateEvent::RoomJoinRules(rules) => {
                 if let Some(ee) = &self.event_emitter {
                     if let Some(room) = self.get_room(&room_id) {
-                        ee.lock()
+                        ee.on_state_join_rules(Arc::clone(&room), &rules).await;
-                            .await
-                            .on_state_join_rules(
-                                Arc::clone(&room),
-                                Arc::new(Mutex::new(rules.clone())),
-                            )
-                            .await;
                     }
                 }
             }
@@ -778,34 +699,19 @@ impl Client {
         }
     }
 
-    pub(crate) async fn emit_account_data_event(
+    pub(crate) async fn emit_account_data_event(&self, room_id: &RoomId, event: &NonRoomEvent) {
-        &mut self,
-        room_id: &RoomId,
-        event: &mut NonRoomEvent,
-    ) {
         match event {
             NonRoomEvent::Presence(presence) => {
                 if let Some(ee) = &self.event_emitter {
                     if let Some(room) = self.get_room(&room_id) {
-                        ee.lock()
+                        ee.on_account_presence(Arc::clone(&room), &presence).await;
-                            .await
-                            .on_account_presence(
-                                Arc::clone(&room),
-                                Arc::new(Mutex::new(presence.clone())),
-                            )
-                            .await;
                     }
                 }
             }
             NonRoomEvent::IgnoredUserList(ignored) => {
                 if let Some(ee) = &self.event_emitter {
                     if let Some(room) = self.get_room(&room_id) {
-                        ee.lock()
+                        ee.on_account_ignored_users(Arc::clone(&room), &ignored)
-                            .await
-                            .on_account_ignored_users(
-                                Arc::clone(&room),
-                                Arc::new(Mutex::new(ignored.clone())),
-                            )
                             .await;
                     }
                 }
@@ -813,25 +719,14 @@ impl Client {
             NonRoomEvent::PushRules(rules) => {
                 if let Some(ee) = &self.event_emitter {
                     if let Some(room) = self.get_room(&room_id) {
-                        ee.lock()
+                        ee.on_account_push_rules(Arc::clone(&room), &rules).await;
-                            .await
-                            .on_account_push_rules(
-                                Arc::clone(&room),
-                                Arc::new(Mutex::new(rules.clone())),
-                            )
-                            .await;
                     }
                 }
             }
             NonRoomEvent::FullyRead(full_read) => {
                 if let Some(ee) = &self.event_emitter {
                     if let Some(room) = self.get_room(&room_id) {
-                        ee.lock()
+                        ee.on_account_data_fully_read(Arc::clone(&room), &full_read)
-                            .await
-                            .on_account_data_fully_read(
-                                Arc::clone(&room),
-                                Arc::new(Mutex::new(full_read.clone())),
-                            )
                             .await;
                     }
                 }
@@ -840,34 +735,19 @@ impl Client {
         }
     }
 
-    pub(crate) async fn emit_ephemeral_event(
+    pub(crate) async fn emit_ephemeral_event(&self, room_id: &RoomId, event: &NonRoomEvent) {
-        &mut self,
-        room_id: &RoomId,
-        event: &mut NonRoomEvent,
-    ) {
         match event {
             NonRoomEvent::Presence(presence) => {
                 if let Some(ee) = &self.event_emitter {
                     if let Some(room) = self.get_room(&room_id) {
-                        ee.lock()
+                        ee.on_account_presence(Arc::clone(&room), &presence).await;
-                            .await
-                            .on_account_presence(
-                                Arc::clone(&room),
-                                Arc::new(Mutex::new(presence.clone())),
-                            )
-                            .await;
                     }
                 }
             }
             NonRoomEvent::IgnoredUserList(ignored) => {
                 if let Some(ee) = &self.event_emitter {
                     if let Some(room) = self.get_room(&room_id) {
-                        ee.lock()
+                        ee.on_account_ignored_users(Arc::clone(&room), &ignored)
-                            .await
-                            .on_account_ignored_users(
-                                Arc::clone(&room),
-                                Arc::new(Mutex::new(ignored.clone())),
-                            )
                             .await;
                     }
                 }
@@ -875,25 +755,14 @@ impl Client {
             NonRoomEvent::PushRules(rules) => {
                 if let Some(ee) = &self.event_emitter {
                     if let Some(room) = self.get_room(&room_id) {
-                        ee.lock()
+                        ee.on_account_push_rules(Arc::clone(&room), &rules).await;
-                            .await
-                            .on_account_push_rules(
-                                Arc::clone(&room),
-                                Arc::new(Mutex::new(rules.clone())),
-                            )
-                            .await;
                     }
                 }
             }
             NonRoomEvent::FullyRead(full_read) => {
                 if let Some(ee) = &self.event_emitter {
                     if let Some(room) = self.get_room(&room_id) {
-                        ee.lock()
+                        ee.on_account_data_fully_read(Arc::clone(&room), &full_read)
-                            .await
-                            .on_account_data_fully_read(
-                                Arc::clone(&room),
-                                Arc::new(Mutex::new(full_read.clone())),
-                            )
                             .await;
                     }
                 }
@@ -902,17 +771,10 @@ impl Client {
         }
     }
 
-    pub(crate) async fn emit_presence_event(
+    pub(crate) async fn emit_presence_event(&self, room_id: &RoomId, event: &PresenceEvent) {
-        &mut self,
-        room_id: &RoomId,
-        event: &mut PresenceEvent,
-    ) {
         if let Some(ee) = &self.event_emitter {
             if let Some(room) = self.get_room(&room_id) {
-                ee.lock()
+                ee.on_presence_event(Arc::clone(&room), &event).await;
-                    .await
-                    .on_presence_event(Arc::clone(&room), Arc::new(Mutex::new(event.clone())))
-                    .await;
             }
         }
     }
@@ -949,7 +811,7 @@ mod test {
         .with_body_from_file("tests/data/sync.json")
         .create();
 
-        let mut client = AsyncClient::new(homeserver, Some(session)).unwrap();
+        let client = AsyncClient::new(homeserver, Some(session)).unwrap();
 
         let sync_settings = SyncSettings::new().timeout(Duration::from_millis(3000));
 

src/crypto/device.rs

@@ -18,9 +18,9 @@ use std::sync::Arc;
 
 use atomic::Atomic;
 
-use ruma_client_api::r0::keys::{DeviceKeys, KeyAlgorithm};
+use crate::api::r0::keys::{DeviceKeys, KeyAlgorithm};
-use ruma_events::Algorithm;
+use crate::events::Algorithm;
-use ruma_identifiers::{DeviceId, UserId};
+use crate::identifiers::{DeviceId, UserId};
 
 #[derive(Debug, Clone)]
 pub struct Device {
@@ -33,12 +33,6 @@ pub struct Device {
     trust_state: Arc<Atomic<TrustState>>,
 }
 
-impl Device {
-    pub fn keys(&self, algorithm: &KeyAlgorithm) -> Option<&String> {
-        self.keys.get(algorithm)
-    }
-}
-
 #[derive(Debug, Clone, Copy)]
 pub enum TrustState {
     Verified,
@@ -55,6 +49,10 @@ impl Device {
     pub fn user_id(&self) -> &UserId {
         &self.user_id
     }
+
+    pub fn keys(&self, algorithm: &KeyAlgorithm) -> Option<&String> {
+        self.keys.get(algorithm)
+    }
 }
 
 impl From<&DeviceKeys> for Device {
@@ -82,3 +80,69 @@ impl From<&DeviceKeys> for Device {
         }
     }
 }
+
+impl PartialEq for Device {
+    fn eq(&self, other: &Self) -> bool {
+        self.user_id() == other.user_id() && self.device_id() == other.device_id()
+    }
+}
+
+#[cfg(test)]
+pub(crate) mod test {
+    use serde_json::json;
+    use std::convert::{From, TryFrom};
+
+    use crate::api::r0::keys::{DeviceKeys, KeyAlgorithm};
+    use crate::crypto::device::Device;
+    use crate::identifiers::UserId;
+
+    pub(crate) fn get_device() -> Device {
+        let user_id = UserId::try_from("@alice:example.org").unwrap();
+        let device_id = "DEVICEID";
+
+        let device_keys = json!({
+          "algorithms": vec![
+              "m.olm.v1.curve25519-aes-sha2",
+              "m.megolm.v1.aes-sha2"
+          ],
+          "device_id": device_id,
+          "user_id": user_id.to_string(),
+          "keys": {
+              "curve25519:DEVICEID": "wjLpTLRqbqBzLs63aYaEv2Boi6cFEbbM/sSRQ2oAKk4",
+              "ed25519:DEVICEID": "nE6W2fCblxDcOFmeEtCHNl8/l8bXcu7GKyAswA4r3mM"
+          },
+          "signatures": {
+              user_id.to_string(): {
+                  "ed25519:DEVICEID": "m53Wkbh2HXkc3vFApZvCrfXcX3AI51GsDHustMhKwlv3TuOJMj4wistcOTM8q2+e/Ro7rWFUb9ZfnNbwptSUBA"
+              }
+          },
+          "unsigned": {
+              "device_display_name": "Alice's mobile phone"
+          }
+        });
+
+        let device_keys: DeviceKeys = serde_json::from_value(device_keys).unwrap();
+
+        Device::from(&device_keys)
+    }
+
+    #[test]
+    fn create_a_device() {
+        let user_id = UserId::try_from("@alice:example.org").unwrap();
+        let device_id = "DEVICEID";
+
+        let device = get_device();
+
+        assert_eq!(&user_id, device.user_id());
+        assert_eq!(device_id, device.device_id());
+        assert_eq!(device.algorithms.len(), 2);
+        assert_eq!(
+            device.keys(&KeyAlgorithm::Curve25519).unwrap(),
+            "wjLpTLRqbqBzLs63aYaEv2Boi6cFEbbM/sSRQ2oAKk4"
+        );
+        assert_eq!(
+            device.keys(&KeyAlgorithm::Ed25519).unwrap(),
+            "nE6W2fCblxDcOFmeEtCHNl8/l8bXcu7GKyAswA4r3mM"
+        );
+    }
+}

src/crypto/machine.rs

@@ -18,7 +18,6 @@ use std::mem;
 #[cfg(feature = "sqlite-cryptostore")]
 use std::path::Path;
 use std::result::Result as StdResult;
-use std::sync::Arc;
 use uuid::Uuid;
 
 use super::error::{OlmError, Result, SignatureError, VerificationResult};
@@ -34,7 +33,6 @@ use api::r0::keys;
 use cjson;
 use olm_rs::{session::OlmMessage, utility::OlmUtility};
 use serde_json::{json, Value};
-use tokio::sync::Mutex;
 use tracing::{debug, error, info, instrument, trace, warn};
 
 use ruma_client_api::r0::client_exchange::{
@@ -658,19 +656,17 @@ impl OlmMachine {
             return Ok(None);
         };
 
-        for session in &*sessions.lock().await {
+        for session in &mut *sessions.lock().await {
             let mut matches = false;
 
-            let mut session_lock = session.lock().await;
-
             if let OlmMessage::PreKey(m) = &message {
-                matches = session_lock.matches(sender_key, m.clone())?;
+                matches = session.matches(sender_key, m.clone()).await?;
                 if !matches {
                     continue;
                 }
             }
 
-            let ret = session_lock.decrypt(message.clone());
+            let ret = session.decrypt(message.clone()).await;
 
             if let Ok(p) = ret {
                 self.store.save_session(session.clone()).await?;
@@ -706,7 +702,7 @@ impl OlmMachine {
                 }
             };
 
-            let plaintext = session.decrypt(message)?;
+            let plaintext = session.decrypt(message).await?;
             self.store.add_and_save_session(session).await?;
             plaintext
         };
@@ -861,7 +857,7 @@ impl OlmMachine {
 
     async fn olm_encrypt(
         &mut self,
-        session: Arc<Mutex<Session>>,
+        mut session: Session,
         recipient_device: &Device,
         event_type: EventType,
         content: Value,
@@ -892,7 +888,7 @@ impl OlmMachine {
         let plaintext = cjson::to_string(&payload)
             .unwrap_or_else(|_| panic!(format!("Can't serialize {} to canonical JSON", payload)));
 
-        let ciphertext = session.lock().await.encrypt(&plaintext).to_tuple();
+        let ciphertext = session.encrypt(&plaintext).await.to_tuple();
         self.store.save_session(session).await?;
 
         let message_type: usize = ciphertext.0.into();

src/crypto/memory_stores.rs

@@ -22,54 +22,59 @@ use super::device::Device;
 use super::olm::{InboundGroupSession, Session};
 use crate::identifiers::{DeviceId, RoomId, UserId};
 
+/// In-memory store for Olm Sessions.
 #[derive(Debug)]
 pub struct SessionStore {
-    entries: HashMap<String, Arc<Mutex<Vec<Arc<Mutex<Session>>>>>>,
+    entries: HashMap<String, Arc<Mutex<Vec<Session>>>>,
 }
 
 impl SessionStore {
+    /// Create a new empty Session store.
     pub fn new() -> Self {
         SessionStore {
             entries: HashMap::new(),
         }
     }
 
-    pub async fn add(&mut self, session: Session) -> Arc<Mutex<Session>> {
+    /// Add a session to the store.
-        if !self.entries.contains_key(&session.sender_key) {
+    pub async fn add(&mut self, session: Session) {
+        if !self.entries.contains_key(&*session.sender_key) {
             self.entries.insert(
-                session.sender_key.to_owned(),
+                session.sender_key.to_string(),
                 Arc::new(Mutex::new(Vec::new())),
             );
         }
-        let sessions = self.entries.get_mut(&session.sender_key).unwrap();
+        let sessions = self.entries.get_mut(&*session.sender_key).unwrap();
-        let session = Arc::new(Mutex::new(session));
+        sessions.lock().await.push(session);
-        sessions.lock().await.push(session.clone());
-
-        session
     }
 
-    pub fn get(&self, sender_key: &str) -> Option<Arc<Mutex<Vec<Arc<Mutex<Session>>>>>> {
+    /// Get all the sessions that belong to the given sender key.
+    pub fn get(&self, sender_key: &str) -> Option<Arc<Mutex<Vec<Session>>>> {
         self.entries.get(sender_key).cloned()
     }
 
-    pub fn set_for_sender(&mut self, sender_key: &str, sessions: Vec<Arc<Mutex<Session>>>) {
+    /// Add a list of sessions belonging to the sender key.
+    pub fn set_for_sender(&mut self, sender_key: &str, sessions: Vec<Session>) {
         self.entries
             .insert(sender_key.to_owned(), Arc::new(Mutex::new(sessions)));
     }
 }
 
 #[derive(Debug)]
+/// In-memory store that houlds inbound group sessions.
 pub struct GroupSessionStore {
     entries: HashMap<RoomId, HashMap<String, HashMap<String, InboundGroupSession>>>,
 }
 
 impl GroupSessionStore {
+    /// Create a new empty store.
     pub fn new() -> Self {
         GroupSessionStore {
             entries: HashMap::new(),
         }
     }
 
+    /// Add a inbound group session to the store.
     pub fn add(&mut self, session: InboundGroupSession) -> bool {
         if !self.entries.contains_key(&session.room_id) {
             let room_id = &*session.room_id;
@@ -89,6 +94,14 @@ impl GroupSessionStore {
         ret.is_some()
     }
 
+    /// Get a inbound group session from our store.
+    ///
+    /// # Arguments
+    /// * `room_id` - The room id of the room that the session belongs to.
+    ///
+    /// * `sender_key` - The sender key that sent us the session.
+    ///
+    /// * `session_id` - The unique id of the session.
     pub fn get(
         &self,
         room_id: &RoomId,
@@ -101,36 +114,45 @@ impl GroupSessionStore {
     }
 }
 
+/// In-memory store holding the devices of users.
 #[derive(Clone, Debug)]
 pub struct DeviceStore {
     entries: Arc<DashMap<UserId, DashMap<String, Device>>>,
 }
 
+/// A read only view over all devices belonging to a user.
 pub struct UserDevices {
     entries: ReadOnlyView<DeviceId, Device>,
 }
 
 impl UserDevices {
+    /// Get the specific device with the given device id.
     pub fn get(&self, device_id: &str) -> Option<Device> {
         self.entries.get(device_id).cloned()
     }
 
+    /// Iterator over all the device ids of the user devices.
     pub fn keys(&self) -> impl Iterator<Item = &DeviceId> {
         self.entries.keys()
     }
 
+    /// Iterator over all the devices of the user devices.
     pub fn devices(&self) -> impl Iterator<Item = &Device> {
         self.entries.values()
     }
 }
 
 impl DeviceStore {
+    /// Create a new empty device store.
     pub fn new() -> Self {
         DeviceStore {
             entries: Arc::new(DashMap::new()),
         }
     }
 
+    /// Add a device to the store.
+    ///
+    /// Returns true if the device was already in the store, false otherwise.
     pub fn add(&self, device: Device) -> bool {
         let user_id = device.user_id();
 
@@ -144,12 +166,14 @@ impl DeviceStore {
             .is_some()
     }
 
+    /// Get the device with the given device_id and belonging to the given user.
     pub fn get(&self, user_id: &UserId, device_id: &str) -> Option<Device> {
         self.entries
             .get(user_id)
             .and_then(|m| m.get(device_id).map(|d| d.value().clone()))
     }
 
+    /// Get a read-only view over all devices of the given user.
     pub fn user_devices(&self, user_id: &UserId) -> UserDevices {
         if !self.entries.contains_key(user_id) {
             self.entries.insert(user_id.clone(), DashMap::new());
@@ -159,3 +183,124 @@ impl DeviceStore {
         }
     }
 }
+
+#[cfg(test)]
+mod test {
+    use std::collections::HashMap;
+    use std::convert::TryFrom;
+
+    use crate::api::r0::keys::SignedKey;
+    use crate::crypto::device::test::get_device;
+    use crate::crypto::memory_stores::{DeviceStore, GroupSessionStore, SessionStore};
+    use crate::crypto::olm::{Account, InboundGroupSession, OutboundGroupSession, Session};
+    use crate::identifiers::RoomId;
+
+    async fn get_account_and_session() -> (Account, Session) {
+        let alice = Account::new();
+
+        let bob = Account::new();
+
+        bob.generate_one_time_keys(1).await;
+        let one_time_key = bob
+            .one_time_keys()
+            .await
+            .curve25519()
+            .iter()
+            .nth(0)
+            .unwrap()
+            .1
+            .to_owned();
+        let one_time_key = SignedKey {
+            key: one_time_key,
+            signatures: HashMap::new(),
+        };
+        let sender_key = bob.identity_keys().curve25519().to_owned();
+        let session = alice
+            .create_outbound_session(&sender_key, &one_time_key)
+            .await
+            .unwrap();
+
+        (alice, session)
+    }
+
+    #[tokio::test]
+    async fn test_session_store() {
+        let (account, session) = get_account_and_session().await;
+
+        let mut store = SessionStore::new();
+        store.add(session.clone()).await;
+
+        let sessions = store.get(&session.sender_key).unwrap();
+        let sessions = sessions.lock().await;
+
+        let loaded_session = &sessions[0];
+
+        assert_eq!(&session, loaded_session);
+    }
+
+    #[tokio::test]
+    async fn test_session_store_bulk_storing() {
+        let (account, session) = get_account_and_session().await;
+
+        let mut store = SessionStore::new();
+        store.set_for_sender(&session.sender_key, vec![session.clone()]);
+
+        let sessions = store.get(&session.sender_key).unwrap();
+        let sessions = sessions.lock().await;
+
+        let loaded_session = &sessions[0];
+
+        assert_eq!(&session, loaded_session);
+    }
+
+    #[tokio::test]
+    async fn test_group_session_store() {
+        let alice = Account::new();
+        let room_id = RoomId::try_from("!test:localhost").unwrap();
+
+        let outbound = OutboundGroupSession::new(&room_id);
+
+        assert_eq!(0, outbound.message_index().await);
+        assert!(!outbound.shared());
+        outbound.mark_as_shared();
+        assert!(outbound.shared());
+
+        let inbound = InboundGroupSession::new(
+            "test_key",
+            "test_key",
+            &room_id,
+            outbound.session_key().await,
+        )
+        .unwrap();
+
+        let mut store = GroupSessionStore::new();
+        store.add(inbound.clone());
+
+        let loaded_session = store
+            .get(&room_id, "test_key", outbound.session_id())
+            .unwrap();
+        assert_eq!(inbound, loaded_session);
+    }
+
+    #[tokio::test]
+    async fn test_device_store() {
+        let device = get_device();
+        let store = DeviceStore::new();
+
+        assert!(!store.add(device.clone()));
+        assert!(store.add(device.clone()));
+
+        let loaded_device = store.get(device.user_id(), device.device_id()).unwrap();
+
+        assert_eq!(device, loaded_device);
+
+        let user_devices = store.user_devices(device.user_id());
+
+        assert_eq!(user_devices.keys().nth(0).unwrap(), device.device_id());
+        assert_eq!(user_devices.devices().nth(0).unwrap(), &device);
+
+        let loaded_device = user_devices.get(device.device_id()).unwrap();
+
+        assert_eq!(device, loaded_device);
+    }
+}

src/crypto/olm.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use std::fmt;
+use std::mem;
 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::sync::Arc;
 use std::time::Instant;
@@ -38,17 +39,16 @@ use crate::identifiers::RoomId;
 pub struct Account {
     inner: Arc<Mutex<OlmAccount>>,
     identity_keys: Arc<IdentityKeys>,
-    pub(crate) shared: Arc<AtomicBool>,
+    shared: Arc<AtomicBool>,
 }
 
+#[cfg_attr(tarpaulin, skip)]
 impl fmt::Debug for Account {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        write!(
+        f.debug_struct("Account")
-            f,
+            .field("identity_keys", self.identity_keys())
-            "Olm Account: {:?}, shared: {}",
+            .field("shared", &self.shared())
-            self.identity_keys(),
+            .finish()
-            self.shared()
-        )
     }
 }
 
@@ -170,12 +170,14 @@ impl Account {
             .create_outbound_session(their_identity_key, &their_one_time_key.key)?;
 
         let now = Instant::now();
+        let session_id = session.session_id();
 
         Ok(Session {
-            inner: session,
+            inner: Arc::new(Mutex::new(session)),
-            sender_key: their_identity_key.to_owned(),
+            session_id: Arc::new(session_id),
-            creation_time: now.clone(),
+            sender_key: Arc::new(their_identity_key.to_owned()),
-            last_use_time: now,
+            creation_time: Arc::new(now.clone()),
+            last_use_time: Arc::new(now),
         })
     }
 
@@ -209,12 +211,14 @@ impl Account {
         );
 
         let now = Instant::now();
+        let session_id = session.session_id();
 
         Ok(Session {
-            inner: session,
+            inner: Arc::new(Mutex::new(session)),
-            sender_key: their_identity_key.to_owned(),
+            session_id: Arc::new(session_id),
-            creation_time: now.clone(),
+            sender_key: Arc::new(their_identity_key.to_owned()),
-            last_use_time: now,
+            creation_time: Arc::new(now.clone()),
+            last_use_time: Arc::new(now),
         })
     }
 }
@@ -225,16 +229,27 @@ impl PartialEq for Account {
     }
 }
 
-#[derive(Debug)]
 /// The Olm Session.
 ///
 /// Sessions are used to exchange encrypted messages between two
 /// accounts/devices.
+#[derive(Clone)]
 pub struct Session {
-    inner: OlmSession,
+    inner: Arc<Mutex<OlmSession>>,
-    pub(crate) sender_key: String,
+    session_id: Arc<String>,
-    pub(crate) creation_time: Instant,
+    pub(crate) sender_key: Arc<String>,
-    pub(crate) last_use_time: Instant,
+    pub(crate) creation_time: Arc<Instant>,
+    pub(crate) last_use_time: Arc<Instant>,
+}
+
+#[cfg_attr(tarpaulin, skip)]
+impl fmt::Debug for Session {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("Session")
+            .field("session_id", &self.session_id())
+            .field("sender_key", &self.sender_key)
+            .finish()
+    }
 }
 
 impl Session {
@@ -246,9 +261,9 @@ impl Session {
     /// # Arguments
     ///
     /// * `message` - The Olm message that should be decrypted.
-    pub fn decrypt(&mut self, message: OlmMessage) -> Result<String, OlmSessionError> {
+    pub async fn decrypt(&mut self, message: OlmMessage) -> Result<String, OlmSessionError> {
-        let plaintext = self.inner.decrypt(message)?;
+        let plaintext = self.inner.lock().await.decrypt(message)?;
-        self.last_use_time = Instant::now();
+        mem::replace(&mut self.last_use_time, Arc::new(Instant::now()));
         Ok(plaintext)
     }
 
@@ -259,9 +274,9 @@ impl Session {
     /// # Arguments
     ///
     /// * `plaintext` - The plaintext that should be encrypted.
-    pub fn encrypt(&mut self, plaintext: &str) -> OlmMessage {
+    pub async fn encrypt(&mut self, plaintext: &str) -> OlmMessage {
-        let message = self.inner.encrypt(plaintext);
+        let message = self.inner.lock().await.encrypt(plaintext);
-        self.last_use_time = Instant::now();
+        mem::replace(&mut self.last_use_time, Arc::new(Instant::now()));
         message
     }
 
@@ -276,18 +291,20 @@ impl Session {
     /// that encrypted this Olm message.
     ///
     /// * `message` - The pre-key Olm message that should be checked.
-    pub fn matches(
+    pub async fn matches(
         &self,
         their_identity_key: &str,
         message: PreKeyMessage,
     ) -> Result<bool, OlmSessionError> {
         self.inner
+            .lock()
+            .await
             .matches_inbound_session_from(their_identity_key, message)
     }
 
     /// Returns the unique identifier for this session.
-    pub fn session_id(&self) -> String {
+    pub fn session_id(&self) -> &str {
-        self.inner.session_id()
+        &self.session_id
     }
 
     /// Store the session as a base64 encoded string.
@@ -296,8 +313,8 @@ impl Session {
     ///
     /// * `pickle_mode` - The mode that was used to pickle the session, either
     /// an unencrypted mode or an encrypted using passphrase.
-    pub fn pickle(&self, pickle_mode: PicklingMode) -> String {
+    pub async fn pickle(&self, pickle_mode: PicklingMode) -> String {
-        self.inner.pickle(pickle_mode)
+        self.inner.lock().await.pickle(pickle_mode)
     }
 
     /// Restore a Session from a previously pickled string.
@@ -328,11 +345,14 @@ impl Session {
         last_use_time: Instant,
     ) -> Result<Self, OlmSessionError> {
         let session = OlmSession::unpickle(pickle, pickle_mode)?;
+        let session_id = session.session_id();
+
         Ok(Session {
-            inner: session,
+            inner: Arc::new(Mutex::new(session)),
-            sender_key,
+            session_id: Arc::new(session_id),
-            creation_time,
+            sender_key: Arc::new(sender_key),
-            last_use_time,
+            creation_time: Arc::new(creation_time),
+            last_use_time: Arc::new(last_use_time),
         })
     }
 }
@@ -471,6 +491,7 @@ impl InboundGroupSession {
     }
 }
 
+#[cfg_attr(tarpaulin, skip)]
 impl fmt::Debug for InboundGroupSession {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("InboundGroupSession")
@@ -479,6 +500,12 @@ impl fmt::Debug for InboundGroupSession {
     }
 }
 
+impl PartialEq for InboundGroupSession {
+    fn eq(&self, other: &Self) -> bool {
+        self.session_id() == other.session_id()
+    }
+}
+
 /// Outbound group session.
 ///
 /// Outbound group sessions are used to exchange room messages between a group
@@ -573,6 +600,7 @@ impl OutboundGroupSession {
     }
 }
 
+#[cfg_attr(tarpaulin, skip)]
 impl std::fmt::Debug for OutboundGroupSession {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("OutboundGroupSession")
@@ -586,10 +614,12 @@ impl std::fmt::Debug for OutboundGroupSession {
 
 #[cfg(test)]
 mod test {
-    use crate::crypto::olm::Account;
+    use crate::crypto::olm::{Account, InboundGroupSession, OutboundGroupSession};
+    use crate::identifiers::RoomId;
     use olm_rs::session::OlmMessage;
     use ruma_client_api::r0::keys::SignedKey;
     use std::collections::HashMap;
+    use std::convert::TryFrom;
 
     #[test]
     fn account_creation() {
@@ -607,6 +637,9 @@ mod test {
             identyty_keys.get("ed25519").unwrap()
         );
         assert!(!identyty_keys.curve25519().is_empty());
+
+        account.mark_as_shared();
+        assert!(account.shared());
     }
 
     #[tokio::test]
@@ -665,7 +698,7 @@ mod test {
 
         let plaintext = "Hello world";
 
-        let message = bob_session.encrypt(plaintext);
+        let message = bob_session.encrypt(plaintext).await;
 
         let prekey_message = match message.clone() {
             OlmMessage::PreKey(m) => m,
@@ -674,13 +707,48 @@ mod test {
 
         let bob_keys = bob.identity_keys();
         let mut alice_session = alice
-            .create_inbound_session(bob_keys.curve25519(), prekey_message)
+            .create_inbound_session(bob_keys.curve25519(), prekey_message.clone())
             .await
             .unwrap();
 
+        assert!(alice_session
+            .matches(bob_keys.curve25519(), prekey_message)
+            .await
+            .unwrap());
+
         assert_eq!(bob_session.session_id(), alice_session.session_id());
 
-        let decyrpted = alice_session.decrypt(message).unwrap();
+        let decyrpted = alice_session.decrypt(message).await.unwrap();
         assert_eq!(plaintext, decyrpted);
     }
+
+    #[tokio::test]
+    async fn group_session_creation() {
+        let alice = Account::new();
+        let room_id = RoomId::try_from("!test:localhost").unwrap();
+
+        let outbound = OutboundGroupSession::new(&room_id);
+
+        assert_eq!(0, outbound.message_index().await);
+        assert!(!outbound.shared());
+        outbound.mark_as_shared();
+        assert!(outbound.shared());
+
+        let inbound = InboundGroupSession::new(
+            "test_key",
+            "test_key",
+            &room_id,
+            outbound.session_key().await,
+        )
+        .unwrap();
+
+        assert_eq!(0, inbound.first_known_index().await);
+
+        assert_eq!(outbound.session_id(), inbound.session_id());
+
+        let plaintext = "This is a secret to everybody".to_owned();
+        let ciphertext = outbound.encrypt(plaintext.clone()).await;
+
+        assert_eq!(plaintext, inbound.decrypt(ciphertext).await.unwrap().0);
+    }
 }

src/crypto/store/memorystore.rs

@@ -52,7 +52,7 @@ impl CryptoStore for MemoryStore {
         Ok(())
     }
 
-    async fn save_session(&mut self, _: Arc<Mutex<Session>>) -> Result<()> {
+    async fn save_session(&mut self, _: Session) -> Result<()> {
         Ok(())
     }
 
@@ -61,10 +61,7 @@ impl CryptoStore for MemoryStore {
         Ok(())
     }
 
-    async fn get_sessions(
+    async fn get_sessions(&mut self, sender_key: &str) -> Result<Option<Arc<Mutex<Vec<Session>>>>> {
-        &mut self,
-        sender_key: &str,
-    ) -> Result<Option<Arc<Mutex<Vec<Arc<Mutex<Session>>>>>>> {
         Ok(self.sessions.get(sender_key))
     }
 

src/crypto/store/mod.rs

@@ -68,12 +68,9 @@ pub trait CryptoStore: Debug + Send + Sync {
     async fn load_account(&mut self) -> Result<Option<Account>>;
     async fn save_account(&mut self, account: Account) -> Result<()>;
 
-    async fn save_session(&mut self, session: Arc<Mutex<Session>>) -> Result<()>;
+    async fn save_session(&mut self, session: Session) -> Result<()>;
     async fn add_and_save_session(&mut self, session: Session) -> Result<()>;
-    async fn get_sessions(
+    async fn get_sessions(&mut self, sender_key: &str) -> Result<Option<Arc<Mutex<Vec<Session>>>>>;
-        &mut self,
-        sender_key: &str,
-    ) -> Result<Option<Arc<Mutex<Vec<Arc<Mutex<Session>>>>>>>;
 
     async fn save_inbound_group_session(&mut self, session: InboundGroupSession) -> Result<bool>;
     async fn get_inbound_group_session(

src/crypto/store/sqlite.rs

@@ -155,7 +155,7 @@ impl SqliteStore {
     async fn get_sessions_for(
         &mut self,
         sender_key: &str,
-    ) -> Result<Option<Arc<Mutex<Vec<Arc<Mutex<Session>>>>>>> {
+    ) -> Result<Option<Arc<Mutex<Vec<Session>>>>> {
         let loaded_sessions = self.sessions.get(sender_key).is_some();
 
         if !loaded_sessions {
@@ -169,7 +169,7 @@ impl SqliteStore {
         Ok(self.sessions.get(sender_key))
     }
 
-    async fn load_sessions_for(&mut self, sender_key: &str) -> Result<Vec<Arc<Mutex<Session>>>> {
+    async fn load_sessions_for(&mut self, sender_key: &str) -> Result<Vec<Session>> {
         let account_id = self.account_id.ok_or(CryptoStoreError::AccountUnset)?;
         let mut connection = self.connection.lock().await;
 
@@ -196,15 +196,15 @@ impl SqliteStore {
                     .checked_sub(serde_json::from_str::<Duration>(&row.3)?)
                     .ok_or(CryptoStoreError::SessionTimestampError)?;
 
-                Ok(Arc::new(Mutex::new(Session::from_pickle(
+                Ok(Session::from_pickle(
                     pickle.to_string(),
                     self.get_pickle_mode(),
                     sender_key.to_string(),
                     creation_time,
                     last_use_time,
-                )?)))
+                )?)
             })
-            .collect::<Result<Vec<Arc<Mutex<Session>>>>>()?)
+            .collect::<Result<Vec<Session>>>()?)
     }
 
     async fn load_inbound_group_sessions(&self) -> Result<Vec<InboundGroupSession>> {
@@ -322,15 +322,13 @@ impl CryptoStore for SqliteStore {
         Ok(())
     }
 
-    async fn save_session(&mut self, session: Arc<Mutex<Session>>) -> Result<()> {
+    async fn save_session(&mut self, session: Session) -> Result<()> {
         let account_id = self.account_id.ok_or(CryptoStoreError::AccountUnset)?;
 
-        let session = session.lock().await;
-
         let session_id = session.session_id();
         let creation_time = serde_json::to_string(&session.creation_time.elapsed())?;
         let last_use_time = serde_json::to_string(&session.last_use_time.elapsed())?;
-        let pickle = session.pickle(self.get_pickle_mode());
+        let pickle = session.pickle(self.get_pickle_mode()).await;
 
         let mut connection = self.connection.lock().await;
 
@@ -341,9 +339,9 @@ impl CryptoStore for SqliteStore {
         )
         .bind(&session_id)
         .bind(&account_id)
-        .bind(&creation_time)
+        .bind(&*creation_time)
-        .bind(&last_use_time)
+        .bind(&*last_use_time)
-        .bind(&session.sender_key)
+        .bind(&*session.sender_key)
         .bind(&pickle)
         .execute(&mut *connection)
         .await?;
@@ -352,15 +350,12 @@ impl CryptoStore for SqliteStore {
     }
 
     async fn add_and_save_session(&mut self, session: Session) -> Result<()> {
-        let session = self.sessions.add(session).await;
+        self.sessions.add(session.clone()).await;
         self.save_session(session).await?;
         Ok(())
     }
 
-    async fn get_sessions(
+    async fn get_sessions(&mut self, sender_key: &str) -> Result<Option<Arc<Mutex<Vec<Session>>>>> {
-        &mut self,
-        sender_key: &str,
-    ) -> Result<Option<Arc<Mutex<Vec<Arc<Mutex<Session>>>>>>> {
         Ok(self.get_sessions_for(sender_key).await?)
     }
 
@@ -440,9 +435,7 @@ mod test {
     use olm_rs::outbound_group_session::OlmOutboundGroupSession;
     use ruma_client_api::r0::keys::SignedKey;
     use std::collections::HashMap;
-    use std::sync::Arc;
     use tempfile::tempdir;
-    use tokio::sync::Mutex;
 
     use super::{
         Account, CryptoStore, InboundGroupSession, RoomId, Session, SqliteStore, TryFrom, UserId,
@@ -565,7 +558,6 @@ mod test {
     async fn save_session() {
         let mut store = get_store().await;
         let (account, session) = get_account_and_session().await;
-        let session = Arc::new(Mutex::new(session));
 
         assert!(store.save_session(session.clone()).await.is_err());
 
@@ -581,22 +573,19 @@ mod test {
     async fn load_sessions() {
         let mut store = get_store().await;
         let (account, session) = get_account_and_session().await;
-        let session = Arc::new(Mutex::new(session));
         store
             .save_account(account.clone())
             .await
             .expect("Can't save account");
         store.save_session(session.clone()).await.unwrap();
 
-        let sess = session.lock().await;
-
         let sessions = store
-            .load_sessions_for(&sess.sender_key)
+            .load_sessions_for(&session.sender_key)
             .await
             .expect("Can't load sessions");
         let loaded_session = &sessions[0];
 
-        assert_eq!(*sess, *loaded_session.lock().await);
+        assert_eq!(&session, loaded_session);
     }
 
     #[tokio::test]
@@ -604,7 +593,7 @@ mod test {
         let mut store = get_store().await;
         let (account, session) = get_account_and_session().await;
         let sender_key = session.sender_key.to_owned();
-        let session_id = session.session_id();
+        let session_id = session.session_id().to_owned();
 
         store
             .save_account(account.clone())
@@ -616,7 +605,7 @@ mod test {
         let sessions_lock = sessions.lock().await;
         let session = &sessions_lock[0];
 
-        assert_eq!(session_id, *session.lock().await.session_id());
+        assert_eq!(session_id, session.session_id());
     }
 
     #[tokio::test]

src/event_emitter/mod.rs

@@ -12,7 +12,6 @@
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.
-
 use std::sync::Arc;
 
 use crate::events::{
@@ -33,8 +32,8 @@ use crate::events::{
     },
 };
 use crate::models::Room;
+use tokio::sync::RwLock;
 
-use tokio::sync::Mutex;
 /// This trait allows any type implementing `EventEmitter` to specify event callbacks for each event.
 /// The `AsyncClient` calls each method when the corresponding event is received.
 ///
@@ -51,29 +50,29 @@ use tokio::sync::Mutex;
 ///     },
 ///     AsyncClient, AsyncClientConfig, EventEmitter, Room, SyncSettings,
 /// };
-/// use tokio::sync::Mutex;
+/// use tokio::sync::RwLock;
 ///
 /// struct EventCallback;
 ///
 /// #[async_trait::async_trait]
 /// impl EventEmitter for EventCallback {
-///     async fn on_room_message(&mut self, room: Arc<Mutex<Room>>, event: Arc<Mutex<MessageEvent>>) {
+///     async fn on_room_message(&self, room: Arc<RwLock<Room>>, event: &MessageEvent) {
 ///         if let MessageEvent {
 ///             content: MessageEventContent::Text(TextMessageEventContent { body: msg_body, .. }),
 ///             sender,
 ///             ..
-///         } = event.lock().await.deref()
+///         } = event
 ///         {
-///             let rooms = room.lock().await;
+///             let name = {
-///             let member = rooms.members.get(&sender).unwrap();
+///                let room = room.read().await;
-///             println!(
+///                let member = room.members.get(&sender).unwrap();
-///                 "{}: {}",
+///                member
-///                 member
+///                    .display_name
-///                     .display_name
+///                    .as_ref()
-///                     .as_ref()
+///                    .map(ToString::to_string)
-///                     .unwrap_or(&sender.to_string()),
+///                    .unwrap_or(sender.to_string())
-///                 msg_body
+///            };
-///             );
+///             println!("{}: {}", name, msg_body);
 ///         }
 ///     }
 /// }
@@ -82,193 +81,128 @@ use tokio::sync::Mutex;
 pub trait EventEmitter: Send + Sync {
     // ROOM EVENTS from `IncomingTimeline`
     /// Fires when `AsyncClient` receives a `RoomEvent::RoomMember` event.
-    async fn on_room_member(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<MemberEvent>>) {}
+    async fn on_room_member(&self, _: Arc<RwLock<Room>>, _: &MemberEvent) {}
     /// Fires when `AsyncClient` receives a `RoomEvent::RoomName` event.
-    async fn on_room_name(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<NameEvent>>) {}
+    async fn on_room_name(&self, _: Arc<RwLock<Room>>, _: &NameEvent) {}
     /// Fires when `AsyncClient` receives a `RoomEvent::RoomCanonicalAlias` event.
-    async fn on_room_canonical_alias(
+    async fn on_room_canonical_alias(&self, _: Arc<RwLock<Room>>, _: &CanonicalAliasEvent) {}
-        &mut self,
-        _: Arc<Mutex<Room>>,
-        _: Arc<Mutex<CanonicalAliasEvent>>,
-    ) {
-    }
     /// Fires when `AsyncClient` receives a `RoomEvent::RoomAliases` event.
-    async fn on_room_aliases(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<AliasesEvent>>) {}
+    async fn on_room_aliases(&self, _: Arc<RwLock<Room>>, _: &AliasesEvent) {}
     /// Fires when `AsyncClient` receives a `RoomEvent::RoomAvatar` event.
-    async fn on_room_avatar(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<AvatarEvent>>) {}
+    async fn on_room_avatar(&self, _: Arc<RwLock<Room>>, _: &AvatarEvent) {}
     /// Fires when `AsyncClient` receives a `RoomEvent::RoomMessage` event.
-    async fn on_room_message(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<MessageEvent>>) {}
+    async fn on_room_message(&self, _: Arc<RwLock<Room>>, _: &MessageEvent) {}
     /// Fires when `AsyncClient` receives a `RoomEvent::RoomMessageFeedback` event.
-    async fn on_room_message_feedback(
+    async fn on_room_message_feedback(&self, _: Arc<RwLock<Room>>, _: &FeedbackEvent) {}
-        &mut self,
-        _: Arc<Mutex<Room>>,
-        _: Arc<Mutex<FeedbackEvent>>,
-    ) {
-    }
     /// Fires when `AsyncClient` receives a `RoomEvent::RoomRedaction` event.
-    async fn on_room_redaction(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<RedactionEvent>>) {}
+    async fn on_room_redaction(&self, _: Arc<RwLock<Room>>, _: &RedactionEvent) {}
     /// Fires when `AsyncClient` receives a `RoomEvent::RoomPowerLevels` event.
-    async fn on_room_power_levels(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<PowerLevelsEvent>>) {
+    async fn on_room_power_levels(&self, _: Arc<RwLock<Room>>, _: &PowerLevelsEvent) {}
-    }
 
     // `RoomEvent`s from `IncomingState`
     /// Fires when `AsyncClient` receives a `StateEvent::RoomMember` event.
-    async fn on_state_member(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<MemberEvent>>) {}
+    async fn on_state_member(&self, _: Arc<RwLock<Room>>, _: &MemberEvent) {}
     /// Fires when `AsyncClient` receives a `StateEvent::RoomName` event.
-    async fn on_state_name(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<NameEvent>>) {}
+    async fn on_state_name(&self, _: Arc<RwLock<Room>>, _: &NameEvent) {}
     /// Fires when `AsyncClient` receives a `StateEvent::RoomCanonicalAlias` event.
-    async fn on_state_canonical_alias(
+    async fn on_state_canonical_alias(&self, _: Arc<RwLock<Room>>, _: &CanonicalAliasEvent) {}
-        &mut self,
-        _: Arc<Mutex<Room>>,
-        _: Arc<Mutex<CanonicalAliasEvent>>,
-    ) {
-    }
     /// Fires when `AsyncClient` receives a `StateEvent::RoomAliases` event.
-    async fn on_state_aliases(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<AliasesEvent>>) {}
+    async fn on_state_aliases(&self, _: Arc<RwLock<Room>>, _: &AliasesEvent) {}
     /// Fires when `AsyncClient` receives a `StateEvent::RoomAvatar` event.
-    async fn on_state_avatar(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<AvatarEvent>>) {}
+    async fn on_state_avatar(&self, _: Arc<RwLock<Room>>, _: &AvatarEvent) {}
     /// Fires when `AsyncClient` receives a `StateEvent::RoomPowerLevels` event.
-    async fn on_state_power_levels(
+    async fn on_state_power_levels(&self, _: Arc<RwLock<Room>>, _: &PowerLevelsEvent) {}
-        &mut self,
-        _: Arc<Mutex<Room>>,
-        _: Arc<Mutex<PowerLevelsEvent>>,
-    ) {
-    }
     /// Fires when `AsyncClient` receives a `StateEvent::RoomJoinRules` event.
-    async fn on_state_join_rules(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<JoinRulesEvent>>) {}
+    async fn on_state_join_rules(&self, _: Arc<RwLock<Room>>, _: &JoinRulesEvent) {}
 
     // `NonRoomEvent` (this is a type alias from ruma_events) from `IncomingAccountData`
     /// Fires when `AsyncClient` receives a `NonRoomEvent::RoomMember` event.
-    async fn on_account_presence(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<PresenceEvent>>) {}
+    async fn on_account_presence(&self, _: Arc<RwLock<Room>>, _: &PresenceEvent) {}
     /// Fires when `AsyncClient` receives a `NonRoomEvent::RoomName` event.
-    async fn on_account_ignored_users(
+    async fn on_account_ignored_users(&self, _: Arc<RwLock<Room>>, _: &IgnoredUserListEvent) {}
-        &mut self,
-        _: Arc<Mutex<Room>>,
-        _: Arc<Mutex<IgnoredUserListEvent>>,
-    ) {
-    }
     /// Fires when `AsyncClient` receives a `NonRoomEvent::RoomCanonicalAlias` event.
-    async fn on_account_push_rules(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<PushRulesEvent>>) {}
+    async fn on_account_push_rules(&self, _: Arc<RwLock<Room>>, _: &PushRulesEvent) {}
     /// Fires when `AsyncClient` receives a `NonRoomEvent::RoomAliases` event.
-    async fn on_account_data_fully_read(
+    async fn on_account_data_fully_read(&self, _: Arc<RwLock<Room>>, _: &FullyReadEvent) {}
-        &mut self,
-        _: Arc<Mutex<Room>>,
-        _: Arc<Mutex<FullyReadEvent>>,
-    ) {
-    }
 
     // `PresenceEvent` is a struct so there is only the one method
     /// Fires when `AsyncClient` receives a `NonRoomEvent::RoomAliases` event.
-    async fn on_presence_event(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<PresenceEvent>>) {}
+    async fn on_presence_event(&self, _: Arc<RwLock<Room>>, _: &PresenceEvent) {}
 }
 
 #[cfg(test)]
 mod test {
     use super::*;
-
+    use std::sync::Arc;
+    use tokio::sync::Mutex;
+    #[derive(Clone)]
     pub struct EvEmitterTest(Arc<Mutex<Vec<String>>>);
 
     #[async_trait::async_trait]
     impl EventEmitter for EvEmitterTest {
-        async fn on_room_member(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<MemberEvent>>) {
+        async fn on_room_member(&self, _: Arc<RwLock<Room>>, _: &MemberEvent) {
             self.0.lock().await.push("member".to_string())
         }
-        async fn on_room_name(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<NameEvent>>) {
+        async fn on_room_name(&self, _: Arc<RwLock<Room>>, _: &NameEvent) {
             self.0.lock().await.push("name".to_string())
         }
-        async fn on_room_canonical_alias(
+        async fn on_room_canonical_alias(&self, _: Arc<RwLock<Room>>, _: &CanonicalAliasEvent) {
-            &mut self,
-            _: Arc<Mutex<Room>>,
-            _: Arc<Mutex<CanonicalAliasEvent>>,
-        ) {
             self.0.lock().await.push("canonical".to_string())
         }
-        async fn on_room_aliases(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<AliasesEvent>>) {
+        async fn on_room_aliases(&self, _: Arc<RwLock<Room>>, _: &AliasesEvent) {
             self.0.lock().await.push("aliases".to_string())
         }
-        async fn on_room_avatar(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<AvatarEvent>>) {
+        async fn on_room_avatar(&self, _: Arc<RwLock<Room>>, _: &AvatarEvent) {
             self.0.lock().await.push("avatar".to_string())
         }
-        async fn on_room_message(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<MessageEvent>>) {
+        async fn on_room_message(&self, _: Arc<RwLock<Room>>, _: &MessageEvent) {
             self.0.lock().await.push("message".to_string())
         }
-        async fn on_room_message_feedback(
+        async fn on_room_message_feedback(&self, _: Arc<RwLock<Room>>, _: &FeedbackEvent) {
-            &mut self,
-            _: Arc<Mutex<Room>>,
-            _: Arc<Mutex<FeedbackEvent>>,
-        ) {
             self.0.lock().await.push("feedback".to_string())
         }
-        async fn on_room_redaction(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<RedactionEvent>>) {
+        async fn on_room_redaction(&self, _: Arc<RwLock<Room>>, _: &RedactionEvent) {
             self.0.lock().await.push("redaction".to_string())
         }
-        async fn on_room_power_levels(
+        async fn on_room_power_levels(&self, _: Arc<RwLock<Room>>, _: &PowerLevelsEvent) {
-            &mut self,
-            _: Arc<Mutex<Room>>,
-            _: Arc<Mutex<PowerLevelsEvent>>,
-        ) {
             self.0.lock().await.push("power".to_string())
         }
 
-        async fn on_state_member(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<MemberEvent>>) {
+        async fn on_state_member(&self, _: Arc<RwLock<Room>>, _: &MemberEvent) {
             self.0.lock().await.push("state member".to_string())
         }
-        async fn on_state_name(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<NameEvent>>) {
+        async fn on_state_name(&self, _: Arc<RwLock<Room>>, _: &NameEvent) {
             self.0.lock().await.push("state name".to_string())
         }
-        async fn on_state_canonical_alias(
+        async fn on_state_canonical_alias(&self, _: Arc<RwLock<Room>>, _: &CanonicalAliasEvent) {
-            &mut self,
-            _: Arc<Mutex<Room>>,
-            _: Arc<Mutex<CanonicalAliasEvent>>,
-        ) {
             self.0.lock().await.push("state canonical".to_string())
         }
-        async fn on_state_aliases(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<AliasesEvent>>) {
+        async fn on_state_aliases(&self, _: Arc<RwLock<Room>>, _: &AliasesEvent) {
             self.0.lock().await.push("state aliases".to_string())
         }
-        async fn on_state_avatar(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<AvatarEvent>>) {
+        async fn on_state_avatar(&self, _: Arc<RwLock<Room>>, _: &AvatarEvent) {
             self.0.lock().await.push("state avatar".to_string())
         }
-        async fn on_state_power_levels(
+        async fn on_state_power_levels(&self, _: Arc<RwLock<Room>>, _: &PowerLevelsEvent) {
-            &mut self,
-            _: Arc<Mutex<Room>>,
-            _: Arc<Mutex<PowerLevelsEvent>>,
-        ) {
             self.0.lock().await.push("state power".to_string())
         }
-        async fn on_state_join_rules(
+        async fn on_state_join_rules(&self, _: Arc<RwLock<Room>>, _: &JoinRulesEvent) {
-            &mut self,
-            _: Arc<Mutex<Room>>,
-            _: Arc<Mutex<JoinRulesEvent>>,
-        ) {
             self.0.lock().await.push("state rules".to_string())
         }
 
-        async fn on_account_presence(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<PresenceEvent>>) {
+        async fn on_account_presence(&self, _: Arc<RwLock<Room>>, _: &PresenceEvent) {
             self.0.lock().await.push("account presence".to_string())
         }
-        async fn on_account_ignored_users(
+        async fn on_account_ignored_users(&self, _: Arc<RwLock<Room>>, _: &IgnoredUserListEvent) {
-            &mut self,
-            _: Arc<Mutex<Room>>,
-            _: Arc<Mutex<IgnoredUserListEvent>>,
-        ) {
             self.0.lock().await.push("account ignore".to_string())
         }
-        async fn on_account_push_rules(
+        async fn on_account_push_rules(&self, _: Arc<RwLock<Room>>, _: &PushRulesEvent) {
-            &mut self,
-            _: Arc<Mutex<Room>>,
-            _: Arc<Mutex<PushRulesEvent>>,
-        ) {
             self.0.lock().await.push("".to_string())
         }
-        async fn on_account_data_fully_read(
+        async fn on_account_data_fully_read(&self, _: Arc<RwLock<Room>>, _: &FullyReadEvent) {
-            &mut self,
-            _: Arc<Mutex<Room>>,
-            _: Arc<Mutex<FullyReadEvent>>,
-        ) {
             self.0.lock().await.push("account read".to_string())
         }
-        async fn on_presence_event(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<PresenceEvent>>) {
+        async fn on_presence_event(&self, _: Arc<RwLock<Room>>, _: &PresenceEvent) {
             self.0.lock().await.push("presence event".to_string())
         }
     }
@@ -303,11 +237,9 @@ mod test {
 
         let vec = Arc::new(Mutex::new(Vec::new()));
         let test_vec = Arc::clone(&vec);
-        let emitter = Arc::new(Mutex::new(
+        let emitter = Box::new(EvEmitterTest(vec)) as Box<(dyn EventEmitter)>;
-            Box::new(EvEmitterTest(vec)) as Box<(dyn EventEmitter)>
-        ));
         let mut client = AsyncClient::new(homeserver, Some(session)).unwrap();
-        client.add_event_emitter(Arc::clone(&emitter)).await;
+        client.add_event_emitter(emitter).await;
 
         let sync_settings = SyncSettings::new().timeout(Duration::from_millis(3000));
         let _response = client.sync(sync_settings).await.unwrap();

src/lib.rs

@@ -38,6 +38,7 @@ mod base_client;
 mod error;
 mod event_emitter;
 mod models;
+mod request_builder;
 mod session;
 mod state;
 
@@ -51,6 +52,7 @@ pub use async_client::{AsyncClient, AsyncClientConfig, SyncSettings};
 pub use base_client::Client;
 pub use event_emitter::EventEmitter;
 pub use models::Room;
-pub use state::{StateStore, JsonStore};
+pub use request_builder::{MessagesRequestBuilder, RoomBuilder};
+pub use state::{JsonStore, StateStore};
 
 pub(crate) const VERSION: &str = env!("CARGO_PKG_VERSION");

src/models/room.rs

@@ -417,7 +417,7 @@ mod test {
         .with_body_from_file("tests/data/sync.json")
         .create();
 
-        let mut client = AsyncClient::new(homeserver, Some(session)).unwrap();
+        let client = AsyncClient::new(homeserver, Some(session)).unwrap();
 
         let sync_settings = SyncSettings::new().timeout(Duration::from_millis(3000));
 
@@ -427,7 +427,7 @@ mod test {
         let room = &rooms
             .get(&RoomId::try_from("!SVkFJHzfwvuaIEawgC:localhost").unwrap())
             .unwrap()
-            .lock()
+            .read()
             .await;
 
         assert_eq!(2, room.members.len());

src/request_builder.rs

@@ -0,0 +1,379 @@
+use crate::api;
+use crate::events::room::power_levels::PowerLevelsEventContent;
+use crate::identifiers::{RoomId, UserId};
+use api::r0::filter::RoomEventFilter;
+use api::r0::membership::Invite3pid;
+use api::r0::message::get_message_events::{self, Direction};
+use api::r0::room::{
+    create_room::{self, CreationContent, InitialStateEvent, RoomPreset},
+    Visibility,
+};
+
+use js_int::UInt;
+
+/// A builder used to create rooms.
+///
+/// # Examples
+/// ```
+/// # use std::convert::TryFrom;
+/// # use matrix_sdk::{AsyncClient, RoomBuilder};
+/// # use matrix_sdk::api::r0::room::Visibility;
+/// # use matrix_sdk::identifiers::UserId;
+/// # use url::Url;
+/// # let homeserver = Url::parse("http://example.com").unwrap();
+/// # let mut rt = tokio::runtime::Runtime::new().unwrap();
+/// # rt.block_on(async {
+/// let mut builder = RoomBuilder::default();
+/// builder.creation_content(false)
+///     .initial_state(vec![])
+///     .visibility(Visibility::Public)
+///     .name("name")
+///     .room_version("v1.0");
+/// let mut cli = AsyncClient::new(homeserver, None).unwrap();
+/// cli.create_room(builder).await;
+/// # })
+/// ```
+#[derive(Clone, Default)]
+pub struct RoomBuilder {
+    /// Extra keys to be added to the content of the `m.room.create`.
+    creation_content: Option<CreationContent>,
+    /// List of state events to send to the new room.
+    ///
+    /// Takes precedence over events set by preset, but gets overriden by
+    /// name and topic keys.
+    initial_state: Vec<InitialStateEvent>,
+    /// A list of user IDs to invite to the room.
+    ///
+    /// This will tell the server to invite everyone in the list to the newly created room.
+    invite: Vec<UserId>,
+    /// List of third party IDs of users to invite.
+    invite_3pid: Vec<Invite3pid>,
+    /// If set, this sets the `is_direct` flag on room invites.
+    is_direct: Option<bool>,
+    /// If this is included, an `m.room.name` event will be sent into the room to indicate
+    /// the name of the room.
+    name: Option<String>,
+    /// Power level content to override in the default power level event.
+    power_level_content_override: Option<PowerLevelsEventContent>,
+    /// Convenience parameter for setting various default state events based on a preset.
+    preset: Option<RoomPreset>,
+    /// The desired room alias local part.
+    room_alias_name: Option<String>,
+    /// Room version to set for the room. Defaults to homeserver's default if not specified.
+    room_version: Option<String>,
+    /// If this is included, an `m.room.topic` event will be sent into the room to indicate
+    /// the topic for the room.
+    topic: Option<String>,
+    /// A public visibility indicates that the room will be shown in the published room
+    /// list. A private visibility will hide the room from the published room list. Rooms
+    /// default to private visibility if this key is not included.
+    visibility: Option<Visibility>,
+}
+
+impl RoomBuilder {
+    /// Returns an empty `RoomBuilder` for creating rooms.
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Set the `CreationContent`.
+    ///
+    /// Weather users on other servers can join this room.
+    pub fn creation_content(&mut self, federate: bool) -> &mut Self {
+        let federate = Some(federate);
+        self.creation_content = Some(CreationContent { federate });
+        self
+    }
+
+    /// Set the `InitialStateEvent` vector.
+    pub fn initial_state(&mut self, state: Vec<InitialStateEvent>) -> &mut Self {
+        self.initial_state = state;
+        self
+    }
+
+    /// Set the vec of `UserId`s.
+    pub fn invite(&mut self, invite: Vec<UserId>) -> &mut Self {
+        self.invite = invite;
+        self
+    }
+
+    /// Set the vec of `Invite3pid`s.
+    pub fn invite_3pid(&mut self, invite: Vec<Invite3pid>) -> &mut Self {
+        self.invite_3pid = invite;
+        self
+    }
+
+    /// Set the vec of `Invite3pid`s.
+    pub fn is_direct(&mut self, direct: bool) -> &mut Self {
+        self.is_direct = Some(direct);
+        self
+    }
+
+    /// Set the room name. A `m.room.name` event will be sent to the room.
+    pub fn name<S: Into<String>>(&mut self, name: S) -> &mut Self {
+        self.name = Some(name.into());
+        self
+    }
+
+    /// Set the room's power levels.
+    pub fn power_level_override(&mut self, power: PowerLevelsEventContent) -> &mut Self {
+        self.power_level_content_override = Some(power);
+        self
+    }
+
+    /// Convenience for setting various default state events based on a preset.
+    pub fn preset(&mut self, preset: RoomPreset) -> &mut Self {
+        self.preset = Some(preset);
+        self
+    }
+
+    ///  The local part of a room alias.
+    pub fn room_alias_name<S: Into<String>>(&mut self, alias: S) -> &mut Self {
+        self.room_alias_name = Some(alias.into());
+        self
+    }
+
+    /// Room version, defaults to homeserver's version if left unspecified.
+    pub fn room_version<S: Into<String>>(&mut self, version: S) -> &mut Self {
+        self.room_version = Some(version.into());
+        self
+    }
+
+    /// If included, a `m.room.topic` event will be sent to the room.
+    pub fn topic<S: Into<String>>(&mut self, topic: S) -> &mut Self {
+        self.topic = Some(topic.into());
+        self
+    }
+
+    /// A public visibility indicates that the room will be shown in the published
+    /// room list. A private visibility will hide the room from the published room list.
+    /// Rooms default to private visibility if this key is not included.
+    pub fn visibility(&mut self, vis: Visibility) -> &mut Self {
+        self.visibility = Some(vis);
+        self
+    }
+}
+
+impl Into<create_room::Request> for RoomBuilder {
+    fn into(self) -> create_room::Request {
+        create_room::Request {
+            creation_content: self.creation_content,
+            initial_state: self.initial_state,
+            invite: self.invite,
+            invite_3pid: self.invite_3pid,
+            is_direct: self.is_direct,
+            name: self.name,
+            power_level_content_override: self.power_level_content_override,
+            preset: self.preset,
+            room_alias_name: self.room_alias_name,
+            room_version: self.room_version,
+            topic: self.topic,
+            visibility: self.visibility,
+        }
+    }
+}
+
+/// Create a builder for making get_message_event requests.
+///
+/// # Examples
+/// ```
+/// # use matrix_sdk::{AsyncClient, MessagesRequestBuilder};
+/// # use matrix_sdk::api::r0::message::get_message_events::{self, Direction};
+/// # use matrix_sdk::identifiers::RoomId;
+/// # use url::Url;
+/// # let homeserver = Url::parse("http://example.com").unwrap();
+/// # let mut rt = tokio::runtime::Runtime::new().unwrap();
+/// # rt.block_on(async {
+/// # let room_id = RoomId::new(homeserver.as_str()).unwrap();
+/// # let last_sync_token = "".to_string();;
+/// let mut cli = AsyncClient::new(homeserver, None).unwrap();
+///
+/// let mut builder = MessagesRequestBuilder::new();
+/// builder.room_id(room_id)
+///     .from(last_sync_token)
+///     .direction(Direction::Forward);
+///
+/// cli.room_messages(builder).await.is_err();
+/// # })
+/// ```
+#[derive(Clone, Default)]
+pub struct MessagesRequestBuilder {
+    /// The room to get events from.
+    room_id: Option<RoomId>,
+    /// The token to start returning events from.
+    ///
+    /// This token can be obtained from a
+    /// prev_batch token returned for each room by the sync API, or from a start or end token
+    /// returned by a previous request to this endpoint.
+    from: Option<String>,
+    /// The token to stop returning events at.
+    ///
+    /// This token can be obtained from a prev_batch
+    /// token returned for each room by the sync endpoint, or from a start or end token returned
+    /// by a previous request to this endpoint.
+    to: Option<String>,
+    /// The direction to return events from.
+    direction: Option<Direction>,
+    /// The maximum number of events to return.
+    ///
+    /// Default: 10.
+    limit: Option<UInt>,
+    /// A filter of the returned events with.
+    filter: Option<RoomEventFilter>,
+}
+
+impl MessagesRequestBuilder {
+    /// Create a `MessagesRequestBuilder` builder to make a `get_message_events::Request`.
+    ///
+    /// The `room_id` and `from`` fields **need to be set** to create the request.
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// RoomId is required to create a `get_message_events::Request`.
+    pub fn room_id(&mut self, room_id: RoomId) -> &mut Self {
+        self.room_id = Some(room_id);
+        self
+    }
+
+    /// A `next_batch` token or `start` or `end` from a previous `get_message_events` request.
+    ///
+    /// This is required to create a `get_message_events::Request`.
+    pub fn from(&mut self, from: String) -> &mut Self {
+        self.from = Some(from);
+        self
+    }
+
+    /// A `next_batch` token or `start` or `end` from a previous `get_message_events` request.
+    ///
+    /// This token signals when to stop receiving events.
+    pub fn to(&mut self, to: String) -> &mut Self {
+        self.to = Some(to);
+        self
+    }
+
+    /// The direction to return events from.
+    ///
+    /// If not specified `Direction::Backward` is used.
+    pub fn direction(&mut self, direction: Direction) -> &mut Self {
+        self.direction = Some(direction);
+        self
+    }
+
+    /// The maximum number of events to return.
+    pub fn limit(&mut self, limit: UInt) -> &mut Self {
+        self.limit = Some(limit);
+        self
+    }
+
+    /// Filter events by the given `RoomEventFilter`.
+    pub fn filter(&mut self, filter: RoomEventFilter) -> &mut Self {
+        self.filter = Some(filter);
+        self
+    }
+}
+
+impl Into<get_message_events::Request> for MessagesRequestBuilder {
+    fn into(self) -> get_message_events::Request {
+        get_message_events::Request {
+            room_id: self.room_id.expect("`room_id` and `from` need to be set"),
+            from: self.from.expect("`room_id` and `from` need to be set"),
+            to: self.to,
+            dir: self.direction.unwrap_or(Direction::Backward),
+            limit: self.limit,
+            filter: self.filter,
+        }
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use std::collections::HashMap;
+
+    use super::*;
+    use crate::events::room::power_levels::NotificationPowerLevels;
+    use crate::{identifiers::RoomId, AsyncClient, Session};
+
+    use js_int::Int;
+    use mockito::{mock, Matcher};
+    use std::convert::TryFrom;
+    use url::Url;
+
+    #[tokio::test]
+    async fn create_room_builder() {
+        let homeserver = Url::parse(&mockito::server_url()).unwrap();
+
+        let _m = mock("POST", "/_matrix/client/r0/createRoom")
+            .with_status(200)
+            .with_body_from_file("./tests/data/room_id.json")
+            .create();
+
+        let session = Session {
+            access_token: "1234".to_owned(),
+            user_id: UserId::try_from("@example:localhost").unwrap(),
+            device_id: "DEVICEID".to_owned(),
+        };
+
+        let mut builder = RoomBuilder::new();
+        builder
+            .creation_content(false)
+            .initial_state(vec![])
+            .visibility(Visibility::Public)
+            .name("room_name")
+            .room_version("v1.0")
+            .invite_3pid(vec![])
+            .is_direct(true)
+            .power_level_override(PowerLevelsEventContent {
+                ban: Int::max_value(),
+                events: HashMap::default(),
+                events_default: Int::min_value(),
+                invite: Int::min_value(),
+                kick: Int::min_value(),
+                redact: Int::max_value(),
+                state_default: Int::min_value(),
+                users_default: Int::min_value(),
+                notifications: NotificationPowerLevels {
+                    room: Int::min_value(),
+                },
+                users: HashMap::default(),
+            })
+            .preset(RoomPreset::PrivateChat)
+            .room_alias_name("room_alias")
+            .topic("room topic")
+            .visibility(Visibility::Private);
+        let cli = AsyncClient::new(homeserver, Some(session)).unwrap();
+        assert!(cli.create_room(builder).await.is_ok());
+    }
+
+    #[tokio::test]
+    async fn get_message_events() {
+        let homeserver = Url::parse(&mockito::server_url()).unwrap();
+
+        let _m = mock(
+            "GET",
+            Matcher::Regex(r"^/_matrix/client/r0/rooms/.*/messages".to_string()),
+        )
+        .with_status(200)
+        .with_body_from_file("./tests/data/room_messages.json")
+        .create();
+
+        let session = Session {
+            access_token: "1234".to_owned(),
+            user_id: UserId::try_from("@example:localhost").unwrap(),
+            device_id: "DEVICEID".to_owned(),
+        };
+
+        let mut builder = MessagesRequestBuilder::new();
+        builder
+            .room_id(RoomId::try_from("!roomid:example.com").unwrap())
+            .from("t47429-4392820_219380_26003_2265".to_string())
+            .to("t4357353_219380_26003_2265".to_string())
+            .direction(Direction::Backward)
+            .limit(UInt::new(10).unwrap());
+        // TODO this makes ruma error `Err(IntoHttp(IntoHttpError(Query(Custom("unsupported value")))))`??
+        // .filter(RoomEventFilter::default());
+
+        let cli = AsyncClient::new(homeserver, Some(session)).unwrap();
+        assert!(cli.room_messages(builder).await.is_ok());
+    }
+}

src/state/mod.rs

@@ -55,13 +55,14 @@ use crate::session::Session;
 use crate::VERSION;
 use crate::{Error, EventEmitter, Result};
 /// Abstraction around the data store to avoid unnecessary request on client initialization.
-/// 
+///
 pub trait StateStore {
+    ///
     fn load_state(&self) -> sync_events::IncomingResponse;
-
+    ///
     fn save_state_events(&mut self, events: Vec<StateEvent>) -> Result<()>;
+    ///
     fn save_room_events(&mut self, events: Vec<RoomEvent>) -> Result<()>;
+    ///
     fn save_non_room_events(&mut self, events: Vec<NonRoomEvent>) -> Result<()>;
-
-    
 }

src/state/state_store.rs

@@ -2,11 +2,8 @@ use super::StateStore;
 
 /// A default `StateStore` implementation that serializes state as json
 /// and saves it to disk.
-pub struct JsonStore {
+pub struct JsonStore {}
-    
-}
 
-impl StateStore for JsonStore {
+// impl StateStore for JsonStore {
-
-}
 
+// }

tests/async_client_tests.rs

@@ -17,7 +17,7 @@ async fn login() {
         .with_body_from_file("tests/data/login_response.json")
         .create();
 
-    let mut client = AsyncClient::new(homeserver, None).unwrap();
+    let client = AsyncClient::new(homeserver, None).unwrap();
 
     client
         .login("example", "wordpass", None, None)
@@ -46,7 +46,7 @@ async fn sync() {
     .with_body_from_file("tests/data/sync.json")
     .create();
 
-    let mut client = AsyncClient::new(homeserver, Some(session)).unwrap();
+    let client = AsyncClient::new(homeserver, Some(session)).unwrap();
 
     let sync_settings = SyncSettings::new().timeout(Duration::from_millis(3000));
 
@@ -75,7 +75,7 @@ async fn room_names() {
     .with_body_from_file("tests/data/sync.json")
     .create();
 
-    let mut client = AsyncClient::new(homeserver, Some(session)).unwrap();
+    let client = AsyncClient::new(homeserver, Some(session)).unwrap();
 
     let sync_settings = SyncSettings::new().timeout(Duration::from_millis(3000));
 

tests/data/events/room_avatar.json

@@ -18,4 +18,3 @@
         "age": 1234
     }
 }
-

tests/data/sync2.json

@@ -1,160 +0,0 @@
-{
-    "next_batch": "s72595_4483_1934",
-    "presence": {
-      "events": [
-        {
-          "content": {
-            "avatar_url": "mxc://localhost:wefuiwegh8742w",
-            "last_active_ago": 2478593,
-            "presence": "online",
-            "currently_active": false,
-            "status_msg": "Making cupcakes"
-          },
-          "type": "m.presence",
-          "sender": "@example:localhost"
-        }
-      ]
-    },
-    "account_data": {
-      "events": [
-        {
-          "type": "org.example.custom.config",
-          "content": {
-            "custom_config_key": "custom_config_value"
-          }
-        }
-      ]
-    },
-    "rooms": {
-      "join": {
-        "!726s6s6q:example.com": {
-          "summary": {
-            "m.heroes": [
-              "@alice:example.com",
-              "@bob:example.com"
-            ],
-            "m.joined_member_count": 2,
-            "m.invited_member_count": 0
-          },
-          "state": {
-            "events": [
-              {
-                "content": {
-                  "membership": "join",
-                  "avatar_url": "mxc://example.org/SEsfnsuifSDFSSEF",
-                  "displayname": "Alice Margatroid"
-                },
-                "type": "m.room.member",
-                "event_id": "$143273582443PhrSn:example.org",
-                "room_id": "!726s6s6q:example.com",
-                "sender": "@example:example.org",
-                "origin_server_ts": 1432735824653,
-                "unsigned": {
-                  "age": 1234
-                },
-                "state_key": "@alice:example.org"
-              }
-            ]
-          },
-          "timeline": {
-            "events": [
-              {
-                "content": {
-                  "membership": "join",
-                  "avatar_url": "mxc://example.org/SEsfnsuifSDFSSEF",
-                  "displayname": "Alice Margatroid"
-                },
-                "type": "m.room.member",
-                "event_id": "$143273582443PhrSn:example.org",
-                "room_id": "!726s6s6q:example.com",
-                "sender": "@example:example.org",
-                "origin_server_ts": 1432735824653,
-                "unsigned": {
-                  "age": 1234
-                },
-                "state_key": "@alice:example.org"
-              },
-              {
-                "content": {
-                  "body": "This is an example text message",
-                  "msgtype": "m.text",
-                  "format": "org.matrix.custom.html",
-                  "formatted_body": "<b>This is an example text message</b>"
-                },
-                "type": "m.room.message",
-                "event_id": "$143273582443PhrSn:example.org",
-                "room_id": "!726s6s6q:example.com",
-                "sender": "@example:example.org",
-                "origin_server_ts": 1432735824653,
-                "unsigned": {
-                  "age": 1234
-                }
-              }
-            ],
-            "limited": true,
-            "prev_batch": "t34-23535_0_0"
-          },
-          "ephemeral": {
-            "events": [
-              {
-                "content": {
-                  "user_ids": [
-                    "@alice:matrix.org",
-                    "@bob:example.com"
-                  ]
-                },
-                "type": "m.typing",
-                "room_id": "!jEsUZKDJdhlrceRyVU:example.org"
-              }
-            ]
-          },
-          "account_data": {
-            "events": [
-              {
-                "content": {
-                  "tags": {
-                    "u.work": {
-                      "order": 0.9
-                    }
-                  }
-                },
-                "type": "m.tag"
-              },
-              {
-                "type": "org.example.custom.room.config",
-                "content": {
-                  "custom_config_key": "custom_config_value"
-                }
-              }
-            ]
-          }
-        }
-      },
-      "invite": {
-        "!696r7674:example.com": {
-          "invite_state": {
-            "events": [
-              {
-                "sender": "@alice:example.com",
-                "type": "m.room.name",
-                "state_key": "",
-                "content": {
-                  "name": "My Room Name"
-                }
-              },
-              {
-                "sender": "@alice:example.com",
-                "type": "m.room.member",
-                "state_key": "@bob:example.com",
-                "content": {
-                  "membership": "invite"
-                }
-              }
-            ]
-          }
-        }
-      },
-      "leave": {}
-    }
-  }
-