Merge branch 'ev-emitter'

Cargo.toml

@@ -12,7 +12,7 @@ version = "0.1.0"
 
 [features]
 default = []
-encryption = ["olm-rs", "serde/derive", "serde_json", "cjson", "async-trait"]
+encryption = ["olm-rs", "serde/derive", "serde_json", "cjson"]
 sqlite-cryptostore = ["sqlx", "zeroize"]
 
 [dependencies]
@@ -20,11 +20,13 @@ futures = "0.3.4"
 reqwest = "0.10.4"
 http = "0.2.1"
 url = "2.1.1"
+async-trait = "0.1.26"
+
 
 # Ruma dependencies
 js_int = "0.1.3"
 ruma-api = "0.15.0-dev.1"
-ruma-client-api = { version = "0.6.0", git = "https://github.com/matrix-org/ruma-client-api/" }
+ruma-client-api = { git = "https://github.com/matrix-org/ruma-client-api/", version = "0.6.0" }
 ruma-events = { git = "https://github.com/matrix-org/ruma-events", version = "0.17.0" }
 ruma-identifiers = "0.14.1"
 
@@ -37,7 +39,6 @@ zeroize = { version = "1.1.0", optional = true }
 
 # Misc dependencies
 thiserror = "1.0.13"
-async-trait = { version = "0.1.26", optional = true }
 tracing = "0.1.13"
 atomic = "0.4.5"
 dashmap = "3.9.1"

examples/login.rs

@@ -1,31 +1,28 @@
-use std::sync::{Arc, RwLock};
+use std::ops::Deref;
+use std::sync::Arc;
 use std::{env, process::exit};
 use url::Url;
 
 use matrix_sdk::{
     self,
-    events::{
+    events::room::message::{MessageEvent, MessageEventContent, TextMessageEventContent},
-        collections::all::RoomEvent,
+    AsyncClient, AsyncClientConfig, EventEmitter, Room, SyncSettings,
-        room::message::{MessageEvent, MessageEventContent, TextMessageEventContent},
-        EventResult,
-    },
-    AsyncClient, AsyncClientConfig, Room, SyncSettings,
 };
+use tokio::sync::Mutex;
 
