Merge pull request #161 from DioxusLabs/jk/coroutine-coroutineoverhaul

Overhaul async hooks: use_future, use_coroutine, use_effect (new)

examples/dog_app.rs

@@ -16,7 +16,9 @@ struct ListBreeds {
 }
 
 fn app(cx: Scope) -> Element {
-    let breeds = use_future(&cx, || async move {
+    let (breed, set_breed) = use_state(&cx, || None);
+
+    let breeds = use_future(&cx, (), |_| async move {
         reqwest::get("https://dog.ceo/api/breeds/list/all")
             .await
             .unwrap()
@@ -24,13 +26,10 @@ fn app(cx: Scope) -> Element {
             .await
     });
 
-    let (breed, set_breed) = use_state(&cx, || None);
-
     match breeds.value() {
         Some(Ok(breeds)) => cx.render(rsx! {
             div {
-                h1 {"Select a dog breed!"}
-
+                h1 { "Select a dog breed!" }
                 div { display: "flex",
                     ul { flex: "50%",
                         breeds.message.keys().map(|breed| rsx!(
@@ -51,34 +50,23 @@ fn app(cx: Scope) -> Element {
                 }
             }
         }),
-        Some(Err(_e)) => cx.render(rsx! {
-            div { "Error fetching breeds" }
-        }),
-        None => cx.render(rsx! {
-            div { "Loading dogs..." }
-        }),
+        Some(Err(_e)) => cx.render(rsx! { div { "Error fetching breeds" } }),
+        None => cx.render(rsx! { div { "Loading dogs..." } }),
     }
 }
 
+#[derive(serde::Deserialize, Debug)]
+struct DogApi {
+    message: String,
+}
+
 #[inline_props]
 fn Breed(cx: Scope, breed: String) -> Element {
-    #[derive(serde::Deserialize, Debug)]
-    struct DogApi {
-        message: String,
-    }
-
-    let endpoint = format!("https://dog.ceo/api/breed/{}/images/random", breed);
-
-    let fut = use_future(&cx, || async move {
+    let fut = use_future(&cx, (breed,), |(breed,)| async move {
+        let endpoint = format!("https://dog.ceo/api/breed/{}/images/random", breed);
         reqwest::get(endpoint).await.unwrap().json::<DogApi>().await
     });
 
-    let (name, set_name) = use_state(&cx, || breed.clone());
-    if name != breed {
-        set_name(breed.clone());
-        fut.restart();
-    }
-
     cx.render(match fut.value() {
         Some(Ok(resp)) => rsx! {
             button {

examples/suspense.rs

@@ -35,8 +35,8 @@ fn app(cx: Scope) -> Element {
         div {
             h1 {"Dogs are very important"}
             p {
-                "The dog or domestic dog (Canis familiaris[4][5] or Canis lupus familiaris[5])" 
-                "is a domesticated descendant of the wolf which is characterized by an upturning tail." 
+                "The dog or domestic dog (Canis familiaris[4][5] or Canis lupus familiaris[5])"
+                "is a domesticated descendant of the wolf which is characterized by an upturning tail."
                 "The dog derived from an ancient, extinct wolf,[6][7] and the modern grey wolf is the"
                 "dog's nearest living relative.[8] The dog was the first species to be domesticated,[9][8]"
                 "by hunter–gatherers over 15,000 years ago,[7] before the development of agriculture.[1]"
@@ -52,7 +52,7 @@ fn app(cx: Scope) -> Element {
 /// Suspense is achieved my moving the future into only the component that
 /// actually renders the data.
 fn Doggo(cx: Scope) -> Element {
-    let fut = use_future(&cx, || async move {
+    let fut = use_future(&cx, (), |_| async move {
         reqwest::get("https://dog.ceo/api/breeds/image/random/")
             .await
             .unwrap()

examples/tasks.rs

@@ -12,8 +12,8 @@ fn main() {
 fn app(cx: Scope) -> Element {
     let (count, set_count) = use_state(&cx, || 0);
 
-    use_future(&cx, move || {
-        let set_count = set_count.to_owned();
+    use_future(&cx, (), move |_| {
+        let set_count = set_count.clone();
         async move {
             loop {
                 tokio::time::sleep(Duration::from_millis(1000)).await;

packages/hooks/Cargo.toml

@@ -13,3 +13,10 @@ keywords = ["dom", "ui", "gui", "react", "wasm"]
 
 [dependencies]
 dioxus-core = { path = "../../packages/core", version = "^0.1.9" }
+futures-channel = "0.3.21"
+log = { version = "0.4", features = ["release_max_level_off"] }
+
+
+[dev-dependencies]
+futures-util = { version = "0.3", default-features = false }
+dioxus-core = { path = "../../packages/core", version = "^0.1.9" }

packages/hooks/src/lib.rs

@@ -16,8 +16,8 @@ pub use usecoroutine::*;
 mod usefuture;
 pub use usefuture::*;
 
-mod usesuspense;
-pub use usesuspense::*;
+mod useeffect;
+pub use useeffect::*;
 
 #[macro_export]
 /// A helper macro for using hooks in async environements.

packages/hooks/src/usecoroutine.rs

@@ -1,122 +1,144 @@
 use dioxus_core::{ScopeState, TaskId};
+pub use futures_channel::mpsc::{UnboundedReceiver, UnboundedSender};
 use std::future::Future;
-use std::{cell::Cell, rc::Rc};
-/*
-
-
-
-let g = use_coroutine(&cx, || {
-    // clone the items in
-    async move {
-
-    }
-})
-
-
-
-*/
-pub fn use_coroutine<F>(cx: &ScopeState, create_future: impl FnOnce() -> F) -> CoroutineHandle<'_>
+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 an integrated channel system.
+/// Specifically, the coroutine generated here comes with an [`UnboundedChannel`]
+/// built into it - saving you the hassle of building your own.
+///
+/// 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
+///
+/// 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<Action>| 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<M, G, F>(cx: &ScopeState, init: G) -> &CoroutineHandle<M>
 where
+    M: 'static,
+    G: FnOnce(UnboundedReceiver<M>) -> F,
     F: Future<Output = ()> + 'static,
 {
-    let state = cx.use_hook(move |_| {
-        let f = create_future();
-        let id = cx.push_future(f);
-        State {
-                running: Default::default(),
-                _id: id
-                // pending_fut: Default::default(),
-                // running_fut: Default::default(),
-            }
-    });
+    cx.use_hook(|_| {
+        let (tx, rx) = futures_channel::mpsc::unbounded();
+        let task = cx.push_future(init(rx));
+        cx.provide_context(CoroutineHandle { tx, task })
+    })
+}
 
-    // state.pending_fut.set(Some(Box::pin(f)));
+/// Get a handle to a coroutine higher in the tree
+///
+/// See the docs for [`use_coroutine`] for more details.
+pub fn use_coroutine_handle<M: 'static>(cx: &ScopeState) -> Option<&Rc<CoroutineHandle<M>>> {
+    cx.use_hook(|_| cx.consume_context::<CoroutineHandle<M>>())
+        .as_ref()
+}
 
-    // if let Some(fut) = state.running_fut.as_mut() {
-    //     cx.push_future(fut);
-    // }
+pub struct CoroutineHandle<T> {
+    tx: UnboundedSender<T>,
+    task: TaskId,
+}
 
-    // if let Some(fut) = state.running_fut.take() {
-    // state.running.set(true);
-    // fut.resume();
-    // }
+impl<T> CoroutineHandle<T> {
+    /// Get the ID of this coroutine
+    #[must_use]
+    pub fn task_id(&self) -> TaskId {
+        self.task
+    }
 
-    // let submit: Box<dyn FnOnce() + 'a> = Box::new(move || {
-    //     let g = async move {
-    //         running.set(true);
-    //         create_future().await;
-    //         running.set(false);
-    //     };
-    //     let p: Pin<Box<dyn Future<Output = ()>>> = Box::pin(g);
-    //     fut_slot
-    //         .borrow_mut()
-    //         .replace(unsafe { std::mem::transmute(p) });
-    // });
+    /// Send a message to the coroutine
+    pub fn send(&self, msg: T) {
+        let _ = self.tx.unbounded_send(msg);
+    }
+}
 
-    // let submit = unsafe { std::mem::transmute(submit) };
-    // state.submit.get_mut().replace(submit);
+#[cfg(test)]
+mod tests {
+    #![allow(unused)]
 
-    // if state.running.get() {
-    //     // let mut fut = state.fut.borrow_mut();
-    //     // cx.push_task(|| fut.as_mut().unwrap().as_mut());
-    // } else {
-    //     // make sure to drop the old future
-    //     if let Some(fut) = state.fut.borrow_mut().take() {
-    //         drop(fut);
-    //     }
-    // }
-    CoroutineHandle { cx, inner: state }
-}
+    use super::*;
+    use dioxus_core::exports::futures_channel::mpsc::unbounded;
+    use dioxus_core::prelude::*;
+    use futures_util::StreamExt;
 
-struct State {
-    running: Rc<Cell<bool>>,
-    _id: TaskId,
-    // the way this is structure, you can toggle the coroutine without re-rendering the comppnent
-    // this means every render *generates* the future, which is a bit of a waste
-    // todo: allocate pending futures in the bump allocator and then have a true promotion
-    // pending_fut: Cell<Option<Pin<Box<dyn Future<Output = ()> + 'static>>>>,
-    // running_fut: Option<Pin<Box<dyn Future<Output = ()> + 'static>>>,
-    // running_fut: Rc<RefCell<Option<Pin<Box<dyn Future<Output = ()> + 'static>>>>>
-}
+    fn app(cx: Scope, name: String) -> Element {
+        let task = use_coroutine(&cx, |mut rx: UnboundedReceiver<i32>| async move {
+            while let Some(msg) = rx.next().await {
+                println!("got message: {}", msg);
+            }
+        });
 
-pub struct CoroutineHandle<'a> {
-    cx: &'a ScopeState,
-    inner: &'a State,
-}
+        let task2 = use_coroutine(&cx, view_task);
 
-impl Clone for CoroutineHandle<'_> {
-    fn clone(&self) -> Self {
-        CoroutineHandle {
-            cx: self.cx,
-            inner: self.inner,
-        }
+        let task3 = use_coroutine(&cx, |rx| complex_task(rx, 10));
+
+        None
     }
-}
-impl Copy for CoroutineHandle<'_> {}
 
-impl<'a> CoroutineHandle<'a> {
-    #[allow(clippy::needless_return)]
-    pub fn start(&self) {
-        if self.is_running() {
-            return;
+    async fn view_task(mut rx: UnboundedReceiver<i32>) {
+        while let Some(msg) = rx.next().await {
+            println!("got message: {}", msg);
         }
-
-        // if let Some(submit) = self.inner.pending_fut.take() {
-        // submit();
-        // let inner = self.inner;
-        // self.cx.push_task(submit());
-        // }
     }
 
-    pub fn is_running(&self) -> bool {
-        self.inner.running.get()
+    enum Actions {
+        CloseAll,
+        OpenAll,
     }
 
-    pub fn resume(&self) {
-        // self.cx.push_task(fut)
+    async fn complex_task(mut rx: UnboundedReceiver<Actions>, name: i32) {
+        while let Some(msg) = rx.next().await {
+            match msg {
+                Actions::CloseAll => todo!(),
+                Actions::OpenAll => todo!(),
+            }
+        }
     }
-
-    pub fn stop(&self) {}
-
-    pub fn restart(&self) {}
 }

packages/hooks/src/useeffect.rs

@@ -0,0 +1,92 @@
+use dioxus_core::{ScopeState, TaskId};
+use std::{any::Any, cell::Cell, future::Future};
+
+use crate::UseFutureDep;
+
+/// A hook that provides a future that executes after the hooks have been applied
+///
+/// Whenever the hooks dependencies change, the future will be re-evaluated.
+/// If a future is pending when the dependencies change, the previous future
+/// will be allowed to continue
+///
+/// - dependencies: a tuple of references to values that are PartialEq + Clone
+///
+/// ## Examples
+///
+/// ```rust, ignore
+///
+/// #[inline_props]
+/// fn app(cx: Scope, name: &str) -> Element {
+///     use_effect(&cx, (name,), |(name,)| async move {
+///         set_title(name);
+///     }))
+/// }
+/// ```
+pub fn use_effect<T, F, D>(cx: &ScopeState, dependencies: D, future: impl FnOnce(D::Out) -> F)
+where
+    T: 'static,
+    F: Future<Output = T> + 'static,
+    D: UseFutureDep,
+{
+    struct UseEffect {
+        needs_regen: bool,
+        task: Cell<Option<TaskId>>,
+        dependencies: Vec<Box<dyn Any>>,
+    }
+
+    let state = cx.use_hook(move |_| UseEffect {
+        needs_regen: true,
+        task: Cell::new(None),
+        dependencies: Vec::new(),
+    });
+
+    if dependencies.clone().apply(&mut state.dependencies) || state.needs_regen {
+        // We don't need regen anymore
+        state.needs_regen = false;
+
+        // Create the new future
+        let fut = future(dependencies.out());
+
+        state.task.set(Some(cx.push_future(async move {
+            fut.await;
+        })));
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[allow(unused)]
+    #[test]
+    fn test_use_future() {
+        use dioxus_core::prelude::*;
+
+        struct MyProps {
+            a: String,
+            b: i32,
+            c: i32,
+            d: i32,
+            e: i32,
+        }
+
+        fn app(cx: Scope<MyProps>) -> Element {
+            // should only ever run once
+            use_effect(&cx, (), |_| async move {
+                //
+            });
+
+            // runs when a is changed
+            use_effect(&cx, (&cx.props.a,), |(a,)| async move {
+                //
+            });
+
+            // runs when a or b is changed
+            use_effect(&cx, (&cx.props.a, &cx.props.b), |(a, b)| async move {
+                //
+            });
+
+            None
+        }
+    }
+}

packages/hooks/src/usefuture.rs

@@ -1,6 +1,6 @@
 #![allow(missing_docs)]
 use dioxus_core::{ScopeState, TaskId};
-use std::{cell::Cell, future::Future, rc::Rc, sync::Arc};
+use std::{any::Any, cell::Cell, future::Future, rc::Rc, sync::Arc};
 
 /// A future that resolves to a value.
 ///
@@ -10,45 +10,56 @@ use std::{cell::Cell, future::Future, rc::Rc, sync::Arc};
 /// This is commonly used for components that cannot be rendered until some
 /// asynchronous operation has completed.
 ///
+/// Whenever the hooks dependencies change, the future will be re-evaluated.
+/// If a future is pending when the dependencies change, the previous future
+/// will be allowed to continue
 ///
-///
-///
-///
-pub fn use_future<'a, T: 'static, F: Future<Output = T> + 'static>(
-    cx: &'a ScopeState,
-    new_fut: impl FnOnce() -> F,
-) -> &'a UseFuture<T> {
+/// - dependencies: a tuple of references to values that are PartialEq + Clone
+pub fn use_future<T, F, D>(
+    cx: &ScopeState,
+    dependencies: D,
+    future: impl FnOnce(D::Out) -> F,
+) -> &UseFuture<T>
+where
+    T: 'static,
+    F: Future<Output = T> + 'static,
+    D: UseFutureDep,
+{
     let state = cx.use_hook(move |_| UseFuture {
         update: cx.schedule_update(),
         needs_regen: Cell::new(true),
         slot: Rc::new(Cell::new(None)),
         value: None,
-        task: None,
-        pending: true,
+        task: Cell::new(None),
+        dependencies: Vec::new(),
     });
 
     if let Some(value) = state.slot.take() {
         state.value = Some(value);
-        state.task = None;
+        state.task.set(None);
     }
 
-    if state.needs_regen.get() {
+    if dependencies.clone().apply(&mut state.dependencies) || state.needs_regen.get() {
         // We don't need regen anymore
         state.needs_regen.set(false);
-        state.pending = false;
 
         // Create the new future
-        let fut = new_fut();
+        let fut = future(dependencies.out());
 
         // Clone in our cells
         let slot = state.slot.clone();
-        let updater = state.update.clone();
+        let schedule_update = state.update.clone();
 
-        state.task = Some(cx.push_future(async move {
+        // Cancel the current future
+        if let Some(current) = state.task.take() {
+            cx.remove_future(current);
+        }
+
+        state.task.set(Some(cx.push_future(async move {
             let res = fut.await;
             slot.set(Some(res));
-            updater();
-        }));
+            schedule_update();
+        })));
     }
 
     state
@@ -64,17 +75,34 @@ pub struct UseFuture<T> {
     update: Arc<dyn Fn()>,
     needs_regen: Cell<bool>,
     value: Option<T>,
-    pending: bool,
     slot: Rc<Cell<Option<T>>>,
-    task: Option<TaskId>,
+    task: Cell<Option<TaskId>>,
+    dependencies: Vec<Box<dyn Any>>,
+}
+
+pub enum UseFutureState<'a, T> {
+    Pending,
+    Complete(&'a T),
+    Reloading(&'a T),
 }
 
 impl<T> UseFuture<T> {
+    /// Restart the future with new dependencies.
+    ///
+    /// Will not cancel the previous future, but will ignore any values that it
+    /// generates.
     pub fn restart(&self) {
         self.needs_regen.set(true);
         (self.update)();
     }
 
+    /// Forcefully cancel a future
+    pub fn cancel(&self, cx: &ScopeState) {
+        if let Some(task) = self.task.take() {
+            cx.remove_future(task);
+        }
+    }
+
     // clears the value in the future slot without starting the future over
     pub fn clear(&self) -> Option<T> {
         (self.update)();
@@ -88,12 +116,163 @@ impl<T> UseFuture<T> {
         (self.update)();
     }
 
+    /// Return any value, even old values if the future has not yet resolved.
+    ///
+    /// If the future has never completed, the returned value will be `None`.
     pub fn value(&self) -> Option<&T> {
         self.value.as_ref()
     }
 
-    pub fn state(&self) -> FutureState<T> {
-        // self.value.as_ref()
-        FutureState::Pending
+    /// Get the ID of the future in Dioxus' internal scheduler
+    pub fn task(&self) -> Option<TaskId> {
+        self.task.get()
+    }
+
+    /// Get the current stateof the future.
+    pub fn state(&self) -> UseFutureState<T> {
+        match (&self.task.get(), &self.value) {
+            // If we have a task and an existing value, we're reloading
+            (Some(_), Some(val)) => UseFutureState::Reloading(val),
+
+            // no task, but value - we're done
+            (None, Some(val)) => UseFutureState::Complete(val),
+
+            // no task, no value - something's wrong? return pending
+            (None, None) => UseFutureState::Pending,
+
+            // Task, no value - we're still pending
+            (Some(_), None) => UseFutureState::Pending,
+        }
+    }
+}
+
+pub trait UseFutureDep: Sized + Clone {
+    type Out;
+    fn out(&self) -> Self::Out;
+    fn apply(self, state: &mut Vec<Box<dyn Any>>) -> bool;
+}
+
+impl UseFutureDep for () {
+    type Out = ();
+    fn out(&self) -> Self::Out {}
+    fn apply(self, _state: &mut Vec<Box<dyn Any>>) -> bool {
+        false
+    }
+}
+
+pub trait Dep: 'static + PartialEq + Clone {}
+impl<T> Dep for T where T: 'static + PartialEq + Clone {}
+
+impl<A: Dep> UseFutureDep for &A {
+    type Out = A;
+    fn out(&self) -> Self::Out {
+        (*self).clone()
+    }
+    fn apply(self, state: &mut Vec<Box<dyn Any>>) -> bool {
+        match state.get_mut(0).and_then(|f| f.downcast_mut::<A>()) {
+            Some(val) => {
+                if *val != *self {
+                    *val = self.clone();
+                    return true;
+                }
+            }
+            None => {
+                state.push(Box::new(self.clone()));
+                return true;
+            }
+        }
+        false
+    }
+}
+
+macro_rules! impl_dep {
+    (
+        $($el:ident=$name:ident,)*
+    ) => {
+        impl< $($el),* > UseFutureDep for ($(&$el,)*)
+        where
+            $(
+                $el: Dep
+            ),*
+        {
+            type Out = ($($el,)*);
+
+            fn out(&self) -> Self::Out {
+                let ($($name,)*) = self;
+                ($((*$name).clone(),)*)
+            }
+
+            #[allow(unused)]
+            fn apply(self, state: &mut Vec<Box<dyn Any>>) -> bool {
+                let ($($name,)*) = self;
+                let mut idx = 0;
+                let mut needs_regen = false;
+
+                $(
+                    match state.get_mut(idx).map(|f| f.downcast_mut::<$el>()).flatten() {
+                        Some(val) => {
+                            if *val != *$name {
+                                *val = $name.clone();
+                                needs_regen = true;
+                            }
+                        }
+                        None => {
+                            state.push(Box::new($name.clone()));
+                            needs_regen = true;
+                        }
+                    }
+                    idx += 1;
+                )*
+
+                needs_regen
+            }
+        }
+    };
+}
+
+impl_dep!(A = a,);
+impl_dep!(A = a, B = b,);
+impl_dep!(A = a, B = b, C = c,);
+impl_dep!(A = a, B = b, C = c, D = d,);
+impl_dep!(A = a, B = b, C = c, D = d, E = e,);
+impl_dep!(A = a, B = b, C = c, D = d, E = e, F = f,);
+impl_dep!(A = a, B = b, C = c, D = d, E = e, F = f, G = g,);
+impl_dep!(A = a, B = b, C = c, D = d, E = e, F = f, G = g, H = h,);
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[allow(unused)]
+    #[test]
+    fn test_use_future() {
+        use dioxus_core::prelude::*;
+
+        struct MyProps {
+            a: String,
+            b: i32,
+            c: i32,
+            d: i32,
+            e: i32,
+        }
+
+        fn app(cx: Scope<MyProps>) -> Element {
+            // should only ever run once
+            let fut = use_future(&cx, (), |_| async move {
+                //
+            });
+
+            // runs when a is changed
+            let fut = use_future(&cx, (&cx.props.a,), |(a,)| async move {
+                //
+            });
+
+            // runs when a or b is changed
+            let fut = use_future(&cx, (&cx.props.a, &cx.props.b), |(a, b)| async move {
+                //
+            });
+
+            None
+        }
     }
 }