matrix_sdk/event_handler/
mod.rs

1// Copyright 2021 Jonas Platte
2// Copyright 2022 Famedly GmbH
3//
4// Licensed under the Apache License, Version 2.0 (the "License");
5// you may not use this file except in compliance with the License.
6// You may obtain a copy of the License at
7//
8//     http://www.apache.org/licenses/LICENSE-2.0
9//
10// Unless required by applicable law or agreed to in writing, software
11// distributed under the License is distributed on an "AS IS" BASIS,
12// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13// See the License for the specific language governing permissions and
14// limitations under the License.
15
16//! Types and traits related for event handlers. For usage, see
17//! [`Client::add_event_handler`].
18//!
19//! ### How it works
20//!
21//! The `add_event_handler` method registers event handlers of different
22//! signatures by actually storing boxed closures that all have the same
23//! signature of `async (EventHandlerData) -> ()` where `EventHandlerData` is a
24//! private type that contains all of the data an event handler *might* need.
25//!
26//! The stored closure takes care of deserializing the event which the
27//! `EventHandlerData` contains as a (borrowed) [`serde_json::value::RawValue`],
28//! extracting the context arguments from other fields of `EventHandlerData` and
29//! calling / `.await`ing the event handler if the previous steps succeeded.
30//! It also logs any errors from the above chain of function calls.
31//!
32//! For more details, see the [`EventHandler`] trait.
33
34#[cfg(any(feature = "anyhow", feature = "eyre"))]
35use std::any::TypeId;
36use std::{
37    borrow::Cow,
38    fmt,
39    future::Future,
40    pin::Pin,
41    sync::{
42        Arc, RwLock, Weak,
43        atomic::{AtomicU64, Ordering::SeqCst},
44    },
45    task::{Context, Poll},
46};
47
48use anymap3::CloneAny;
49use eyeball::{SharedObservable, Subscriber};
50use futures_core::Stream;
51use futures_util::stream::{FuturesUnordered, StreamExt};
52use matrix_sdk_base::{
53    SendOutsideWasm, SyncOutsideWasm,
54    deserialized_responses::{EncryptionInfo, TimelineEvent},
55    sync::State,
56};
57use matrix_sdk_common::deserialized_responses::ProcessedToDeviceEvent;
58use pin_project_lite::pin_project;
59use ruma::{OwnedRoomId, events::BooleanType, push::Action, serde::Raw};
60use serde::{Deserialize, de::DeserializeOwned};
61use serde_json::value::RawValue as RawJsonValue;
62use tracing::{debug, error, field::debug, instrument, warn};
63
64use self::maps::EventHandlerMaps;
65use crate::{Client, Room};
66
67mod context;
68mod maps;
69mod static_events;
70
71pub use self::context::{Ctx, EventHandlerContext, RawEvent};
72
73#[cfg(not(target_family = "wasm"))]
74type EventHandlerFut = Pin<Box<dyn Future<Output = ()> + Send>>;
75#[cfg(target_family = "wasm")]
76type EventHandlerFut = Pin<Box<dyn Future<Output = ()>>>;
77
78#[cfg(not(target_family = "wasm"))]
79type EventHandlerFn = dyn Fn(EventHandlerData<'_>) -> EventHandlerFut + Send + Sync;
80#[cfg(target_family = "wasm")]
81type EventHandlerFn = dyn Fn(EventHandlerData<'_>) -> EventHandlerFut;
82
83#[cfg(not(target_family = "wasm"))]
84type AnyMap = anymap3::Map<dyn CloneAny + Send + Sync>;
85#[cfg(target_family = "wasm")]
86type AnyMap = anymap3::Map<dyn CloneAny>;
87
88#[derive(Default)]
89pub(crate) struct EventHandlerStore {
90    handlers: RwLock<EventHandlerMaps>,
91    context: RwLock<AnyMap>,
92    counter: AtomicU64,
93}
94
95impl EventHandlerStore {
96    pub fn add_handler(&self, handle: EventHandlerHandle, handler_fn: Box<EventHandlerFn>) {
97        self.handlers.write().unwrap().add(handle, handler_fn);
98    }
99
100    pub fn add_context<T>(&self, ctx: T)
101    where
102        T: Clone + Send + Sync + 'static,
103    {
104        self.context.write().unwrap().insert(ctx);
105    }
106
107    pub fn remove(&self, handle: EventHandlerHandle) {
108        self.handlers.write().unwrap().remove(handle);
109    }
110
111    #[cfg(test)]
112    fn len(&self) -> usize {
113        self.handlers.read().unwrap().len()
114    }
115}
116
117#[doc(hidden)]
118#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
119pub enum HandlerKind {
120    GlobalAccountData,
121    RoomAccountData,
122    EphemeralRoomData,
123    Timeline,
124    MessageLike,
125    OriginalMessageLike,
126    RedactedMessageLike,
127    State,
128    OriginalState,
129    RedactedState,
130    StrippedState,
131    ToDevice,
132    Presence,
133}
134
135impl HandlerKind {
136    fn message_like_redacted(redacted: bool) -> Self {
137        if redacted { Self::RedactedMessageLike } else { Self::OriginalMessageLike }
138    }
139
140    fn state_redacted(redacted: bool) -> Self {
141        if redacted { Self::RedactedState } else { Self::OriginalState }
142    }
143}
144
145/// A statically-known event kind/type that can be retrieved from an event sync.
146pub trait SyncEvent {
147    #[doc(hidden)]
148    const KIND: HandlerKind;
149    #[doc(hidden)]
150    const TYPE: Option<&'static str>;
151    #[doc(hidden)]
152    type IsPrefix: BooleanType;
153}
154
155pub(crate) struct EventHandlerWrapper {
156    handler_fn: Box<EventHandlerFn>,
157    pub handler_id: u64,
158}
159
160/// Handle to remove a registered event handler by passing it to
161/// [`Client::remove_event_handler`].
162#[derive(Clone, Debug)]
163pub struct EventHandlerHandle {
164    pub(crate) ev_kind: HandlerKind,
165    pub(crate) ev_type: Option<StaticEventTypePart>,
166    pub(crate) room_id: Option<OwnedRoomId>,
167    pub(crate) handler_id: u64,
168}
169
170/// The static part of an event type.
171#[derive(Clone, Copy, Debug)]
172pub(crate) enum StaticEventTypePart {
173    /// The full event type is static.
174    Full(&'static str),
175    /// Only the prefix of the event type is static.
176    Prefix(&'static str),
177}
178
179/// Interface for event handlers.
180///
181/// This trait is an abstraction for a certain kind of functions / closures,
182/// specifically:
183///
184/// - They must have at least one argument, which is the event itself, a type
185///   that implements [`SyncEvent`]. Any additional arguments need to implement
186///   the [`EventHandlerContext`] trait.
187/// - Their return type has to be one of: `()`, `Result<(), impl Display + Debug
188///   - 'static>` (if you are using `anyhow::Result` or `eyre::Result` you can
189///   additionally enable the `anyhow` / `eyre` feature to get the verbose
190///   `Debug` output printed on error)
191///
192/// ### How it works
193///
194/// This trait is basically a very constrained version of `Fn`: It requires at
195/// least one argument, which is represented as its own generic parameter `Ev`
196/// with the remaining parameter types being represented by the second generic
197/// parameter `Ctx`; they have to be stuffed into one generic parameter as a
198/// tuple because Rust doesn't have variadic generics.
199///
200/// `Ev` and `Ctx` are generic parameters rather than associated types because
201/// the argument list is a generic parameter for the `Fn` traits too, so a
202/// single type could implement `Fn` multiple times with different argument
203/// lists¹. Luckily, when calling [`Client::add_event_handler`] with a closure
204/// argument the trait solver takes into account that only a single one of the
205/// implementations applies (even though this could theoretically change through
206/// a dependency upgrade) and uses that rather than raising an ambiguity error.
207/// This is the same trick used by web frameworks like actix-web and axum.
208///
209/// ¹ the only thing stopping such types from existing in stable Rust is that
210/// all manual implementations of the `Fn` traits require a Nightly feature
211pub trait EventHandler<Ev, Ctx>: Clone + SendOutsideWasm + SyncOutsideWasm + 'static {
212    /// The future returned by `handle_event`.
213    #[doc(hidden)]
214    type Future: EventHandlerFuture;
215
216    /// Create a future for handling the given event.
217    ///
218    /// `data` provides additional data about the event, for example the room it
219    /// appeared in.
220    ///
221    /// Returns `None` if one of the context extractors failed.
222    #[doc(hidden)]
223    fn handle_event(self, ev: Ev, data: EventHandlerData<'_>) -> Option<Self::Future>;
224}
225
226#[doc(hidden)]
227pub trait EventHandlerFuture:
228    Future<Output = <Self as EventHandlerFuture>::Output> + SendOutsideWasm + 'static
229{
230    type Output: EventHandlerResult;
231}
232
233impl<T> EventHandlerFuture for T
234where
235    T: Future + SendOutsideWasm + 'static,
236    <T as Future>::Output: EventHandlerResult,
237{
238    type Output = <T as Future>::Output;
239}
240
241#[doc(hidden)]
242#[derive(Debug)]
243pub struct EventHandlerData<'a> {
244    client: Client,
245    room: Option<Room>,
246    raw: &'a RawJsonValue,
247    encryption_info: Option<&'a EncryptionInfo>,
248    push_actions: &'a [Action],
249    handle: EventHandlerHandle,
250}
251
252/// Return types supported for event handlers implement this trait.
253///
254/// It is not meant to be implemented outside of matrix-sdk.
255pub trait EventHandlerResult: Sized {
256    #[doc(hidden)]
257    fn print_error(&self, event_type: Option<&str>);
258}
259
260impl EventHandlerResult for () {
261    fn print_error(&self, _event_type: Option<&str>) {}
262}
263
264impl<E: fmt::Debug + fmt::Display + 'static> EventHandlerResult for Result<(), E> {
265    fn print_error(&self, event_type: Option<&str>) {
266        let msg_fragment = match event_type {
267            Some(event_type) => format!(" for `{event_type}`"),
268            None => "".to_owned(),
269        };
270
271        match self {
272            #[cfg(feature = "anyhow")]
273            Err(e) if TypeId::of::<E>() == TypeId::of::<anyhow::Error>() => {
274                error!("Event handler{msg_fragment} failed: {e:?}");
275            }
276            #[cfg(feature = "eyre")]
277            Err(e) if TypeId::of::<E>() == TypeId::of::<eyre::Report>() => {
278                error!("Event handler{msg_fragment} failed: {e:?}");
279            }
280            Err(e) => {
281                error!("Event handler{msg_fragment} failed: {e}");
282            }
283            Ok(_) => {}
284        }
285    }
286}
287
288#[derive(Deserialize)]
289struct UnsignedDetails {
290    redacted_because: Option<serde::de::IgnoredAny>,
291}
292
293/// Event handling internals.
294impl Client {
295    pub(crate) fn add_event_handler_impl<Ev, Ctx, H>(
296        &self,
297        handler: H,
298        room_id: Option<OwnedRoomId>,
299    ) -> EventHandlerHandle
300    where
301        Ev: SyncEvent + DeserializeOwned + SendOutsideWasm + 'static,
302        H: EventHandler<Ev, Ctx>,
303    {
304        let handler_fn: Box<EventHandlerFn> = Box::new(move |data| {
305            let maybe_fut = serde_json::from_str(data.raw.get())
306                .map(|ev| handler.clone().handle_event(ev, data));
307
308            Box::pin(async move {
309                match maybe_fut {
310                    Ok(Some(fut)) => {
311                        fut.await.print_error(Ev::TYPE);
312                    }
313                    Ok(None) => {
314                        error!(
315                            event_type = Ev::TYPE, event_kind = ?Ev::KIND,
316                            "Event handler has an invalid context argument",
317                        );
318                    }
319                    Err(e) => {
320                        warn!(
321                            event_type = Ev::TYPE, event_kind = ?Ev::KIND,
322                            "Failed to deserialize event, skipping event handler.\n
323                             Deserialization error: {e}",
324                        );
325                    }
326                }
327            })
328        });
329
330        let handler_id = self.inner.event_handlers.counter.fetch_add(1, SeqCst);
331        let ev_type = Ev::TYPE.map(|ev_type| {
332            if Ev::IsPrefix::as_bool() {
333                StaticEventTypePart::Prefix(ev_type)
334            } else {
335                StaticEventTypePart::Full(ev_type)
336            }
337        });
338        let handle = EventHandlerHandle { ev_kind: Ev::KIND, ev_type, room_id, handler_id };
339
340        self.inner.event_handlers.add_handler(handle.clone(), handler_fn);
341
342        handle
343    }
344
345    pub(crate) async fn handle_sync_events<T>(
346        &self,
347        kind: HandlerKind,
348        room: Option<&Room>,
349        events: &[Raw<T>],
350    ) -> serde_json::Result<()> {
351        #[derive(Deserialize)]
352        struct ExtractType<'a> {
353            #[serde(borrow, rename = "type")]
354            event_type: Cow<'a, str>,
355        }
356
357        for raw_event in events {
358            let event_type = raw_event.deserialize_as_unchecked::<ExtractType<'_>>()?.event_type;
359            self.call_event_handlers(room, raw_event.json(), kind, &event_type, None, &[]).await;
360        }
361
362        Ok(())
363    }
364
365    pub(crate) async fn handle_sync_to_device_events(
366        &self,
367        events: &[ProcessedToDeviceEvent],
368    ) -> serde_json::Result<()> {
369        #[derive(Deserialize)]
370        struct ExtractType<'a> {
371            #[serde(borrow, rename = "type")]
372            event_type: Cow<'a, str>,
373        }
374
375        for processed_to_device in events {
376            let (raw_event, encryption_info) = match processed_to_device {
377                ProcessedToDeviceEvent::Decrypted { raw, encryption_info } => {
378                    (raw, Some(encryption_info))
379                }
380                other => (&other.to_raw(), None),
381            };
382            let event_type = raw_event.deserialize_as_unchecked::<ExtractType<'_>>()?.event_type;
383            self.call_event_handlers(
384                None,
385                raw_event.json(),
386                HandlerKind::ToDevice,
387                &event_type,
388                encryption_info,
389                &[],
390            )
391            .await;
392        }
393
394        Ok(())
395    }
396
397    pub(crate) async fn handle_sync_state_events(
398        &self,
399        room: Option<&Room>,
400        state: &State,
401    ) -> serde_json::Result<()> {
402        #[derive(Deserialize)]
403        struct StateEventDetails<'a> {
404            #[serde(borrow, rename = "type")]
405            event_type: Cow<'a, str>,
406            unsigned: Option<UnsignedDetails>,
407        }
408
409        let state_events = match state {
410            State::Before(events) => events,
411            State::After(events) => events,
412        };
413
414        // Event handlers for possibly-redacted state events
415        self.handle_sync_events(HandlerKind::State, room, state_events).await?;
416
417        // Event handlers specifically for redacted OR unredacted state events
418        for raw_event in state_events {
419            let StateEventDetails { event_type, unsigned } =
420                raw_event.deserialize_as_unchecked()?;
421            let redacted = unsigned.and_then(|u| u.redacted_because).is_some();
422            let handler_kind = HandlerKind::state_redacted(redacted);
423
424            self.call_event_handlers(room, raw_event.json(), handler_kind, &event_type, None, &[])
425                .await;
426        }
427
428        Ok(())
429    }
430
431    pub(crate) async fn handle_sync_timeline_events(
432        &self,
433        room: Option<&Room>,
434        timeline_events: &[TimelineEvent],
435    ) -> serde_json::Result<()> {
436        #[derive(Deserialize)]
437        struct TimelineEventDetails<'a> {
438            #[serde(borrow, rename = "type")]
439            event_type: Cow<'a, str>,
440            state_key: Option<serde::de::IgnoredAny>,
441            unsigned: Option<UnsignedDetails>,
442        }
443
444        for item in timeline_events {
445            let TimelineEventDetails { event_type, state_key, unsigned } =
446                item.raw().deserialize_as_unchecked()?;
447
448            let redacted = unsigned.and_then(|u| u.redacted_because).is_some();
449            let (handler_kind_g, handler_kind_r) = match state_key {
450                Some(_) => (HandlerKind::State, HandlerKind::state_redacted(redacted)),
451                None => (HandlerKind::MessageLike, HandlerKind::message_like_redacted(redacted)),
452            };
453
454            let raw_event = item.raw().json();
455            let encryption_info = item.encryption_info().map(|i| &**i);
456            let push_actions = item.push_actions().unwrap_or(&[]);
457
458            // Event handlers for possibly-redacted timeline events
459            self.call_event_handlers(
460                room,
461                raw_event,
462                handler_kind_g,
463                &event_type,
464                encryption_info,
465                push_actions,
466            )
467            .await;
468
469            // Event handlers specifically for redacted OR unredacted timeline
470            // events
471            self.call_event_handlers(
472                room,
473                raw_event,
474                handler_kind_r,
475                &event_type,
476                encryption_info,
477                push_actions,
478            )
479            .await;
480
481            // Event handlers for `AnySyncTimelineEvent`
482            let kind = HandlerKind::Timeline;
483            self.call_event_handlers(
484                room,
485                raw_event,
486                kind,
487                &event_type,
488                encryption_info,
489                push_actions,
490            )
491            .await;
492        }
493
494        Ok(())
495    }
496
497    #[instrument(skip_all, fields(?event_kind, ?event_type, room_id))]
498    async fn call_event_handlers(
499        &self,
500        room: Option<&Room>,
501        raw: &RawJsonValue,
502        event_kind: HandlerKind,
503        event_type: &str,
504        encryption_info: Option<&EncryptionInfo>,
505        push_actions: &[Action],
506    ) {
507        let room_id = room.map(|r| r.room_id());
508        if let Some(room_id) = room_id {
509            tracing::Span::current().record("room_id", debug(room_id));
510        }
511
512        // Construct event handler futures
513        let mut futures: FuturesUnordered<_> = self
514            .inner
515            .event_handlers
516            .handlers
517            .read()
518            .unwrap()
519            .get_handlers(event_kind, event_type, room_id)
520            .map(|(handle, handler_fn)| {
521                let data = EventHandlerData {
522                    client: self.clone(),
523                    room: room.cloned(),
524                    raw,
525                    encryption_info,
526                    push_actions,
527                    handle,
528                };
529
530                (handler_fn)(data)
531            })
532            .collect();
533
534        if !futures.is_empty() {
535            debug!(amount = futures.len(), "Calling event handlers");
536
537            // Run the event handler futures with the
538            // `self.event_handlers.handlers` lock no longer being held.
539            while let Some(()) = futures.next().await {}
540        }
541    }
542}
543
544/// A guard type that removes an event handler when it drops (goes out of
545/// scope).
546///
547/// Created with [`Client::event_handler_drop_guard`].
548#[derive(Debug)]
549pub struct EventHandlerDropGuard {
550    handle: EventHandlerHandle,
551    client: Client,
552}
553
554impl EventHandlerDropGuard {
555    pub(crate) fn new(handle: EventHandlerHandle, client: Client) -> Self {
556        Self { handle, client }
557    }
558}
559
560impl Drop for EventHandlerDropGuard {
561    fn drop(&mut self) {
562        self.client.remove_event_handler(self.handle.clone());
563    }
564}
565
566macro_rules! impl_event_handler {
567    ($($ty:ident),* $(,)?) => {
568        impl<Ev, Fun, Fut, $($ty),*> EventHandler<Ev, ($($ty,)*)> for Fun
569        where
570            Ev: SyncEvent,
571            Fun: FnOnce(Ev, $($ty),*) -> Fut + Clone + SendOutsideWasm + SyncOutsideWasm + 'static,
572            Fut: EventHandlerFuture,
573            $($ty: EventHandlerContext),*
574        {
575            type Future = Fut;
576
577            fn handle_event(self, ev: Ev, _d: EventHandlerData<'_>) -> Option<Self::Future> {
578                Some((self)(ev, $($ty::from_data(&_d)?),*))
579            }
580        }
581    };
582}
583
584impl_event_handler!();
585impl_event_handler!(A);
586impl_event_handler!(A, B);
587impl_event_handler!(A, B, C);
588impl_event_handler!(A, B, C, D);
589impl_event_handler!(A, B, C, D, E);
590impl_event_handler!(A, B, C, D, E, F);
591impl_event_handler!(A, B, C, D, E, F, G);
592impl_event_handler!(A, B, C, D, E, F, G, H);
593
594/// An observer of events (may be tailored to a room).
595///
596/// Only the most recent value can be observed. Subscribers are notified when a
597/// new value is sent, but there is no guarantee that they will see all values.
598///
599/// To create such observer, use [`Client::observe_events`] or
600/// [`Client::observe_room_events`].
601#[derive(Debug)]
602pub struct ObservableEventHandler<T> {
603    /// This type is actually nothing more than a thin glue layer between the
604    /// [`EventHandler`] mechanism and the reactive programming types from
605    /// [`eyeball`]. Here, we use a [`SharedObservable`] that is updated by the
606    /// [`EventHandler`].
607    shared_observable: SharedObservable<Option<T>>,
608
609    /// This type owns the [`EventHandlerDropGuard`]. As soon as this type goes
610    /// out of scope, the event handler is unregistered/removed.
611    ///
612    /// [`EventHandlerSubscriber`] holds a weak, non-owning reference, to this
613    /// guard. It is useful to detect when to close the [`Stream`]: as soon as
614    /// this type goes out of scope, the subscriber will close itself on poll.
615    event_handler_guard: Arc<EventHandlerDropGuard>,
616}
617
618impl<T> ObservableEventHandler<T> {
619    pub(crate) fn new(
620        shared_observable: SharedObservable<Option<T>>,
621        event_handler_guard: EventHandlerDropGuard,
622    ) -> Self {
623        Self { shared_observable, event_handler_guard: Arc::new(event_handler_guard) }
624    }
625
626    /// Subscribe to this observer.
627    ///
628    /// It returns an [`EventHandlerSubscriber`], which implements [`Stream`].
629    /// See its documentation to learn more.
630    pub fn subscribe(&self) -> EventHandlerSubscriber<T> {
631        EventHandlerSubscriber::new(
632            self.shared_observable.subscribe(),
633            // The subscriber holds a weak non-owning reference to the event
634            // handler guard, so that it can detect when this observer is
635            // dropped, and can close the subscriber's stream.
636            Arc::downgrade(&self.event_handler_guard),
637        )
638    }
639}
640
641pin_project! {
642    /// The subscriber of an [`ObservableEventHandler`].
643    ///
644    /// To create such subscriber, use [`ObservableEventHandler::subscribe`].
645    ///
646    /// This type implements [`Stream`], which means it is possible to poll the
647    /// next value asynchronously. In other terms, polling this type will return
648    /// the new event as soon as they are synced. See [`Client::observe_events`]
649    /// to learn more.
650    #[derive(Debug)]
651    pub struct EventHandlerSubscriber<T> {
652        // The `Subscriber` associated to the `SharedObservable` inside
653        // `ObservableEventHandle`.
654        //
655        // Keep in mind all this API is just a thin glue layer between
656        // `EventHandle` and `SharedObservable`, that's… maagiic!
657        #[pin]
658        subscriber: Subscriber<Option<T>>,
659
660        // A weak non-owning reference to the event handler guard from
661        // `ObservableEventHandler`. When this type is polled (via its `Stream`
662        // implementation), it is possible to detect whether the observable has
663        // been dropped by upgrading this weak reference, and close the `Stream`
664        // if it needs to.
665        event_handler_guard: Weak<EventHandlerDropGuard>,
666    }
667}
668
669impl<T> EventHandlerSubscriber<T> {
670    fn new(
671        subscriber: Subscriber<Option<T>>,
672        event_handler_handle: Weak<EventHandlerDropGuard>,
673    ) -> Self {
674        Self { subscriber, event_handler_guard: event_handler_handle }
675    }
676}
677
678impl<T> Stream for EventHandlerSubscriber<T>
679where
680    T: Clone,
681{
682    type Item = T;
683
684    fn poll_next(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Option<Self::Item>> {
685        let mut this = self.project();
686
687        let Some(_) = this.event_handler_guard.upgrade() else {
688            // The `EventHandlerHandle` has been dropped via
689            // `EventHandlerDropGuard`. It means the `ObservableEventHandler`
690            // has been dropped. It's time to close this stream.
691            return Poll::Ready(None);
692        };
693
694        // First off, the subscriber is of type `Subscriber<Option<T>>` because
695        // the `SharedObservable` starts with a `None` value to indicate it has
696        // no yet received any update. We want the `Stream` to return `T`, not
697        // `Option<T>`. We then filter out all `None` value.
698        //
699        // Second, when a `None` value is met, we want to poll again (hence the
700        // `loop`). At best, there is a new value to return. At worst, the
701        // subscriber will return `Poll::Pending` and will register the wakers
702        // accordingly.
703
704        loop {
705            match this.subscriber.as_mut().poll_next(context) {
706                // Stream has been closed somehow.
707                Poll::Ready(None) => return Poll::Ready(None),
708
709                // The initial value (of the `SharedObservable` behind
710                // `self.subscriber`) has been polled. We want to filter it out.
711                Poll::Ready(Some(None)) => {
712                    // Loop over.
713                    continue;
714                }
715
716                // We have a new value!
717                Poll::Ready(Some(Some(value))) => return Poll::Ready(Some(value)),
718
719                // Classical pending.
720                Poll::Pending => return Poll::Pending,
721            }
722        }
723    }
724}
725
726#[cfg(test)]
727mod tests {
728    use matrix_sdk_test::{
729        DEFAULT_TEST_ROOM_ID, InvitedRoomBuilder, JoinedRoomBuilder, async_test,
730        event_factory::{EventFactory, PreviousMembership},
731    };
732    use serde::Serialize;
733    use stream_assert::{assert_closed, assert_pending, assert_ready};
734    #[cfg(target_family = "wasm")]
735    wasm_bindgen_test::wasm_bindgen_test_configure!(run_in_browser);
736    use std::{
737        future,
738        sync::{
739            Arc, LazyLock,
740            atomic::{AtomicU8, Ordering::SeqCst},
741        },
742    };
743
744    use matrix_sdk_common::{deserialized_responses::EncryptionInfo, locks::Mutex};
745    use matrix_sdk_test::SyncResponseBuilder;
746    use ruma::{
747        event_id,
748        events::{
749            AnySyncStateEvent, AnySyncTimelineEvent, AnyToDeviceEvent,
750            macros::EventContent,
751            room::{
752                member::{MembershipState, OriginalSyncRoomMemberEvent, StrippedRoomMemberEvent},
753                name::OriginalSyncRoomNameEvent,
754                power_levels::OriginalSyncRoomPowerLevelsEvent,
755            },
756            secret_storage::key::SecretStorageKeyEvent,
757            typing::SyncTypingEvent,
758        },
759        mxc_uri,
760        room::JoinRule,
761        room_id,
762        serde::Raw,
763        user_id,
764    };
765    use serde_json::json;
766    use strass::assert_let;
767
768    use crate::{
769        Client, Room,
770        event_handler::Ctx,
771        test_utils::{logged_in_client, no_retry_test_client},
772    };
773
774    static MEMBER_EVENT: LazyLock<Raw<AnySyncTimelineEvent>> = LazyLock::new(|| {
775        EventFactory::new()
776            .member(user_id!("@example:localhost"))
777            .membership(MembershipState::Join)
778            .display_name("example")
779            .event_id(event_id!("$151800140517rfvjc:localhost"))
780            .previous(PreviousMembership::new(MembershipState::Invite).display_name("example"))
781            .into()
782    });
783
784    #[async_test]
785    async fn test_add_event_handler() -> crate::Result<()> {
786        let client = logged_in_client(None).await;
787
788        let member_count = Arc::new(AtomicU8::new(0));
789        let typing_count = Arc::new(AtomicU8::new(0));
790        let power_levels_count = Arc::new(AtomicU8::new(0));
791        let invited_member_count = Arc::new(AtomicU8::new(0));
792
793        client.add_event_handler({
794            let member_count = member_count.clone();
795            move |_ev: OriginalSyncRoomMemberEvent, _room: Room| async move {
796                member_count.fetch_add(1, SeqCst);
797            }
798        });
799        client.add_event_handler({
800            let typing_count = typing_count.clone();
801            move |_ev: SyncTypingEvent| async move {
802                typing_count.fetch_add(1, SeqCst);
803            }
804        });
805        client.add_event_handler({
806            let power_levels_count = power_levels_count.clone();
807            move |_ev: OriginalSyncRoomPowerLevelsEvent, _client: Client, _room: Room| async move {
808                power_levels_count.fetch_add(1, SeqCst);
809            }
810        });
811        client.add_event_handler({
812            let invited_member_count = invited_member_count.clone();
813            move |_ev: StrippedRoomMemberEvent| async move {
814                invited_member_count.fetch_add(1, SeqCst);
815            }
816        });
817
818        let f = EventFactory::new().sender(user_id!("@example:localhost"));
819        let response = SyncResponseBuilder::default()
820            .add_joined_room(
821                JoinedRoomBuilder::default()
822                    .add_timeline_event(MEMBER_EVENT.clone())
823                    .add_typing(
824                        f.typing(vec![user_id!("@alice:matrix.org"), user_id!("@bob:example.com")]),
825                    )
826                    .add_state_event(f.default_power_levels()),
827            )
828            .add_invited_room(
829                InvitedRoomBuilder::new(room_id!("!test_invited:example.org")).add_state_event({
830                    let bob = user_id!("@bob:example.org");
831                    EventFactory::new()
832                        .sender(user_id!("@example:example.org"))
833                        .member(user_id!("@alice:example.org"))
834                        .membership(MembershipState::Invite)
835                        .display_name("Alice")
836                        .avatar_url(mxc_uri!("mxc://example.org/SEsfnsuifSDFSSEF"))
837                        .age(1234_i32)
838                        .invite_room_state(vec![
839                            Raw::from(f.room_name("Example Room").sender(bob)),
840                            Raw::from(f.room_join_rules(JoinRule::Invite).sender(bob)),
841                        ])
842                }),
843            )
844            .build_sync_response();
845        client.process_sync(response).await?;
846
847        assert_eq!(member_count.load(SeqCst), 1);
848        assert_eq!(typing_count.load(SeqCst), 1);
849        assert_eq!(power_levels_count.load(SeqCst), 1);
850        assert_eq!(invited_member_count.load(SeqCst), 1);
851
852        Ok(())
853    }
854
855    #[async_test]
856    async fn test_add_to_device_event_handler() -> crate::Result<()> {
857        let client = logged_in_client(None).await;
858
859        let captured_event: Arc<Mutex<Option<AnyToDeviceEvent>>> = Arc::new(Mutex::new(None));
860        let captured_info: Arc<Mutex<Option<EncryptionInfo>>> = Arc::new(Mutex::new(None));
861
862        client.add_event_handler({
863            let captured = captured_event.clone();
864            let captured_info = captured_info.clone();
865            move |ev: AnyToDeviceEvent, encryption_info: Option<EncryptionInfo>| {
866                let mut captured_lock = captured.lock();
867                *captured_lock = Some(ev);
868                let mut captured_info_lock = captured_info.lock();
869                *captured_info_lock = encryption_info;
870                future::ready(())
871            }
872        });
873
874        let response = SyncResponseBuilder::default()
875            .add_to_device_event(json!({
876              "sender": "@alice:example.com",
877              "type": "m.custom.to.device.type",
878              "content": {
879                "a": "test",
880              }
881            }))
882            .build_sync_response();
883        client.process_sync(response).await?;
884
885        let captured = captured_event.lock().clone();
886        assert_let!(Some(received_event) = captured);
887        assert_eq!(received_event.event_type().to_string(), "m.custom.to.device.type");
888        let info = captured_info.lock().clone();
889        assert!(info.is_none());
890        Ok(())
891    }
892
893    #[async_test]
894    async fn test_add_room_event_handler() -> crate::Result<()> {
895        let client = logged_in_client(None).await;
896
897        let room_id_a = room_id!("!foo:example.org");
898        let room_id_b = room_id!("!bar:matrix.org");
899
900        let member_count = Arc::new(AtomicU8::new(0));
901        let power_levels_count = Arc::new(AtomicU8::new(0));
902
903        // Room event handlers for member events in both rooms
904        client.add_room_event_handler(room_id_a, {
905            let member_count = member_count.clone();
906            move |_ev: OriginalSyncRoomMemberEvent, _room: Room| {
907                member_count.fetch_add(1, SeqCst);
908                future::ready(())
909            }
910        });
911        client.add_room_event_handler(room_id_b, {
912            let member_count = member_count.clone();
913            move |_ev: OriginalSyncRoomMemberEvent, _room: Room| {
914                member_count.fetch_add(1, SeqCst);
915                future::ready(())
916            }
917        });
918
919        // Power levels event handlers for member events in room A
920        client.add_room_event_handler(room_id_a, {
921            let power_levels_count = power_levels_count.clone();
922            move |_ev: OriginalSyncRoomPowerLevelsEvent, _client: Client, _room: Room| {
923                power_levels_count.fetch_add(1, SeqCst);
924                future::ready(())
925            }
926        });
927
928        // Room name event handler for room name events in room B
929        client.add_room_event_handler(
930            room_id_b,
931            // lint is buggy: rustc wants the explicit conversion from ! to ()
932            // here, but clippy thinks it's useless.
933            #[allow(clippy::unused_unit)]
934            async move |_ev: OriginalSyncRoomNameEvent| -> () {
935                unreachable!("No room event in room B")
936            },
937        );
938
939        let f = EventFactory::new().sender(user_id!("@example:localhost"));
940        let response = SyncResponseBuilder::default()
941            .add_joined_room(
942                JoinedRoomBuilder::new(room_id_a)
943                    .add_timeline_event(MEMBER_EVENT.clone())
944                    .add_state_event(f.default_power_levels())
945                    .add_state_event(f.room_name("room name")),
946            )
947            .add_joined_room(
948                JoinedRoomBuilder::new(room_id_b)
949                    .add_timeline_event(MEMBER_EVENT.clone())
950                    .add_state_event(f.default_power_levels()),
951            )
952            .build_sync_response();
953        client.process_sync(response).await?;
954
955        assert_eq!(member_count.load(SeqCst), 2);
956        assert_eq!(power_levels_count.load(SeqCst), 1);
957
958        Ok(())
959    }
960
961    #[async_test]
962    async fn test_add_event_handler_with_tuples() -> crate::Result<()> {
963        let client = logged_in_client(None).await;
964
965        client.add_event_handler(
966            |_ev: OriginalSyncRoomMemberEvent, (_room, _client): (Room, Client)| future::ready(()),
967        );
968
969        // If it compiles, it works. No need to assert anything.
970
971        Ok(())
972    }
973
974    #[async_test]
975    async fn test_remove_event_handler() -> crate::Result<()> {
976        let client = logged_in_client(None).await;
977
978        let member_count = Arc::new(AtomicU8::new(0));
979
980        client.add_event_handler({
981            let member_count = member_count.clone();
982            move |_ev: OriginalSyncRoomMemberEvent| async move {
983                member_count.fetch_add(1, SeqCst);
984            }
985        });
986
987        let handle_a = client.add_event_handler(
988            // lint is buggy: rustc wants the explicit conversion from ! to ()
989            // here, but clippy thinks it's useless.
990            #[allow(clippy::unused_unit)]
991            async move |_ev: OriginalSyncRoomMemberEvent| -> () {
992                panic!("handler should have been removed");
993            },
994        );
995        let handle_b = client.add_room_event_handler(
996            #[allow(unknown_lints, clippy::explicit_auto_deref)] // lint is buggy
997            *DEFAULT_TEST_ROOM_ID,
998            // lint is buggy: rustc wants the explicit conversion from ! to ()
999            // here, but clippy thinks it's useless.
1000            #[allow(clippy::unused_unit)]
1001            async move |_ev: OriginalSyncRoomMemberEvent| -> () {
1002                panic!("handler should have been removed");
1003            },
1004        );
1005
1006        client.add_event_handler({
1007            let member_count = member_count.clone();
1008            move |_ev: OriginalSyncRoomMemberEvent| async move {
1009                member_count.fetch_add(1, SeqCst);
1010            }
1011        });
1012
1013        let response = SyncResponseBuilder::default()
1014            .add_joined_room(JoinedRoomBuilder::default().add_timeline_event(MEMBER_EVENT.clone()))
1015            .build_sync_response();
1016
1017        client.remove_event_handler(handle_a);
1018        client.remove_event_handler(handle_b);
1019
1020        client.process_sync(response).await?;
1021
1022        assert_eq!(member_count.load(SeqCst), 2);
1023
1024        Ok(())
1025    }
1026
1027    #[async_test]
1028    async fn test_event_handler_drop_guard() {
1029        let client = no_retry_test_client(None).await;
1030
1031        let handle = client.add_event_handler(|_ev: OriginalSyncRoomMemberEvent| async {});
1032        assert_eq!(client.inner.event_handlers.len(), 1);
1033
1034        {
1035            let _guard = client.event_handler_drop_guard(handle);
1036            assert_eq!(client.inner.event_handlers.len(), 1);
1037            // guard dropped here
1038        }
1039
1040        assert_eq!(client.inner.event_handlers.len(), 0);
1041    }
1042
1043    #[async_test]
1044    async fn test_use_client_in_handler() {
1045        // This used to not work because we were requiring `Send` of event
1046        // handler futures even on WASM, where practically all futures that do
1047        // I/O aren't.
1048        let client = no_retry_test_client(None).await;
1049
1050        client.add_event_handler(|_ev: OriginalSyncRoomMemberEvent, client: Client| async move {
1051            // All of Client's async methods that do network requests (and
1052            // possibly some that don't) are `!Send` on wasm. We obviously want
1053            // to be able to use them in event handlers.
1054            client
1055                .homeserver_capabilities()
1056                .refresh()
1057                .await
1058                .map_err(|e| anyhow::anyhow!("{}", e))?;
1059            anyhow::Ok(())
1060        });
1061    }
1062
1063    #[async_test]
1064    async fn test_raw_event_handler() -> crate::Result<()> {
1065        let client = logged_in_client(None).await;
1066        let counter = Arc::new(AtomicU8::new(0));
1067        client.add_event_handler_context(counter.clone());
1068        client.add_event_handler(
1069            |_ev: Raw<OriginalSyncRoomMemberEvent>, counter: Ctx<Arc<AtomicU8>>| async move {
1070                counter.fetch_add(1, SeqCst);
1071            },
1072        );
1073
1074        let response = SyncResponseBuilder::default()
1075            .add_joined_room(JoinedRoomBuilder::default().add_timeline_event(MEMBER_EVENT.clone()))
1076            .build_sync_response();
1077        client.process_sync(response).await?;
1078
1079        assert_eq!(counter.load(SeqCst), 1);
1080        Ok(())
1081    }
1082
1083    #[async_test]
1084    async fn test_enum_event_handler() -> crate::Result<()> {
1085        let client = logged_in_client(None).await;
1086        let counter = Arc::new(AtomicU8::new(0));
1087        client.add_event_handler_context(counter.clone());
1088        client.add_event_handler(
1089            |_ev: AnySyncStateEvent, counter: Ctx<Arc<AtomicU8>>| async move {
1090                counter.fetch_add(1, SeqCst);
1091            },
1092        );
1093
1094        let response = SyncResponseBuilder::default()
1095            .add_joined_room(JoinedRoomBuilder::default().add_timeline_event(MEMBER_EVENT.clone()))
1096            .build_sync_response();
1097        client.process_sync(response).await?;
1098
1099        assert_eq!(counter.load(SeqCst), 1);
1100        Ok(())
1101    }
1102
1103    #[async_test]
1104    async fn test_observe_events() -> crate::Result<()> {
1105        let client = logged_in_client(None).await;
1106
1107        let room_id_0 = room_id!("!r0.matrix.org");
1108        let room_id_1 = room_id!("!r1.matrix.org");
1109
1110        let observable = client.observe_events::<OriginalSyncRoomNameEvent, Room>();
1111
1112        let mut subscriber = observable.subscribe();
1113
1114        assert_pending!(subscriber);
1115
1116        let f = EventFactory::new().sender(user_id!("@mnt_io:matrix.org"));
1117        let mut response_builder = SyncResponseBuilder::new();
1118        let response = response_builder
1119            .add_joined_room(
1120                JoinedRoomBuilder::new(room_id_0)
1121                    .add_state_event(f.room_name("Name 0").event_id(event_id!("$ev0"))),
1122            )
1123            .build_sync_response();
1124        client.process_sync(response).await?;
1125
1126        let (room_name, room) = assert_ready!(subscriber);
1127
1128        assert_eq!(room_name.event_id.as_str(), "$ev0");
1129        assert_eq!(room.room_id(), room_id_0);
1130        assert_eq!(room.name().unwrap(), "Name 0");
1131
1132        assert_pending!(subscriber);
1133
1134        let response = response_builder
1135            .add_joined_room(
1136                JoinedRoomBuilder::new(room_id_1)
1137                    .add_state_event(f.room_name("Name 1").event_id(event_id!("$ev1"))),
1138            )
1139            .build_sync_response();
1140        client.process_sync(response).await?;
1141
1142        let (room_name, room) = assert_ready!(subscriber);
1143
1144        assert_eq!(room_name.event_id.as_str(), "$ev1");
1145        assert_eq!(room.room_id(), room_id_1);
1146        assert_eq!(room.name().unwrap(), "Name 1");
1147
1148        assert_pending!(subscriber);
1149
1150        drop(observable);
1151        assert_closed!(subscriber);
1152
1153        Ok(())
1154    }
1155
1156    #[async_test]
1157    async fn test_observe_room_events() -> crate::Result<()> {
1158        let client = logged_in_client(None).await;
1159
1160        let room_id = room_id!("!r0.matrix.org");
1161
1162        let observable_for_room =
1163            client.observe_room_events::<OriginalSyncRoomNameEvent, (Room, Client)>(room_id);
1164
1165        let mut subscriber_for_room = observable_for_room.subscribe();
1166
1167        assert_pending!(subscriber_for_room);
1168
1169        let f = EventFactory::new().sender(user_id!("@mnt_io:matrix.org"));
1170        let mut response_builder = SyncResponseBuilder::new();
1171        let response = response_builder
1172            .add_joined_room(
1173                JoinedRoomBuilder::new(room_id)
1174                    .add_state_event(f.room_name("Name 0").event_id(event_id!("$ev0"))),
1175            )
1176            .build_sync_response();
1177        client.process_sync(response).await?;
1178
1179        let (room_name, (room, _client)) = assert_ready!(subscriber_for_room);
1180
1181        assert_eq!(room_name.event_id.as_str(), "$ev0");
1182        assert_eq!(room.name().unwrap(), "Name 0");
1183
1184        assert_pending!(subscriber_for_room);
1185
1186        let response = response_builder
1187            .add_joined_room(
1188                JoinedRoomBuilder::new(room_id)
1189                    .add_state_event(f.room_name("Name 1").event_id(event_id!("$ev1"))),
1190            )
1191            .build_sync_response();
1192        client.process_sync(response).await?;
1193
1194        let (room_name, (room, _client)) = assert_ready!(subscriber_for_room);
1195
1196        assert_eq!(room_name.event_id.as_str(), "$ev1");
1197        assert_eq!(room.name().unwrap(), "Name 1");
1198
1199        assert_pending!(subscriber_for_room);
1200
1201        drop(observable_for_room);
1202        assert_closed!(subscriber_for_room);
1203
1204        Ok(())
1205    }
1206
1207    #[async_test]
1208    async fn test_observe_several_room_events() -> crate::Result<()> {
1209        let client = logged_in_client(None).await;
1210
1211        let room_id = room_id!("!r0.matrix.org");
1212
1213        let observable_for_room =
1214            client.observe_room_events::<OriginalSyncRoomNameEvent, (Room, Client)>(room_id);
1215
1216        let mut subscriber_for_room = observable_for_room.subscribe();
1217
1218        assert_pending!(subscriber_for_room);
1219
1220        let f = EventFactory::new().sender(user_id!("@mnt_io:matrix.org"));
1221        let mut response_builder = SyncResponseBuilder::new();
1222        let response = response_builder
1223            .add_joined_room(
1224                JoinedRoomBuilder::new(room_id)
1225                    .add_state_event(f.room_name("Name 0").event_id(event_id!("$ev0")))
1226                    .add_state_event(f.room_name("Name 1").event_id(event_id!("$ev1")))
1227                    .add_state_event(f.room_name("Name 2").event_id(event_id!("$ev2"))),
1228            )
1229            .build_sync_response();
1230        client.process_sync(response).await?;
1231
1232        let (room_name, (room, _client)) = assert_ready!(subscriber_for_room);
1233
1234        // Check we only get notified about the latest received event
1235        assert_eq!(room_name.event_id.as_str(), "$ev2");
1236        assert_eq!(room.name().unwrap(), "Name 2");
1237
1238        assert_pending!(subscriber_for_room);
1239
1240        drop(observable_for_room);
1241        assert_closed!(subscriber_for_room);
1242
1243        Ok(())
1244    }
1245
1246    #[async_test]
1247    async fn test_observe_events_with_type_prefix() -> crate::Result<()> {
1248        let client = logged_in_client(None).await;
1249
1250        let observable = client.observe_events::<SecretStorageKeyEvent, ()>();
1251
1252        let mut subscriber = observable.subscribe();
1253
1254        assert_pending!(subscriber);
1255
1256        let mut response_builder = SyncResponseBuilder::new();
1257        let response = response_builder
1258            .add_custom_global_account_data(json!({
1259                "content": {
1260                    "algorithm": "m.secret_storage.v1.aes-hmac-sha2",
1261                    "iv": "gH2iNpiETFhApvW6/FFEJQ",
1262                    "mac": "9Lw12m5SKDipNghdQXKjgpfdj1/K7HFI2brO+UWAGoM",
1263                    "passphrase": {
1264                        "algorithm": "m.pbkdf2",
1265                        "salt": "IuLnH7S85YtZmkkBJKwNUKxWF42g9O1H",
1266                        "iterations": 10,
1267                    },
1268                },
1269                "type": "m.secret_storage.key.foobar",
1270            }))
1271            .build_sync_response();
1272        client.process_sync(response).await?;
1273
1274        let (secret_storage_key, ()) = assert_ready!(subscriber);
1275
1276        assert_eq!(secret_storage_key.content.key_id, "foobar");
1277
1278        assert_pending!(subscriber);
1279
1280        drop(observable);
1281        assert_closed!(subscriber);
1282
1283        Ok(())
1284    }
1285
1286    #[async_test]
1287    async fn test_observe_room_events_with_type_prefix() -> crate::Result<()> {
1288        // To create an event handler for a room account data event type with
1289        // prefix, we need to create a custom event type, none exist in the
1290        // Matrix specification yet.
1291        #[derive(Debug, Clone, EventContent, Serialize)]
1292        #[ruma_event(type = "fake.event.*", kind = RoomAccountData)]
1293        struct AccountDataWithPrefixEventContent {
1294            #[ruma_event(type_fragment)]
1295            #[serde(skip)]
1296            key_id: String,
1297        }
1298
1299        let room_id = room_id!("!r0.matrix.org");
1300        let client = logged_in_client(None).await;
1301
1302        let observable = client.observe_room_events::<AccountDataWithPrefixEvent, Room>(room_id);
1303
1304        let mut subscriber = observable.subscribe();
1305
1306        assert_pending!(subscriber);
1307
1308        let mut response_builder = SyncResponseBuilder::new();
1309        let response = response_builder
1310            .add_joined_room(
1311                JoinedRoomBuilder::new(room_id).add_account_data_bulk([Raw::new(&json!({
1312                    "content": {},
1313                    "type": "fake.event.foobar",
1314                }))
1315                .unwrap()
1316                .cast_unchecked()]),
1317            )
1318            .build_sync_response();
1319        client.process_sync(response).await?;
1320
1321        let (secret_storage_key, _room) = assert_ready!(subscriber);
1322
1323        assert_eq!(secret_storage_key.content.key_id, "foobar");
1324
1325        assert_pending!(subscriber);
1326
1327        drop(observable);
1328        assert_closed!(subscriber);
1329
1330        Ok(())
1331    }
1332}
matrix_sdk/event_handler/mod.rs

matrix_sdk/event_handler/
mod.rs
