widget: change signature of the ClientApi

Return `Self` and the channel to process incoming actions.

Author: daniel-abramov

crates/matrix-sdk/src/widget/client/mod.rs

@@ -23,6 +23,7 @@ use ruma::{
     OwnedEventId,
 };
 use serde_json::Value as JsonValue;
+use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver};
 
 use super::Permissions;
 
@@ -31,16 +32,18 @@ pub struct ClientApi;
 
 impl ClientApi {
     /// Creates a new instance of a client widget API state machine.
-    pub fn new() -> Self {
-        Self
+    /// Returns the client api handler as well as the channel to receive
+    /// actions (commands) from the client.
+    pub fn new() -> (Self, UnboundedReceiver<Action>) {
+        let (_tx, rx) = unbounded_channel();
+        (Self, rx)
     }
 
     /// Processes an incoming event (an incoming raw message from a widget,
     /// or a data produced as a result of a previously sent `Action`).
     /// Produceses a list of actions that the client must perform.
-    pub fn process(&mut self, _event: Event) -> Vec<Action> {
+    pub fn process(&mut self, _event: Event) {
         // TODO: Process the event.
-        Vec::new()
     }
 }
 

crates/matrix-sdk/src/widget/mod.rs

@@ -119,75 +119,78 @@ impl WidgetDriver {
         room: Room,
         permissions_provider: impl PermissionsProvider,
     ) -> Result<(), ()> {
+        let (mut client_api, mut actions) = ClientApi::new();
+
         // Create a channel so that we can conveniently send all events to it.
         let (events_tx, mut events_rx) = unbounded_channel();
 
-        // `from.map(|m| Ok(Event::MessageFromWidget(msg)).forward(events_tx)`,
-        // but apparently `UnboundedSender<T>` does not implement `Sink<T>`.
+        // Forward all of the incoming messages from the widget to the `events_tx`.
         let tx = events_tx.clone();
         tokio::spawn(async move {
             while let Ok(msg) = self.from_widget_rx.recv().await {
                 let _ = tx.send(Event::MessageFromWidget(msg));
             }
         });
 
-        // Our "state" (everything we need to process events).
-        let (mut client_api, matrix) = (ClientApi::new(), MatrixDriver::new(room));
-        let mut event_forwarding_guard: Option<DropGuard> = None;
+        // Forward all of the incoming events to the `ClientApi` implementation.
+        tokio::spawn(async move {
+            while let Some(event) = events_rx.recv().await {
+                client_api.process(event);
+            }
+        });
 
-        // Process events by passing them to the `ClientApi` implementation.
-        while let Some(event) = events_rx.recv().await {
-            for action in client_api.process(event) {
-                match action {
-                    Action::SendToWidget(msg) => {
-                        self.to_widget_tx.send(msg).await.map_err(|_| ())?
-                    }
-                    Action::AcquirePermissions(cmd) => {
-                        let obtained = permissions_provider.acquire_permissions(cmd.clone()).await;
-                        let event = Event::PermissionsAcquired(cmd.ok(obtained));
-                        events_tx.send(event).map_err(|_| ())?;
-                    }
-                    Action::GetOpenId(cmd) => {
-                        let result = cmd.result(matrix.get_open_id().await);
-                        events_tx.send(Event::OpenIdReceived(result)).map_err(|_| ())?;
-                    }
-                    Action::ReadMatrixEvent(cmd) => {
-                        let matrix_events = matrix.read(cmd.event_type.clone(), cmd.limit).await;
-                        let event = Event::MatrixEventRead(cmd.result(matrix_events));
-                        events_tx.send(event).map_err(|_| ())?;
-                    }
-                    Action::SendMatrixEvent(cmd) => {
-                        let SendEventCommand { event_type, state_key, content } = cmd.clone();
-                        let matrix_event_id = matrix.send(event_type, state_key, content).await;
-                        let event = Event::MatrixEventSent(cmd.result(matrix_event_id));
-                        events_tx.send(event).map_err(|_| ())?;
-                    }
-                    Action::Subscribe => {
-                        // Only subscribe if we are not already subscribed.
-                        if event_forwarding_guard.is_none() {
-                            let (stop_forwarding, guard) = {
-                                let token = CancellationToken::new();
-                                (token.child_token(), token.drop_guard())
-                            };
-
-                            event_forwarding_guard = Some(guard);
-                            let (mut matrix, events_tx) = (matrix.events(), events_tx.clone());
-                            tokio::spawn(async move {
-                                loop {
-                                    tokio::select! {
-                                        _ = stop_forwarding.cancelled() => { return }
-                                        Some(event) = matrix.recv() => {
-                                            let _ = events_tx.send(Event::MatrixEventReceived(event));
-                                        }
+        // Process events that we receive **from** the client api implementation,
+        // i.e. the commands (actions) that the client sends to us.
+        let matrix = MatrixDriver::new(room);
+        let mut event_forwarding_guard: Option<DropGuard> = None;
+        while let Some(action) = actions.recv().await {
+            match action {
+                Action::SendToWidget(msg) => self.to_widget_tx.send(msg).await.map_err(|_| ())?,
+                Action::AcquirePermissions(cmd) => {
+                    let obtained = permissions_provider.acquire_permissions(cmd.clone()).await;
+                    let event = Event::PermissionsAcquired(cmd.ok(obtained));
+                    events_tx.send(event).map_err(|_| ())?;
+                }
+                Action::GetOpenId(cmd) => {
+                    let result = cmd.result(matrix.get_open_id().await);
+                    events_tx.send(Event::OpenIdReceived(result)).map_err(|_| ())?;
+                }
+                Action::ReadMatrixEvent(cmd) => {
+                    let matrix_events = matrix.read(cmd.event_type.clone(), cmd.limit).await;
+                    let event = Event::MatrixEventRead(cmd.result(matrix_events));
+                    events_tx.send(event).map_err(|_| ())?;
+                }
+                Action::SendMatrixEvent(cmd) => {
+                    let SendEventCommand { event_type, state_key, content } = cmd.clone();
+                    let matrix_event_id = matrix.send(event_type, state_key, content).await;
+                    let event = Event::MatrixEventSent(cmd.result(matrix_event_id));
+                    events_tx.send(event).map_err(|_| ())?;
+                }
+                Action::Subscribe => {
+                    // Only subscribe if we are not already subscribed.
+                    if event_forwarding_guard.is_none() {
+                        let (stop_forwarding, guard) = {
+                            let token = CancellationToken::new();
+                            (token.child_token(), token.drop_guard())
+                        };
+
+                        event_forwarding_guard = Some(guard);
+                        let (mut matrix, events_tx) = (matrix.events(), events_tx.clone());
+                        tokio::spawn(async move {
+                            loop {
+                                tokio::select! {
+                                    _ = stop_forwarding.cancelled() => { return }
+                                    Some(event) = matrix.recv() => {
+                                        let _ = events_tx.send(Event::MatrixEventReceived(event));
                                     }
                                 }
-                            });
-                        }
-                    }
-                    Action::Unsubscribe => {
-                        event_forwarding_guard = None;
+                            }
+                        });
                     }
                 }
+                Action::Unsubscribe => {
+                    event_forwarding_guard = None;
+                }
             }
         }