-async fn async_cb(room: Arc<RwLock<Room>>, event: Arc<EventResult<RoomEvent>>) {
+struct EventCallback;
-    let room = room.read().unwrap();
+
-    let event = if let EventResult::Ok(event) = &*event {
+#[async_trait::async_trait]
-        event
+impl EventEmitter for EventCallback {
-    } else {
+    async fn on_room_message(&mut self, room: Arc<Mutex<Room>>, event: Arc<Mutex<MessageEvent>>) {
-        return;
+        if let MessageEvent {
-    };
-    if let RoomEvent::RoomMessage(MessageEvent {
             content: MessageEventContent::Text(TextMessageEventContent { body: msg_body, .. }),
             sender,
             ..
-    }) = event
+        } = event.lock().await.deref()
         {
-        let member = room.members.get(&sender.to_string()).unwrap();
+            let rooms = room.lock().await;
+            let member = rooms.members.get(&sender.to_string()).unwrap();
             println!(
                 "{}: {}",
                 member
@@ -37,6 +34,7 @@ async fn async_cb(room: Arc<RwLock<Room>>, event: Arc<EventResult<RoomEvent>>) {
             );
         }
     }
+}
 
 async fn login(
     homeserver_url: String,
@@ -49,7 +47,9 @@ async fn login(
     let homeserver_url = Url::parse(&homeserver_url)?;
     let mut client = AsyncClient::new_with_config(homeserver_url, None, client_config).unwrap();
 
-    client.add_event_callback(async_cb);
+    client
+        .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

@@ -17,10 +17,10 @@ use std::collections::HashMap;
 use std::convert::{TryFrom, TryInto};
 use std::result::Result as StdResult;
 use std::sync::atomic::{AtomicU64, Ordering};
-use std::sync::{Arc, Mutex, RwLock as SyncLock};
+use std::sync::Arc;
 use std::time::{Duration, Instant};
 
-use futures::future::{BoxFuture, Future, FutureExt};
+use futures::future::Future;
 use tokio::sync::RwLock;
 use tokio::time::delay_for as sleep;
 use tracing::{debug, info, instrument, trace};
@@ -31,8 +31,6 @@ use reqwest::header::{HeaderValue, InvalidHeaderValue};
 use url::Url;
 
 use ruma_api::{Endpoint, Outgoing};
-use ruma_events::collections::all::RoomEvent;
-use ruma_events::presence::PresenceEvent;
 use ruma_events::room::message::MessageEventContent;
 use ruma_events::EventResult;
 pub use ruma_events::EventType;
@@ -46,16 +44,7 @@ use crate::base_client::Client as BaseClient;
 use crate::models::Room;
 use crate::session::Session;
 use crate::VERSION;
-use crate::{Error, Result};
+use crate::{Error, EventEmitter, Result};
-
-type RoomEventCallback = Box<
-    dyn FnMut(Arc<SyncLock<Room>>, Arc<EventResult<RoomEvent>>) -> BoxFuture<'static, ()> + Send,
->;
-
-type PresenceEventCallback = Box<
-    dyn FnMut(Arc<SyncLock<Room>>, Arc<EventResult<PresenceEvent>>) -> BoxFuture<'static, ()>
-        + Send,
->;
 
 const DEFAULT_SYNC_TIMEOUT: Duration = Duration::from_secs(30);
 
@@ -70,9 +59,6 @@ pub struct AsyncClient {
     pub(crate) base_client: Arc<RwLock<BaseClient>>,
     /// The transaction id.
     transaction_id: Arc<AtomicU64>,
-    /// Event callbacks
-    event_callbacks: Arc<Mutex<Vec<RoomEventCallback>>>,
-    presence_callbacks: Arc<Mutex<Vec<PresenceEventCallback>>>,
 }
 
 impl std::fmt::Debug for AsyncClient {
@@ -224,6 +210,7 @@ impl AsyncClient {
         session: Option<Session>,
         config: AsyncClientConfig,
     ) -> Result<Self> {
+        #[allow(clippy::match_wild_err_arm)]
         let homeserver: Url = match homeserver_url.try_into() {
             Ok(u) => u,
             Err(_e) => panic!("Error parsing homeserver url"),
@@ -258,8 +245,6 @@ impl AsyncClient {
             http_client,
             base_client: Arc::new(RwLock::new(BaseClient::new(session)?)),
             transaction_id: Arc::new(AtomicU64::new(0)),
-            event_callbacks: Arc::new(Mutex::new(Vec::new())),
-            presence_callbacks: Arc::new(Mutex::new(Vec::new())),
         })
     }
 
@@ -275,154 +260,39 @@ impl AsyncClient {
         &self.homeserver
     }
 
-    /// Calculate the room name from a `RoomId`, returning a string.
+    /// Add `EventEmitter` to `AsyncClient`.
+    ///
+    /// The methods of `EventEmitter` are called when the respective `RoomEvents` occur.
+    pub async fn add_event_emitter(
+        &mut self,
+        emitter: Arc<tokio::sync::Mutex<Box<dyn EventEmitter>>>,
+    ) {
+        self.base_client.write().await.event_emitter = Some(emitter);
+    }
+
+    /// Returns an `Option` of the room name from a `RoomId`.
+    ///
+    /// This is a human readable room name.
     pub async fn get_room_name(&self, room_id: &str) -> Option<String> {
-        self.base_client.read().await.calculate_room_name(room_id)
+        self.base_client
+            .read()
+            .await
+            .calculate_room_name(room_id)
+            .await
     }
 
-    /// Calculate the room names this client knows about.
+    /// Returns a `Vec` of the room names this client knows about.
+    ///
+    /// This is a human readable list of room names.
     pub async fn get_room_names(&self) -> Vec<String> {
-        self.base_client.read().await.calculate_room_names()
+        self.base_client.read().await.calculate_room_names().await
     }
 
-    /// Calculate the room names this client knows about.
+    /// Returns the rooms this client knows about.
-    pub async fn current_room_id(&self) -> Option<RoomId> {
-        self.base_client.read().await.current_room_id()
-    }
-
-    /// Add a callback that will be called every time the client receives a room
-    /// event
     ///
-    /// # Arguments
+    /// A `HashMap` of room id to `matrix::models::Room`
-    ///
+    pub async fn get_rooms(&self) -> HashMap<String, Arc<tokio::sync::Mutex<Room>>> {
-    /// * `callback` - The callback that should be called once a RoomEvent is
+        self.base_client.read().await.joined_rooms.clone()
-    ///     received.
-    ///
-    /// # Examples
-    /// ```
-    /// # use matrix_sdk::events::{
-    /// #     collections::all::RoomEvent,
-    /// #     room::message::{MessageEvent, MessageEventContent, TextMessageEventContent},
-    /// #     EventResult,
-    /// # };
-    /// # use matrix_sdk::Room;
-    /// # use std::sync::{Arc, RwLock};
-    /// # use matrix_sdk::AsyncClient;
-    /// # use url::Url;
-    ///
-    /// async fn async_cb(room: Arc<RwLock<Room>>, event: Arc<EventResult<RoomEvent>>) {
-    ///     let room = room.read().unwrap();
-    ///     let event = if let EventResult::Ok(event) = &*event {
-    ///         event
-    ///     } else {
-    ///         return;
-    ///     };
-    ///     if let RoomEvent::RoomMessage(MessageEvent {
-    ///         content: MessageEventContent::Text(TextMessageEventContent { body: msg_body, .. }),
-    ///         sender,
-    ///         ..
-    ///     }) = event
-    ///     {
-    ///         let member = room.members.get(&sender.to_string()).unwrap();
-    ///         println!(
-    ///             "{}: {}",
-    ///             member.user.display_name.as_ref().unwrap_or(&sender.to_string()),
-    ///             msg_body
-    ///         );
-    ///     }
-    /// }
-    /// # fn main() -> Result<(), matrix_sdk::Error> {
-    /// let homeserver = Url::parse("http://localhost:8080")?;
-    ///
-    /// let mut client = AsyncClient::new(homeserver, None)?;
-    ///
-    /// client.add_event_callback(async_cb);
-    /// # Ok(())
-    /// # }
-    /// ```
-    pub fn add_event_callback<C: 'static>(
-        &mut self,
-        mut callback: impl FnMut(Arc<SyncLock<Room>>, Arc<EventResult<RoomEvent>>) -> C + 'static + Send,
-    ) where
-        C: Future<Output = ()> + Send,
-    {
-        let mut futures = self.event_callbacks.lock().unwrap();
-
-        let future = move |room, event| callback(room, event).boxed();
-
-        futures.push(Box::new(future));
-    }
-
-    /// Add a callback that will be called every time the client receives a presence
-    /// event
-    ///
-    /// # Arguments
-    ///
-    /// * `callback` - The callback that should be called once a RoomEvent is
-    ///     received.
-    ///
-    /// # Examples
-    /// ```
-    /// # use matrix_sdk::events::{
-    /// #     collections::all::RoomEvent,
-    /// #     room::message::{MessageEvent, MessageEventContent, TextMessageEventContent},
-    /// #     presence::{PresenceEvent, PresenceEventContent},
-    /// #     EventResult,
-    /// # };
-    /// # use matrix_sdk::Room;
-    /// # use std::sync::{Arc, RwLock};
-    /// # use matrix_sdk::AsyncClient;
-    /// # use url::Url;
-    ///
-    /// async fn async_cb(room: Arc<RwLock<Room>>, event: Arc<EventResult<PresenceEvent>>) {
-    ///     let room = room.read().unwrap();
-    ///     let event = if let EventResult::Ok(event) = &*event {
-    ///         event
-    ///     } else {
-    ///         return;
-    ///     };
-    ///     let PresenceEvent {
-    ///         content: PresenceEventContent {
-    ///             avatar_url,
-    ///             currently_active,
-    ///             displayname,
-    ///             last_active_ago,
-    ///             presence,
-    ///             status_msg,
-    ///         },
-    ///         sender,
-    ///     } = event;
-    ///     {
-    ///         let member = room.members.get(&sender.to_string()).unwrap();
-    ///         println!(
-    ///             "{} is {}",
-    ///             displayname.as_deref().unwrap_or(&sender.to_string()),
-    ///             status_msg.as_deref().unwrap_or("not here")
-    ///         );
-    ///     }
-    /// }
-    /// # fn main() -> Result<(), matrix_sdk::Error> {
-    /// let homeserver = Url::parse("http://localhost:8080")?;
-    ///
-    /// let mut client = AsyncClient::new(homeserver, None)?;
-    ///
-    /// client.add_presence_callback(async_cb);
-    /// # Ok(())
-    /// # }
-    /// ```
-    pub fn add_presence_callback<C: 'static>(
-        &mut self,
-        mut callback: impl FnMut(Arc<SyncLock<Room>>, Arc<EventResult<PresenceEvent>>) -> C
-            + 'static
-            + Send,
-    ) where
-        C: Future<Output = ()> + Send,
-    {
-        let mut futures = self.presence_callbacks.lock().unwrap();
-
-        let future = move |room, event| callback(room, event).boxed();
-
-        futures.push(Box::new(future));
     }
 
     /// Login to the server.
@@ -486,21 +356,27 @@ impl AsyncClient {
         for (room_id, room) in &mut response.rooms.join {
             let room_id_string = room_id.to_string();
 
-            let matrix_room = {
             let mut client = self.base_client.write().await;
 
+            let _matrix_room = {
                 for event in &room.state.events {
                     if let EventResult::Ok(e) = event {
-                        client.receive_joined_state_event(&room_id_string, &e);
+                        client.receive_joined_state_event(&room_id_string, &e).await;
                     }
                 }
 
                 client.get_or_create_room(&room_id_string).clone()
             };
 
+            // re looping is not ideal here
+            for event in &mut room.state.events {
+                if let EventResult::Ok(e) = event {
+                    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
@@ -510,35 +386,19 @@ impl AsyncClient {
                     *event = e;
                 }
 
-                let callbacks = {
+                if let EventResult::Ok(e) = event {
-                    let mut cb_futures = self.event_callbacks.lock().unwrap();
+                    client.emit_timeline_event(room_id, e).await;
-
-                    let event = if !cb_futures.is_empty() {
-                        Arc::new(event.clone())
-                    } else {
-                        continue;
-                    };
-
-                    let mut callbacks = Vec::new();
-
-                    for cb in &mut cb_futures.iter_mut() {
-                        callbacks.push(cb(matrix_room.clone(), Arc::clone(&event)));
-                    }
-
-                    callbacks
-                };
-
-                for cb in callbacks {
-                    cb.await;
                 }
             }
 
             // look at AccountData to further cut down users by collecting ignored users
-            // TODO actually use the ignored users
+            for account_data in &mut room.account_data.events {
-            for account_data in &room.account_data.events {
+                {
-                let mut client = self.base_client.write().await;
                     if let EventResult::Ok(e) = account_data {
-                    client.receive_account_data(&room_id_string, e);
+                        client.receive_account_data(&room_id_string, e).await;
+
+                        client.emit_account_data_event(room_id, e).await;
+                    }
                 }
             }
 
@@ -547,28 +407,13 @@ impl AsyncClient {
             // After the room has been created and state/timeline events accounted for we use the room_id of the newly created
             // room to add any presence events that relate to a user in the current room. This is not super
             // efficient but we need a room_id so we would loop through now or later.
-            for presence in &response.presence.events {
+            for presence in &mut response.presence.events {
-                let mut client = self.base_client.write().await;
+                {
                     if let EventResult::Ok(e) = presence {
-                    client.receive_presence_event(&room_id_string, e);
+                        client.receive_presence_event(&room_id_string, e).await;
-                }
 
-                let callbacks = {
+                        client.emit_presence_event(room_id, e).await;
-                    let mut cb_futures = self.presence_callbacks.lock().unwrap();
-                    let event = if !cb_futures.is_empty() {
-                        Arc::new(presence.clone())
-                    } else {
-                        continue;
-                    };
-                    let mut callbacks = Vec::new();
-                    for cb in &mut cb_futures.iter_mut() {
-                        callbacks.push(cb(matrix_room.clone(), Arc::clone(&event)));
                     }
-
-                    callbacks
-                };
-                for cb in callbacks {
-                    cb.await;
                 }
             }
         }

src/base_client.rs

@@ -14,7 +14,8 @@
 // limitations under the License.
 
 use std::collections::{HashMap, HashSet};
-use std::convert::TryFrom;
+use std::fmt;
+use std::sync::Arc;
 
 #[cfg(feature = "encryption")]
 use std::result::Result as StdResult;
@@ -27,21 +28,12 @@ use crate::events::presence::PresenceEvent;
 use crate::events::collections::only::Event as NonRoomEvent;
 use crate::events::ignored_user_list::IgnoredUserListEvent;
 use crate::events::push_rules::{PushRulesEvent, Ruleset};
-use crate::events::room::{
-    aliases::AliasesEvent,
-    canonical_alias::CanonicalAliasEvent,
-    member::{MemberEvent, MembershipState},
-    name::NameEvent,
-};
 use crate::events::EventResult;
-use crate::identifiers::{RoomAliasId, UserId as Uid};
+use crate::identifiers::RoomAliasId;
 use crate::models::Room;
 use crate::session::Session;
-use std::sync::{Arc, RwLock};
+use crate::EventEmitter;
 
-use js_int::UInt;
-
-#[cfg(feature = "encryption")]
 use tokio::sync::Mutex;
 
 #[cfg(feature = "encryption")]
@@ -67,34 +59,6 @@ pub struct RoomName {
     aliases: Vec<RoomAliasId>,
 }
 
-#[derive(Clone, Debug, Default)]
-pub struct CurrentRoom {
-    last_active: Option<UInt>,
-    current_room_id: Option<RoomId>,
-}
-
-impl CurrentRoom {
-    // TODO when UserId is isomorphic to &str clean this up.
-    pub(crate) fn comes_after(&self, user: &Uid, event: &PresenceEvent) -> bool {
-        if user == &event.sender {
-            if self.last_active.is_none() {
-                true
-            } else {
-                event.content.last_active_ago < self.last_active
-            }
-        } else {
-            false
-        }
-    }
-
-    pub(crate) fn update(&mut self, room_id: &str, event: &PresenceEvent) {
-        self.last_active = event.content.last_active_ago;
-        self.current_room_id =
-            Some(RoomId::try_from(room_id).expect("room id failed CurrentRoom::update"));
-    }
-}
-
-#[derive(Debug)]
 /// A no IO Client implementation.
 ///
 /// This Client is a state machine that receives responses and events and
@@ -106,18 +70,32 @@ 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<String, Arc<RwLock<Room>>>,
+    pub joined_rooms: HashMap<String, Arc<Mutex<Room>>>,
-    /// The most recent room the logged in user used by `RoomId`.
-    pub current_room_id: CurrentRoom,
     /// 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>>>>,
 
     #[cfg(feature = "encryption")]
     olm: Arc<Mutex<Option<OlmMachine>>>,
 }
 
+impl fmt::Debug for Client {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("Client")
+            .field("session", &self.session)
+            .field("sync_token", &self.sync_token)
+            .field("joined_rooms", &self.joined_rooms)
+            .field("ignored_users", &self.ignored_users)
+            .field("push_ruleset", &self.push_ruleset)
+            .field("event_emitter", &"EventEmitter<...>")
+            .finish()
+    }
+}
+
 impl Client {
     /// Create a new client.
     ///
@@ -136,9 +114,9 @@ impl Client {
             session,
             sync_token: None,
             joined_rooms: HashMap::new(),
-            current_room_id: CurrentRoom::default(),
             ignored_users: Vec::new(),
             push_ruleset: None,
+            event_emitter: None,
             #[cfg(feature = "encryption")]
             olm: Arc::new(Mutex::new(olm)),
         })
@@ -149,6 +127,16 @@ impl Client {
         self.session.is_some()
     }
 
+    /// Add `EventEmitter` to `Client`.
+    ///
+    /// The methods of `EventEmitter` are called when the respective `RoomEvents` occur.
+    pub async fn add_event_emitter(
+        &mut self,
+        emitter: Arc<tokio::sync::Mutex<Box<dyn EventEmitter>>>,
+    ) {
+        self.event_emitter = Some(emitter);
+    }
+
     /// Receive a login response and update the session of the client.
     ///
     /// # Arguments
@@ -175,34 +163,29 @@ impl Client {
         Ok(())
     }
 
-    pub(crate) fn calculate_room_name(&self, room_id: &str) -> Option<String> {
+    pub(crate) async fn calculate_room_name(&self, room_id: &str) -> Option<String> {
-        self.joined_rooms.get(room_id).and_then(|r| {
+        if let Some(room) = self.joined_rooms.get(room_id) {
-            r.read()
+            let room = room.lock().await;
-                .map(|r| r.room_name.calculate_name(room_id, &r.members))
+            Some(room.room_name.calculate_name(room_id, &room.members))
-                .ok()
+        } else {
-        })
+            None
+        }
     }
 
-    pub(crate) fn calculate_room_names(&self) -> Vec<String> {
+    pub(crate) async fn calculate_room_names(&self) -> Vec<String> {
-        self.joined_rooms
+        let mut res = Vec::new();
-            .iter()
+        for (id, room) in &self.joined_rooms {
-            .flat_map(|(id, room)| {
+            let room = room.lock().await;
-                room.read()
+            res.push(room.room_name.calculate_name(id, &room.members))
-                    .map(|r| r.room_name.calculate_name(id, &r.members))
+        }
-                    .ok()
+        res
-            })
-            .collect()
     }
 
-    pub(crate) fn current_room_id(&self) -> Option<RoomId> {
+    pub(crate) fn get_or_create_room(&mut self, room_id: &str) -> &mut Arc<Mutex<Room>> {
-        self.current_room_id.current_room_id.clone()
-    }
-
-    pub(crate) fn get_or_create_room(&mut self, room_id: &str) -> &mut Arc<RwLock<Room>> {
         #[allow(clippy::or_fun_call)]
         self.joined_rooms
             .entry(room_id.to_string())
-            .or_insert(Arc::new(RwLock::new(Room::new(
+            .or_insert(Arc::new(Mutex::new(Room::new(
                 room_id,
                 &self
                     .session
@@ -213,8 +196,8 @@ impl Client {
             ))))
     }
 
-    pub(crate) fn get_room(&mut self, room_id: &str) -> Option<&mut Arc<RwLock<Room>>> {
+    pub(crate) fn get_room(&self, room_id: &str) -> Option<&Arc<Mutex<Room>>> {
-        self.joined_rooms.get_mut(room_id)
+        self.joined_rooms.get(room_id)
     }
 
     /// Handle a m.ignored_user_list event, updating the room state if necessary.
@@ -292,10 +275,7 @@ impl Client {
                     }
                 }
 
-                let mut room = self
+                let mut room = self.get_or_create_room(&room_id.to_string()).lock().await;
-                    .get_or_create_room(&room_id.to_string())
-                    .write()
-                    .unwrap();
                 room.receive_timeline_event(e);
                 decrypted_event
             }
@@ -313,8 +293,8 @@ impl Client {
     /// * `room_id` - The unique id of the room the event belongs to.
     ///
     /// * `event` - The event that should be handled by the client.
-    pub fn receive_joined_state_event(&mut self, room_id: &str, event: &StateEvent) -> bool {
+    pub async fn receive_joined_state_event(&mut self, room_id: &str, event: &StateEvent) -> bool {
-        let mut room = self.get_or_create_room(room_id).write().unwrap();
+        let mut room = self.get_or_create_room(room_id).lock().await;
         room.receive_state_event(event)
     }
 
@@ -328,19 +308,10 @@ impl Client {
     /// * `room_id` - The unique id of the room the event belongs to.
     ///
     /// * `event` - The event that should be handled by the client.
-    pub fn receive_presence_event(&mut self, room_id: &str, event: &PresenceEvent) -> bool {
+    pub async fn receive_presence_event(&mut self, room_id: &str, event: &PresenceEvent) -> bool {
-        let user_id = &self
-            .session
-            .as_ref()
-            .expect("to receive events you must be logged in")
-            .user_id;
-
-        if self.current_room_id.comes_after(user_id, event) {
-            self.current_room_id.update(room_id, event);
-        }
         // 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.write().unwrap();
+            let mut room = room.lock().await;
             room.receive_presence_event(event)
         } else {
             false
@@ -354,11 +325,13 @@ impl Client {
     ///
     /// # Arguments
     ///
+    /// * `room_id` - The unique id of the room the event belongs to.
+    ///
     /// * `event` - The presence event for a specified room member.
-    pub fn receive_account_data(&mut self, room_id: &str, event: &NonRoomEvent) -> bool {
+    pub async fn receive_account_data(&mut self, room_id: &str, event: &NonRoomEvent) -> bool {
         match event {
             NonRoomEvent::IgnoredUserList(iu) => self.handle_ignored_users(iu),
-            NonRoomEvent::Presence(p) => self.receive_presence_event(room_id, p),
+            NonRoomEvent::Presence(p) => self.receive_presence_event(room_id, p).await,
             NonRoomEvent::PushRules(pr) => self.handle_push_rules(pr),
             _ => false,
         }
@@ -386,7 +359,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.read().unwrap();
+                    let room = room.lock().await;
                     if !room.is_encrypted() {
                         continue;
                     }
@@ -489,18 +462,297 @@ impl Client {
         // TODO notify our callers of new devices via some callback.
         Ok(())
     }
+
+    pub(crate) async fn emit_timeline_event(&mut self, room_id: &RoomId, event: &mut RoomEvent) {
+        match event {
+            RoomEvent::RoomMember(mem) => {
+                if let Some(ee) = &self.event_emitter {
+                    if let Some(room) = self.get_room(&room_id.to_string()) {
+                        ee.lock()
+                            .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.to_string()) {
+                        ee.lock()
+                            .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.to_string()) {
+                        ee.lock()
+                            .await
+                            .on_room_canonical_alias(
+                                Arc::clone(&room),
+                                Arc::new(Mutex::new(canonical.clone())),
+                            )
+                            .await;
+                    }
+                }
+            }
+            RoomEvent::RoomAliases(aliases) => {
+                if let Some(ee) = &self.event_emitter {
+                    if let Some(room) = self.get_room(&room_id.to_string()) {
+                        ee.lock()
+                            .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.to_string()) {
+                        ee.lock()
+                            .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.to_string()) {
+                        ee.lock()
+                            .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.to_string()) {
+                        ee.lock()
+                            .await
+                            .on_room_message_feedback(
+                                Arc::clone(&room),
+                                Arc::new(Mutex::new(msg_feedback.clone())),
+                            )
+                            .await;
+                    }
+                }
+            }
+            RoomEvent::RoomRedaction(redaction) => {
+                if let Some(ee) = &self.event_emitter {
+                    if let Some(room) = self.get_room(&room_id.to_string()) {
+                        ee.lock()
+                            .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.to_string()) {
+                        ee.lock()
+                            .await
+                            .on_room_power_levels(
+                                Arc::clone(&room),
+                                Arc::new(Mutex::new(power.clone())),
+                            )
+                            .await;
+                    }
+                }
+            }
+            _ => {}
+        }
+    }
+
+    pub(crate) async fn emit_state_event(&mut self, room_id: &RoomId, event: &mut StateEvent) {
+        match event {
+            StateEvent::RoomMember(member) => {
+                if let Some(ee) = &self.event_emitter {
+                    if let Some(room) = self.get_room(&room_id.to_string()) {
+                        ee.lock()
+                            .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.to_string()) {
+                        ee.lock()
+                            .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.to_string()) {
+                        ee.lock()
+                            .await
+                            .on_state_canonical_alias(
+                                Arc::clone(&room),
+                                Arc::new(Mutex::new(canonical.clone())),
+                            )
+                            .await;
+                    }
+                }
+            }
+            StateEvent::RoomAliases(aliases) => {
+                if let Some(ee) = &self.event_emitter {
+                    if let Some(room) = self.get_room(&room_id.to_string()) {
+                        ee.lock()
+                            .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.to_string()) {
+                        ee.lock()
+                            .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.to_string()) {
+                        ee.lock()
+                            .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.to_string()) {
+                        ee.lock()
+                            .await
+                            .on_state_join_rules(
+                                Arc::clone(&room),
+                                Arc::new(Mutex::new(rules.clone())),
+                            )
+                            .await;
+                    }
+                }
+            }
+            _ => {}
+        }
+    }
+
+    pub(crate) async fn emit_account_data_event(
+        &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.to_string()) {
+                        ee.lock()
+                            .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.to_string()) {
+                        ee.lock()
+                            .await
+                            .on_account_ignored_users(
+                                Arc::clone(&room),
+                                Arc::new(Mutex::new(ignored.clone())),
+                            )
+                            .await;
+                    }
+                }
+            }
+            NonRoomEvent::PushRules(rules) => {
+                if let Some(ee) = &self.event_emitter {
+                    if let Some(room) = self.get_room(&room_id.to_string()) {
+                        ee.lock()
+                            .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.to_string()) {
+                        ee.lock()
+                            .await
+                            .on_account_data_fully_read(
+                                Arc::clone(&room),
+                                Arc::new(Mutex::new(full_read.clone())),
+                            )
+                            .await;
+                    }
+                }
+            }
+            _ => {}
+        }
+    }
+
+    pub(crate) async fn emit_presence_event(
+        &mut self,
+        room_id: &RoomId,
+        event: &mut PresenceEvent,
+    ) {
+        if let Some(ee) = &self.event_emitter {
+            if let Some(room) = self.get_room(&room_id.to_string()) {
+                ee.lock()
+                    .await
+                    .on_presence_event(Arc::clone(&room), Arc::new(Mutex::new(event.clone())))
+                    .await;
+            }
+        }
+    }
 }
 
 #[cfg(test)]
 mod test {
-    use super::*;
 
-    use crate::events::room::member::MembershipState;
     use crate::identifiers::UserId;
     use crate::{AsyncClient, Session, SyncSettings};
 
     use mockito::{mock, Matcher};
-    use tokio::runtime::Runtime;
     use url::Url;
 
     use std::convert::TryFrom;

src/event_emitter/mod.rs

@@ -0,0 +1,334 @@
+// Copyright 2020 Damir Jelić
+// Copyright 2020 The Matrix.org Foundation C.I.C.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// 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::{
+    fully_read::FullyReadEvent,
+    ignored_user_list::IgnoredUserListEvent,
+    presence::PresenceEvent,
+    push_rules::PushRulesEvent,
+    room::{
+        aliases::AliasesEvent,
+        avatar::AvatarEvent,
+        canonical_alias::CanonicalAliasEvent,
+        join_rules::JoinRulesEvent,
+        member::MemberEvent,
+        message::{feedback::FeedbackEvent, MessageEvent},
+        name::NameEvent,
+        power_levels::PowerLevelsEvent,
+        redaction::RedactionEvent,
+    },
+};
+use crate::models::Room;
+
+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.
+///
+/// # Examples
+/// ```
+/// # use std::ops::Deref;
+/// # use std::sync::Arc;
+/// # use std::{env, process::exit};
+/// # use url::Url;
+/// use matrix_sdk::{
+///     self,
+///     events::{
+///         room::message::{MessageEvent, MessageEventContent, TextMessageEventContent},
+///     },
+///     AsyncClient, AsyncClientConfig, EventEmitter, Room, SyncSettings,
+/// };
+/// use tokio::sync::Mutex;
+///
+/// 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>>) {
+///         if let MessageEvent {
+///             content: MessageEventContent::Text(TextMessageEventContent { body: msg_body, .. }),
+///             sender,
+///             ..
+///         } = event.lock().await.deref()
+///         {
+///             let rooms = room.lock().await;
+///             let member = rooms.members.get(&sender.to_string()).unwrap();
+///             println!(
+///                 "{}: {}",
+///                 member
+///                     .user
+///                     .display_name
+///                     .as_ref()
+///                     .unwrap_or(&sender.to_string()),
+///                 msg_body
+///             );
+///         }
+///     }
+/// }
+/// ```
+#[async_trait::async_trait]
+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>>) {}
+    /// Fires when `AsyncClient` receives a `RoomEvent::RoomName` event.
+    async fn on_room_name(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<NameEvent>>) {}
+    /// Fires when `AsyncClient` receives a `RoomEvent::RoomCanonicalAlias` event.
+    async fn on_room_canonical_alias(
+        &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>>) {}
+    /// Fires when `AsyncClient` receives a `RoomEvent::RoomAvatar` event.
+    async fn on_room_avatar(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<AvatarEvent>>) {}
+    /// Fires when `AsyncClient` receives a `RoomEvent::RoomMessage` event.
+    async fn on_room_message(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<MessageEvent>>) {}
+    /// Fires when `AsyncClient` receives a `RoomEvent::RoomMessageFeedback` event.
+    async fn on_room_message_feedback(
+        &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>>) {}
+    /// Fires when `AsyncClient` receives a `RoomEvent::RoomPowerLevels` event.
+    async fn on_room_power_levels(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<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>>) {}
+    /// Fires when `AsyncClient` receives a `StateEvent::RoomName` event.
+    async fn on_state_name(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<NameEvent>>) {}
+    /// Fires when `AsyncClient` receives a `StateEvent::RoomCanonicalAlias` event.
+    async fn on_state_canonical_alias(
+        &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>>) {}
+    /// Fires when `AsyncClient` receives a `StateEvent::RoomAvatar` event.
+    async fn on_state_avatar(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<AvatarEvent>>) {}
+    /// Fires when `AsyncClient` receives a `StateEvent::RoomPowerLevels` event.
+    async fn on_state_power_levels(
+        &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>>) {}
+
+    // `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>>) {}
+    /// Fires when `AsyncClient` receives a `NonRoomEvent::RoomName` event.
+    async fn on_account_ignored_users(
+        &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>>) {}
+    /// Fires when `AsyncClient` receives a `NonRoomEvent::RoomAliases` event.
+    async fn on_account_data_fully_read(
+        &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>>) {}
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    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>>) {
+            self.0.lock().await.push("member".to_string())
+        }
+        async fn on_room_name(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<NameEvent>>) {
+            self.0.lock().await.push("name".to_string())
+        }
+        async fn on_room_canonical_alias(
+            &mut self,
+            r: 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>>) {
+            self.0.lock().await.push("aliases".to_string())
+        }
+        async fn on_room_avatar(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<AvatarEvent>>) {
+            self.0.lock().await.push("avatar".to_string())
+        }
+        async fn on_room_message(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<MessageEvent>>) {
+            self.0.lock().await.push("message".to_string())
+        }
+        async fn on_room_message_feedback(
+            &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>>) {
+            self.0.lock().await.push("redaction".to_string())
+        }
+        async fn on_room_power_levels(
+            &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>>) {
+            self.0.lock().await.push("state member".to_string())
+        }
+        async fn on_state_name(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<NameEvent>>) {
+            self.0.lock().await.push("state name".to_string())
+        }
+        async fn on_state_canonical_alias(
+            &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>>) {
+            self.0.lock().await.push("state aliases".to_string())
+        }
+        async fn on_state_avatar(&mut self, _: Arc<Mutex<Room>>, _: Arc<Mutex<AvatarEvent>>) {
+            self.0.lock().await.push("state avatar".to_string())
+        }
+        async fn on_state_power_levels(
+            &mut self,
+            _: Arc<Mutex<Room>>,
+            _: Arc<Mutex<PowerLevelsEvent>>,
+        ) {
+            self.0.lock().await.push("state power".to_string())
+        }
+        async fn on_state_join_rules(
+            &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>>) {
+            self.0.lock().await.push("account presence".to_string())
+        }
+        async fn on_account_ignored_users(
+            &mut self,
+            _: Arc<Mutex<Room>>,
+            _: Arc<Mutex<IgnoredUserListEvent>>,
+        ) {
+            self.0.lock().await.push("account ignore".to_string())
+        }
+        async fn on_account_push_rules(
+            &mut self,
+            _: Arc<Mutex<Room>>,
+            _: Arc<Mutex<PushRulesEvent>>,
+        ) {
+            self.0.lock().await.push("".to_string())
+        }
+        async fn on_account_data_fully_read(
+            &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>>) {
+            self.0.lock().await.push("presence event".to_string())
+        }
+    }
+
+    use crate::identifiers::UserId;
+    use crate::{AsyncClient, Session, SyncSettings};
+
+    use mockito::{mock, Matcher};
+    use url::Url;
+
+    use std::convert::TryFrom;
+    use std::str::FromStr;
+    use std::time::Duration;
+
+    #[tokio::test]
+    async fn event_emitter() {
+        let homeserver = Url::from_str(&mockito::server_url()).unwrap();
+
+        let session = Session {
+            access_token: "1234".to_owned(),
+            user_id: UserId::try_from("@example:example.com").unwrap(),
+            device_id: "DEVICEID".to_owned(),
+        };
+
+        let _m = mock(
+            "GET",
+            Matcher::Regex(r"^/_matrix/client/r0/sync\?.*$".to_string()),
+        )
+        .with_status(200)
+        .with_body_from_file("tests/data/sync.json")
+        .create();
+
+        let vec = Arc::new(Mutex::new(Vec::new()));
+        let test_vec = Arc::clone(&vec);
+        let mut emitter = Arc::new(Mutex::new(
+            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;
+
+        let sync_settings = SyncSettings::new().timeout(Duration::from_millis(3000));
+        let _response = client.sync(sync_settings).await.unwrap();
+
+        let v = test_vec.lock().await;
+        assert_eq!(
+            v.as_slice(),
+            [
+                "state rules",
+                "state member",
+                "state aliases",
+                "state power",
+                "state canonical",
+                "state member",
+                "state member",
+                "message",
+                "account read",
+                "account ignore",
+                "presence event",
+            ],
+        )
+    }
+}

src/lib.rs

@@ -28,6 +28,7 @@
 
 pub use crate::{error::Error, error::Result, session::Session};
 pub use reqwest::header::InvalidHeaderValue;
+pub use ruma_api;
 pub use ruma_client_api as api;
 pub use ruma_events as events;
 pub use ruma_identifiers as identifiers;
@@ -35,6 +36,7 @@ pub use ruma_identifiers as identifiers;
 mod async_client;
 mod base_client;
 mod error;
+mod event_emitter;
 mod models;
 mod session;
 
@@ -43,6 +45,7 @@ mod crypto;
 
 pub use async_client::{AsyncClient, AsyncClientConfig, SyncSettings};
 pub use base_client::Client;
+pub use event_emitter::EventEmitter;
 pub use models::Room;
 
 pub(crate) const VERSION: &str = env!("CARGO_PKG_VERSION");

src/models/mod.rs

@@ -6,5 +6,6 @@ pub use room::{Room, RoomName};
 pub use room_member::RoomMember;
 pub use user::User;
 
+#[allow(dead_code)]
 pub type Token = String;
 pub type UserId = String;

src/models/room.rs

@@ -14,11 +14,11 @@
 // limitations under the License.
 
 use std::collections::HashMap;
-use std::sync::{Arc, RwLock};
 
-use super::{RoomMember, User, UserId};
+use super::{RoomMember, UserId};
 
 use crate::events::collections::all::{RoomEvent, StateEvent};
+use crate::events::presence::PresenceEvent;
 use crate::events::room::{
     aliases::AliasesEvent,
     canonical_alias::CanonicalAliasEvent,
@@ -27,12 +27,7 @@ use crate::events::room::{
     name::NameEvent,
     power_levels::PowerLevelsEvent,
 };
-use crate::events::{
-    presence::{PresenceEvent, PresenceEventContent},
-    EventResult,
-};
 use crate::identifiers::RoomAliasId;
-use crate::session::Session;
 
 use js_int::UInt;
 
@@ -145,7 +140,7 @@ impl Room {
         }
     }
 
-    /// Calculate and return the display name of the room.
+    /// Return the display name of the room.
     pub fn calculate_name(&self) -> String {
         self.room_name.calculate_name(&self.room_id, &self.members)
     }
@@ -242,8 +237,7 @@ impl Room {
         }
     }
 
-    fn handle_encryption_event(&mut self, _: &EncryptionEvent) -> bool {
+    fn handle_encryption_event(&mut self, _event: &EncryptionEvent) -> bool {
-        // TODO store the encryption settings.
         self.encrypted = true;
         true
     }
@@ -318,14 +312,11 @@ impl Room {
 
 #[cfg(test)]
 mod test {
-    use super::*;
-
     use crate::events::room::member::MembershipState;
     use crate::identifiers::UserId;
     use crate::{AsyncClient, Session, SyncSettings};
 
     use mockito::{mock, Matcher};
-    use tokio::runtime::Runtime;
     use url::Url;
 
     use std::convert::TryFrom;
@@ -360,11 +351,11 @@ mod test {
         let room = &rooms
             .get("!SVkFJHzfwvuaIEawgC:localhost")
             .unwrap()
-            .read()
+            .lock()
-            .unwrap();
+            .await;
 
         assert_eq!(2, room.members.len());
-        for (id, member) in &room.members {
+        for (_id, member) in &room.members {
             assert_eq!(MembershipState::Join, member.membership);
         }
     }

src/models/room_member.rs

@@ -16,20 +16,14 @@
 use std::convert::TryFrom;
 
 use super::User;
-use crate::api::r0 as api;
+use crate::events::collections::all::Event;
-use crate::events::collections::all::{Event, RoomEvent, StateEvent};
 use crate::events::room::{
-    aliases::AliasesEvent,
+    member::{MemberEvent, MembershipChange, MembershipState},
-    canonical_alias::CanonicalAliasEvent,
-    member::{MemberEvent, MemberEventContent, MembershipChange, MembershipState},
-    name::NameEvent,
     power_levels::PowerLevelsEvent,
 };
-use crate::events::EventResult;
+use crate::identifiers::UserId;
-use crate::identifiers::{RoomAliasId, UserId};
-use crate::session::Session;
 
-use js_int::{Int, UInt};
+use js_int::Int;
 #[cfg(feature = "encryption")]
 use tokio::sync::Mutex;
 
@@ -108,7 +102,7 @@ impl RoomMember {
             max_power = *power.max(&max_power);
         }
 
-        let mut changed = false;
+        let changed;
         if let Some(user_power) = event.content.users.get(&self.user_id) {
             changed = self.power_level != Some(*user_power);
             self.power_level = Some(*user_power);
@@ -134,7 +128,6 @@ mod test {
 
     use js_int::{Int, UInt};
     use mockito::{mock, Matcher};
-    use tokio::runtime::Runtime;
     use url::Url;
 
     use std::collections::HashMap;
@@ -174,10 +167,10 @@ mod test {
         let mut room = rooms
             .get_mut("!SVkFJHzfwvuaIEawgC:localhost")
             .unwrap()
-            .write()
+            .lock()
-            .unwrap();
+            .await;
 
-        for (id, member) in &mut room.members {
+        for (_id, member) in &mut room.members {
             let power = power_levels();
             assert!(member.update_power(&power));
             assert_eq!(MembershipState::Join, member.membership);

src/models/user.rs

@@ -13,22 +13,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::collections::HashMap;
+use crate::events::collections::all::Event;
-use std::sync::{Arc, RwLock};
-
-use super::UserId;
-use crate::api::r0 as api;
-use crate::events::collections::all::{Event, RoomEvent, StateEvent};
 use crate::events::presence::{PresenceEvent, PresenceEventContent, PresenceState};
-use crate::events::room::{
+use crate::events::room::member::MemberEvent;
-    aliases::AliasesEvent,
-    canonical_alias::CanonicalAliasEvent,
-    member::{MemberEvent, MembershipState},
-    name::NameEvent,
-};
-use crate::events::EventResult;
-use crate::identifiers::RoomAliasId;
-use crate::session::Session;
 
 use js_int::UInt;
 #[cfg(feature = "encryption")]

tests/async_client_tests.rs

@@ -2,7 +2,6 @@ use matrix_sdk::identifiers::UserId;
 use matrix_sdk::{AsyncClient, Session, SyncSettings};
 
 use mockito::{mock, Matcher};
-use tokio::runtime::Runtime;
 use url::Url;
 
 use std::convert::TryFrom;
@@ -88,33 +87,3 @@ async fn room_names() {
         client.get_room_name("!SVkFJHzfwvuaIEawgC:localhost").await
     );
 }
-
-#[tokio::test]
-async fn current_room() {
-    let homeserver = Url::from_str(&mockito::server_url()).unwrap();
-
-    let session = Session {
-        access_token: "1234".to_owned(),
-        user_id: UserId::try_from("@example:localhost").unwrap(),
-        device_id: "DEVICEID".to_owned(),
-    };
-
-    let _m = mock(
-        "GET",
-        Matcher::Regex(r"^/_matrix/client/r0/sync\?.*$".to_string()),
-    )
-    .with_status(200)
-    .with_body_from_file("tests/data/sync.json")
-    .create();
-
-    let mut client = AsyncClient::new(homeserver, Some(session)).unwrap();
-
-    let sync_settings = SyncSettings::new().timeout(Duration::from_millis(3000));
-
-    let _response = client.sync(sync_settings).await.unwrap();
-
-    assert_eq!(
-        Some("!SVkFJHzfwvuaIEawgC:localhost".into()),
-        client.current_room_id().await.map(|id| id.to_string())
-    );
-}