feat: configurable keyboard shortcuts (#185)

This adds the ability to provide custom keybinds in your config.toml.
Examples of the default config.toml and a config.toml with vim remapped keybinds are added to ./youtui/config/
A large refactoring of the action handling system was required to implement this.

async-callback-manager/examples/ratatui_example.rs

@@ -2,7 +2,7 @@
 #![allow(clippy::unwrap_used)]
 
 use async_callback_manager::{
-    AsyncCallbackManager, AsyncCallbackSender, BackendStreamingTask, BackendTask,
+    AsyncCallbackManager, AsyncTask, BackendStreamingTask, BackendTask, TaskOutcome,
 };
 use crossterm::event::{Event, EventStream, KeyCode, KeyEvent, KeyEventKind};
 use futures::{stream, FutureExt};
@@ -47,7 +47,6 @@ struct State {
     word: String,
     number: String,
     mode: Mode,
-    callback_handle: AsyncCallbackSender<reqwest::Client, Self, ()>,
 }
 impl State {
     fn draw(&self, f: &mut Frame) {
@@ -72,25 +71,21 @@ impl State {
     fn handle_toggle_mode(&mut self) {
         self.mode = self.mode.toggle()
     }
-    async fn handle_get_word(&mut self) {
+    async fn handle_get_word(&mut self) -> AsyncTask<Self, reqwest::Client, ()> {
         self.word = "Loading".to_string();
-        self.callback_handle
-            .add_callback(
-                GetWordRequest,
-                |state, word| state.word = word,
-                (&self.mode).into(),
-            )
-            .unwrap()
+        AsyncTask::new_future(
+            GetWordRequest,
+            |state: &mut Self, word| state.word = word,
+            (&self.mode).into(),
+        )
     }
-    async fn handle_start_counter(&mut self) {
+    async fn handle_start_counter(&mut self) -> AsyncTask<Self, reqwest::Client, ()> {
         self.number = "Loading".to_string();
-        self.callback_handle
-            .add_stream_callback(
-                CounterStream,
-                |state, num| state.number = num,
-                (&self.mode).into(),
-            )
-            .unwrap()
+        AsyncTask::new_stream(
+            CounterStream,
+            |state: &mut Self, num| state.number = num,
+            (&self.mode).into(),
+        )
     }
 }
 
@@ -103,27 +98,35 @@ async fn main() {
     let mut state = State {
         word: String::new(),
         number: String::new(),
-        callback_handle: manager.new_sender(50),
         mode: Default::default(),
     };
     loop {
         terminal.draw(|f| state.draw(f)).unwrap();
         tokio::select! {
             Some(action) = events.next() => match action {
                 Action::Quit => break,
-                Action::GetWord => state.handle_get_word().await,
-                Action::StartCounter => state.handle_start_counter().await,
+                Action::GetWord => {
+                    manager.spawn_task(&backend,
+                    state.handle_get_word().await)
+                },
+                Action::StartCounter => {
+                    manager.spawn_task(&backend,
+                    state.handle_start_counter().await)
+                },
                 Action::ToggleMode => state.handle_toggle_mode(),
             },
-            Some(manager_event) = manager.manage_next_event(&backend) => if manager_event.is_spawned_task() {
-                continue
+            Some(outcome) = manager.get_next_response() => match outcome {
+                TaskOutcome::StreamClosed => continue,
+                TaskOutcome::TaskPanicked {error,..} => std::panic::resume_unwind(error.into_panic()),
+                TaskOutcome::MutationReceived { mutation, ..} =>
+                    manager.spawn_task(&backend, mutation(&mut state)),
             },
-            mutations = state.callback_handle.get_next_mutations(10) => mutations.apply(&mut state),
         };
     }
     ratatui::restore();
 }
 
+#[derive(Debug)]
 struct GetWordRequest;
 impl BackendTask<reqwest::Client> for GetWordRequest {
     type MetadataType = ();
@@ -146,6 +149,7 @@ impl BackendTask<reqwest::Client> for GetWordRequest {
     }
 }
 
+#[derive(Debug)]
 struct CounterStream;
 impl<T> BackendStreamingTask<T> for CounterStream {
     type Output = String;

async-callback-manager/src/adaptors.rs

@@ -1,6 +1,6 @@
-use crate::{BackendStreamingTask, BackendTask};
+use crate::{BackendStreamingTask, BackendTask, DEFAULT_STREAM_CHANNEL_SIZE};
 use futures::{Stream, StreamExt};
-use std::future::Future;
+use std::{fmt::Debug, future::Future};
 use tokio_stream::wrappers::ReceiverStream;
 
 impl<Bkend, T: BackendTask<Bkend>> BackendTaskExt<Bkend> for T {}
@@ -54,6 +54,19 @@ pub struct Map<T, F> {
     create_next: F,
 }
 
+impl<T, F> Debug for Map<T, F>
+where
+    T: Debug,
+{
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("Map")
+            .field("first", &self.first)
+            // TODO: we could deduce the type name returned by the closure
+            .field("create_next", &"..closure..")
+            .finish()
+    }
+}
+
 impl<Bkend, T, S, F, Ct, O, E> BackendStreamingTask<Bkend> for Map<T, F>
 where
     Bkend: Clone + Sync + Send + 'static,
@@ -73,8 +86,7 @@ where
     ) -> impl Stream<Item = Self::Output> + Send + Unpin + 'static {
         let Map { first, create_next } = self;
         let backend = backend.clone();
-        // TODO: Channel size
-        let (tx, rx) = tokio::sync::mpsc::channel(30);
+        let (tx, rx) = tokio::sync::mpsc::channel(DEFAULT_STREAM_CHANNEL_SIZE);
         tokio::task::spawn(async move {
             let seed = first.into_future(&backend).await;
             match seed {
@@ -149,8 +161,7 @@ where
     ) -> impl Stream<Item = Self::Output> + Send + Unpin + 'static {
         let Then { first, create_next } = self;
         let backend = backend.clone();
-        // TODO: Channel size
-        let (tx, rx) = tokio::sync::mpsc::channel(30);
+        let (tx, rx) = tokio::sync::mpsc::channel(DEFAULT_STREAM_CHANNEL_SIZE);
         tokio::task::spawn(async move {
             let seed = first.into_future(&backend).await;
             let mut stream = create_next(seed).into_stream(&backend);

async-callback-manager/src/lib.rs

@@ -1,20 +1,20 @@
 use futures::Future;
-use futures::FutureExt;
 use futures::Stream;
 use std::any::Any;
-use tokio::sync::oneshot;
 
 mod adaptors;
 mod error;
 mod manager;
-mod sender;
 mod task;
 
 pub use adaptors::*;
 pub use error::*;
 pub use manager::*;
-pub use sender::*;
-pub use task::Constraint;
+pub use task::{AsyncTask, Constraint, TaskOutcome};
+
+// Size of the channel used for each stream task.
+// In future, this could be settable.
+pub(crate) const DEFAULT_STREAM_CHANNEL_SIZE: usize = 20;
 
 pub trait BkendMap<Bkend> {
     fn map(backend: &Bkend) -> &Self;
@@ -50,32 +50,3 @@ pub trait BackendStreamingTask<Bkend>: Send + Any {
         vec![]
     }
 }
-
-struct KillHandle(Option<oneshot::Sender<()>>);
-struct KillSignal(oneshot::Receiver<()>);
-
-impl KillHandle {
-    fn kill(&mut self) -> Result<()> {
-        if let Some(tx) = self.0.take() {
-            return tx.send(()).map_err(|_| Error::ReceiverDropped);
-        }
-        Ok(())
-    }
-}
-impl Future for KillSignal {
-    type Output = Result<()>;
-    fn poll(
-        mut self: std::pin::Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-    ) -> std::task::Poll<Self::Output> {
-        self.0.poll_unpin(cx).map_err(|_| Error::ReceiverDropped)
-    }
-}
-fn kill_channel() -> (KillHandle, KillSignal) {
-    let (tx, rx) = oneshot::channel();
-    (KillHandle(Some(tx)), KillSignal(rx))
-}
-
-type DynFallibleFuture = Box<dyn Future<Output = Result<()>> + Unpin + Send>;
-type DynCallbackFn<Frntend> = Box<dyn FnOnce(&mut Frntend) + Send>;
-type DynBackendTask<Bkend> = Box<dyn FnOnce(&Bkend) -> DynFallibleFuture>;