nio: Reintroduce futures as callbacks.

examples/login.rs

@@ -1,6 +1,10 @@
 #![feature(async_closure)]
 
 use std::{env, process::exit};
+use std::pin::Pin;
+use std::future::Future;
+use std::rc::Rc;
+use std::cell::RefCell;
 
 use matrix_nio::{
     self,
@@ -12,8 +16,22 @@ use matrix_nio::{
     AsyncClient, AsyncClientConfig, SyncSettings, Room
 };
 
-fn callback(room: &Room, event: &RoomEvent) {
-    println!("{:?}", event);
+async fn async_helper(room: Rc<RefCell<Room>>, event: Rc<RoomEvent>) {
+    let room = room.borrow();
+    if let RoomEvent::RoomMessage(MessageEvent {
+        content: MessageEventContent::Text(TextMessageEventContent { body: msg_body, .. }),
+        sender,
+        ..
+    }) = &*event
+    {
+        let user = room.members.get(&sender.to_string()).unwrap();
+        println!("{}: {}", user.display_name.as_ref().unwrap_or(&sender.to_string()), msg_body);
+    }
+
+}
+
+fn async_callback(room: Rc<RefCell<Room>>, event: Rc<RoomEvent>) -> Pin<Box<dyn Future<Output = ()>>> {
+    Box::pin(async_helper(room, event))
 }
 
 async fn login(
@@ -26,20 +44,7 @@ async fn login(
         .disable_ssl_verification();
     let mut client = AsyncClient::new_with_config(&homeserver_url, None, client_config).unwrap();
 
-    client.add_event_callback(EventType::RoomMessage, Box::new(callback));
-
-    // client.add_event_callback(EventType::RoomMessage, Box::new(|event| {
-    //     Box::pin(async {
-    //         if let RoomEvent::RoomMessage(MessageEvent {
-    //             content: MessageEventContent::Text(TextMessageEventContent { body: msg_body, .. }),
-    //             sender,
-    //             ..
-    //         }) = event
-    //         {
-    //             println!("{}: {}", sender, msg_body);
-    //         }
-    //     })
-    // }));
+    client.add_event_future(EventType::RoomMessage, Box::new(async_callback));
 
     client.login(username, password, None).await?;
     let response = client.sync(SyncSettings::new()).await?;

src/async_client.rs

@@ -1,6 +1,8 @@
 use std::convert::{TryFrom, TryInto};
 use std::future::Future;
 use std::pin::Pin;
+use std::rc::Rc;
+use std::cell::RefCell;
 
 use http::Method as HttpMethod;
 use http::Response as HttpResponse;
@@ -20,6 +22,7 @@ use crate::error::{Error, InnerError};
 use crate::session::Session;
 
 type RoomEventCallback = Box::<dyn FnMut(&Room, &RoomEvent)>;
+type RoomEventCallbackF = Box::<dyn FnMut(Rc<RefCell<Room>>, Rc<RoomEvent>) -> Pin<Box<dyn Future<Output = ()>>>>;
 
 pub struct AsyncClient {
     /// The URL of the homeserver to connect to.
@@ -30,6 +33,8 @@ pub struct AsyncClient {
     base_client: BaseClient,
     /// Event callbacks
     event_callbacks: Vec<RoomEventCallback>,
+    /// Event futures
+    event_futures: Vec<RoomEventCallbackF>,
 }
 
 #[derive(Default, Debug)]
@@ -162,6 +167,7 @@ impl AsyncClient {
             http_client,
             base_client: BaseClient::new(session),
             event_callbacks: Vec::new(),
+            event_futures: Vec::new(),
         })
     }
 
@@ -173,6 +179,14 @@ impl AsyncClient {
         self.event_callbacks.push(callback);
     }
 
+    pub fn add_event_future(
+        &mut self,
+        event_type: EventType,
+        callback: RoomEventCallbackF,
+    ) {
+        self.event_futures.push(callback);
+    }
+
     pub async fn login<S: Into<String>>(
         &mut self,
         user: S,
@@ -232,13 +246,12 @@ impl AsyncClient {
                 let room = self.base_client.joined_rooms.get(&room_id).unwrap();
 
                 for mut cb in &mut self.event_callbacks {
-                    cb(&room, &event);
+                    cb(&room.borrow(), &event);
                 }
 
-                // if self.event_callbacks.contains_key(&event_type) {
-                //     let cb = self.event_callbacks.get_mut(&event_type).unwrap();
-                //     cb(&event).await;
-                // }
+                for mut cb in &mut self.event_futures {
+                    cb(room.clone(), Rc::new(event.clone())).await;
+                }
             }
         }
 

src/base_client.rs

@@ -4,6 +4,8 @@ use crate::api::r0 as api;
 use crate::events::collections::all::{RoomEvent, StateEvent};
 use crate::events::room::member::{MemberEvent, MembershipState};
 use crate::session::Session;
+use std::rc::Rc;
+use std::cell::RefCell;
 
 pub type Token = String;
 pub type RoomId = String;
@@ -13,30 +15,30 @@ pub type UserId = String;
 /// A Matrix room member.
 pub struct RoomMember {
     /// The unique mxid of the user.
-    user_id: UserId,
+    pub user_id: UserId,
     /// The human readable name of the user.
-    display_name: Option<String>,
+    pub display_name: Option<String>,
     /// The matrix url of the users avatar.
-    avatar_url: Option<String>,
+    pub avatar_url: Option<String>,
     /// The users power level.
-    power_level: u8,
+    pub power_level: u8,
 }
 
 #[derive(Debug)]
 /// A Matrix rooom.
 pub struct Room {
     /// The unique id of the room.
-    room_id: RoomId,
+    pub room_id: RoomId,
     /// The mxid of our own user.
-    own_user_id: UserId,
+    pub own_user_id: UserId,
     /// The mxid of the room creator.
-    creator: Option<UserId>,
+    pub creator: Option<UserId>,
     /// The map of room members.
-    members: HashMap<UserId, RoomMember>,
+    pub members: HashMap<UserId, RoomMember>,
     /// A list of users that are currently typing.
-    typing_users: Vec<UserId>,
+    pub typing_users: Vec<UserId>,
     /// A flag indicating if the room is encrypted.
-    encrypted: bool,
+    pub encrypted: bool,
 }
 
 impl Room {
@@ -148,7 +150,7 @@ pub struct Client {
     /// The current sync token that should be used for the next sync call.
     pub sync_token: Option<Token>,
     /// A map of the rooms our user is joined in.
-    pub joined_rooms: HashMap<RoomId, Room>,
+    pub joined_rooms: HashMap<RoomId, Rc<RefCell<Room>>>,
 }
 
 impl Client {
@@ -184,18 +186,19 @@ impl Client {
         self.session = Some(session);
     }
 
-    fn get_or_create_room(&mut self, room_id: &RoomId) -> &mut Room {
+    fn get_or_create_room(&mut self, room_id: &RoomId) -> &mut Rc<RefCell<Room>> {
         self.joined_rooms
             .entry(room_id.to_string())
-            .or_insert(Room::new(
-                room_id,
-                &self
-                    .session
-                    .as_ref()
-                    .expect("Receiving events while not being logged in")
-                    .user_id
-                    .to_string(),
-            ))
+            .or_insert(
+                Rc::new(RefCell::new(Room::new(
+                    room_id,
+                    &self
+                        .session
+                        .as_ref()
+                        .expect("Receiving events while not being logged in")
+                        .user_id
+                        .to_string(),
+            ))))
     }
 
     /// Receive a timeline event for a joined room and update the client state.
@@ -207,7 +210,7 @@ impl Client {
     /// Returns true if the membership list of the room changed, false
     /// otherwise.
     pub fn receive_joined_timeline_event(&mut self, room_id: &RoomId, event: &RoomEvent) -> bool {
-        let mut room = self.get_or_create_room(room_id);
+        let mut room = self.get_or_create_room(room_id).borrow_mut();
         room.receive_timeline_event(event)
     }
 
@@ -220,7 +223,7 @@ impl Client {
     /// Returns true if the membership list of the room changed, false
     /// otherwise.
     pub fn receive_joined_state_event(&mut self, room_id: &RoomId, event: &StateEvent) -> bool {
-        let mut room = self.get_or_create_room(room_id);
+        let mut room = self.get_or_create_room(room_id).borrow_mut();
         room.receive_state_event(event)
     }
 }