[WIP] Fetch events from Matrix

src/app.rs

@@ -1,5 +1,26 @@
+/*
+ * Copyright (C) 2023  Valentin Lorentz
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License version 3,
+ * as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+use std::cell::RefCell;
+use std::sync::atomic::{AtomicBool, Ordering};
+
 use color_eyre::eyre::{Result, WrapErr};
 use crossterm::event::KeyEvent;
+use futures::stream::FuturesUnordered;
+use futures::StreamExt;
 use nonempty::NonEmpty;
 use ratatui::prelude::Rect;
 use serde::{Deserialize, Serialize};
@@ -21,10 +42,10 @@ pub struct App {
   pub tick_rate: f64,
   pub frame_rate: f64,
   pub components: Vec<Box<dyn Component>>,
-  pub should_quit: bool,
+  pub should_quit: AtomicBool,
   pub should_suspend: bool,
   pub mode: Mode,
-  pub last_tick_key_events: Vec<KeyEvent>,
+  pub last_tick_key_events: RefCell<Vec<KeyEvent>>,
 }
 
 impl App {
@@ -64,13 +85,13 @@ impl App {
       tick_rate,
       frame_rate,
       components: vec![Box::new(home), Box::new(fps)],
-      should_quit: false,
+      should_quit: AtomicBool::new(false),
       should_suspend: false,
       clients,
       config,
       commands: RataCommands::new(),
       mode,
-      last_tick_key_events: Vec::new(),
+      last_tick_key_events: RefCell::new(Vec::new()),
     };
     crate::plugins::load_all(&mut app).context("Could not load plugins")?;
     Ok(app)
@@ -78,6 +99,7 @@ impl App {
 
   pub async fn run(&mut self) -> Result<()> {
     let (action_tx, mut action_rx) = mpsc::unbounded_channel();
+    let (sync_responses_tx, mut sync_responses_rx) = mpsc::unbounded_channel();
 
     let mut tui = tui::Tui::new()?
       .tick_rate(self.tick_rate)
@@ -97,36 +119,55 @@ impl App {
       component.init(tui.size()?)?;
     }
 
-    loop {
+    let sync_clients = self.clients.clone();
-      if let Some(e) = tui.next().await {
+    let mut sync_results = FuturesUnordered::new();
-        match e {
+    for client in sync_clients.iter() {
-          tui::Event::Quit => action_tx.send(Action::Quit)?,
+      client.add_event_handler(|ev: matrix_sdk::ruma::events::room::message::SyncRoomMessageEvent| async move {
-          tui::Event::Tick => action_tx.send(Action::Tick)?,
+        println!("Received a message {:?}", ev);
-          tui::Event::Render => action_tx.send(Action::Render)?,
+      });
-          tui::Event::Resize(x, y) => action_tx.send(Action::Resize(x, y))?,
+      let server_name = client.user_id().expect("missing user id").server_name();
-          tui::Event::Key(key) => {
+      let mut sync_settings = matrix_sdk::config::SyncSettings::default();
-            if let Some(keymap) = self.config.keybindings.get(&self.mode) {
+      if server_name == "matrix.org" {
-              if let Some(command_line) = keymap.get(&vec![key]) {
+        // matrix.org is slow and frequently hits the timeout on initial sync,
-                log::info!("Got command: {command_line}");
+        // let's quadruple the default
-                crate::commands::run_command(command_line, &self, &action_tx)?;
+        sync_settings = sync_settings.timeout(std::time::Duration::from_secs(120));
-              } else {
+      }
-                // If the key was not handled as a single key action,
+      sync_results.push(client.sync_with_result_callback(
-                // then consider it for multi-key combinations.
+        sync_settings,
-                self.last_tick_key_events.push(key);
+        |res| {
+          let client2 = client.clone();
+          async {
+            let sync_response = res.expect("Failed sync");
+            println!("sync response {:?}", sync_response);
+            sync_responses_tx.send((client2, sync_response)).expect("could not send sync response");
+            if self.should_quit.load(Ordering::Acquire) {
+              Ok(matrix_sdk::LoopCtrl::Break)
+            } else {
+              Ok(matrix_sdk::LoopCtrl::Continue)
+            }
+          }
+        },
+      ));
+    }
 
-                // Check for multi-key combinations
+    loop {
-                if let Some(command_line) = keymap.get(&self.last_tick_key_events) {
+      tokio::select! {
-                  log::info!("Got command: {command_line}");
+        e = tui.next() => {
-                  crate::commands::run_command(command_line, &self, &action_tx)?;
+          if let Some(e) = e {
-                }
+            self.handle_tui_event(&action_tx, e.clone())?;
+            for component in self.components.iter_mut() {
+              if let Some(action) = component.handle_events(Some(e.clone()))? {
+                action_tx.send(action)?;
               }
-            };
+            }
-          },
+          }
-          _ => {},
         }
-        for component in self.components.iter_mut() {
+        sync_response = sync_responses_rx.recv() => {
-          if let Some(action) = component.handle_events(Some(e.clone()))? {
+          todo!("handle {:?}", sync_response);
-            action_tx.send(action)?;
+        }
+        sync_result = sync_results.next() => {
+          if !self.should_quit.load(Ordering::Acquire) {
+            panic!("Sync ended unexpected: {:?}", sync_result);
           }
         }
       }
@@ -137,9 +178,9 @@ impl App {
         }
         match action {
           Action::Tick => {
-            self.last_tick_key_events.drain(..);
+            self.last_tick_key_events.borrow_mut().drain(..);
           },
-          Action::Quit => self.should_quit = true,
+          Action::Quit => self.should_quit.store(true, Ordering::Release),
           Action::Suspend => self.should_suspend = true,
           Action::Resume => self.should_suspend = false,
           Action::Resize(w, h) => {
@@ -183,7 +224,7 @@ impl App {
           .frame_rate(self.frame_rate);
         // tui.mouse(true);
         tui.enter()?;
-      } else if self.should_quit {
+      } else if self.should_quit.load(Ordering::Acquire) {
         tui.stop()?;
         break;
       }
@@ -191,4 +232,40 @@ impl App {
     tui.exit()?;
     Ok(())
   }
+
+  fn handle_tui_event(
+    &self,
+    action_tx: &mpsc::UnboundedSender<Action>,
+    e: tui::Event,
+  ) -> Result<()> {
+    match e {
+      tui::Event::Quit => action_tx.send(Action::Quit)?,
+      tui::Event::Tick => action_tx.send(Action::Tick)?,
+      tui::Event::Render => action_tx.send(Action::Render)?,
+      tui::Event::Resize(x, y) => action_tx.send(Action::Resize(x, y))?,
+      tui::Event::Key(key) => {
+        if let Some(keymap) = self.config.keybindings.get(&self.mode) {
+          if let Some(command_line) = keymap.get(&vec![key]) {
+            log::info!("Got command: {command_line}");
+            crate::commands::run_command(command_line, &self, &action_tx)?;
+          } else {
+            let mut last_tick_key_events = self.last_tick_key_events.borrow_mut();
+
+            // If the key was not handled as a single key action,
+            // then consider it for multi-key combinations.
+            last_tick_key_events.push(key);
+
+            // Check for multi-key combinations
+            if let Some(command_line) = keymap.get(&*last_tick_key_events) {
+              log::info!("Got command: {command_line}");
+              crate::commands::run_command(command_line, &self, &action_tx)?;
+            }
+          }
+        };
+      },
+      _ => {},
+    }
+
+    Ok(())
+  }
 }