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

Rewrite event handler abstraction

master
Jonas Platte 2021-06-22 11:36:33 +02:00
parent 623408913c
commit 2fdad12521
No known key found for this signature in database
GPG Key ID: CC154DE0E30B7C67
12 changed files with 741 additions and 1225 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
matrix_sdk/Cargo.toml
View File

@ -31,9 +31,11 @@ appservice = ["ruma/appservice-api-s", "ruma/appservice-api-helper", "ruma/rand"
docs = ["encryption", "sled_cryptostore", "sled_state_store", "sso_login"] docs = ["encryption", "sled_cryptostore", "sled_state_store", "sso_login"]
[dependencies] [dependencies]
anyhow = { version = "1.0.42", optional = true }
dashmap = "4.0.2" dashmap = "4.0.2"
futures = "0.3.15" futures = "0.3.15"
http = "0.2.4" http = "0.2.4"
serde = "1.0.126"
serde_json = "1.0.64" serde_json = "1.0.64"
thiserror = "1.0.25" thiserror = "1.0.25"
tracing = "0.1.26" tracing = "0.1.26"

34
matrix_sdk/examples/autojoin.rs
View File

@ -1,33 +1,19 @@
use std::{env, process::exit}; use std::{env, process::exit};
use matrix_sdk::{ use matrix_sdk::{
async_trait,
room::Room, room::Room,
ruma::events::{room::member::MemberEventContent, StrippedStateEvent}, ruma::events::{room::member::MemberEventContent, StrippedStateEvent},
Client, ClientConfig, EventHandler, SyncSettings, Client, ClientConfig, SyncSettings,
}; };
use tokio::time::{sleep, Duration}; use tokio::time::{sleep, Duration};
use url::Url; use url::Url;
struct AutoJoinBot { async fn on_stripped_state_member(
room_member: StrippedStateEvent<MemberEventContent>,
client: Client, client: Client,
}
impl AutoJoinBot {
pub fn new(client: Client) -> Self {
Self { client }
}
}
#[async_trait]
impl EventHandler for AutoJoinBot {
async fn on_stripped_state_member(
&self,
room: Room, room: Room,
room_member: &StrippedStateEvent<MemberEventContent>, ) {
_: Option<MemberEventContent>, if room_member.state_key != client.user_id().await.unwrap() {
) {
if room_member.state_key != self.client.user_id().await.unwrap() {
return; return;
} }
@ -39,12 +25,7 @@ impl EventHandler for AutoJoinBot {
// retry autojoin due to synapse sending invites, before the // retry autojoin due to synapse sending invites, before the
// invited user can join for more information see // invited user can join for more information see
// https://github.com/matrix-org/synapse/issues/4345 // https://github.com/matrix-org/synapse/issues/4345
eprintln!( eprintln!("Failed to join room {} ({:?}), retrying in {}s", room.room_id(), err, delay);
"Failed to join room {} ({:?}), retrying in {}s",
room.room_id(),
err,
delay
);
sleep(Duration::from_secs(delay)).await; sleep(Duration::from_secs(delay)).await;
delay *= 2; delay *= 2;
@ -56,7 +37,6 @@ impl EventHandler for AutoJoinBot {
} }
println!("Successfully joined room {}", room.room_id()); println!("Successfully joined room {}", room.room_id());
} }
}
} }
async fn login_and_sync( async fn login_and_sync(
@ -76,7 +56,7 @@ async fn login_and_sync(
println!("logged in as {}", username); println!("logged in as {}", username);
client.set_event_handler(Box::new(AutoJoinBot::new(client.clone()))).await; client.register_event_handler(on_stripped_state_member).await;
client.sync(SyncSettings::default()).await; client.sync(SyncSettings::default()).await;

18
matrix_sdk/examples/command_bot.rs
View File

@ -1,27 +1,16 @@
use std::{env, process::exit}; use std::{env, process::exit};
use matrix_sdk::{ use matrix_sdk::{
async_trait,
room::Room, room::Room,
ruma::events::{ ruma::events::{
room::message::{MessageEventContent, MessageType, TextMessageEventContent}, room::message::{MessageEventContent, MessageType, TextMessageEventContent},
AnyMessageEventContent, SyncMessageEvent, AnyMessageEventContent, SyncMessageEvent,
}, },
Client, ClientConfig, EventHandler, SyncSettings, Client, ClientConfig, SyncSettings,
}; };
use url::Url; use url::Url;
struct CommandBot; async fn on_room_message(event: SyncMessageEvent<MessageEventContent>, room: Room) {
impl CommandBot {
pub fn new() -> Self {
Self {}
}
}
#[async_trait]
impl EventHandler for CommandBot {
async fn on_room_message(&self, room: Room, event: &SyncMessageEvent<MessageEventContent>) {
if let Room::Joined(room) = room { if let Room::Joined(room) = room {
let msg_body = if let SyncMessageEvent { let msg_body = if let SyncMessageEvent {
content: content:
@ -51,7 +40,6 @@ impl EventHandler for CommandBot {
println!("message sent"); println!("message sent");
} }
} }
}
} }
async fn login_and_sync( async fn login_and_sync(
@ -79,7 +67,7 @@ async fn login_and_sync(
client.sync_once(SyncSettings::default()).await.unwrap(); client.sync_once(SyncSettings::default()).await.unwrap();
// add our CommandBot to be notified of incoming messages, we do this after the // 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. // initial sync to avoid responding to messages before the bot was running.
client.set_event_handler(Box::new(CommandBot::new())).await; client.register_event_handler(on_room_message).await;
// since we called `sync_once` before we entered our sync loop we must pass // since we called `sync_once` before we entered our sync loop we must pass
// that sync token to `sync` // that sync token to `sync`

28
matrix_sdk/examples/image_bot.rs
View File

@ -8,31 +8,22 @@ use std::{
}; };
use matrix_sdk::{ use matrix_sdk::{
self, async_trait, self,
room::Room, room::Room,
ruma::events::{ ruma::events::{
room::message::{MessageEventContent, MessageType, TextMessageEventContent}, room::message::{MessageEventContent, MessageType, TextMessageEventContent},
SyncMessageEvent, SyncMessageEvent,
}, },
Client, EventHandler, SyncSettings, Client, SyncSettings,
}; };
use tokio::sync::Mutex; use tokio::sync::Mutex;
use url::Url; use url::Url;
struct ImageBot { async fn on_room_message(
event: SyncMessageEvent<MessageEventContent>,
room: Room,
image: Arc<Mutex<File>>, image: Arc<Mutex<File>>,
} ) {
impl ImageBot {
pub fn new(image: File) -> Self {
let image = Arc::new(Mutex::new(image));
Self { image }
}
}
#[async_trait]
impl EventHandler for ImageBot {
async fn on_room_message(&self, room: Room, event: &SyncMessageEvent<MessageEventContent>) {
if let Room::Joined(room) = room { if let Room::Joined(room) = room {
let msg_body = if let SyncMessageEvent { let msg_body = if let SyncMessageEvent {
content: content:
@ -50,7 +41,7 @@ impl EventHandler for ImageBot {
if msg_body.contains("!image") { if msg_body.contains("!image") {
println!("sending image"); println!("sending image");
let mut image = self.image.lock().await; let mut image = image.lock().await;
room.send_attachment("cat", &mime::IMAGE_JPEG, &mut *image, None).await.unwrap(); room.send_attachment("cat", &mime::IMAGE_JPEG, &mut *image, None).await.unwrap();
@ -59,7 +50,6 @@ impl EventHandler for ImageBot {
println!("message sent"); println!("message sent");
} }
} }
}
} }
async fn login_and_sync( async fn login_and_sync(
@ -74,7 +64,9 @@ async fn login_and_sync(
client.login(&username, &password, None, Some("command bot")).await?; client.login(&username, &password, None, Some("command bot")).await?;
client.sync_once(SyncSettings::default()).await.unwrap(); client.sync_once(SyncSettings::default()).await.unwrap();
client.set_event_handler(Box::new(ImageBot::new(image))).await;
let image = Arc::new(Mutex::new(image));
client.register_event_handler(move |ev, room| on_room_message(ev, room, image.clone())).await;
let settings = SyncSettings::default().token(client.sync_token().await.unwrap()); let settings = SyncSettings::default().token(client.sync_token().await.unwrap());
client.sync(settings).await; client.sync(settings).await;

15
matrix_sdk/examples/login.rs
View File

@ -1,21 +1,17 @@
use std::{env, process::exit}; use std::{env, process::exit};
use matrix_sdk::{ use matrix_sdk::{
self, async_trait, self,
room::Room, room::Room,
ruma::events::{ ruma::events::{
room::message::{MessageEventContent, MessageType, TextMessageEventContent}, room::message::{MessageEventContent, MessageType, TextMessageEventContent},
SyncMessageEvent, SyncMessageEvent,
}, },
Client, EventHandler, SyncSettings, Client, SyncSettings,
}; };
use url::Url; use url::Url;
struct EventCallback; async fn on_room_message(event: SyncMessageEvent<MessageEventContent>, room: Room) {
#[async_trait]
impl EventHandler for EventCallback {
async fn on_room_message(&self, room: Room, event: &SyncMessageEvent<MessageEventContent>) {
if let Room::Joined(room) = room { if let Room::Joined(room) = room {
if let SyncMessageEvent { if let SyncMessageEvent {
content: content:
@ -27,12 +23,11 @@ impl EventHandler for EventCallback {
.. ..
} = event } = event
{ {
let member = room.get_member(sender).await.unwrap().unwrap(); let member = room.get_member(&sender).await.unwrap().unwrap();
let name = member.display_name().unwrap_or_else(|| member.user_id().as_str()); let name = member.display_name().unwrap_or_else(|| member.user_id().as_str());
println!("{}: {}", name, msg_body); println!("{}: {}", name, msg_body);
} }
} }
}
} }
async fn login( async fn login(
@ -43,7 +38,7 @@ async fn login(
let homeserver_url = Url::parse(&homeserver_url).expect("Couldn't parse the homeserver URL"); let homeserver_url = Url::parse(&homeserver_url).expect("Couldn't parse the homeserver URL");
let client = Client::new(homeserver_url).unwrap(); let client = Client::new(homeserver_url).unwrap();
client.set_event_handler(Box::new(EventCallback)).await; client.register_event_handler(on_room_message).await;
client.login(username, password, None, Some("rust-sdk")).await?; client.login(username, password, None, Some("rust-sdk")).await?;
client.sync(SyncSettings::new()).await; client.sync(SyncSettings::new()).await;

221
matrix_sdk/src/client.rs
View File

@ -13,12 +13,19 @@
// 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.
#[cfg(feature = "encryption")]
use std::io::{Cursor, Write};
#[cfg(all(feature = "encryption", not(target_arch = "wasm32")))] #[cfg(all(feature = "encryption", not(target_arch = "wasm32")))]
use std::path::PathBuf; use std::path::PathBuf;
#[cfg(feature = "encryption")]
use std::{ use std::{
collections::BTreeMap, collections::BTreeMap,
io::{Cursor, Write}, fmt::{self, Debug},
future::Future,
io::Read,
path::Path,
pin::Pin,
result::Result as StdResult,
sync::Arc,
}; };
#[cfg(feature = "sso_login")] #[cfg(feature = "sso_login")]
use std::{ use std::{
@ -26,16 +33,9 @@ use std::{
io::{Error as IoError, ErrorKind as IoErrorKind}, io::{Error as IoError, ErrorKind as IoErrorKind},
ops::Range, ops::Range,
}; };
use std::{
fmt::{self, Debug},
future::Future,
io::Read,
path::Path,
result::Result as StdResult,
sync::Arc,
};
use dashmap::DashMap; use dashmap::DashMap;
use futures::FutureExt;
use futures_timer::Delay as sleep; use futures_timer::Delay as sleep;
use http::HeaderValue; use http::HeaderValue;
#[cfg(feature = "sso_login")] #[cfg(feature = "sso_login")]
@ -50,7 +50,7 @@ use matrix_sdk_base::crypto::{
#[cfg(feature = "encryption")] #[cfg(feature = "encryption")]
use matrix_sdk_base::deserialized_responses::RoomEvent; use matrix_sdk_base::deserialized_responses::RoomEvent;
use matrix_sdk_base::{ use matrix_sdk_base::{
deserialized_responses::SyncResponse, deserialized_responses::{JoinedRoom, LeftRoom, SyncResponse},
media::{MediaEventContent, MediaFormat, MediaRequest, MediaThumbnailSize, MediaType}, media::{MediaEventContent, MediaFormat, MediaRequest, MediaThumbnailSize, MediaType},
BaseClient, BaseClientConfig, Session, Store, BaseClient, BaseClientConfig, Session, Store,
}; };
@ -60,14 +60,19 @@ use rand::{thread_rng, Rng};
use reqwest::header::InvalidHeaderValue; use reqwest::header::InvalidHeaderValue;
#[cfg(feature = "encryption")] #[cfg(feature = "encryption")]
use ruma::events::{AnyMessageEvent, AnyRoomEvent, AnySyncMessageEvent}; use ruma::events::{AnyMessageEvent, AnyRoomEvent, AnySyncMessageEvent};
use ruma::{api::SendAccessToken, events::AnyMessageEventContent, MxcUri}; use ruma::{
api::{client::r0::push::get_notifications::Notification, SendAccessToken},
events::AnyMessageEventContent,
MxcUri,
};
use serde::de::DeserializeOwned;
#[cfg(feature = "sso_login")] #[cfg(feature = "sso_login")]
use tokio::{net::TcpListener, sync::oneshot}; use tokio::{net::TcpListener, sync::oneshot};
#[cfg(feature = "sso_login")] #[cfg(feature = "sso_login")]
use tokio_stream::wrappers::TcpListenerStream; use tokio_stream::wrappers::TcpListenerStream;
#[cfg(feature = "encryption")] #[cfg(feature = "encryption")]
use tracing::{debug, warn}; use tracing::debug;
use tracing::{error, info, instrument}; use tracing::{error, info, instrument, warn};
use url::Url; use url::Url;
#[cfg(feature = "sso_login")] #[cfg(feature = "sso_login")]
use warp::Filter; use warp::Filter;
@ -138,9 +143,9 @@ use crate::{
}; };
use crate::{ use crate::{
error::HttpError, error::HttpError,
event_handler::Handler, event_handler::{EventHandler, EventHandlerData, EventHandlerResult, EventKind, SyncEvent},
http_client::{client_with_config, HttpClient, HttpSend}, http_client::{client_with_config, HttpClient, HttpSend},
room, Error, EventHandler, Result, room, Error, Result,
}; };
const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::from_secs(10); const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
@ -156,6 +161,14 @@ const SSO_SERVER_BIND_RANGE: Range<u16> = 20000..30000;
#[cfg(feature = "sso_login")] #[cfg(feature = "sso_login")]
const SSO_SERVER_BIND_TRIES: u8 = 10; const SSO_SERVER_BIND_TRIES: u8 = 10;
type EventHandlerFut = Pin<Box<dyn Future<Output = ()> + Send>>;
type EventHandlerFn = Box<dyn Fn(EventHandlerData<'_>) -> EventHandlerFut + Send + Sync>;
type EventHandlerMap = BTreeMap<(EventKind, &'static str), Vec<EventHandlerFn>>;
type NotificationHandlerFut = EventHandlerFut;
type NotificationHandlerFn =
Box<dyn Fn(Notification, room::Room, Client) -> NotificationHandlerFut + Send + Sync>;
/// An async/await enabled Matrix client. /// An async/await enabled Matrix client.
/// ///
/// All of the state is held in an `Arc` so the `Client` can be cloned freely. /// All of the state is held in an `Arc` so the `Client` can be cloned freely.
@ -176,9 +189,10 @@ pub struct Client {
key_claim_lock: Arc<Mutex<()>>, key_claim_lock: Arc<Mutex<()>>,
pub(crate) members_request_locks: Arc<DashMap<RoomId, Arc<Mutex<()>>>>, pub(crate) members_request_locks: Arc<DashMap<RoomId, Arc<Mutex<()>>>>,
pub(crate) typing_notice_times: Arc<DashMap<RoomId, Instant>>, pub(crate) typing_notice_times: Arc<DashMap<RoomId, Instant>>,
/// Any implementor of EventHandler will act as the callbacks for various /// Event handlers. See `register_event_handler`.
/// events. pub(crate) event_handlers: Arc<RwLock<EventHandlerMap>>,
event_handler: Arc<RwLock<Option<Handler>>>, /// Notification handlers. See `register_notification_handler`.
notification_handlers: Arc<RwLock<Vec<NotificationHandlerFn>>>,
/// Whether the client should operate in application service style mode. /// Whether the client should operate in application service style mode.
/// This is low-level functionality. For an high-level API check the /// This is low-level functionality. For an high-level API check the
/// `matrix_sdk_appservice` crate. /// `matrix_sdk_appservice` crate.
@ -562,7 +576,8 @@ impl Client {
key_claim_lock: Arc::new(Mutex::new(())), key_claim_lock: Arc::new(Mutex::new(())),
members_request_locks: Arc::new(DashMap::new()), members_request_locks: Arc::new(DashMap::new()),
typing_notice_times: Arc::new(DashMap::new()), typing_notice_times: Arc::new(DashMap::new()),
event_handler: Arc::new(RwLock::new(None)), event_handlers: Arc::new(RwLock::new(BTreeMap::new())),
notification_handlers: Arc::new(RwLock::new(Vec::new())),
appservice_mode: config.appservice_mode, appservice_mode: config.appservice_mode,
}) })
} }
@ -666,12 +681,7 @@ impl Client {
) -> Result<()> { ) -> Result<()> {
let txn_id = incoming_transaction.txn_id.clone(); let txn_id = incoming_transaction.txn_id.clone();
let response = incoming_transaction.try_into_sync_response(txn_id)?; let response = incoming_transaction.try_into_sync_response(txn_id)?;
let base_client = self.base_client.clone(); self.process_sync(response).await?;
let sync_response = base_client.receive_sync_response(response).await?;
if let Some(handler) = self.event_handler.read().await.as_ref() {
handler.handle_sync(&sync_response).await;
}
Ok(()) Ok(())
} }
@ -867,13 +877,76 @@ impl Client {
Ok(()) Ok(())
} }
/// Add `EventHandler` to `Client`. /// Register a handler for a specific event type.
/// ///
/// The methods of `EventHandler` are called when the respective /// The handler is a function or closue with one or more arguments. The
/// `RoomEvents` occur. /// first argument is the event itself. All additional arguments are
pub async fn set_event_handler(&self, handler: Box<dyn EventHandler>) { /// "context" arguments: They have to implement [`EventHandlerContext`][].
let handler = Handler { inner: handler, client: self.clone() }; /// This trait is named that way because most of the types implementing it
*self.event_handler.write().await = Some(handler); /// give additional context about an event: The room it was in, its raw form
/// and other similar things. As an exception to this,
/// [`matrix_sdk::Client`] also implements the `EventHandlerContext` trait
/// so you don't have to clone your client into the event handler manually.
///
/// Invalid context arguments, for example a [`Room`][room::Room] as an
/// argument to an account data event handler, will result in the event
/// handler being skipped and an error logged.
///
/// [`EventHandlerContext`]: crate::event_handler::EventHandlerContext
///
/// # Examples
///
/// ```no_run
/// # let client: matrix_sdk::Client = unimplemented!();
/// use matrix_sdk::{
/// room::Room,
/// ruma::events::{
/// push_rules::PushRulesEvent,
/// room::{message::MessageEventContent, topic::TopicEventContent},
/// SyncMessageEvent, SyncStateEvent,
/// },
/// Client,
/// };
///
/// client.register_event_handler(
/// |msg: SyncMessageEvent<MessageEventContent>, room: Room, client: Client| async move {
/// // Common usage: Room event plus room and client.
/// },
/// );
/// client.register_event_handler(|ev: SyncStateEvent<TopicEventContent>| async move {
/// // Also possible: Omit any or all arguments after the first.
/// });
/// ```
pub async fn register_event_handler<Ev, Ctx, H>(&self, handler: H)
where
Ev: SyncEvent + DeserializeOwned + Send + 'static,
H: EventHandler<Ev, Ctx>,
<H::Future as Future>::Output: EventHandlerResult,
{
let event_type = H::ID.1;
self.event_handlers.write().await.entry(H::ID).or_default().push(Box::new(move |data| {
let maybe_fut = serde_json::from_str(data.raw.get())
.map(|ev| handler.clone().handle_event(ev, data));
async move {
match maybe_fut {
Ok(Some(fut)) => {
fut.await.print_error(event_type);
}
Ok(None) => {
error!("Event handler for {} has an invalid context argument", event_type);
}
Err(e) => {
warn!(
"Failed to deserialize `{}` event, skipping event handler.\n\
Deserialization error: {}",
event_type, e,
);
}
}
}
.boxed()
}));
} }
/// Get all the rooms the client knows about. /// Get all the rooms the client knows about.
@ -1957,13 +2030,87 @@ impl Client {
); );
let response = self.send(request, Some(request_config)).await?; let response = self.send(request, Some(request_config)).await?;
let sync_response = self.base_client.receive_sync_response(response).await?; self.process_sync(response).await
if let Some(handler) = self.event_handler.read().await.as_ref() {
handler.handle_sync(&sync_response).await;
} }
Ok(sync_response) async fn process_sync(&self, response: sync_events::Response) -> Result<SyncResponse> {
let response = self.base_client.receive_sync_response(response).await?;
let SyncResponse {
next_batch: _,
rooms,
presence,
account_data,
to_device: _,
device_lists: _,
device_one_time_keys_count: _,
ambiguity_changes: _,
notifications,
} = &response;
self.handle_sync_events(EventKind::GlobalAccountData, &None, &account_data.events).await?;
self.handle_sync_events(EventKind::Presence, &None, &presence.events).await?;
for (room_id, room_info) in &rooms.join {
let room = self.get_room(room_id);
if room.is_none() {
// raise warning?
continue;
}
let JoinedRoom { unread_notifications: _, timeline, state, account_data, ephemeral } =
room_info;
self.handle_sync_events(EventKind::EphemeralRoomData, &room, &ephemeral.events).await?;
self.handle_sync_events(EventKind::RoomAccountData, &room, &account_data.events)
.await?;
self.handle_sync_state_events(&room, &state.events).await?;
self.handle_sync_timeline_events(&room, &timeline.events).await?;
}
for (room_id, room_info) in &rooms.leave {
let room = self.get_room(room_id);
if room.is_none() {
// raise warning?
continue;
}
let LeftRoom { timeline, state, account_data } = room_info;
self.handle_sync_events(EventKind::RoomAccountData, &room, &account_data.events)
.await?;
self.handle_sync_state_events(&room, &state.events).await?;
self.handle_sync_timeline_events(&room, &timeline.events).await?;
}
for (room_id, room_info) in &rooms.invite {
let room = self.get_room(room_id);
if room.is_none() {
// raise warning?
continue;
}
// FIXME: Destructure room_info
self.handle_sync_events(EventKind::InitialState, &room, &room_info.invite_state.events)
.await?;
}
for handler in &*self.notification_handlers.read().await {
for (room_id, room_notifications) in notifications {
let room = match self.get_room(room_id) {
Some(room) => room,
None => {
// raise warning?
continue;
}
};
for notification in room_notifications {
(handler)(notification.clone(), room.clone(), self.clone()).await;
}
}
}
Ok(response)
} }
/// Repeatedly call sync to synchronize the client state with the server. /// Repeatedly call sync to synchronize the client state with the server.

397
matrix_sdk/src/event_handler.rs Normal file
View File

@ -0,0 +1,397 @@
// 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.
//! Types and traits related for event handlers. For usage, see
//! [`Client::register_event_handler`].
use std::{borrow::Cow, future::Future, ops::Deref};
use matrix_sdk_base::deserialized_responses::SyncRoomEvent;
use ruma::{events::AnySyncStateEvent, serde::Raw};
use serde::Deserialize;
use serde_json::value::RawValue as RawJsonValue;
use crate::{room, Client};
#[doc(hidden)]
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
pub enum EventKind {
GlobalAccountData,
RoomAccountData,
EphemeralRoomData,
Message { redacted: bool },
State { redacted: bool },
StrippedState { redacted: bool },
InitialState,
ToDevice,
Presence,
}
/// A statically-known event kind/type that can be retrieved from an event sync.
pub trait SyncEvent {
#[doc(hidden)]
const ID: (EventKind, &'static str);
}
/// Interface for event handlers.
///
/// This trait is an abstraction for a certain kind of functions / closures,
/// specifically:
///
/// * They must have at least one argument, which is the event itself, a type
/// that implements [`StaticEvent`]. Any additional arguments need to
/// implement the [`EventHandlerContext`] trait.
/// * Their return type has to be one of: `()`, `Result<(), impl
/// std::error::Error>` or `anyhow::Result<()>` (requires the `anyhow` Cargo
/// feature to be enabled)
pub trait EventHandler<Ev, Ctx>: Clone + Send + Sync + 'static {
/// The future returned by `handle_event`.
#[doc(hidden)]
type Future: Future + Send + 'static;
/// The event type being handled, for example a message event of type
/// `m.room.message`.
#[doc(hidden)]
const ID: (EventKind, &'static str);
/// Create a future for handling the given event.
///
/// `data` provides additional data about the event, for example the room it
/// appeared in.
///
/// Returns `None` if one of the context extractors failed.
#[doc(hidden)]
fn handle_event(&self, ev: Ev, data: EventHandlerData<'_>) -> Option<Self::Future>;
}
#[doc(hidden)]
#[derive(Debug)]
pub struct EventHandlerData<'a> {
pub client: Client,
pub room: Option<room::Room>,
pub raw: &'a RawJsonValue,
}
/// Context for an event handler.
///
/// This trait defines the set of types that may be used as additional arguments
/// in event handler functions after the event itself.
pub trait EventHandlerContext: Sized {
#[doc(hidden)]
fn from_data(_: &EventHandlerData<'_>) -> Option<Self>;
}
impl EventHandlerContext for Client {
fn from_data(data: &EventHandlerData<'_>) -> Option<Self> {
Some(data.client.clone())
}
}
impl EventHandlerContext for room::Room {
fn from_data(data: &EventHandlerData<'_>) -> Option<Self> {
data.room.clone()
}
}
/// The raw JSON form of an event.
///
/// Used as a context argument for event handlers (see
/// [`Client::register_event_handler`]).
// FIXME: This could be made to not own the raw JSON value with some changes to
// the traits above, but only with GATs.
#[derive(Clone, Debug)]
pub struct RawEvent(pub Box<RawJsonValue>);
impl Deref for RawEvent {
type Target = RawJsonValue;
fn deref(&self) -> &Self::Target {
&self.0
}
}
impl EventHandlerContext for RawEvent {
fn from_data(data: &EventHandlerData<'_>) -> Option<Self> {
Some(Self(data.raw.to_owned()))
}
}
/// Return types supported for event handlers implement this trait.
///
/// It is not meant to be implemented outside of matrix-sdk.
pub trait EventHandlerResult: Sized {
#[doc(hidden)]
fn print_error(&self, event_type: &str);
}
impl EventHandlerResult for () {
fn print_error(&self, _event_type: &str) {}
}
impl<E: std::error::Error> EventHandlerResult for Result<(), E> {
fn print_error(&self, event_type: &str) {
if let Err(e) = self {
tracing::error!("Event handler for `{}` failed: {}", event_type, e);
}
}
}
#[cfg(feature = "anyhow")]
impl EventHandlerResult for anyhow::Result<()> {
fn print_error(&self, event_type: &str) {
if let Err(e) = self {
tracing::error!("Event handler for `{}` failed: {:?}", event_type, e);
}
}
}
#[derive(Deserialize)]
struct UnsignedDetails {
redacted_because: Option<serde::de::IgnoredAny>,
}
impl Client {
pub(crate) async fn handle_sync_events<T>(
&self,
kind: EventKind,
room: &Option<room::Room>,
events: &[Raw<T>],
) -> serde_json::Result<()> {
self.handle_sync_events_wrapped(kind, room, events, |x| x).await
}
pub(crate) async fn handle_sync_state_events(
&self,
room: &Option<room::Room>,
state_events: &[Raw<AnySyncStateEvent>],
) -> serde_json::Result<()> {
#[derive(Deserialize)]
struct StateEventDetails<'a> {
#[serde(borrow, rename = "type")]
event_type: Cow<'a, str>,
unsigned: Option<UnsignedDetails>,
}
self.handle_sync_events_wrapped_with(room, state_events, std::convert::identity, |raw| {
let StateEventDetails { event_type, unsigned } = raw.deserialize_as()?;
let redacted = unsigned.and_then(|u| u.redacted_because).is_some();
Ok((EventKind::State { redacted }, event_type))
})
.await
}
pub(crate) async fn handle_sync_timeline_events(
&self,
room: &Option<room::Room>,
timeline_events: &[SyncRoomEvent],
) -> serde_json::Result<()> {
// FIXME: add EncryptionInfo to context
#[derive(Deserialize)]
struct TimelineEventDetails<'a> {
#[serde(borrow, rename = "type")]
event_type: Cow<'a, str>,
state_key: Option<serde::de::IgnoredAny>,
unsigned: Option<UnsignedDetails>,
}
self.handle_sync_events_wrapped_with(
room,
timeline_events,
|e| &e.event,
|raw| {
let TimelineEventDetails { event_type, state_key, unsigned } =
raw.deserialize_as()?;
let redacted = unsigned.and_then(|u| u.redacted_because).is_some();
let kind = match state_key {
Some(_) => EventKind::State { redacted },
None => EventKind::Message { redacted },
};
Ok((kind, event_type))
},
)
.await
}
async fn handle_sync_events_wrapped<'a, T: 'a, U: 'a>(
&self,
kind: EventKind,
room: &Option<room::Room>,
events: &'a [U],
get_event: impl Fn(&'a U) -> &'a Raw<T>,
) -> Result<(), serde_json::Error> {
#[derive(Deserialize)]
struct ExtractType<'a> {
#[serde(borrow, rename = "type")]
event_type: Cow<'a, str>,
}
self.handle_sync_events_wrapped_with(room, events, get_event, |raw| {
Ok((kind, raw.deserialize_as::<ExtractType>()?.event_type))
})
.await
}
async fn handle_sync_events_wrapped_with<'a, T: 'a, U: 'a>(
&self,
room: &Option<room::Room>,
list: &'a [U],
get_event: impl Fn(&'a U) -> &'a Raw<T>,
get_id: impl Fn(&Raw<T>) -> serde_json::Result<(EventKind, Cow<'_, str>)>,
) -> serde_json::Result<()> {
for x in list {
let event = get_event(x);
let (ev_kind, ev_type) = get_id(event)?;
let event_handler_id = (ev_kind, &*ev_type);
if let Some(handlers) = self.event_handlers.read().await.get(&event_handler_id) {
for handler in &*handlers {
let data = EventHandlerData {
client: self.clone(),
room: room.clone(),
raw: event.json(),
};
(handler)(data).await;
}
}
}
Ok(())
}
}
macro_rules! impl_event_handler {
($($ty:ident),* $(,)?) => {
impl<Ev, Fun, Fut, $($ty),*> EventHandler<Ev, ($($ty,)*)> for Fun
where
Ev: SyncEvent,
Fun: Fn(Ev, $($ty),*) -> Fut + Clone + Send + Sync + 'static,
Fut: Future + Send + 'static,
Fut::Output: EventHandlerResult,
$($ty: EventHandlerContext),*
{
type Future = Fut;
const ID: (EventKind, &'static str) = Ev::ID;
fn handle_event(&self, ev: Ev, _d: EventHandlerData<'_>) -> Option<Self::Future> {
Some((self)(ev, $($ty::from_data(&_d)?),*))
}
}
};
}
impl_event_handler!();
impl_event_handler!(A);
impl_event_handler!(A, B);
impl_event_handler!(A, B, C);
impl_event_handler!(A, B, C, D);
impl_event_handler!(A, B, C, D, E);
impl_event_handler!(A, B, C, D, E, F);
impl_event_handler!(A, B, C, D, E, F, G);
impl_event_handler!(A, B, C, D, E, F, G, H);
mod static_events {
use ruma::events::{
self,
presence::{PresenceEvent, PresenceEventContent},
StaticEventContent,
};
use super::{EventKind, SyncEvent};
impl<C> SyncEvent for events::GlobalAccountDataEvent<C>
where
C: StaticEventContent + events::GlobalAccountDataEventContent,
{
const ID: (EventKind, &'static str) = (EventKind::GlobalAccountData, C::TYPE);
}
impl<C> SyncEvent for events::RoomAccountDataEvent<C>
where
C: StaticEventContent + events::RoomAccountDataEventContent,
{
const ID: (EventKind, &'static str) = (EventKind::RoomAccountData, C::TYPE);
}
impl<C> SyncEvent for events::SyncEphemeralRoomEvent<C>
where
C: StaticEventContent + events::EphemeralRoomEventContent,
{
const ID: (EventKind, &'static str) = (EventKind::EphemeralRoomData, C::TYPE);
}
impl<C> SyncEvent for events::SyncMessageEvent<C>
where
C: StaticEventContent + events::MessageEventContent,
{
const ID: (EventKind, &'static str) = (EventKind::Message { redacted: false }, C::TYPE);
}
impl<C> SyncEvent for events::SyncStateEvent<C>
where
C: StaticEventContent + events::StateEventContent,
{
const ID: (EventKind, &'static str) = (EventKind::State { redacted: false }, C::TYPE);
}
impl<C> SyncEvent for events::StrippedStateEvent<C>
where
C: StaticEventContent + events::StateEventContent,
{
const ID: (EventKind, &'static str) =
(EventKind::StrippedState { redacted: false }, C::TYPE);
}
impl<C> SyncEvent for events::InitialStateEvent<C>
where
C: StaticEventContent + events::StateEventContent,
{
const ID: (EventKind, &'static str) = (EventKind::InitialState, C::TYPE);
}
impl<C> SyncEvent for events::ToDeviceEvent<C>
where
C: StaticEventContent + events::ToDeviceEventContent,
{
const ID: (EventKind, &'static str) = (EventKind::ToDevice, C::TYPE);
}
impl SyncEvent for PresenceEvent {
const ID: (EventKind, &'static str) = (EventKind::Presence, PresenceEventContent::TYPE);
}
impl<C> SyncEvent for events::RedactedSyncMessageEvent<C>
where
C: StaticEventContent + events::RedactedMessageEventContent,
{
const ID: (EventKind, &'static str) = (EventKind::Message { redacted: true }, C::TYPE);
}
impl<C> SyncEvent for events::RedactedSyncStateEvent<C>
where
C: StaticEventContent + events::RedactedStateEventContent,
{
const ID: (EventKind, &'static str) = (EventKind::State { redacted: true }, C::TYPE);
}
impl<C> SyncEvent for events::RedactedStrippedStateEvent<C>
where
C: StaticEventContent + events::RedactedStateEventContent,
{
const ID: (EventKind, &'static str) =
(EventKind::StrippedState { redacted: true }, C::TYPE);
}
}

964
matrix_sdk/src/event_handler/mod.rs
View File

@ -1,964 +0,0 @@
// 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::ops::Deref;
use matrix_sdk_base::{hoist_and_deserialize_state_event, hoist_room_event_prev_content};
use matrix_sdk_common::async_trait;
use ruma::{
api::client::r0::push::get_notifications::Notification,
events::{
call::{
answer::AnswerEventContent, candidates::CandidatesEventContent,
hangup::HangupEventContent, invite::InviteEventContent,
},
custom::CustomEventContent,
fully_read::FullyReadEventContent,
ignored_user_list::IgnoredUserListEventContent,
presence::PresenceEvent,
push_rules::PushRulesEventContent,
reaction::ReactionEventContent,
receipt::ReceiptEventContent,
room::{
aliases::AliasesEventContent,
avatar::AvatarEventContent,
canonical_alias::CanonicalAliasEventContent,
join_rules::JoinRulesEventContent,
member::MemberEventContent,
message::{feedback::FeedbackEventContent, MessageEventContent as MsgEventContent},
name::NameEventContent,
power_levels::PowerLevelsEventContent,
redaction::SyncRedactionEvent,
tombstone::TombstoneEventContent,
},
typing::TypingEventContent,
AnyGlobalAccountDataEvent, AnyRoomAccountDataEvent, AnyStrippedStateEvent,
AnySyncEphemeralRoomEvent, AnySyncMessageEvent, AnySyncRoomEvent, AnySyncStateEvent,
GlobalAccountDataEvent, RoomAccountDataEvent, StrippedStateEvent, SyncEphemeralRoomEvent,
SyncMessageEvent, SyncStateEvent,
},
serde::Raw,
RoomId,
};
use serde_json::value::RawValue as RawJsonValue;
use crate::{deserialized_responses::SyncResponse, room::Room, Client};
pub(crate) struct Handler {
pub(crate) inner: Box<dyn EventHandler>,
pub(crate) client: Client,
}
impl Deref for Handler {
type Target = dyn EventHandler;
fn deref(&self) -> &Self::Target {
&*self.inner
}
}
impl Handler {
fn get_room(&self, room_id: &RoomId) -> Option<Room> {
self.client.get_room(room_id)
}
pub(crate) async fn handle_sync(&self, response: &SyncResponse) {
for event in response.account_data.events.iter().filter_map(|e| e.deserialize().ok()) {
self.handle_account_data_event(&event).await;
}
for (room_id, room_info) in &response.rooms.join {
if let Some(room) = self.get_room(room_id) {
for event in room_info.ephemeral.events.iter().filter_map(|e| e.deserialize().ok())
{
self.handle_ephemeral_event(room.clone(), &event).await;
}
for event in
room_info.account_data.events.iter().filter_map(|e| e.deserialize().ok())
{
self.handle_room_account_data_event(room.clone(), &event).await;
}
for (raw_event, event) in room_info.state.events.iter().filter_map(|e| {
if let Ok(d) = hoist_and_deserialize_state_event(e) {
Some((e, d))
} else {
None
}
}) {
self.handle_state_event(room.clone(), &event, raw_event).await;
}
for (raw_event, event) in room_info.timeline.events.iter().filter_map(|e| {
if let Ok(d) = hoist_room_event_prev_content(&e.event) {
Some((&e.event, d))
} else {
None
}
}) {
self.handle_timeline_event(room.clone(), &event, raw_event).await;
}
}
}
for (room_id, room_info) in &response.rooms.leave {
if let Some(room) = self.get_room(room_id) {
for event in
room_info.account_data.events.iter().filter_map(|e| e.deserialize().ok())
{
self.handle_room_account_data_event(room.clone(), &event).await;
}
for (raw_event, event) in room_info.state.events.iter().filter_map(|e| {
if let Ok(d) = hoist_and_deserialize_state_event(e) {
Some((e, d))
} else {
None
}
}) {
self.handle_state_event(room.clone(), &event, raw_event).await;
}
for (raw_event, event) in room_info.timeline.events.iter().filter_map(|e| {
if let Ok(d) = hoist_room_event_prev_content(&e.event) {
Some((&e.event, d))
} else {
None
}
}) {
self.handle_timeline_event(room.clone(), &event, raw_event).await;
}
}
}
for (room_id, room_info) in &response.rooms.invite {
if let Some(room) = self.get_room(room_id) {
for event in
room_info.invite_state.events.iter().filter_map(|e| e.deserialize().ok())
{
self.handle_stripped_state_event(room.clone(), &event).await;
}
}
}
for event in response.presence.events.iter().filter_map(|e| e.deserialize().ok()) {
self.on_presence_event(&event).await;
}
for (room_id, notifications) in &response.notifications {
if let Some(room) = self.get_room(room_id) {
for notification in notifications {
self.on_room_notification(room.clone(), notification.clone()).await;
}
}
}
}
async fn handle_timeline_event(
&self,
room: Room,
event: &AnySyncRoomEvent,
raw_event: &Raw<AnySyncRoomEvent>,
) {
match event {
AnySyncRoomEvent::State(event) => match event {
AnySyncStateEvent::RoomMember(e) => self.on_room_member(room, e).await,
AnySyncStateEvent::RoomName(e) => self.on_room_name(room, e).await,
AnySyncStateEvent::RoomCanonicalAlias(e) => {
self.on_room_canonical_alias(room, e).await
}
AnySyncStateEvent::RoomAliases(e) => self.on_room_aliases(room, e).await,
AnySyncStateEvent::RoomAvatar(e) => self.on_room_avatar(room, e).await,
AnySyncStateEvent::RoomPowerLevels(e) => self.on_room_power_levels(room, e).await,
AnySyncStateEvent::RoomTombstone(e) => self.on_room_tombstone(room, e).await,
AnySyncStateEvent::RoomJoinRules(e) => self.on_room_join_rules(room, e).await,
AnySyncStateEvent::PolicyRuleRoom(_)
| AnySyncStateEvent::PolicyRuleServer(_)
| AnySyncStateEvent::PolicyRuleUser(_)
| AnySyncStateEvent::RoomCreate(_)
| AnySyncStateEvent::RoomEncryption(_)
| AnySyncStateEvent::RoomGuestAccess(_)
| AnySyncStateEvent::RoomHistoryVisibility(_)
| AnySyncStateEvent::RoomPinnedEvents(_)
| AnySyncStateEvent::RoomServerAcl(_)
| AnySyncStateEvent::RoomThirdPartyInvite(_)
| AnySyncStateEvent::RoomTopic(_)
| AnySyncStateEvent::SpaceChild(_)
| AnySyncStateEvent::SpaceParent(_) => {}
_ => {
if let Ok(e) = raw_event.deserialize_as::<SyncStateEvent<CustomEventContent>>()
{
self.on_custom_event(room, &CustomEvent::State(&e)).await;
}
}
},
AnySyncRoomEvent::Message(event) => match event {
AnySyncMessageEvent::RoomMessage(e) => self.on_room_message(room, e).await,
AnySyncMessageEvent::RoomMessageFeedback(e) => {
self.on_room_message_feedback(room, e).await
}
AnySyncMessageEvent::RoomRedaction(e) => self.on_room_redaction(room, e).await,
AnySyncMessageEvent::Reaction(e) => self.on_room_reaction(room, e).await,
AnySyncMessageEvent::CallInvite(e) => self.on_room_call_invite(room, e).await,
AnySyncMessageEvent::CallAnswer(e) => self.on_room_call_answer(room, e).await,
AnySyncMessageEvent::CallCandidates(e) => {
self.on_room_call_candidates(room, e).await
}
AnySyncMessageEvent::CallHangup(e) => self.on_room_call_hangup(room, e).await,
AnySyncMessageEvent::KeyVerificationReady(_)
| AnySyncMessageEvent::KeyVerificationStart(_)
| AnySyncMessageEvent::KeyVerificationCancel(_)
| AnySyncMessageEvent::KeyVerificationAccept(_)
| AnySyncMessageEvent::KeyVerificationKey(_)
| AnySyncMessageEvent::KeyVerificationMac(_)
| AnySyncMessageEvent::KeyVerificationDone(_)
| AnySyncMessageEvent::RoomEncrypted(_)
| AnySyncMessageEvent::Sticker(_) => {}
_ => {
if let Ok(e) =
raw_event.deserialize_as::<SyncMessageEvent<CustomEventContent>>()
{
self.on_custom_event(room, &CustomEvent::Message(&e)).await;
}
}
},
AnySyncRoomEvent::RedactedState(_event) => {}
AnySyncRoomEvent::RedactedMessage(_event) => {}
}
}
async fn handle_state_event(
&self,
room: Room,
event: &AnySyncStateEvent,
raw_event: &Raw<AnySyncStateEvent>,
) {
match event {
AnySyncStateEvent::RoomMember(member) => self.on_state_member(room, member).await,
AnySyncStateEvent::RoomName(name) => self.on_state_name(room, name).await,
AnySyncStateEvent::RoomCanonicalAlias(canonical) => {
self.on_state_canonical_alias(room, canonical).await
}
AnySyncStateEvent::RoomAliases(aliases) => self.on_state_aliases(room, aliases).await,
AnySyncStateEvent::RoomAvatar(avatar) => self.on_state_avatar(room, avatar).await,
AnySyncStateEvent::RoomPowerLevels(power) => {
self.on_state_power_levels(room, power).await
}
AnySyncStateEvent::RoomJoinRules(rules) => self.on_state_join_rules(room, rules).await,
AnySyncStateEvent::RoomTombstone(tomb) => {
// TODO make `on_state_tombstone` method
self.on_room_tombstone(room, tomb).await
}
AnySyncStateEvent::PolicyRuleRoom(_)
| AnySyncStateEvent::PolicyRuleServer(_)
| AnySyncStateEvent::PolicyRuleUser(_)
| AnySyncStateEvent::RoomCreate(_)
| AnySyncStateEvent::RoomEncryption(_)
| AnySyncStateEvent::RoomGuestAccess(_)
| AnySyncStateEvent::RoomHistoryVisibility(_)
| AnySyncStateEvent::RoomPinnedEvents(_)
| AnySyncStateEvent::RoomServerAcl(_)
| AnySyncStateEvent::RoomThirdPartyInvite(_)
| AnySyncStateEvent::RoomTopic(_)
| AnySyncStateEvent::SpaceChild(_)
| AnySyncStateEvent::SpaceParent(_) => {}
_ => {
if let Ok(e) = raw_event.deserialize_as::<SyncStateEvent<CustomEventContent>>() {
self.on_custom_event(room, &CustomEvent::State(&e)).await;
}
}
}
}
pub(crate) async fn handle_stripped_state_event(
&self,
// TODO these events are only handled in invited rooms.
room: Room,
event: &AnyStrippedStateEvent,
) {
match event {
AnyStrippedStateEvent::RoomMember(member) => {
self.on_stripped_state_member(room, member, None).await
}
AnyStrippedStateEvent::RoomName(name) => self.on_stripped_state_name(room, name).await,
AnyStrippedStateEvent::RoomCanonicalAlias(canonical) => {
self.on_stripped_state_canonical_alias(room, canonical).await
}
AnyStrippedStateEvent::RoomAliases(aliases) => {
self.on_stripped_state_aliases(room, aliases).await
}
AnyStrippedStateEvent::RoomAvatar(avatar) => {
self.on_stripped_state_avatar(room, avatar).await
}
AnyStrippedStateEvent::RoomPowerLevels(power) => {
self.on_stripped_state_power_levels(room, power).await
}
AnyStrippedStateEvent::RoomJoinRules(rules) => {
self.on_stripped_state_join_rules(room, rules).await
}
_ => {}
}
}
pub(crate) async fn handle_room_account_data_event(
&self,
room: Room,
event: &AnyRoomAccountDataEvent,
) {
if let AnyRoomAccountDataEvent::FullyRead(event) = event {
self.on_non_room_fully_read(room, event).await
}
}
pub(crate) async fn handle_account_data_event(&self, event: &AnyGlobalAccountDataEvent) {
match event {
AnyGlobalAccountDataEvent::IgnoredUserList(ignored) => {
self.on_non_room_ignored_users(ignored).await
}
AnyGlobalAccountDataEvent::PushRules(rules) => self.on_non_room_push_rules(rules).await,
_ => {}
}
}
pub(crate) async fn handle_ephemeral_event(
&self,
room: Room,
event: &AnySyncEphemeralRoomEvent,
) {
match event {
AnySyncEphemeralRoomEvent::Typing(typing) => {
self.on_non_room_typing(room, typing).await
}
AnySyncEphemeralRoomEvent::Receipt(receipt) => {
self.on_non_room_receipt(room, receipt).await
}
_ => {}
}
}
}
/// This represents the various "unrecognized" events.
#[derive(Clone, Copy, Debug)]
pub enum CustomEvent<'c> {
/// A custom basic event.
Basic(&'c GlobalAccountDataEvent<CustomEventContent>),
/// A custom basic event.
EphemeralRoom(&'c SyncEphemeralRoomEvent<CustomEventContent>),
/// A custom room event.
Message(&'c SyncMessageEvent<CustomEventContent>),
/// A custom state event.
State(&'c SyncStateEvent<CustomEventContent>),
/// A custom stripped state event.
StrippedState(&'c StrippedStateEvent<CustomEventContent>),
}
/// This trait allows any type implementing `EventHandler` to specify event
/// callbacks for each event. The `Client` calls each method when the
/// corresponding event is received.
///
/// # Examples
/// ```
/// # use std::ops::Deref;
/// # use std::sync::Arc;
/// # use std::{env, process::exit};
/// # use matrix_sdk::{
/// # async_trait,
/// # EventHandler,
/// # ruma::events::{
/// # room::message::{MessageEventContent, MessageType, TextMessageEventContent},
/// # SyncMessageEvent
/// # },
/// # locks::RwLock,
/// # room::Room,
/// # };
///
/// struct EventCallback;
///
/// #[async_trait]
/// impl EventHandler for EventCallback {
/// async fn on_room_message(&self, room: Room, event: &SyncMessageEvent<MessageEventContent>) {
/// if let Room::Joined(room) = room {
/// if let SyncMessageEvent {
/// content:
/// MessageEventContent {
/// msgtype: MessageType::Text(TextMessageEventContent { body: msg_body, .. }),
/// ..
/// },
/// sender,
/// ..
/// } = event
/// {
/// let member = room.get_member(&sender).await.unwrap().unwrap();
/// let name = member
/// .display_name()
/// .unwrap_or_else(|| member.user_id().as_str());
/// println!("{}: {}", name, msg_body);
/// }
/// }
/// }
/// }
/// ```
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
pub trait EventHandler: Send + Sync {
// ROOM EVENTS from `IncomingTimeline`
/// Fires when `Client` receives a `RoomEvent::RoomMember` event.
async fn on_room_member(&self, _: Room, _: &SyncStateEvent<MemberEventContent>) {}
/// Fires when `Client` receives a `RoomEvent::RoomName` event.
async fn on_room_name(&self, _: Room, _: &SyncStateEvent<NameEventContent>) {}
/// Fires when `Client` receives a `RoomEvent::RoomCanonicalAlias` event.
async fn on_room_canonical_alias(
&self,
_: Room,
_: &SyncStateEvent<CanonicalAliasEventContent>,
) {
}
/// Fires when `Client` receives a `RoomEvent::RoomAliases` event.
async fn on_room_aliases(&self, _: Room, _: &SyncStateEvent<AliasesEventContent>) {}
/// Fires when `Client` receives a `RoomEvent::RoomAvatar` event.
async fn on_room_avatar(&self, _: Room, _: &SyncStateEvent<AvatarEventContent>) {}
/// Fires when `Client` receives a `RoomEvent::RoomMessage` event.
async fn on_room_message(&self, _: Room, _: &SyncMessageEvent<MsgEventContent>) {}
/// Fires when `Client` receives a `RoomEvent::RoomMessageFeedback` event.
async fn on_room_message_feedback(&self, _: Room, _: &SyncMessageEvent<FeedbackEventContent>) {}
/// Fires when `Client` receives a `RoomEvent::Reaction` event.
async fn on_room_reaction(&self, _: Room, _: &SyncMessageEvent<ReactionEventContent>) {}
/// Fires when `Client` receives a `RoomEvent::CallInvite` event
async fn on_room_call_invite(&self, _: Room, _: &SyncMessageEvent<InviteEventContent>) {}
/// Fires when `Client` receives a `RoomEvent::CallAnswer` event
async fn on_room_call_answer(&self, _: Room, _: &SyncMessageEvent<AnswerEventContent>) {}
/// Fires when `Client` receives a `RoomEvent::CallCandidates` event
async fn on_room_call_candidates(&self, _: Room, _: &SyncMessageEvent<CandidatesEventContent>) {
}
/// Fires when `Client` receives a `RoomEvent::CallHangup` event
async fn on_room_call_hangup(&self, _: Room, _: &SyncMessageEvent<HangupEventContent>) {}
/// Fires when `Client` receives a `RoomEvent::RoomRedaction` event.
async fn on_room_redaction(&self, _: Room, _: &SyncRedactionEvent) {}
/// Fires when `Client` receives a `RoomEvent::RoomPowerLevels` event.
async fn on_room_power_levels(&self, _: Room, _: &SyncStateEvent<PowerLevelsEventContent>) {}
/// Fires when `Client` receives a `RoomEvent::RoomJoinRules` event.
async fn on_room_join_rules(&self, _: Room, _: &SyncStateEvent<JoinRulesEventContent>) {}
/// Fires when `Client` receives a `RoomEvent::Tombstone` event.
async fn on_room_tombstone(&self, _: Room, _: &SyncStateEvent<TombstoneEventContent>) {}
/// Fires when `Client` receives room events that trigger notifications
/// according to the push rules of the user.
async fn on_room_notification(&self, _: Room, _: Notification) {}
// `RoomEvent`s from `IncomingState`
/// Fires when `Client` receives a `StateEvent::RoomMember` event.
async fn on_state_member(&self, _: Room, _: &SyncStateEvent<MemberEventContent>) {}
/// Fires when `Client` receives a `StateEvent::RoomName` event.
async fn on_state_name(&self, _: Room, _: &SyncStateEvent<NameEventContent>) {}
/// Fires when `Client` receives a `StateEvent::RoomCanonicalAlias` event.
async fn on_state_canonical_alias(
&self,
_: Room,
_: &SyncStateEvent<CanonicalAliasEventContent>,
) {
}
/// Fires when `Client` receives a `StateEvent::RoomAliases` event.
async fn on_state_aliases(&self, _: Room, _: &SyncStateEvent<AliasesEventContent>) {}
/// Fires when `Client` receives a `StateEvent::RoomAvatar` event.
async fn on_state_avatar(&self, _: Room, _: &SyncStateEvent<AvatarEventContent>) {}
/// Fires when `Client` receives a `StateEvent::RoomPowerLevels` event.
async fn on_state_power_levels(&self, _: Room, _: &SyncStateEvent<PowerLevelsEventContent>) {}
/// Fires when `Client` receives a `StateEvent::RoomJoinRules` event.
async fn on_state_join_rules(&self, _: Room, _: &SyncStateEvent<JoinRulesEventContent>) {}
// `AnyStrippedStateEvent`s
/// Fires when `Client` receives a
/// `AnyStrippedStateEvent::StrippedRoomMember` event.
async fn on_stripped_state_member(
&self,
_: Room,
_: &StrippedStateEvent<MemberEventContent>,
_: Option<MemberEventContent>,
) {
}
/// Fires when `Client` receives a `AnyStrippedStateEvent::StrippedRoomName`
/// event.
async fn on_stripped_state_name(&self, _: Room, _: &StrippedStateEvent<NameEventContent>) {}
/// Fires when `Client` receives a
/// `AnyStrippedStateEvent::StrippedRoomCanonicalAlias` event.
async fn on_stripped_state_canonical_alias(
&self,
_: Room,
_: &StrippedStateEvent<CanonicalAliasEventContent>,
) {
}
/// Fires when `Client` receives a
/// `AnyStrippedStateEvent::StrippedRoomAliases` event.
async fn on_stripped_state_aliases(
&self,
_: Room,
_: &StrippedStateEvent<AliasesEventContent>,
) {
}
/// Fires when `Client` receives a
/// `AnyStrippedStateEvent::StrippedRoomAvatar` event.
async fn on_stripped_state_avatar(&self, _: Room, _: &StrippedStateEvent<AvatarEventContent>) {}
/// Fires when `Client` receives a
/// `AnyStrippedStateEvent::StrippedRoomPowerLevels` event.
async fn on_stripped_state_power_levels(
&self,
_: Room,
_: &StrippedStateEvent<PowerLevelsEventContent>,
) {
}
/// Fires when `Client` receives a
/// `AnyStrippedStateEvent::StrippedRoomJoinRules` event.
async fn on_stripped_state_join_rules(
&self,
_: Room,
_: &StrippedStateEvent<JoinRulesEventContent>,
) {
}
// `NonRoomEvent` (this is a type alias from ruma_events)
/// Fires when `Client` receives a `NonRoomEvent::RoomPresence` event.
async fn on_non_room_presence(&self, _: Room, _: &PresenceEvent) {}
/// Fires when `Client` receives a `NonRoomEvent::RoomName` event.
async fn on_non_room_ignored_users(
&self,
_: &GlobalAccountDataEvent<IgnoredUserListEventContent>,
) {
}
/// Fires when `Client` receives a `NonRoomEvent::RoomCanonicalAlias` event.
async fn on_non_room_push_rules(&self, _: &GlobalAccountDataEvent<PushRulesEventContent>) {}
/// Fires when `Client` receives a `NonRoomEvent::RoomAliases` event.
async fn on_non_room_fully_read(
&self,
_: Room,
_: &RoomAccountDataEvent<FullyReadEventContent>,
) {
}
/// Fires when `Client` receives a `NonRoomEvent::Typing` event.
async fn on_non_room_typing(&self, _: Room, _: &SyncEphemeralRoomEvent<TypingEventContent>) {}
/// Fires when `Client` receives a `NonRoomEvent::Receipt` event.
///
/// This is always a read receipt.
async fn on_non_room_receipt(&self, _: Room, _: &SyncEphemeralRoomEvent<ReceiptEventContent>) {}
// `PresenceEvent` is a struct so there is only the one method
/// Fires when `Client` receives a `NonRoomEvent::RoomAliases` event.
async fn on_presence_event(&self, _: &PresenceEvent) {}
/// Fires when `Client` receives a `Event::Custom` event or if
/// deserialization fails because the event was unknown to ruma.
///
/// The only guarantee this method can give about the event is that it is
/// valid JSON.
async fn on_unrecognized_event(&self, _: Room, _: &RawJsonValue) {}
/// Fires when `Client` receives a `Event::Custom` event or if
/// deserialization fails because the event was unknown to ruma.
///
/// The only guarantee this method can give about the event is that it is in
/// the shape of a valid matrix event.
async fn on_custom_event(&self, _: Room, _: &CustomEvent<'_>) {}
}
#[cfg(test)]
mod test {
use std::{sync::Arc, time::Duration};
use matrix_sdk_common::{async_trait, locks::Mutex};
use matrix_sdk_test::{async_test, test_json};
use mockito::{mock, Matcher};
use ruma::user_id;
#[cfg(target_arch = "wasm32")]
pub use wasm_bindgen_test::*;
use super::*;
#[derive(Clone)]
pub struct EvHandlerTest(Arc<Mutex<Vec<String>>>);
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
impl EventHandler for EvHandlerTest {
async fn on_room_member(&self, _: Room, _: &SyncStateEvent<MemberEventContent>) {
self.0.lock().await.push("member".to_string())
}
async fn on_room_name(&self, _: Room, _: &SyncStateEvent<NameEventContent>) {
self.0.lock().await.push("name".to_string())
}
async fn on_room_canonical_alias(
&self,
_: Room,
_: &SyncStateEvent<CanonicalAliasEventContent>,
) {
self.0.lock().await.push("canonical".to_string())
}
async fn on_room_aliases(&self, _: Room, _: &SyncStateEvent<AliasesEventContent>) {
self.0.lock().await.push("aliases".to_string())
}
async fn on_room_avatar(&self, _: Room, _: &SyncStateEvent<AvatarEventContent>) {
self.0.lock().await.push("avatar".to_string())
}
async fn on_room_message(&self, _: Room, _: &SyncMessageEvent<MsgEventContent>) {
self.0.lock().await.push("message".to_string())
}
async fn on_room_message_feedback(
&self,
_: Room,
_: &SyncMessageEvent<FeedbackEventContent>,
) {
self.0.lock().await.push("feedback".to_string())
}
async fn on_room_call_invite(&self, _: Room, _: &SyncMessageEvent<InviteEventContent>) {
self.0.lock().await.push("call invite".to_string())
}
async fn on_room_call_answer(&self, _: Room, _: &SyncMessageEvent<AnswerEventContent>) {
self.0.lock().await.push("call answer".to_string())
}
async fn on_room_call_candidates(
&self,
_: Room,
_: &SyncMessageEvent<CandidatesEventContent>,
) {
self.0.lock().await.push("call candidates".to_string())
}
async fn on_room_call_hangup(&self, _: Room, _: &SyncMessageEvent<HangupEventContent>) {
self.0.lock().await.push("call hangup".to_string())
}
async fn on_room_redaction(&self, _: Room, _: &SyncRedactionEvent) {
self.0.lock().await.push("redaction".to_string())
}
async fn on_room_power_levels(&self, _: Room, _: &SyncStateEvent<PowerLevelsEventContent>) {
self.0.lock().await.push("power".to_string())
}
async fn on_room_tombstone(&self, _: Room, _: &SyncStateEvent<TombstoneEventContent>) {
self.0.lock().await.push("tombstone".to_string())
}
async fn on_state_member(&self, _: Room, _: &SyncStateEvent<MemberEventContent>) {
self.0.lock().await.push("state member".to_string())
}
async fn on_state_name(&self, _: Room, _: &SyncStateEvent<NameEventContent>) {
self.0.lock().await.push("state name".to_string())
}
async fn on_state_canonical_alias(
&self,
_: Room,
_: &SyncStateEvent<CanonicalAliasEventContent>,
) {
self.0.lock().await.push("state canonical".to_string())
}
async fn on_state_aliases(&self, _: Room, _: &SyncStateEvent<AliasesEventContent>) {
self.0.lock().await.push("state aliases".to_string())
}
async fn on_state_avatar(&self, _: Room, _: &SyncStateEvent<AvatarEventContent>) {
self.0.lock().await.push("state avatar".to_string())
}
async fn on_state_power_levels(
&self,
_: Room,
_: &SyncStateEvent<PowerLevelsEventContent>,
) {
self.0.lock().await.push("state power".to_string())
}
async fn on_state_join_rules(&self, _: Room, _: &SyncStateEvent<JoinRulesEventContent>) {
self.0.lock().await.push("state rules".to_string())
}
// `AnyStrippedStateEvent`s
/// Fires when `Client` receives a
/// `AnyStrippedStateEvent::StrippedRoomMember` event.
async fn on_stripped_state_member(
&self,
_: Room,
_: &StrippedStateEvent<MemberEventContent>,
_: Option<MemberEventContent>,
) {
self.0.lock().await.push("stripped state member".to_string())
}
/// Fires when `Client` receives a
/// `AnyStrippedStateEvent::StrippedRoomName` event.
async fn on_stripped_state_name(&self, _: Room, _: &StrippedStateEvent<NameEventContent>) {
self.0.lock().await.push("stripped state name".to_string())
}
/// Fires when `Client` receives a
/// `AnyStrippedStateEvent::StrippedRoomCanonicalAlias` event.
async fn on_stripped_state_canonical_alias(
&self,
_: Room,
_: &StrippedStateEvent<CanonicalAliasEventContent>,
) {
self.0.lock().await.push("stripped state canonical".to_string())
}
/// Fires when `Client` receives a
/// `AnyStrippedStateEvent::StrippedRoomAliases` event.
async fn on_stripped_state_aliases(
&self,
_: Room,
_: &StrippedStateEvent<AliasesEventContent>,
) {
self.0.lock().await.push("stripped state aliases".to_string())
}
/// Fires when `Client` receives a
/// `AnyStrippedStateEvent::StrippedRoomAvatar` event.
async fn on_stripped_state_avatar(
&self,
_: Room,
_: &StrippedStateEvent<AvatarEventContent>,
) {
self.0.lock().await.push("stripped state avatar".to_string())
}
/// Fires when `Client` receives a
/// `AnyStrippedStateEvent::StrippedRoomPowerLevels` event.
async fn on_stripped_state_power_levels(
&self,
_: Room,
_: &StrippedStateEvent<PowerLevelsEventContent>,
) {
self.0.lock().await.push("stripped state power".to_string())
}
/// Fires when `Client` receives a
/// `AnyStrippedStateEvent::StrippedRoomJoinRules` event.
async fn on_stripped_state_join_rules(
&self,
_: Room,
_: &StrippedStateEvent<JoinRulesEventContent>,
) {
self.0.lock().await.push("stripped state rules".to_string())
}
async fn on_non_room_presence(&self, _: Room, _: &PresenceEvent) {
self.0.lock().await.push("presence".to_string())
}
async fn on_non_room_ignored_users(
&self,
_: &GlobalAccountDataEvent<IgnoredUserListEventContent>,
) {
self.0.lock().await.push("account ignore".to_string())
}
async fn on_non_room_push_rules(&self, _: &GlobalAccountDataEvent<PushRulesEventContent>) {
self.0.lock().await.push("account push rules".to_string())
}
async fn on_non_room_fully_read(
&self,
_: Room,
_: &RoomAccountDataEvent<FullyReadEventContent>,
) {
self.0.lock().await.push("account read".to_string())
}
async fn on_non_room_typing(
&self,
_: Room,
_: &SyncEphemeralRoomEvent<TypingEventContent>,
) {
self.0.lock().await.push("typing event".to_string())
}
async fn on_non_room_receipt(
&self,
_: Room,
_: &SyncEphemeralRoomEvent<ReceiptEventContent>,
) {
self.0.lock().await.push("receipt event".to_string())
}
async fn on_presence_event(&self, _: &PresenceEvent) {
self.0.lock().await.push("presence event".to_string())
}
async fn on_unrecognized_event(&self, _: Room, _: &RawJsonValue) {
self.0.lock().await.push("unrecognized event".to_string())
}
async fn on_custom_event(&self, _: Room, _: &CustomEvent<'_>) {
self.0.lock().await.push("custom event".to_string())
}
async fn on_room_notification(&self, _: Room, _: Notification) {
self.0.lock().await.push("notification".to_string())
}
}
use crate::{Client, Session, SyncSettings};
async fn get_client() -> Client {
let session = Session {
access_token: "1234".to_owned(),
user_id: user_id!("@example:localhost"),
device_id: "DEVICEID".into(),
};
let homeserver = url::Url::parse(&mockito::server_url()).unwrap();
let client = Client::new(homeserver).unwrap();
client.restore_login(session).await.unwrap();
client
}
async fn mock_sync(client: &Client, response: String) {
let _m = mock("GET", Matcher::Regex(r"^/_matrix/client/r0/sync\?.*$".to_string()))
.with_status(200)
.match_header("authorization", "Bearer 1234")
.with_body(response)
.create();
let sync_settings = SyncSettings::new().timeout(Duration::from_millis(3000));
let _response = client.sync_once(sync_settings).await.unwrap();
}
#[async_test]
async fn event_handler_joined() {
let vec = Arc::new(Mutex::new(Vec::new()));
let test_vec = Arc::clone(&vec);
let handler = Box::new(EvHandlerTest(vec));
let client = get_client().await;
client.set_event_handler(handler).await;
mock_sync(&client, test_json::SYNC.to_string()).await;
let v = test_vec.lock().await;
assert_eq!(
v.as_slice(),
[
"account ignore",
"receipt event",
"account read",
"state rules",
"state member",
"state aliases",
"state power",
"state canonical",
"state member",
"state member",
"message",
"presence event",
"notification",
],
)
}
#[async_test]
async fn event_handler_invite() {
let vec = Arc::new(Mutex::new(Vec::new()));
let test_vec = Arc::clone(&vec);
let handler = Box::new(EvHandlerTest(vec));
let client = get_client().await;
client.set_event_handler(handler).await;
mock_sync(&client, test_json::INVITE_SYNC.to_string()).await;
let v = test_vec.lock().await;
assert_eq!(v.as_slice(), ["stripped state name", "stripped state member", "presence event"],)
}
#[async_test]
async fn event_handler_leave() {
let vec = Arc::new(Mutex::new(Vec::new()));
let test_vec = Arc::clone(&vec);
let handler = Box::new(EvHandlerTest(vec));
let client = get_client().await;
client.set_event_handler(handler).await;
mock_sync(&client, test_json::LEAVE_SYNC.to_string()).await;
let v = test_vec.lock().await;
assert_eq!(
v.as_slice(),
[
"account ignore",
"state rules",
"state member",
"state aliases",
"state power",
"state canonical",
"state member",
"state member",
"message",
"presence event",
"notification",
],
)
}
#[async_test]
async fn event_handler_more_events() {
let vec = Arc::new(Mutex::new(Vec::new()));
let test_vec = Arc::clone(&vec);
let handler = Box::new(EvHandlerTest(vec));
let client = get_client().await;
client.set_event_handler(handler).await;
mock_sync(&client, test_json::MORE_SYNC.to_string()).await;
let v = test_vec.lock().await;
assert_eq!(
v.as_slice(),
[
"receipt event",
"typing event",
"message",
"message", // this is a message edit event
"redaction",
"message", // this is a notice event
],
)
}
#[async_test]
async fn event_handler_voip() {
let vec = Arc::new(Mutex::new(Vec::new()));
let test_vec = Arc::clone(&vec);
let handler = Box::new(EvHandlerTest(vec));
let client = get_client().await;
client.set_event_handler(handler).await;
mock_sync(&client, test_json::VOIP_SYNC.to_string()).await;
let v = test_vec.lock().await;
assert_eq!(v.as_slice(), ["call invite", "call answer", "call candidates", "call hangup",],)
}
#[async_test]
async fn event_handler_two_syncs() {
let vec = Arc::new(Mutex::new(Vec::new()));
let test_vec = Arc::clone(&vec);
let handler = Box::new(EvHandlerTest(vec));
let client = get_client().await;
client.set_event_handler(handler).await;
mock_sync(&client, test_json::SYNC.to_string()).await;
mock_sync(&client, test_json::MORE_SYNC.to_string()).await;
let v = test_vec.lock().await;
assert_eq!(
v.as_slice(),
[
"account ignore",
"receipt event",
"account read",
"state rules",
"state member",
"state aliases",
"state power",
"state canonical",
"state member",
"state member",
"message",
"presence event",
"notification",
"receipt event",
"typing event",
"message",
"message", // this is a message edit event
"redaction",
"message", // this is a notice event
"notification",
"notification",
"notification",
],
)
}
}

4
matrix_sdk/src/lib.rs
View File

@ -56,6 +56,7 @@
#![deny( #![deny(
missing_debug_implementations, missing_debug_implementations,
missing_docs,
dead_code, dead_code,
missing_docs, missing_docs,
trivial_casts, trivial_casts,
@ -90,7 +91,7 @@ pub use ruma;
mod client; mod client;
mod error; mod error;
mod event_handler; pub mod event_handler;
mod http_client; mod http_client;
/// High-level room API /// High-level room API
pub mod room; pub mod room;
@ -107,7 +108,6 @@ pub use client::{Client, ClientConfig, LoopCtrl, RequestConfig, SyncSettings};
#[cfg_attr(feature = "docs", doc(cfg(encryption)))] #[cfg_attr(feature = "docs", doc(cfg(encryption)))]
pub use device::Device; pub use device::Device;
pub use error::{Error, HttpError, Result}; pub use error::{Error, HttpError, Result};
pub use event_handler::{CustomEvent, EventHandler};
pub use http_client::HttpSend; pub use http_client::HttpSend;
pub use room_member::RoomMember; pub use room_member::RoomMember;
#[cfg(not(target_arch = "wasm32"))] #[cfg(not(target_arch = "wasm32"))]

48
matrix_sdk_appservice/examples/appservice_autojoin.rs
View File

@ -2,7 +2,6 @@ use std::{convert::TryFrom, env};
use matrix_sdk_appservice::{ use matrix_sdk_appservice::{
matrix_sdk::{ matrix_sdk::{
async_trait,
room::Room, room::Room,
ruma::{ ruma::{
events::{ events::{
@ -11,51 +10,27 @@ use matrix_sdk_appservice::{
}, },
UserId, UserId,
}, },
EventHandler,
}, },
AppService, AppServiceRegistration, AppService, AppServiceRegistration, Result,
}; };
use tracing::{error, trace}; use tracing::trace;
struct AppServiceEventHandler { pub async fn handle_room_member(
appservice: AppService, appservice: AppService,
}
impl AppServiceEventHandler {
pub fn new(appservice: AppService) -> Self {
Self { appservice }
}
pub async fn handle_room_member(
&self,
room: Room, room: Room,
event: &SyncStateEvent<MemberEventContent>, event: SyncStateEvent<MemberEventContent>,
) -> Result<(), Box<dyn std::error::Error>> { ) -> Result<()> {
if !self.appservice.user_id_is_in_namespace(&event.state_key)? { if !appservice.user_id_is_in_namespace(&event.state_key)? {
trace!("not an appservice user: {}", event.state_key); trace!("not an appservice user: {}", event.state_key);
} else if let MembershipState::Invite = event.content.membership { } else if let MembershipState::Invite = event.content.membership {
let user_id = UserId::try_from(event.state_key.clone())?; let user_id = UserId::try_from(event.state_key.as_str())?;
let appservice = self.appservice.clone();
appservice.register_virtual_user(user_id.localpart()).await?; appservice.register_virtual_user(user_id.localpart()).await?;
let client = appservice.virtual_user_client(user_id.localpart()).await?; let client = appservice.virtual_user_client(user_id.localpart()).await?;
client.join_room_by_id(room.room_id()).await?; client.join_room_by_id(room.room_id()).await?;
} }
Ok(()) Ok(())
}
}
#[async_trait]
impl EventHandler for AppServiceEventHandler {
async fn on_room_member(&self, room: Room, event: &SyncStateEvent<MemberEventContent>) {
match self.handle_room_member(room, event).await {
Ok(_) => (),
Err(error) => error!("{:?}", error),
}
}
} }
#[tokio::main] #[tokio::main]
@ -68,7 +43,14 @@ pub async fn main() -> Result<(), Box<dyn std::error::Error>> {
let registration = AppServiceRegistration::try_from_yaml_file("./tests/registration.yaml")?; let registration = AppServiceRegistration::try_from_yaml_file("./tests/registration.yaml")?;
let mut appservice = AppService::new(homeserver_url, server_name, registration).await?; let mut appservice = AppService::new(homeserver_url, server_name, registration).await?;
appservice.set_event_handler(Box::new(AppServiceEventHandler::new(appservice.clone()))).await?; appservice
.register_event_handler({
let appservice = appservice.clone();
move |event: SyncStateEvent<MemberEventContent>, room: Room| {
handle_room_member(appservice.clone(), room, event)
}
})
.await?;
let (host, port) = appservice.registration().get_host_and_port()?; let (host, port) = appservice.registration().get_host_and_port()?;
appservice.run(host, port).await?; appservice.run(host, port).await?;

39
matrix_sdk_appservice/src/lib.rs
View File

@ -34,14 +34,10 @@
//! ```no_run //! ```no_run
//! # async { //! # async {
//! # //! #
//! # use matrix_sdk::{async_trait, EventHandler}; //! use matrix_sdk_appservice::{
//! # //! ruma::events::{SyncStateEvent, room::member::MemberEventContent},
//! # struct MyEventHandler; //! AppService, AppServiceRegistration
//! # //! };
//! # #[async_trait]
//! # impl EventHandler for MyEventHandler {}
//! #
//! use matrix_sdk_appservice::{AppService, AppServiceRegistration};
//! //!
//! let homeserver_url = "http://127.0.0.1:8008"; //! let homeserver_url = "http://127.0.0.1:8008";
//! let server_name = "localhost"; //! let server_name = "localhost";
@ -59,7 +55,9 @@
//! ")?; //! ")?;
//! //!
//! let mut appservice = AppService::new(homeserver_url, server_name, registration).await?; //! let mut appservice = AppService::new(homeserver_url, server_name, registration).await?;
//! appservice.set_event_handler(Box::new(MyEventHandler)).await?; //! appservice.register_event_handler(|_ev: SyncStateEvent<MemberEventContent>| async {
//! // do stuff
//! });
//! //!
//! let (host, port) = appservice.registration().get_host_and_port()?; //! let (host, port) = appservice.registration().get_host_and_port()?;
//! appservice.run(host, port).await?; //! appservice.run(host, port).await?;
@ -80,6 +78,7 @@ compile_error!("one webserver feature must be enabled. available ones: `warp`");
use std::{ use std::{
convert::{TryFrom, TryInto}, convert::{TryFrom, TryInto},
fs::File, fs::File,
future::Future,
ops::Deref, ops::Deref,
path::PathBuf, path::PathBuf,
sync::Arc, sync::Arc,
@ -92,7 +91,10 @@ pub use matrix_sdk;
#[doc(no_inline)] #[doc(no_inline)]
pub use matrix_sdk::ruma; pub use matrix_sdk::ruma;
use matrix_sdk::{ use matrix_sdk::{
bytes::Bytes, reqwest::Url, Client, ClientConfig, EventHandler, HttpError, Session, bytes::Bytes,
event_handler::{EventHandler, EventHandlerResult, SyncEvent},
reqwest::Url,
Client, ClientConfig, HttpError, Session,
}; };
use regex::Regex; use regex::Regex;
use ruma::{ use ruma::{
@ -106,12 +108,13 @@ use ruma::{
}, },
assign, identifiers, DeviceId, ServerNameBox, UserId, assign, identifiers, DeviceId, ServerNameBox, UserId,
}; };
use serde::de::DeserializeOwned;
use tracing::{info, warn}; use tracing::{info, warn};
mod error; mod error;
mod webserver; mod webserver;
pub type Result<T> = std::result::Result<T, Error>; pub type Result<T, E = Error> = std::result::Result<T, E>;
pub type Host = String; pub type Host = String;
pub type Port = u16; pub type Port = u16;
@ -354,8 +357,8 @@ impl AppService {
Ok(entry.value().clone()) Ok(entry.value().clone())
} }
/// Convenience wrapper around [`Client::set_event_handler()`] that attaches /// Convenience wrapper around [`Client::register_event_handler()`] that
/// the event handler to the [`MainUser`]'s [`Client`] /// attaches the event handler to the [`MainUser`]'s [`Client`]
/// ///
/// Note that the event handler in the [`AppService`] context only triggers /// Note that the event handler in the [`AppService`] context only triggers
/// [`join` room `timeline` events], so no state events or events from the /// [`join` room `timeline` events], so no state events or events from the
@ -370,10 +373,14 @@ impl AppService {
/// ///
/// [`join` room `timeline` events]: https://spec.matrix.org/unstable/client-server-api/#get_matrixclientr0sync /// [`join` room `timeline` events]: https://spec.matrix.org/unstable/client-server-api/#get_matrixclientr0sync
/// [MSC2409]: https://github.com/matrix-org/matrix-doc/pull/2409 /// [MSC2409]: https://github.com/matrix-org/matrix-doc/pull/2409
pub async fn set_event_handler(&mut self, handler: Box<dyn EventHandler>) -> Result<()> { pub async fn register_event_handler<Ev, Ctx, H>(&mut self, handler: H) -> Result<()>
where
Ev: SyncEvent + DeserializeOwned + Send + 'static,
H: EventHandler<Ev, Ctx>,
<H::Future as Future>::Output: EventHandlerResult,
{
let client = self.get_cached_client(None)?; let client = self.get_cached_client(None)?;
client.register_event_handler(handler).await;
client.set_event_handler(handler).await;
Ok(()) Ok(())
} }

38
matrix_sdk_appservice/tests/tests.rs
View File

@ -1,13 +1,14 @@
use std::sync::{Arc, Mutex}; use std::{
future,
sync::{Arc, Mutex},
};
use matrix_sdk::{ use matrix_sdk::{
async_trait,
room::Room,
ruma::{ ruma::{
api::appservice::Registration, api::appservice::Registration,
events::{room::member::MemberEventContent, SyncStateEvent}, events::{room::member::MemberEventContent, SyncStateEvent},
}, },
ClientConfig, EventHandler, RequestConfig, ClientConfig, RequestConfig,
}; };
use matrix_sdk_appservice::*; use matrix_sdk_appservice::*;
use matrix_sdk_test::{appservice::TransactionBuilder, async_test, EventsJson}; use matrix_sdk_test::{appservice::TransactionBuilder, async_test, EventsJson};
@ -203,28 +204,17 @@ async fn test_no_access_token() -> Result<()> {
async fn test_event_handler() -> Result<()> { async fn test_event_handler() -> Result<()> {
let mut appservice = appservice(None).await?; let mut appservice = appservice(None).await?;
#[derive(Clone)]
struct Example {
pub on_state_member: Arc<Mutex<bool>>,
}
impl Example {
pub fn new() -> Self {
#[allow(clippy::mutex_atomic)] #[allow(clippy::mutex_atomic)]
Self { on_state_member: Arc::new(Mutex::new(false)) } let on_state_member = Arc::new(Mutex::new(false));
} appservice
} .register_event_handler({
let on_state_member = on_state_member.clone();
#[async_trait] move |_ev: SyncStateEvent<MemberEventContent>| {
impl EventHandler for Example {
async fn on_room_member(&self, _: Room, _: &SyncStateEvent<MemberEventContent>) {
let on_state_member = self.on_state_member.clone();
*on_state_member.lock().unwrap() = true; *on_state_member.lock().unwrap() = true;
future::ready(())
} }
} })
.await?;
let example = Example::new();
appservice.set_event_handler(Box::new(example.clone())).await?;
let uri = "/_matrix/app/v1/transactions/1?access_token=hs_token"; let uri = "/_matrix/app/v1/transactions/1?access_token=hs_token";
@ -241,7 +231,7 @@ async fn test_event_handler() -> Result<()> {
.await .await
.unwrap(); .unwrap();
let on_room_member_called = *example.on_state_member.lock().unwrap(); let on_room_member_called = *on_state_member.lock().unwrap();
assert!(on_room_member_called); assert!(on_room_member_called);
Ok(()) Ok(())