yewstack · hamza1311 · Sep 10, 2022 · Aug 4, 2022 · Aug 5, 2022 · Aug 5, 2022
diff --git a/packages/yew/Cargo.toml b/packages/yew/Cargo.toml
@@ -31,9 +31,8 @@ implicit-clone = { version = "0.3", features = ["map"] }
 base64ct = { version = "1.5.0", features = ["std"], optional = true }
 bincode = { version = "1.3.3", optional = true }
 serde = { version = "1", features = ["derive"] }
-tokio = { version = "1.19", features = ["sync"] }
-tokio-stream = { version = "0.1.9", features = ["sync"] }
 tracing = "0.1.36"
+pin-project = "1.0.11"
 
 [dependencies.web-sys]
 version = "^0.3.59"
@@ -77,8 +76,10 @@ wasm-bindgen-futures = "0.4"
 
 [target.'cfg(not(target_arch = "wasm32"))'.dependencies]
 num_cpus = { version = "1.13", optional = true }
-tokio-util = { version = "0.7", features = ["rt"], optional = true }
 once_cell = "1"
+tokio = { version = "1.19", features = ["rt", "time"], optional = true }
+tokio-stream = { version = "0.1", features = ["time"], optional = true }
+tokio-util = { version = "0.7", features = ["rt"], optional = true }
 
 [dev-dependencies]
 wasm-bindgen-test = "0.3"
@@ -95,7 +96,7 @@ features = [
 ]
 
 [features]
-tokio = ["tokio/rt", "tokio/time", "dep:num_cpus", "dep:tokio-util"]
+tokio = ["dep:tokio", "dep:num_cpus", "dep:tokio-util", "dep:tokio-stream"]
 ssr = ["dep:html-escape", "dep:base64ct", "dep:bincode"]
 csr = []
 hydration = ["csr", "dep:bincode"]

diff --git a/packages/yew/src/html/component/lifecycle.rs b/packages/yew/src/html/component/lifecycle.rs
@@ -40,7 +40,7 @@ pub(crate) enum ComponentRenderState {
     },
     #[cfg(feature = "ssr")]
     Ssr {
-        sender: Option<crate::platform::sync::oneshot::Sender<Html>>,
+        sender: Option<futures::channel::oneshot::Sender<Html>>,
     },
 }
 

