server_future uses use_resource
This commit is contained in:
4 files changed
+64
-99
No files matched your search
@@ -1,11 +1,11 @@
|
|||||||
use crate::innerlude::{remove_future, spawn, Runtime};
|
use crate::innerlude::{remove_future, spawn, Runtime};
|
||||||
use crate::ScopeId;
|
use crate::ScopeId;
|
||||||
use futures_util::task::ArcWake;
|
use futures_util::task::ArcWake;
|
||||||
use std::pin::Pin;
|
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
use std::task::Waker;
|
use std::task::Waker;
|
||||||
use std::{cell::Cell, future::Future};
|
use std::{cell::Cell, future::Future};
|
||||||
use std::{cell::RefCell, rc::Rc};
|
use std::{cell::RefCell, rc::Rc};
|
||||||
|
use std::{pin::Pin, task::Poll};
|
||||||
|
|
||||||
/// A task's unique identifier.
|
/// A task's unique identifier.
|
||||||
///
|
///
|
||||||
@@ -50,6 +50,11 @@ impl Task {
|
|||||||
Runtime::with(|rt| _ = rt.sender.unbounded_send(SchedulerMsg::TaskNotified(*self)));
|
Runtime::with(|rt| _ = rt.sender.unbounded_send(SchedulerMsg::TaskNotified(*self)));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Poll the task immediately.
|
||||||
|
pub fn poll_now(&self) -> Poll<()> {
|
||||||
|
Runtime::with(|rt| rt.handle_task_wakeup(*self)).unwrap()
|
||||||
|
}
|
||||||
|
|
||||||
/// Set the task as active or paused.
|
/// Set the task as active or paused.
|
||||||
pub fn set_active(&self, active: bool) {
|
pub fn set_active(&self, active: bool) {
|
||||||
Runtime::with(|rt| rt.tasks.borrow()[self.0].active.set(active));
|
Runtime::with(|rt| rt.tasks.borrow()[self.0].active.set(active));
|
||||||
@@ -122,19 +127,19 @@ impl Runtime {
|
|||||||
self.tasks.borrow().get(task.0)?.parent
|
self.tasks.borrow().get(task.0)?.parent
|
||||||
}
|
}
|
||||||
|
|
||||||
pub(crate) fn handle_task_wakeup(&self, id: Task) {
|
pub(crate) fn handle_task_wakeup(&self, id: Task) -> Poll<()> {
|
||||||
debug_assert!(Runtime::current().is_some(), "Must be in a dioxus runtime");
|
debug_assert!(Runtime::current().is_some(), "Must be in a dioxus runtime");
|
||||||
|
|
||||||
let task = self.tasks.borrow().get(id.0).cloned();
|
let task = self.tasks.borrow().get(id.0).cloned();
|
||||||
|
|
||||||
// The task was removed from the scheduler, so we can just ignore it
|
// The task was removed from the scheduler, so we can just ignore it
|
||||||
let Some(task) = task else {
|
let Some(task) = task else {
|
||||||
return;
|
return Poll::Ready(());
|
||||||
};
|
};
|
||||||
|
|
||||||
// If a task woke up but is paused, we can just ignore it
|
// If a task woke up but is paused, we can just ignore it
|
||||||
if !task.active.get() {
|
if !task.active.get() {
|
||||||
return;
|
return Poll::Pending;
|
||||||
}
|
}
|
||||||
|
|
||||||
let mut cx = std::task::Context::from_waker(&task.waker);
|
let mut cx = std::task::Context::from_waker(&task.waker);
|
||||||
@@ -144,7 +149,9 @@ impl Runtime {
|
|||||||
self.rendering.set(false);
|
self.rendering.set(false);
|
||||||
self.current_task.set(Some(id));
|
self.current_task.set(Some(id));
|
||||||
|
|
||||||
if task.task.borrow_mut().as_mut().poll(&mut cx).is_ready() {
|
let poll_result = task.task.borrow_mut().as_mut().poll(&mut cx);
|
||||||
|
|
||||||
|
if poll_result.is_ready() {
|
||||||
// Remove it from the scope so we dont try to double drop it when the scope dropes
|
// Remove it from the scope so we dont try to double drop it when the scope dropes
|
||||||
self.get_state(task.scope)
|
self.get_state(task.scope)
|
||||||
.unwrap()
|
.unwrap()
|
||||||
@@ -160,6 +167,8 @@ impl Runtime {
|
|||||||
self.scope_stack.borrow_mut().pop();
|
self.scope_stack.borrow_mut().pop();
|
||||||
self.rendering.set(true);
|
self.rendering.set(true);
|
||||||
self.current_task.set(None);
|
self.current_task.set(None);
|
||||||
|
|
||||||
|
poll_result
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Drop the future with the given TaskId
|
/// Drop the future with the given TaskId
|
||||||
|
|||||||
@@ -454,7 +454,7 @@ impl VirtualDom {
|
|||||||
|
|
||||||
match self.rx.next().await.expect("channel should never close") {
|
match self.rx.next().await.expect("channel should never close") {
|
||||||
SchedulerMsg::Immediate(id) => self.mark_dirty(id),
|
SchedulerMsg::Immediate(id) => self.mark_dirty(id),
|
||||||
SchedulerMsg::TaskNotified(id) => self.runtime.handle_task_wakeup(id),
|
SchedulerMsg::TaskNotified(id) => _ = self.runtime.handle_task_wakeup(id),
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -467,7 +467,7 @@ impl VirtualDom {
|
|||||||
while let Ok(Some(msg)) = self.rx.try_next() {
|
while let Ok(Some(msg)) = self.rx.try_next() {
|
||||||
match msg {
|
match msg {
|
||||||
SchedulerMsg::Immediate(id) => self.mark_dirty(id),
|
SchedulerMsg::Immediate(id) => self.mark_dirty(id),
|
||||||
SchedulerMsg::TaskNotified(task) => self.runtime.handle_task_wakeup(task),
|
SchedulerMsg::TaskNotified(task) => _ = self.runtime.handle_task_wakeup(task),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,99 +1,57 @@
|
|||||||
use dioxus_lib::prelude::*;
|
use dioxus_lib::prelude::*;
|
||||||
use serde::{de::DeserializeOwned, Serialize};
|
use serde::{de::DeserializeOwned, Serialize};
|
||||||
use std::cell::Cell;
|
|
||||||
use std::cell::Ref;
|
|
||||||
use std::cell::RefCell;
|
|
||||||
use std::fmt::Debug;
|
|
||||||
use std::future::Future;
|
use std::future::Future;
|
||||||
use std::rc::Rc;
|
|
||||||
use std::sync::Arc;
|
|
||||||
|
|
||||||
/// A future that resolves to a value.
|
/// A future that resolves to a value.
|
||||||
///
|
|
||||||
///
|
|
||||||
///
|
|
||||||
/// ```rust
|
|
||||||
/// fn User(id: String) -> Element {
|
|
||||||
/// let data = use_sever_future(move || fetch_user(id)).suspend()?;
|
|
||||||
///
|
|
||||||
///
|
|
||||||
/// }
|
|
||||||
///
|
|
||||||
/// ```
|
|
||||||
#[must_use = "Consider using `cx.spawn` to run a future without reading its value"]
|
#[must_use = "Consider using `cx.spawn` to run a future without reading its value"]
|
||||||
pub fn use_server_future<T, F>(_future: impl Fn() -> F) -> UseServerFuture<T>
|
pub fn use_server_future<T, F>(_future: impl Fn() -> F + 'static) -> Option<Resource<T>>
|
||||||
where
|
where
|
||||||
T: Serialize + DeserializeOwned + 'static,
|
T: Serialize + DeserializeOwned + 'static,
|
||||||
F: Future<Output = T> + 'static,
|
F: Future<Output = T> + 'static,
|
||||||
{
|
{
|
||||||
let value: Signal<Option<T>> = use_signal(|| {
|
let mut cb = use_callback(_future);
|
||||||
// Doesn't this need to be keyed by something?
|
let mut gen = use_hook(|| CopyValue::new(0));
|
||||||
// We should try and link these IDs across the server and client
|
|
||||||
// Just the file/line/col span should be fine (or byte index)
|
|
||||||
#[cfg(feature = "ssr")]
|
|
||||||
return crate::html_storage::deserialize::take_server_data::<T>();
|
|
||||||
|
|
||||||
#[cfg(not(feature = "ssr"))]
|
let resource = use_resource(move || {
|
||||||
return None;
|
async move {
|
||||||
|
// this is going to subscribe this resource to any reactivity given to use in the callback
|
||||||
|
// We're doing this regardless so inputs get tracked, even if we drop the future before polling it
|
||||||
|
let user_fut = cb.call();
|
||||||
|
|
||||||
|
// If this is the first run, the data might be cached
|
||||||
|
if gen() == 0 {
|
||||||
|
#[cfg(not(feature = "web"))]
|
||||||
|
if let Some(o) = crate::html_storage::deserialize::take_server_data::<T>() {
|
||||||
|
gen.set(1);
|
||||||
|
return o;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Otherwise just run the future itself
|
||||||
|
let out = user_fut.await;
|
||||||
|
|
||||||
|
// and push the gen forward
|
||||||
|
gen.set(1);
|
||||||
|
|
||||||
|
out
|
||||||
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
// Run the callback regardless, giving us the future without actually polling it
|
// On the first run, force this task to be polled right away in case its value is ready
|
||||||
// This is where use_server_future gets its reactivity from
|
use_hook(|| {
|
||||||
// If the client is using signals to drive the future itself, (say, via args to the server_fn), then we need to know
|
let _ = resource.task().unwrap().poll_now();
|
||||||
// what signals are being used
|
|
||||||
use_future(move || async move {
|
|
||||||
// watch the reactive context
|
|
||||||
// if it changes, restart the future
|
|
||||||
//
|
|
||||||
// if let Err(err) = crate::prelude::server_context().push_html_data(&data) {
|
|
||||||
// tracing::error!("Failed to push HTML data: {}", err);
|
|
||||||
// };
|
|
||||||
});
|
});
|
||||||
|
|
||||||
// if there's no value ready, mark this component as suspended and return early
|
// Suspend if the value isn't ready
|
||||||
if value.peek().is_none() {
|
match resource.state() {
|
||||||
suspend();
|
UseResourceState::Pending => {
|
||||||
|
suspend();
|
||||||
|
None
|
||||||
|
}
|
||||||
|
UseResourceState::Regenerating => {
|
||||||
|
suspend();
|
||||||
|
Some(resource)
|
||||||
|
}
|
||||||
|
UseResourceState::Ready => Some(resource),
|
||||||
}
|
}
|
||||||
|
|
||||||
todo!()
|
|
||||||
}
|
|
||||||
|
|
||||||
pub struct UseServerFuture<T: 'static> {
|
|
||||||
value: Signal<Option<Signal<T>>>,
|
|
||||||
}
|
|
||||||
|
|
||||||
impl<T> UseServerFuture<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) {
|
|
||||||
// if let Some(task) = self.task.take() {
|
|
||||||
// remove_future(task);
|
|
||||||
// }
|
|
||||||
// }
|
|
||||||
|
|
||||||
/// 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<Signal<T>> {
|
|
||||||
todo!()
|
|
||||||
// Ref::map(self.value.read(), |v: &Option<T>| v.as_deref().unwrap())
|
|
||||||
}
|
|
||||||
|
|
||||||
// /// Get the ID of the future in Dioxus' internal scheduler
|
|
||||||
// pub fn task(&self) -> Option<Task> {
|
|
||||||
// self.task.get()
|
|
||||||
// }
|
|
||||||
|
|
||||||
// /// Get the current state of the future.
|
|
||||||
// pub fn reloading(&self) -> bool {
|
|
||||||
// self.task.get().is_some()
|
|
||||||
// }
|
|
||||||
}
|
}
|
||||||
@@ -11,9 +11,9 @@ use std::future::Future;
|
|||||||
|
|
||||||
/// A memo that resolve to a value asynchronously.
|
/// A memo that resolve to a value asynchronously.
|
||||||
///
|
///
|
||||||
/// Regular memos are synchronous and resolve immediately. However, you might want to resolve a memo
|
/// This runs on the server
|
||||||
#[must_use = "Consider using `cx.spawn` to run a future without reading its value"]
|
#[must_use = "Consider using `cx.spawn` to run a future without reading its value"]
|
||||||
pub fn use_resource<T, F>(future: impl Fn() -> F + 'static) -> AsyncMemo<T>
|
pub fn use_resource<T, F>(future: impl Fn() -> F + 'static) -> Resource<T>
|
||||||
where
|
where
|
||||||
T: 'static,
|
T: 'static,
|
||||||
F: Future<Output = T> + 'static,
|
F: Future<Output = T> + 'static,
|
||||||
@@ -28,9 +28,6 @@ where
|
|||||||
|
|
||||||
// Spawn a wrapper task that polls the innner future and watch its dependencies
|
// Spawn a wrapper task that polls the innner future and watch its dependencies
|
||||||
spawn(async move {
|
spawn(async move {
|
||||||
// Wait for the dom the be finished with sync work
|
|
||||||
flush_sync().await;
|
|
||||||
|
|
||||||
// move the future here and pin it so we can poll it
|
// move the future here and pin it so we can poll it
|
||||||
let fut = fut;
|
let fut = fut;
|
||||||
pin_mut!(fut);
|
pin_mut!(fut);
|
||||||
@@ -40,7 +37,7 @@ where
|
|||||||
let res = future::poll_fn(|cx| rc.run_in(|| fut.poll_unpin(cx))).await;
|
let res = future::poll_fn(|cx| rc.run_in(|| fut.poll_unpin(cx))).await;
|
||||||
|
|
||||||
// Set the value and state
|
// Set the value and state
|
||||||
state.set(UseResourceState::Complete);
|
state.set(UseResourceState::Ready);
|
||||||
value.set(Some(Signal::new(res)));
|
value.set(Some(Signal::new(res)));
|
||||||
})
|
})
|
||||||
});
|
});
|
||||||
@@ -62,17 +59,17 @@ where
|
|||||||
})
|
})
|
||||||
});
|
});
|
||||||
|
|
||||||
AsyncMemo { task, value, state }
|
Resource { task, value, state }
|
||||||
}
|
}
|
||||||
|
|
||||||
#[allow(unused)]
|
#[allow(unused)]
|
||||||
pub struct AsyncMemo<T: 'static> {
|
pub struct Resource<T: 'static> {
|
||||||
value: Signal<Option<Signal<T>>>,
|
value: Signal<Option<Signal<T>>>,
|
||||||
task: Signal<Task>,
|
task: Signal<Task>,
|
||||||
state: Signal<UseResourceState>,
|
state: Signal<UseResourceState>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<T> AsyncMemo<T> {
|
impl<T> Resource<T> {
|
||||||
/// Restart the future with new dependencies.
|
/// Restart the future with new dependencies.
|
||||||
///
|
///
|
||||||
/// Will not cancel the previous future, but will ignore any values that it
|
/// Will not cancel the previous future, but will ignore any values that it
|
||||||
@@ -139,8 +136,9 @@ impl<T> AsyncMemo<T> {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[derive(PartialEq, Eq, Clone, Copy, Debug)]
|
||||||
pub enum UseResourceState {
|
pub enum UseResourceState {
|
||||||
Pending,
|
Pending,
|
||||||
Complete,
|
Ready,
|
||||||
Regenerating, // the old value
|
Regenerating, // the old value
|
||||||
}
|
}
|
||||||
Reference in new issue
Block a user