diff --git a/Cargo.lock b/Cargo.lock index 2ae13e609..128e2455b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -664,6 +664,14 @@ dependencies = [ "tokio", ] +[[package]] +name = "deferred-view" +version = "0.5.0" +dependencies = [ + "tokio", + "topcoat", +] + [[package]] name = "deranged" version = "0.5.8" @@ -3194,6 +3202,7 @@ name = "topcoat-view" version = "0.5.0" dependencies = [ "criterion", + "futures-util", "http", "itoa", "memchr", diff --git a/Cargo.toml b/Cargo.toml index e2dbeeff7..ee3edd4a8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -40,6 +40,7 @@ members = [ "examples/context", "examples/cookie", "examples/datastar", + "examples/deferred-view", "examples/font", "examples/hello-world", "examples/htmx", diff --git a/crates/topcoat-core/src/context.rs b/crates/topcoat-core/src/context.rs index a8862aea5..c43dd69d0 100644 --- a/crates/topcoat-core/src/context.rs +++ b/crates/topcoat-core/src/context.rs @@ -19,9 +19,9 @@ use crate::{abort::AbortStore, memoize::MemoizeCache}; pub struct Cx { id: CxId, app_context: Arc, - request_context: ContextMap, - memoize_cache: MemoizeCache, - abort_store: AbortStore, + request_context: Arc, + memoize_cache: Arc, + abort_store: Arc, } impl Cx { @@ -30,17 +30,59 @@ impl Cx { Self { id: CxId::new(), app_context, - request_context, - memoize_cache: MemoizeCache::new(), - abort_store: AbortStore::new(), + request_context: Arc::new(request_context), + memoize_cache: Arc::new(MemoizeCache::new()), + abort_store: Arc::new(AbortStore::new()), } } /// Returns this context's unique [`CxId`]. #[inline] + #[must_use] pub fn id(&self) -> CxId { self.id } + + /// Returns an owned handle to this request context. + /// + /// The handle keeps request and app context values alive while a response + /// stream is still producing deferred views. + #[must_use] + pub fn handle(&self) -> CxHandle { + CxHandle(Self { + id: self.id, + app_context: Arc::clone(&self.app_context), + request_context: Arc::clone(&self.request_context), + memoize_cache: Arc::clone(&self.memoize_cache), + abort_store: Arc::clone(&self.abort_store), + }) + } +} + +/// An owned handle to a request [`Cx`]. +/// +/// It dereferences to `Cx`, so context helpers accept `&handle`. +#[derive(Debug)] +pub struct CxHandle(Cx); + +impl Clone for CxHandle { + fn clone(&self) -> Self { + self.0.handle() + } +} + +impl AsRef for CxHandle { + fn as_ref(&self) -> &Cx { + &self.0 + } +} + +impl Deref for CxHandle { + type Target = Cx; + + fn deref(&self) -> &Self::Target { + &self.0 + } } /// Assembles the request context for an in-flight request. @@ -71,11 +113,17 @@ impl CxBuilder { /// /// A type can hold only one value at a time, so registering a type that is /// already present replaces it and hands back the displaced value. + /// + /// # Panics + /// + /// Panics if the request context was shared before the builder finished. pub fn insert(&mut self, value: T) -> Option where T: Any + Send + Sync, { - self.cx.request_context.insert(value) + Arc::get_mut(&mut self.cx.request_context) + .expect("request context was shared before it was built") + .insert(value) } /// Returns `true` if a value of type `T` has been registered on the request @@ -100,12 +148,18 @@ impl CxBuilder { /// Returns a mutable reference to the request context value of type `T`, or /// `None` if no such value has been registered. + /// + /// # Panics + /// + /// Panics if the request context was shared before the builder finished. #[must_use] pub fn get_mut(&mut self) -> Option<&mut T> where T: Any + Send + Sync, { - self.cx.request_context.get_mut::() + Arc::get_mut(&mut self.cx.request_context) + .expect("request context was shared before it was built") + .get_mut::() } /// Consumes the builder, returning the finished [`Cx`]. @@ -169,12 +223,14 @@ impl CxTestBuilder { #[inline] #[doc(hidden)] +#[must_use] pub fn memoize_cache(cx: &Cx) -> &MemoizeCache { &cx.memoize_cache } #[inline] #[doc(hidden)] +#[must_use] pub fn abort_store(cx: &Cx) -> &AbortStore { &cx.abort_store } diff --git a/crates/topcoat-core/src/context/context_map.rs b/crates/topcoat-core/src/context/context_map.rs index c281d0998..299e6b8a3 100644 --- a/crates/topcoat-core/src/context/context_map.rs +++ b/crates/topcoat-core/src/context/context_map.rs @@ -66,6 +66,7 @@ where /// } /// ``` #[track_caller] +#[must_use] pub fn app_context(cx: &Cx) -> &T where T: Any + Send + Sync, @@ -131,6 +132,7 @@ where /// } /// ``` #[track_caller] +#[must_use] pub fn request_context(cx: &Cx) -> &T where T: Any + Send + Sync, diff --git a/crates/topcoat-router/Cargo.toml b/crates/topcoat-router/Cargo.toml index cd96975c3..75e799968 100644 --- a/crates/topcoat-router/Cargo.toml +++ b/crates/topcoat-router/Cargo.toml @@ -19,7 +19,6 @@ discover = [ "dep:inventory", ] multipart = [ - "dep:futures-util", "dep:multer", ] serve = [ @@ -56,7 +55,7 @@ bytes.workspace = true form_urlencoded.workspace = true futures-core.workspace = true futures-sink = { workspace = true, optional = true } -futures-util = { workspace = true, optional = true } +futures-util.workspace = true heck.workspace = true http.workspace = true http-body.workspace = true diff --git a/crates/topcoat-router/src/page.rs b/crates/topcoat-router/src/page.rs index 6d9374f27..c691c0cb0 100644 --- a/crates/topcoat-router/src/page.rs +++ b/crates/topcoat-router/src/page.rs @@ -78,6 +78,7 @@ impl PageFn { } /// Renders the page, returning a [`Result`]. + #[must_use] pub fn render<'cx>( &self, cx: &'cx Cx, @@ -124,6 +125,7 @@ impl LayoutFn { } /// Renders the layout, embedding the given child content [`Result`]`<`[`View`]`>` as its slot. + #[must_use] pub fn render<'cx>( &self, cx: &'cx Cx, diff --git a/crates/topcoat-router/src/response.rs b/crates/topcoat-router/src/response.rs index f69a78b56..0a124f307 100644 --- a/crates/topcoat-router/src/response.rs +++ b/crates/topcoat-router/src/response.rs @@ -1,16 +1,25 @@ -use std::{borrow::Cow, convert::Infallible}; +use std::{ + borrow::Cow, + convert::Infallible, + future::Future, + pin::Pin, + task::{Context, Poll}, +}; use bytes::{Bytes, BytesMut}; +use futures_util::stream::FuturesUnordered; use http::{ Extensions, HeaderMap, StatusCode, header::{CONTENT_TYPE, HeaderName, HeaderValue}, response::Parts, }; +use http_body::Frame; +use http_body_util::StreamBody; use topcoat_core::{ - context::Cx, + context::{Cx, CxHandle}, error::{Error, Result}, }; -use topcoat_view::View; +use topcoat_view::{DeferredTask, View}; use crate::{Body, BoxError, content::Html}; @@ -19,6 +28,63 @@ pub type Response = http::Response; const TEXT_PLAIN: HeaderValue = HeaderValue::from_static("text/plain; charset=utf-8"); const APPLICATION_OCTET_STREAM: HeaderValue = HeaderValue::from_static("application/octet-stream"); +type PendingDeferred = Pin> + Send + 'static>>; + +struct DeferredResponseStream { + initial: Option, + cx: CxHandle, + pending: FuturesUnordered, +} + +impl DeferredResponseStream { + fn new(html: String, cx: CxHandle, deferred: Vec) -> Self { + let mut stream = Self { + initial: Some(Bytes::from(html)), + cx, + pending: FuturesUnordered::new(), + }; + stream.extend(deferred); + stream + } + + fn extend(&mut self, deferred: Vec) { + for task in deferred { + let cx = self.cx.clone(); + let id = task.id(); + self.pending + .push(Box::pin(async move { Ok((id, task.resolve(cx).await?)) })); + } + } +} + +impl futures_core::Stream for DeferredResponseStream { + type Item = Result>; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + if let Some(initial) = self.initial.take() { + return Poll::Ready(Some(Ok(Frame::data(initial)))); + } + + match futures_core::Stream::poll_next(Pin::new(&mut self.pending), cx) { + Poll::Ready(Some(Ok((id, view)))) => { + let rendered = view.render_response(&self.cx); + self.extend(rendered.deferred); + Poll::Ready(Some(Ok(Frame::data(Bytes::from(deferred_patch( + id, + &rendered.html, + )))))) + } + Poll::Ready(Some(Err(error))) => Poll::Ready(Some(Err(error))), + Poll::Ready(None) => Poll::Ready(None), + Poll::Pending => Poll::Pending, + } + } +} + +fn deferred_patch(id: u64, html: &str) -> String { + format!(r#""#) +} + /// Converts a value into an HTTP [`Response`]. /// /// Route handlers return any type that implements this trait. @@ -236,7 +302,14 @@ impl IntoResponse for Parts { impl IntoResponse for View { fn into_response(self, cx: &Cx) -> Result { let rendered = self.render_response(cx); - let mut response = Html(rendered.html).into_response(cx)?; + let mut response = if rendered.deferred.is_empty() { + Html(rendered.html).into_response(cx)? + } else { + let stream = DeferredResponseStream::new(rendered.html, cx.handle(), rendered.deferred); + let mut response = Html(String::new()).into_response(cx)?; + *response.body_mut() = Body::new(StreamBody::new(stream)); + response + }; if let Some(status_code) = rendered.status_code { *response.status_mut() = status_code; } @@ -420,8 +493,20 @@ impl_into_response_tuples!( #[cfg(test)] mod tests { + use std::{ + future::{pending, poll_fn}, + sync::{ + Arc, Mutex, + atomic::{AtomicBool, Ordering}, + }, + task::Poll, + time::Duration, + }; + + use futures_util::StreamExt as _; use http_body_util::Full; - use topcoat_view::{HtmlContext, NodeViewParts, PartsWriter, ViewParts}; + use topcoat_core::context::request_context; + use topcoat_view::{HtmlContext, NodeViewParts, PartsWriter, ViewParts, defer}; use super::*; use crate::to_bytes; @@ -553,6 +638,14 @@ mod tests { View::new(parts) } + fn compose(views: impl IntoIterator) -> View { + let mut parts = ViewParts::new(); + for view in views { + parts.push_view(view); + } + View::new(parts) + } + #[test] fn view_is_an_html_response() { let (parts, body) = run(view(|_cx, writer| { @@ -593,6 +686,140 @@ mod tests { assert_eq!(header(&parts, "content-type"), "application/xhtml+xml"); } + #[tokio::test] + async fn deferred_views_stream_template_patches_in_completion_order() { + let cx = Cx::default(); + let completed = Arc::new(Mutex::new(Vec::new())); + let slow_completed = Arc::clone(&completed); + let slow = defer( + view(|_cx, writer| { + writer.push_str("slow loading"); + }), + move |_cx| async move { + tokio::time::sleep(Duration::from_millis(20)).await; + slow_completed.lock().unwrap().push("slow"); + Ok(view(|_cx, writer| { + writer.push_str("slow ready"); + })) + }, + ); + let fast_completed = Arc::clone(&completed); + let fast = defer( + view(|_cx, writer| { + writer.push_str("fast loading"); + }), + move |_cx| async move { + fast_completed.lock().unwrap().push("fast"); + Ok(view(|_cx, writer| { + writer.push_str("fast ready"); + })) + }, + ); + let response = compose([slow, fast]).into_response(&cx).unwrap(); + let mut body = response.into_body().into_data_stream(); + + let shell = body.next().await.unwrap().unwrap(); + let shell = String::from_utf8(shell.to_vec()).unwrap(); + assert!(shell.contains("slow loading")); + assert!(shell.contains("fast loading")); + + let first = body.next().await.unwrap().unwrap(); + let first = String::from_utf8(first.to_vec()).unwrap(); + let second = body.next().await.unwrap().unwrap(); + let second = String::from_utf8(second.to_vec()).unwrap(); + assert!(first.contains("fast ready")); + assert!(second.contains("slow ready")); + assert!(first.starts_with("