diff --git a/packages/yew/src/html/component/scope.rs b/packages/yew/src/html/component/scope.rs
@@ -260,12 +260,15 @@ impl<COMP: BaseComponent> Scope<COMP> {
 
 #[cfg(feature = "ssr")]
 mod feat_ssr {
+    use std::fmt::Write;
+
+    use futures::channel::oneshot;
+
     use super::*;
     use crate::html::component::lifecycle::{
         ComponentRenderState, CreateRunner, DestroyRunner, RenderRunner,
     };
-    use crate::platform::io::BufWriter;
-    use crate::platform::sync::oneshot;
+    use crate::platform::fmt::BufWriter;
     use crate::scheduler;
     use crate::virtual_dom::Collectable;
 
@@ -311,9 +314,9 @@ mod feat_ssr {
                 .await;
 
             if let Some(prepared_state) = self.get_component().unwrap().prepare_state() {
-                w.write(r#"<script type="application/x-yew-comp-state">"#.into());
-                w.write(prepared_state.into());
-                w.write(r#"</script>"#.into());
+                let _ = w.write_str(r#"<script type="application/x-yew-comp-state">"#);
+                let _ = w.write_str(&prepared_state);
+                let _ = w.write_str(r#"</script>"#);
             }
 
             if hydratable {

diff --git a/packages/yew/src/platform/fmt.rs b/packages/yew/src/platform/fmt.rs
@@ -0,0 +1,272 @@
+//! This module contains types for I/O functionality.
+
+use std::cell::RefCell;
+use std::fmt::{self, Write};
+use std::future::Future;
+use std::rc::Rc;
+use std::task::{Poll, Waker};
+
+use futures::future::{self, FusedFuture, MaybeDone};
+use futures::stream::{FusedStream, Stream};
+use pin_project::pin_project;
+
+pub(crate) static DEFAULT_BUF_SIZE: usize = 1024;
+
+enum BufStreamInner {
+    Combined {
+        buf: String,
+    },
+    Detached {
+        buf: String,
+        waker: Option<Waker>,
+        done: bool,
+    },
+}
+
+impl BufStreamInner {
+    #[inline]
+    const fn new_detached() -> Self {
+        Self::Detached {
+            buf: String::new(),
+            waker: None,
+            done: false,
+        }
+    }
+
+    #[inline]
+    const fn new_combined() -> Self {
+        Self::Combined { buf: String::new() }
+    }
+
+    #[inline]
+    fn buf(&self) -> &String {
+        match self {
+            Self::Combined { ref buf } => buf,
+            Self::Detached { ref buf, .. } => buf,
+        }
+    }
+
+    #[inline]
+    fn buf_mut(&mut self) -> &mut String {
+        match self {
+            Self::Combined { ref mut buf } => buf,
+            Self::Detached { ref mut buf, .. } => buf,
+        }
+    }
+
+    fn wake(&self) {
+        match self {
+            Self::Combined { .. } => {}
+            Self::Detached { ref waker, .. } => {
+                if let Some(m) = waker {
+                    m.wake_by_ref();
+                }
+            }
+        }
+    }
+
+    fn set_waker(&mut self, waker: Waker) {
+        match self {
+            Self::Combined { .. } => {}
+            Self::Detached {
+                waker: ref mut current_waker,
+                ..
+            } => {
+                *current_waker = Some(waker);
+            }
+        }
+    }
+
+    fn finish(&mut self) {
+        match self {
+            Self::Combined { .. } => {}
+            Self::Detached {
+                ref waker,
+                ref mut done,
+                ..
+            } => {
+                *done = true;
+                if let Some(m) = waker {
+                    m.wake_by_ref();
+                }
+            }
+        }
+    }
+
+    fn is_finished(&self) -> Option<bool> {
+        match self {
+            Self::Combined { .. } => None,
+            Self::Detached { ref buf, done, .. } => Some(buf.is_empty() && *done),
+        }
+    }
+}
+
+pub(crate) struct BufWriter {
+    inner: Rc<RefCell<BufStreamInner>>,
+    capacity: usize,
+}
+
+impl BufWriter {
+    pub fn capacity(&self) -> usize {
+        self.capacity
+    }
+}
+
+impl Write for BufWriter {
+    fn write_str(&mut self, s: &str) -> fmt::Result {
+        if s.is_empty() {
+            return Ok(());
+        }
+
+        let mut inner = self.inner.borrow_mut();
+        inner.wake();
+
+        let buf = inner.buf_mut();
+        if buf.is_empty() {
+            buf.reserve(self.capacity);
+        }
+
+        buf.write_str(s)
+    }
+
+    fn write_char(&mut self, c: char) -> fmt::Result {
+        let mut inner = self.inner.borrow_mut();
+        inner.wake();
+
+        let buf = inner.buf_mut();
+        if buf.is_empty() {
+            buf.reserve(self.capacity);
+        }
+
+        buf.write_char(c)
+    }
+
+    fn write_fmt(&mut self, args: fmt::Arguments<'_>) -> fmt::Result {
+        let mut inner = self.inner.borrow_mut();
+
+        let buf = inner.buf_mut();
+        if buf.is_empty() {
+            buf.reserve(self.capacity);
+        }
+
+        buf.write_fmt(args)
+    }
+}
+
+#[pin_project]
+pub(crate) struct BufStream<F>
+where
+    F: Future<Output = ()>,
+{
+    #[pin]
+    resolver: Option<MaybeDone<F>>,
+    inner: Rc<RefCell<BufStreamInner>>,
+}
+
+impl<F> BufStream<F>
+where
+    F: Future<Output = ()>,
+{
+    pub fn new<C>(capacity: usize, f: C) -> Self
+    where
+        C: FnOnce(BufWriter) -> F,
+    {
+        let inner = Rc::new(RefCell::new(BufStreamInner::new_combined()));
+
+        let resolver = {
+            let inner = inner.clone();
+            let w = BufWriter { inner, capacity };
+
+            f(w)
+        };
+
+        Self {
+            resolver: Some(future::maybe_done(resolver)),
+            inner,
+        }
+    }
+
+    pub fn new_with_resolver<C>(capacity: usize, f: C) -> (BufStream<F>, impl Future<Output = ()>)
+    where
+        C: FnOnce(BufWriter) -> F,
+    {
+        let inner = Rc::new(RefCell::new(BufStreamInner::new_detached()));
+
+        let resolver = {
+            let inner = inner.clone();
+            let w = {
+                let inner = inner.clone();
+                BufWriter { inner, capacity }
+            };
+
+            async move {
+                f(w).await;
+                inner.borrow_mut().finish();
+            }
+        };
+
+        (
+            Self {
+                resolver: None,
+                inner,
+            },
+            resolver,
+        )
+    }
+}
+
+impl<F> Stream for BufStream<F>
+where
+    F: Future<Output = ()>,
+{
+    type Item = String;
+
+    fn poll_next(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Option<Self::Item>> {
+        let this = self.project();
+
+        match this.resolver.as_pin_mut() {
+            Some(mut resolver) => {
+                let _ = resolver.as_mut().poll(cx);
+
+                let mut inner = this.inner.borrow_mut();
+
+                match (inner.buf().is_empty(), resolver.is_terminated()) {
+                    (true, true) => Poll::Ready(None),
+                    (true, false) => Poll::Pending,
+                    (false, _) => Poll::Ready(Some(inner.buf_mut().split_off(0))),
+                }
+            }
+            None => {
+                let mut inner = this.inner.borrow_mut();
+
+                if !inner.buf().is_empty() {
+                    return Poll::Ready(Some(inner.buf_mut().split_off(0)));
+                }
+
+                if Some(true) == inner.is_finished() {
+                    return Poll::Ready(None);
+                }
+
+                inner.set_waker(cx.waker().clone());
+                Poll::Pending
+            }
+        }
+    }
+}
+
+impl<F> FusedStream for BufStream<F>
+where
+    F: Future<Output = ()>,
+{
+    fn is_terminated(&self) -> bool {
+        let inner = self.inner.borrow();
+
+        match self.resolver.as_ref() {
+            Some(resolver) => inner.buf().is_empty() && resolver.is_terminated(),
+            None => inner.is_finished().unwrap_or_default(),
+        }
+    }
+}