use std::{
fmt::Write,
pin::Pin,
task::{Context, Poll},
};
use bytes::Bytes;
use futures_util::future::poll_fn;
use http::{HeaderMap, StatusCode};
use http_body::Frame;
use pin_project_lite::pin_project;
use topcoat_core::{context::Cx, error::Result};
use topcoat_view::{BoxView, Formatter, View, ViewExt, ViewHandle, internal::MoveView};
use crate::{
Body, BoxError,
content::Html,
error::RedirectError,
response::{AsyncIntoResponse, IntoResponse, Response},
};
impl IntoResponse for ViewHandle {
fn into_response(self, cx: &Cx) -> Result<Response> {
let rendered = self.render_response(cx);
html_response(cx, rendered.html, rendered.status_code, rendered.headers)
}
}
impl AsyncIntoResponse for BoxView<'static> {
fn async_into_response(self, cx: &Cx) -> impl Future<Output = Result<Response>> + Send {
stream(self, cx)
}
}
impl<Fut> AsyncIntoResponse for MoveView<Fut>
where
Fut: Future<Output = Result<()>> + Send + 'static,
{
fn async_into_response(self, cx: &Cx) -> impl Future<Output = Result<Response>> + Send {
stream(self.boxed(), cx)
}
}
async fn stream<V: View + Unpin + 'static>(mut view: V, cx: &Cx) -> Result<Response> {
let mut pinned_view = Pin::new(&mut view);
let first = poll_fn(|cx| pinned_view.as_mut().poll_first(cx)).await?;
let rendered = first.content.render_response(cx);
if first.live {
let body = ViewBody {
cx: cx.clone(),
first: Some(rendered.html),
script: Some(SWAP_SCRIPT),
done: false,
view,
};
html_response(cx, Body::new(body), rendered.status_code, rendered.headers)
} else {
html_response(
cx,
Body::new(rendered.html),
rendered.status_code,
rendered.headers,
)
}
}
fn html_response(
cx: &Cx,
body: impl Into<Body>,
status_code: Option<StatusCode>,
headers: HeaderMap,
) -> Result<Response> {
let mut response = Html(body.into()).into_response(cx)?;
if let Some(status_code) = status_code {
*response.status_mut() = status_code;
}
response.headers_mut().extend(headers);
Ok(response)
}
const SWAP_SCRIPT: &str = r"<script>
window.topcoat ??= {
swap(id) {
const script = document.currentScript;
const template = script.previousElementSibling;
let open = null;
let close = null;
const walker = document.createTreeWalker(document.documentElement, NodeFilter.SHOW_COMMENT);
while (walker.nextNode()) {
const comment = walker.currentNode;
if (comment.data === `topcoat::region::start(${id})`) open = comment;
else if (comment.data === `topcoat::region::end(${id})`) close = comment;
}
if (open && close) {
while (open.nextSibling && open.nextSibling !== close) open.nextSibling.remove();
close.parentNode.insertBefore(template.content, close);
}
template.remove();
script.remove();
},
};
</script>";
fn redirect_script(redirect: &RedirectError) -> String {
let uri = redirect
.location()
.to_str()
.expect("redirect location is ASCII");
let mut location = String::with_capacity(uri.len());
for c in uri.chars() {
match c {
'\\' => location.push_str("\\\\"),
'"' => location.push_str("\\\""),
'<' => location.push_str("\\x3C"),
c => location.push(c),
}
}
format!("<script>window.location.replace(\"{location}\")</script>")
}
pin_project! {
struct ViewBody<V> {
cx: Cx,
first: Option<String>,
script: Option<&'static str>,
done: bool,
#[pin]
view: V,
}
}
impl<V: View + 'static> http_body::Body for ViewBody<V> {
type Data = Bytes;
type Error = BoxError;
fn poll_frame(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
let this = self.project();
if let Some(first) = this.first.take() {
return Poll::Ready(Some(Ok(Frame::data(first.into()))));
}
if *this.done {
return Poll::Ready(None);
}
match this.view.poll_swap(cx) {
Poll::Ready(Ok(Some(swap))) => {
let script = this.script.take();
let region = swap.region;
let mut envelope = String::with_capacity(
script.map_or(0, str::len) + swap.replacement.size_hint() + 96,
);
{
let mut f = Formatter::new(&mut envelope);
if let Some(script) = script {
f.write_str(script);
}
write!(f, "<template data-topcoat-swap=\"{region}\">").unwrap();
swap.replacement.render_into(this.cx, &mut f);
write!(f, "</template><script>topcoat.swap({region})</script>").unwrap();
}
Poll::Ready(Some(Ok(Frame::data(envelope.into()))))
}
Poll::Ready(Ok(None)) => {
*this.done = true;
Poll::Ready(None)
}
Poll::Ready(Err(error)) => {
*this.done = true;
match error.downcast::<RedirectError>() {
Ok(redirect) => {
Poll::Ready(Some(Ok(Frame::data(redirect_script(&redirect).into()))))
}
Err(error) => Poll::Ready(Some(Err(error.into()))),
}
}
Poll::Pending => Poll::Pending,
}
}
fn is_end_stream(&self) -> bool {
self.first.is_none() && self.done
}
}
#[cfg(test)]
mod tests {
use std::io;
use futures_util::StreamExt;
use http::header::CONTENT_TYPE;
use topcoat::view::{emit, live, view};
use super::*;
use crate::{
BodyPanicError, LayoutFn, Method, PageFn, Router, RouterBuilder, Slot, error::redirect,
to_bytes,
};
async fn send(router: &Router, path: &str) -> Response {
let request = http::Request::builder()
.method(Method::GET)
.uri(path)
.body(Body::empty())
.unwrap();
router.handle(request).await
}
async fn send_page(render: crate::PageRenderFn) -> Response {
let router = RouterBuilder::new()
.page(PageFn::new(Method::GET, "/p", render))
.build();
send(&router, "/p").await
}
async fn data_frames(body: Body) -> Vec<String> {
let mut frames = body.into_data_stream();
let mut chunks = Vec::new();
while let Some(frame) = frames.next().await {
chunks.push(String::from_utf8(frame.unwrap().to_vec()).unwrap());
}
chunks
}
fn swap_envelope(region: u64, replacement: &str) -> String {
format!(
"<template data-topcoat-swap=\"{region}\">{replacement}</template>\
<script>topcoat.swap({region})</script>"
)
}
fn render_settled_region_page(cx: &Cx, _body: Body) -> BoxView<'_> {
view! { cx => <main>(live! { emit! { <p>"only"</p> } })</main> }.boxed()
}
fn render_live_page(cx: &Cx, _body: Body) -> BoxView<'_> {
view! {
cx =>
<main>
(live! {
emit! { <p>"first"</p> }?;
emit! { <p>"second"</p> }
})
</main>
}
.boxed()
}
fn render_thrice_emitting_page(cx: &Cx, _body: Body) -> BoxView<'_> {
view! {
cx =>
<main>
(live! {
emit! { <p>"one"</p> }?;
emit! { <p>"two"</p> }?;
emit! { <p>"three"</p> }
})
</main>
}
.boxed()
}
fn render_two_region_page(cx: &Cx, _body: Body) -> BoxView<'_> {
view! {
cx =>
<main>
<section>
(live! {
emit! { <p>"a1"</p> }?;
emit! { <p>"a2"</p> }
})
</section>
<section>
(live! {
emit! { <p>"b1"</p> }?;
emit! { <p>"b2"</p> }
})
</section>
</main>
}
.boxed()
}
fn render_live_metadata_page(cx: &Cx, _body: Body) -> BoxView<'_> {
view! {
cx =>
(StatusCode::ACCEPTED)
((
http::HeaderName::from_static("x-test"),
http::HeaderValue::from_static("1"),
))
<main>
(live! {
emit! { <p>"first"</p> }?;
emit! { <p>"second"</p> }
})
</main>
}
.boxed()
}
fn render_settled_metadata_page(cx: &Cx, _body: Body) -> BoxView<'_> {
view! {
cx =>
(StatusCode::CREATED)
((
http::HeaderName::from_static("x-test"),
http::HeaderValue::from_static("1"),
))
<p>"made"</p>
}
.boxed()
}
fn render_failing_page(cx: &Cx, _body: Body) -> BoxView<'_> {
view! { cx => <main>(live! { Err(io::Error::other("boom").into()) })</main> }.boxed()
}
fn render_late_failing_page(cx: &Cx, _body: Body) -> BoxView<'_> {
view! {
cx =>
<main>
(live! {
emit! { <p>"first"</p> }?;
tokio::task::yield_now().await;
Err(io::Error::other("late").into())
})
</main>
}
.boxed()
}
fn render_late_panicking_page(cx: &Cx, _body: Body) -> BoxView<'_> {
view! {
cx =>
<main>
(live! {
emit! { <p>"first"</p> }?;
tokio::task::yield_now().await;
panic!("late");
})
</main>
}
.boxed()
}
fn wrap_layout<'a>(cx: &Cx, slot: Slot<'a>) -> BoxView<'a> {
view! {
cx =>
"R["
(slot)
"]"
}
.boxed()
}
#[tokio::test]
async fn a_page_with_a_settled_region_responds_with_plain_html() {
let response = send_page(render_settled_region_page).await;
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(
response.headers().get(CONTENT_TYPE).unwrap(),
"text/html; charset=utf-8"
);
let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
assert_eq!(&body[..], b"<main><p>only</p></main>");
}
#[tokio::test]
async fn a_live_page_streams_its_first_content_then_its_swaps() {
let response = send_page(render_live_page).await;
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(
response.headers().get(CONTENT_TYPE).unwrap(),
"text/html; charset=utf-8"
);
let frames = data_frames(response.into_body()).await;
assert_eq!(frames.len(), 2);
assert_eq!(
frames[0],
"<main><!--topcoat::region::start(1)--><p>first</p>\
<!--topcoat::region::end(1)--></main>"
);
assert_eq!(
frames[1],
format!("{SWAP_SCRIPT}{}", swap_envelope(1, "<p>second</p>"))
);
}
#[tokio::test]
async fn the_swap_applier_is_sent_once_ahead_of_the_first_swap() {
let response = send_page(render_thrice_emitting_page).await;
let frames = data_frames(response.into_body()).await;
assert_eq!(frames.len(), 3);
assert_eq!(
frames[1],
format!("{SWAP_SCRIPT}{}", swap_envelope(1, "<p>two</p>"))
);
assert_eq!(frames[2], swap_envelope(1, "<p>three</p>"));
}
#[tokio::test]
async fn sibling_regions_stream_their_own_swaps() {
let response = send_page(render_two_region_page).await;
let frames = data_frames(response.into_body()).await;
assert_eq!(
frames[0],
"<main>\
<section><!--topcoat::region::start(1)--><p>a1</p>\
<!--topcoat::region::end(1)--></section>\
<section><!--topcoat::region::start(2)--><p>b1</p>\
<!--topcoat::region::end(2)--></section>\
</main>"
);
let swaps = frames[1..].concat();
assert_eq!(swaps.matches("window.topcoat ??=").count(), 1);
assert!(swaps.contains(&swap_envelope(1, "<p>a2</p>")), "{swaps}");
assert!(swaps.contains(&swap_envelope(2, "<p>b2</p>")), "{swaps}");
}
#[tokio::test]
async fn each_request_numbers_its_regions_from_the_start() {
let router = RouterBuilder::new()
.page(PageFn::new(Method::GET, "/p", render_live_page))
.build();
for _ in 0..2 {
let response = send(&router, "/p").await;
let frames = data_frames(response.into_body()).await;
assert!(
frames[0].contains("<!--topcoat::region::start(1)-->"),
"{}",
frames[0]
);
}
}
#[tokio::test]
async fn a_live_page_streams_below_its_layouts() {
let router = RouterBuilder::new()
.page(PageFn::new(Method::GET, "/p", render_live_page))
.layout(LayoutFn::new("/", wrap_layout))
.build();
let response = send(&router, "/p").await;
let frames = data_frames(response.into_body()).await;
assert_eq!(
frames[0],
"R[<main><!--topcoat::region::start(1)--><p>first</p>\
<!--topcoat::region::end(1)--></main>]"
);
assert_eq!(
frames[1],
format!("{SWAP_SCRIPT}{}", swap_envelope(1, "<p>second</p>"))
);
}
#[tokio::test]
async fn a_live_view_applies_its_declared_status_and_headers() {
let response = send_page(render_live_metadata_page).await;
assert_eq!(response.status(), StatusCode::ACCEPTED);
assert_eq!(response.headers().get("x-test").unwrap(), "1");
let frames = data_frames(response.into_body()).await;
assert_eq!(frames.len(), 2);
assert!(frames[1].ends_with(&swap_envelope(1, "<p>second</p>")));
}
#[tokio::test]
async fn a_settled_view_applies_its_declared_status_and_headers() {
let response = send_page(render_settled_metadata_page).await;
assert_eq!(response.status(), StatusCode::CREATED);
assert_eq!(response.headers().get("x-test").unwrap(), "1");
let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
assert_eq!(&body[..], b"<p>made</p>");
}
#[tokio::test]
async fn a_view_handle_response_carries_its_status_and_headers() {
let cx = &Cx::default();
let handle = view! {
cx =>
(StatusCode::CREATED)
((
http::HeaderName::from_static("x-test"),
http::HeaderValue::from_static("1"),
))
<p>"made"</p>
}
.single()
.await
.unwrap();
let response = handle.into_response(cx).unwrap();
assert_eq!(response.status(), StatusCode::CREATED);
assert_eq!(response.headers().get("x-test").unwrap(), "1");
assert_eq!(
response.headers().get(CONTENT_TYPE).unwrap(),
"text/html; charset=utf-8"
);
let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
assert_eq!(&body[..], b"<p>made</p>");
}
#[tokio::test]
async fn a_view_failing_before_its_first_content_is_a_server_error() {
let response = send_page(render_failing_page).await;
assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR);
let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
assert_eq!(&body[..], b"internal server error");
}
fn render_immediately_failing_page(cx: &Cx, _body: Body) -> BoxView<'_> {
view! {
cx =>
<main>
(live! {
emit! { <p>"first"</p> }?;
Err(io::Error::other("early").into())
})
</main>
}
.boxed()
}
#[tokio::test]
async fn a_failure_in_the_same_poll_as_the_emission_is_a_server_error() {
let response = send_page(render_immediately_failing_page).await;
assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR);
let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
assert_eq!(&body[..], b"internal server error");
}
#[tokio::test]
async fn a_failure_after_the_first_content_ends_the_stream_with_an_error() {
let response = send_page(render_late_failing_page).await;
assert_eq!(response.status(), StatusCode::OK);
let mut frames = response.into_body().into_data_stream();
let first = frames.next().await.unwrap().unwrap();
assert!(first.starts_with(b"<main><!--topcoat::region::start(1)-->"));
let error = frames.next().await.unwrap().unwrap_err();
assert_eq!(error.to_string(), "late");
assert!(frames.next().await.is_none());
}
#[tokio::test]
async fn a_panic_after_the_first_content_ends_the_stream_with_an_error() {
let router = RouterBuilder::new()
.page(PageFn::new(Method::GET, "/p", render_late_panicking_page))
.page(PageFn::new(Method::GET, "/q", render_live_page))
.build();
let response = send(&router, "/p").await;
assert_eq!(response.status(), StatusCode::OK);
let mut frames = response.into_body().into_data_stream();
let first = frames.next().await.unwrap().unwrap();
assert!(first.starts_with(b"<main><!--topcoat::region::start(1)-->"));
let error = frames.next().await.unwrap().unwrap_err();
let error = error.downcast::<BodyPanicError>().unwrap();
assert_eq!(error.message(), Some("late"));
assert!(frames.next().await.is_none());
let response = send(&router, "/q").await;
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(data_frames(response.into_body()).await.len(), 2);
}
fn render_redirecting_page(cx: &Cx, _body: Body) -> BoxView<'_> {
view! { cx => <main>(live! { Err(redirect("/target").into()) })</main> }.boxed()
}
fn render_late_redirecting_page(cx: &Cx, _body: Body) -> BoxView<'_> {
view! {
cx =>
<main>
(live! {
emit! { <p>"first"</p> }?;
tokio::task::yield_now().await;
Err(redirect("/target").into())
})
</main>
}
.boxed()
}
#[tokio::test]
async fn a_redirect_before_the_first_content_is_a_real_redirect() {
let response = send_page(render_redirecting_page).await;
assert_eq!(response.status(), StatusCode::TEMPORARY_REDIRECT);
assert_eq!(
response.headers().get(http::header::LOCATION).unwrap(),
"/target"
);
}
#[tokio::test]
async fn a_redirect_after_the_first_content_streams_a_navigation_script() {
let response = send_page(render_late_redirecting_page).await;
assert_eq!(response.status(), StatusCode::OK);
let frames = data_frames(response.into_body()).await;
assert_eq!(frames.len(), 2);
assert!(frames[0].starts_with("<main><!--topcoat::region::start(1)-->"));
assert_eq!(
frames[1],
"<script>window.location.replace(\"/target\")</script>"
);
}
#[test]
fn the_navigation_script_escapes_the_redirect_target() {
let script = redirect_script(&redirect("/a\"b\\c<d"));
assert_eq!(
script,
"<script>window.location.replace(\"/a\\\"b\\\\c\\x3Cd\")</script>"
);
}
#[test]
fn the_navigation_script_keeps_a_non_ascii_redirect_target() {
let script = redirect_script(&redirect("/caf\u{e9}"));
assert_eq!(
script,
"<script>window.location.replace(\"/caf%C3%A9\")</script>"
);
}
}