From fbaf25dc6b381f0797fc4d38e6d14700739143de Mon Sep 17 00:00:00 2001 From: Jonathan Kelley Date: Mon, 21 Feb 2022 23:34:18 -0500 Subject: [PATCH] feat: pair down usecoroutine and make it much happier --- packages/hooks/src/usecoroutine.rs | 263 +++++++++++++---------------- 1 file changed, 113 insertions(+), 150 deletions(-) diff --git a/packages/hooks/src/usecoroutine.rs b/packages/hooks/src/usecoroutine.rs index 40a12443d..b4b2907ff 100644 --- a/packages/hooks/src/usecoroutine.rs +++ b/packages/hooks/src/usecoroutine.rs @@ -1,160 +1,102 @@ -#![warn(clippy::pedantic)] - -use dioxus_core::exports::bumpalo; -use dioxus_core::{LazyNodes, ScopeState, TaskId}; -use futures_channel::mpsc::{UnboundedReceiver, UnboundedSender}; -use std::any::Any; +use dioxus_core::{ScopeState, TaskId}; +pub use futures_channel::mpsc::{UnboundedReceiver, UnboundedSender}; use std::future::Future; -use std::{cell::Cell, rc::Rc}; +use std::rc::Rc; /// Maintain a handle over a future that can be paused, resumed, and canceled. /// -/// This is an upgraded form of [`use_future`] with lots of bells-and-whistles. +/// This is an upgraded form of [`use_future`] with an integrated channel system. +/// Specifically, the coroutine generated here comes with an [`UnboundedChannel`] +/// built into it - saving you the hassle of building your own. /// -/// [`use_coroutine`] is well suited for long-running tasks and is very customizable. -/// -/// -/// ## Long running tasks -/// -/// -/// -/// ## One-off tasks -/// -/// -/// ## Cancellation +/// Addititionally, coroutines are automatically injected as shared contexts, so +/// downstream components can tap into a coroutine's channel and send messages +/// into a singular async event loop. /// +/// This makes it effective for apps that need to interact with an event loop or +/// some asynchronous code without thinking too hard about state. /// /// ## Global State -#[allow(clippy::mut_from_ref)] -pub fn use_coroutine(cx: &ScopeState) -> &mut UseCoroutine { +/// +/// Typically, writing apps that handle concurrency properly can be difficult, +/// so the intention of this hook is to make it easy to join and poll async tasks +/// concurrently in a centralized place. You'll find that you can have much better +/// control over your app's state if you centralize your async actions, even under +/// the same concurrent context. This makes it easier to prevent undeseriable +/// states in your UI while various async tasks are already running. +/// +/// This hook is especially powerful when combined with Fermi. We can store important +/// global data in a coroutine, and then access display-level values from the rest +/// of our app through atoms. +/// +/// ## UseCallback instead +/// +/// However, you must plan out your own concurrency and synchronization. If you +/// don't care about actions in your app being synchronized, you can use [`use_callback`] +/// hook to spawn multiple tasks and run them concurrently. +/// +/// ## Example +/// +/// ```rust, ignore +/// enum Action { +/// Start, +/// Stop, +/// } +/// +/// let chat_client = use_coroutine(&cx, |rx: UnboundedReceiver| async move { +/// while let Some(action) = rx.next().await { +/// match action { +/// Action::Start => {} +/// Action::Stop => {}, +/// } +/// } +/// }); +/// +/// +/// cx.render(rsx!{ +/// button { +/// onclick: move |_| chat_client.send(Action::Start), +/// "Start Chat Service" +/// } +/// }) +/// ``` +pub fn use_coroutine(cx: &ScopeState, init: G) -> &CoroutineHandle +where + M: 'static, + G: FnOnce(UnboundedReceiver) -> F, + F: Future + 'static, +{ cx.use_hook(|_| { - // - UseCoroutine { - val: Cell::new(None), - rx: Cell::new(None), - tx: None, - first_run: true, - deps: vec![], - dep_cnt: 0, - needs_regen: false, - auto_start: true, - } + let (tx, rx) = futures_channel::mpsc::unbounded(); + let task = cx.push_future(init(rx)); + cx.provide_context(CoroutineHandle { tx, task }) }) } -pub struct UseCoroutine { - val: Cell>, - rx: Cell>>, - tx: Option>, - first_run: bool, - deps: Vec>, - dep_cnt: usize, - needs_regen: bool, - auto_start: bool, +/// Get a handle to a coroutine higher in the tree +/// +/// See the docs for [`use_coroutine`] for more details. +pub fn use_coroutine_handle(cx: &ScopeState) -> Option<&Rc>> { + cx.use_hook(|_| cx.consume_context::>()) + .as_ref() } -pub enum FutureState<'a, T> { - Pending, - Complete(&'a T), - Regenerating(&'a T), // the old value -} - -impl UseCoroutine { - /// explicitly set the type of the channel used by the coroutine - fn with_channel(&mut self) -> &mut UseCoroutine { - if self.first_run { - // self.provide_context() - } - todo!() - } - - /// explicitly set the type of the channel used by the coroutine - fn with_channel_isolate(&mut self) -> &mut UseCoroutine { - todo!() - } -} - -impl UseCoroutine { - pub fn is_running(&self) -> bool { - false - // self.running.get() - } - - pub fn start(&self) { - // if !self.is_running() { - // if let Some(mut fut) = self.create_fut.take() { - // let fut = fut(); - // let ready_handle = self.running.clone(); - - // let task = self.cx.push_future(async move { - // ready_handle.set(true); - // fut.await; - // ready_handle.set(false); - // }); - - // self.task_id.set(Some(task)); - // } - // } - } - - pub fn send(&self, msg: M) { - if let Some(tx) = self.tx.clone() { - if tx.unbounded_send(msg).is_err() { - log::error!("Failed to send message"); - } - } - } - - // todo: wire these up, either into the task system or into the coroutine system itself - // we would have change how we poll the coroutine and how its awaken - - fn build>(&mut self, f: impl FnOnce(UnboundedReceiver) -> F) -> &Self { - self.first_run = false; - if self.auto_start || self.needs_regen { - // - } - - self - } - - pub fn auto_start(mut self, start: bool) -> Self { - // if start && self.run_count.get() == 1 { - // self.start(); - // } - self - } - - /// Add this value to the dependency list - /// - /// This is a hook and should be called during the initial hook process. - /// It should •not• be called in a conditional. - pub fn with_dep(&mut self, dependency: &F) -> &mut Self { - if let Some(dep) = self.deps.get_mut(self.dep_cnt) { - if let Some(saved_dep) = dep.downcast_mut::() { - if dependency != saved_dep { - *saved_dep = dependency.clone(); - self.needs_regen = true; - } - }; - } else { - self.deps.push(Box::new(dependency.to_owned())); - self.needs_regen = true; - } - - self - } - - pub fn restart_if(&self, f: impl FnOnce() -> bool) -> &Self { - self - } - - // pub fn resume(&self) {} - // pub fn stop(&self) {} - // pub fn restart(&self) {} -} - -pub struct CoroutineContext { +pub struct CoroutineHandle { tx: UnboundedSender, + task: TaskId, +} + +impl CoroutineHandle { + /// Get the ID of this coroutine + #[must_use] + pub fn task_id(&self) -> TaskId { + self.task + } + + /// Send a message to the coroutine + pub fn send(&self, msg: T) { + let _ = self.tx.unbounded_send(msg); + } } #[cfg(test)] @@ -167,15 +109,36 @@ mod tests { use futures_util::StreamExt; fn app(cx: Scope, name: String) -> Element { - let task = use_coroutine(&cx) - .with_dep(&name) - .with_channel::() - .build(|mut rx| async move { - while let Some(msg) = rx.next().await { - println!("got message: {}", msg); - } - }); + let task = use_coroutine(&cx, |mut rx: UnboundedReceiver| async move { + while let Some(msg) = rx.next().await { + println!("got message: {}", msg); + } + }); + + let task2 = use_coroutine(&cx, view_task); + + let task3 = use_coroutine(&cx, |rx| complex_task(rx, 10)); None } + + async fn view_task(mut rx: UnboundedReceiver) { + while let Some(msg) = rx.next().await { + println!("got message: {}", msg); + } + } + + enum Actions { + CloseAll, + OpenAll, + } + + async fn complex_task(mut rx: UnboundedReceiver, name: i32) { + while let Some(msg) = rx.next().await { + match msg { + Actions::CloseAll => todo!(), + Actions::OpenAll => todo!(), + } + } + } }