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

Merge branch 'bot-example'

master
Damir Jelić 2020-04-15 16:10:18 +02:00
parent af73ebdf09 9b52b58fea
commit 616a67c1cb
8 changed files with 272 additions and 359 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

118
examples/command_bot.rs Normal file
View File

@ -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(())
}

30
examples/login.rs
View File

@ -1,4 +1,3 @@
use std::ops::Deref;
use std::sync::Arc; use std::sync::Arc;
use std::{env, process::exit}; use std::{env, process::exit};
use url::Url; use url::Url;
@ -8,26 +7,31 @@ use matrix_sdk::{
events::room::message::{MessageEvent, MessageEventContent, TextMessageEventContent}, events::room::message::{MessageEvent, MessageEventContent, TextMessageEventContent},
AsyncClient, AsyncClientConfig, EventEmitter, Room, SyncSettings, AsyncClient, AsyncClientConfig, EventEmitter, Room, SyncSettings,
}; };
use tokio::sync::Mutex; use tokio::sync::RwLock;
struct EventCallback; struct EventCallback;
#[async_trait::async_trait] #[async_trait::async_trait]
impl EventEmitter for EventCallback { 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 { if let MessageEvent {
content: MessageEventContent::Text(TextMessageEventContent { body: msg_body, .. }), content: MessageEventContent::Text(TextMessageEventContent { body: msg_body, .. }),
sender, 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 homeserver_url = Url::parse(&homeserver_url)?;
let mut client = AsyncClient::new_with_config(homeserver_url, None, client_config).unwrap(); 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 client
.login(username, password, None, Some("rust-sdk".to_string())) .login(username, password, None, Some("rust-sdk".to_string()))

49
src/async_client.rs
View File

@ -53,6 +53,8 @@ const DEFAULT_SYNC_TIMEOUT: Duration = Duration::from_secs(30);
#[derive(Clone)] #[derive(Clone)]
/// An async/await enabled Matrix client. /// An async/await enabled Matrix client.
///
/// All of the state is held in an `Arc` so the `AsyncClient` can be cloned freely.
pub struct AsyncClient { pub struct AsyncClient {
/// The URL of the homeserver to connect to. /// The URL of the homeserver to connect to.
homeserver: Url, homeserver: Url,
@ -273,10 +275,7 @@ impl AsyncClient {
/// Add `EventEmitter` to `AsyncClient`. /// Add `EventEmitter` to `AsyncClient`.
/// ///
/// The methods of `EventEmitter` are called when the respective `RoomEvents` occur. /// The methods of `EventEmitter` are called when the respective `RoomEvents` occur.
pub async fn add_event_emitter( 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); self.base_client.write().await.event_emitter = Some(emitter);
} }
@ -301,7 +300,7 @@ impl AsyncClient {
/// Returns the rooms this client knows about. /// Returns the rooms this client knows about.
/// ///
/// A `HashMap` of room id to `matrix::models::Room` /// A `HashMap` of room id to `matrix::models::Room`
pub async fn get_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() self.base_client.read().await.joined_rooms.clone()
} }
@ -319,7 +318,7 @@ impl AsyncClient {
/// only if the client also holds the encryption keys for this device. /// only if the client also holds the encryption keys for this device.
#[instrument(skip(password))] #[instrument(skip(password))]
pub async fn login<S: Into<String> + std::fmt::Debug>( pub async fn login<S: Into<String> + std::fmt::Debug>(
&mut self, &self,
user: S, user: S,
password: S, password: S,
device_id: Option<S>, device_id: Option<S>,
@ -369,7 +368,7 @@ impl AsyncClient {
/// * alias - The `RoomId` or `RoomAliasId` of the room to be joined. /// * alias - The `RoomId` or `RoomAliasId` of the room to be joined.
/// An alias looks like this `#name:example.com` /// An alias looks like this `#name:example.com`
pub async fn join_room_by_id_or_alias( pub async fn join_room_by_id_or_alias(
&mut self, &self,
alias: &RoomIdOrAliasId, alias: &RoomIdOrAliasId,
) -> Result<join_room_by_id_or_alias::Response> { ) -> Result<join_room_by_id_or_alias::Response> {
let request = join_room_by_id_or_alias::Request { let request = join_room_by_id_or_alias::Request {
@ -391,7 +390,7 @@ impl AsyncClient {
/// ///
/// * reason - Optional reason why the room member is being kicked out. /// * reason - Optional reason why the room member is being kicked out.
pub async fn kick_user( pub async fn kick_user(
&mut self, &self,
room_id: &RoomId, room_id: &RoomId,
user_id: &UserId, user_id: &UserId,
reason: Option<String>, reason: Option<String>,
@ -412,7 +411,7 @@ impl AsyncClient {
/// ///
/// * room_id - The `RoomId` of the room to leave. /// * room_id - The `RoomId` of the room to leave.
/// ///
pub async fn leave_room(&mut self, room_id: &RoomId) -> Result<leave_room::Response> { pub async fn leave_room(&self, room_id: &RoomId) -> Result<leave_room::Response> {
let request = leave_room::Request { let request = leave_room::Request {
room_id: room_id.clone(), room_id: room_id.clone(),
}; };
@ -429,7 +428,7 @@ impl AsyncClient {
/// ///
/// * user_id - The `UserId` of the user to invite to the room. /// * user_id - The `UserId` of the user to invite to the room.
pub async fn invite_user_by_id( pub async fn invite_user_by_id(
&mut self, &self,
room_id: &RoomId, room_id: &RoomId,
user_id: &UserId, user_id: &UserId,
) -> Result<invite_user::Response> { ) -> Result<invite_user::Response> {
@ -452,7 +451,7 @@ impl AsyncClient {
/// ///
/// * invite_id - A third party id of a user to invite to the room. /// * invite_id - A third party id of a user to invite to the room.
pub async fn invite_user_by_3pid( pub async fn invite_user_by_3pid(
&mut self, &self,
room_id: &RoomId, room_id: &RoomId,
invite_id: &Invite3pid, invite_id: &Invite3pid,
) -> Result<invite_user::Response> { ) -> Result<invite_user::Response> {
@ -493,7 +492,7 @@ impl AsyncClient {
/// # }); /// # });
/// ``` /// ```
pub async fn create_room<R: Into<create_room::Request>>( pub async fn create_room<R: Into<create_room::Request>>(
&mut self, &self,
room: R, room: R,
) -> Result<create_room::Response> { ) -> Result<create_room::Response> {
let request = room.into(); let request = room.into();
@ -537,7 +536,7 @@ impl AsyncClient {
/// # }); /// # });
/// ``` /// ```
pub async fn room_messages<R: Into<get_message_events::Request>>( pub async fn room_messages<R: Into<get_message_events::Request>>(
&mut self, &self,
request: R, request: R,
) -> Result<get_message_events::IncomingResponse> { ) -> Result<get_message_events::IncomingResponse> {
let req = request.into(); let req = request.into();
@ -550,10 +549,7 @@ impl AsyncClient {
/// ///
/// * `sync_settings` - Settings for the sync call. /// * `sync_settings` - Settings for the sync call.
#[instrument] #[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 { let request = sync_events::Request {
filter: None, filter: None,
since: sync_settings.token, since: sync_settings.token,
@ -565,9 +561,8 @@ impl AsyncClient {
let mut response = self.send(request).await?; let mut response = self.send(request).await?;
for (room_id, room) in &mut response.rooms.join { for (room_id, room) in &mut response.rooms.join {
let mut client = self.base_client.write().await;
let _matrix_room = { let _matrix_room = {
let mut client = self.base_client.write().await;
for event in &room.state.events { for event in &room.state.events {
if let EventResult::Ok(e) = event { if let EventResult::Ok(e) = event {
client.receive_joined_state_event(&room_id, &e).await; client.receive_joined_state_event(&room_id, &e).await;
@ -580,12 +575,14 @@ impl AsyncClient {
// re looping is not ideal here // re looping is not ideal here
for event in &mut room.state.events { for event in &mut room.state.events {
if let EventResult::Ok(e) = event { if let EventResult::Ok(e) = event {
let client = self.base_client.read().await;
client.emit_state_event(room_id, e).await; client.emit_state_event(room_id, e).await;
} }
} }
for mut event in &mut room.timeline.events { for mut event in &mut room.timeline.events {
let decrypted_event = { let decrypted_event = {
let mut client = self.base_client.write().await;
client client
.receive_joined_timeline_event(room_id, &mut event) .receive_joined_timeline_event(room_id, &mut event)
.await .await
@ -596,6 +593,7 @@ impl AsyncClient {
} }
if let EventResult::Ok(e) = event { if let EventResult::Ok(e) = event {
let client = self.base_client.read().await;
client.emit_timeline_event(room_id, e).await; client.emit_timeline_event(room_id, e).await;
} }
} }
@ -604,8 +602,8 @@ impl AsyncClient {
for account_data in &mut room.account_data.events { for account_data in &mut room.account_data.events {
{ {
if let EventResult::Ok(e) = account_data { 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.receive_account_data_event(&room_id, e).await;
client.emit_account_data_event(room_id, e).await; client.emit_account_data_event(room_id, e).await;
} }
} }
@ -617,6 +615,7 @@ impl AsyncClient {
for presence in &mut response.presence.events { for presence in &mut response.presence.events {
{ {
if let EventResult::Ok(e) = presence { if let EventResult::Ok(e) = presence {
let mut client = self.base_client.write().await;
client.receive_presence_event(&room_id, e).await; client.receive_presence_event(&room_id, e).await;
client.emit_presence_event(room_id, e).await; client.emit_presence_event(room_id, e).await;
@ -627,6 +626,7 @@ impl AsyncClient {
for ephemeral in &mut room.ephemeral.events { for ephemeral in &mut room.ephemeral.events {
{ {
if let EventResult::Ok(e) = ephemeral { if let EventResult::Ok(e) = ephemeral {
let mut client = self.base_client.write().await;
client.receive_ephemeral_event(&room_id, e).await; client.receive_ephemeral_event(&room_id, e).await;
client.emit_ephemeral_event(room_id, e).await; client.emit_ephemeral_event(room_id, e).await;
@ -698,7 +698,7 @@ impl AsyncClient {
/// ``` /// ```
#[instrument(skip(callback))] #[instrument(skip(callback))]
pub async fn sync_forever<C>( pub async fn sync_forever<C>(
&mut self, &self,
sync_settings: SyncSettings, sync_settings: SyncSettings,
callback: impl Fn(sync_events::IncomingResponse) -> C + Send, callback: impl Fn(sync_events::IncomingResponse) -> C + Send,
) where ) where
@ -810,7 +810,6 @@ impl AsyncClient {
} else { } else {
request_builder request_builder
}; };
let mut response = request_builder.send().await?; let mut response = request_builder.send().await?;
trace!("Got response: {:?}", response); trace!("Got response: {:?}", response);
@ -877,7 +876,7 @@ impl AsyncClient {
/// }) /// })
/// ``` /// ```
pub async fn room_send( pub async fn room_send(
&mut self, &self,
room_id: &RoomId, room_id: &RoomId,
#[allow(unused_mut)] mut content: MessageEventContent, #[allow(unused_mut)] mut content: MessageEventContent,
txn_id: Option<Uuid>, txn_id: Option<Uuid>,
@ -892,7 +891,7 @@ impl AsyncClient {
let room = client.joined_rooms.get(room_id); let room = client.joined_rooms.get(room_id);
match room { match room {
Some(r) => r.lock().await.is_encrypted(), Some(r) => r.read().await.is_encrypted(),
None => false, None => false,
} }
}; };
@ -901,7 +900,7 @@ impl AsyncClient {
let missing_sessions = { let missing_sessions = {
let client = self.base_client.read().await; let client = self.base_client.read().await;
let room = client.joined_rooms.get(room_id); 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(); let users = room.members.keys();
self.base_client self.base_client
.read() .read()

230
src/base_client.rs
View File

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

184
src/event_emitter/mod.rs
View File

@ -12,7 +12,6 @@
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and // See the License for the specific language governing permissions and
// limitations under the License. // limitations under the License.
use std::sync::Arc; use std::sync::Arc;
use crate::events::{ use crate::events::{
@ -33,8 +32,8 @@ use crate::events::{
}, },
}; };
use crate::models::Room; 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. /// 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. /// The `AsyncClient` calls each method when the corresponding event is received.
/// ///
@ -51,29 +50,29 @@ use tokio::sync::Mutex;
/// }, /// },
/// AsyncClient, AsyncClientConfig, EventEmitter, Room, SyncSettings, /// AsyncClient, AsyncClientConfig, EventEmitter, Room, SyncSettings,
/// }; /// };
/// use tokio::sync::Mutex; /// use tokio::sync::RwLock;
/// ///
/// struct EventCallback; /// struct EventCallback;
/// ///
/// #[async_trait::async_trait] /// #[async_trait::async_trait]
/// impl EventEmitter for EventCallback { /// 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 { /// if let MessageEvent {
/// content: MessageEventContent::Text(TextMessageEventContent { body: msg_body, .. }), /// content: MessageEventContent::Text(TextMessageEventContent { body: msg_body, .. }),
/// sender, /// 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()
/// .unwrap_or(&sender.to_string()), /// .map(ToString::to_string)
/// msg_body /// .unwrap_or(sender.to_string())
/// ); /// };
/// println!("{}: {}", name, msg_body);
/// } /// }
/// } /// }
/// } /// }
@ -82,193 +81,128 @@ use tokio::sync::Mutex;
pub trait EventEmitter: Send + Sync { pub trait EventEmitter: Send + Sync {
// ROOM EVENTS from `IncomingTimeline` // ROOM EVENTS from `IncomingTimeline`
/// Fires when `AsyncClient` receives a `RoomEvent::RoomMember` event. /// 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. /// 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. /// 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. /// 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. /// 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. /// 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. /// 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. /// 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. /// 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` // `RoomEvent`s from `IncomingState`
/// Fires when `AsyncClient` receives a `StateEvent::RoomMember` event. /// 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. /// 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. /// 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. /// 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. /// 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. /// 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. /// 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` // `NonRoomEvent` (this is a type alias from ruma_events) from `IncomingAccountData`
/// Fires when `AsyncClient` receives a `NonRoomEvent::RoomMember` event. /// 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. /// 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. /// 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. /// 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 // `PresenceEvent` is a struct so there is only the one method
/// Fires when `AsyncClient` receives a `NonRoomEvent::RoomAliases` event. /// 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)] #[cfg(test)]
mod test { mod test {
use super::*; use super::*;
use std::sync::Arc;
use tokio::sync::Mutex;
#[derive(Clone)]
pub struct EvEmitterTest(Arc<Mutex<Vec<String>>>); pub struct EvEmitterTest(Arc<Mutex<Vec<String>>>);
#[async_trait::async_trait] #[async_trait::async_trait]
impl EventEmitter for EvEmitterTest { 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()) 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()) 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()) 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()) 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()) 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()) 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()) 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()) 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()) 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()) 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()) 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()) 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()) 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()) 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()) 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()) 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()) 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()) 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()) 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()) 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()) self.0.lock().await.push("presence event".to_string())
} }
} }
@ -303,11 +237,9 @@ mod test {
let vec = Arc::new(Mutex::new(Vec::new())); let vec = Arc::new(Mutex::new(Vec::new()));
let test_vec = Arc::clone(&vec); 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(); 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 sync_settings = SyncSettings::new().timeout(Duration::from_millis(3000));
let _response = client.sync(sync_settings).await.unwrap(); let _response = client.sync(sync_settings).await.unwrap();

4
src/models/room.rs
View File

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

4
src/request_builder.rs
View File

@ -341,7 +341,7 @@ mod test {
.room_alias_name("room_alias") .room_alias_name("room_alias")
.topic("room topic") .topic("room topic")
.visibility(Visibility::Private); .visibility(Visibility::Private);
let mut cli = AsyncClient::new(homeserver, Some(session)).unwrap(); let cli = AsyncClient::new(homeserver, Some(session)).unwrap();
assert!(cli.create_room(builder).await.is_ok()); assert!(cli.create_room(builder).await.is_ok());
} }
@ -373,7 +373,7 @@ mod test {
// TODO this makes ruma error `Err(IntoHttp(IntoHttpError(Query(Custom("unsupported value")))))`?? // TODO this makes ruma error `Err(IntoHttp(IntoHttpError(Query(Custom("unsupported value")))))`??
// .filter(RoomEventFilter::default()); // .filter(RoomEventFilter::default());
let mut cli = AsyncClient::new(homeserver, Some(session)).unwrap(); let cli = AsyncClient::new(homeserver, Some(session)).unwrap();
assert!(cli.room_messages(builder).await.is_ok()); assert!(cli.room_messages(builder).await.is_ok());
} }
} }

6
tests/async_client_tests.rs
View File

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