New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Polled SSR Stream #2824
Merged
Merged
Polled SSR Stream #2824
Changes from 23 commits
Commits
Show all changes
32 commits
Select commit
Hold shift + click to select a range
5bf1ae8
Switch to pinned channels.
futursolo fb2603e
Fix ServerRenderer so it's not blocked until the result is resolved.
futursolo 892b759
Fix tests.
futursolo 29a46bd
Remove unused SendError.
futursolo 20fc4a3
Merge branch 'master' into pinned-channels
futursolo 8a41c3e
Implement a stream to be polled alongside rendering.
futursolo 8688045
Update Buffer Size.
futursolo 3bd9e0a
Make Send renderer work.
futursolo ed36760
Remove pinned channels.
futursolo 2799a22
Unified Naming.
futursolo 02a838d
Optimise code.
futursolo 3de0770
Restore capacity.
futursolo 689980e
merge 'master' into polled-stream
futursolo cbbd765
Remove unused profile.
futursolo 2045c49
Merge branch 'master' into polled-stream
futursolo a470862
Default to separate resolver.
futursolo 3ac02fd
Reduce allocations on string.
futursolo 345b745
Adjust API.
futursolo d1a6ab3
Remove duplicate trait bound.
futursolo f0863a4
Update docs.
futursolo 7dcaf6e
Remove capacity setting.
futursolo ced12ac
Merge branch 'master' into polled-stream
futursolo 475f52e
Merge branch 'master' into polled-stream
futursolo e757d35
Unsafe?
futursolo 0a399ba
Separate files.
futursolo 6ae5a65
Adjust inlining.
futursolo f310575
Merge branch 'master' into polled-stream
futursolo e18e0fd
Fix test.
futursolo 67348d3
Update notice.
futursolo f491d6c
Update documentation.
futursolo 2a300fe
Merge branch 'master' into polled-stream
futursolo a78f0b3
Fix tests.
futursolo File filter
Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,205 @@ | ||
//! Asynchronous utilities to work with `String`s. | ||
|
||
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, MaybeDone}; | ||
use futures::stream::{FusedStream, Stream}; | ||
use futures::StreamExt; | ||
use pin_project::pin_project; | ||
|
||
pub(crate) static BUF_SIZE: usize = 1024; | ||
|
||
enum BufStreamState { | ||
Ready, | ||
Pending(Waker), | ||
Done, | ||
} | ||
|
||
struct BufStreamInner { | ||
buf: String, | ||
state: BufStreamState, | ||
} | ||
|
||
impl BufStreamInner { | ||
#[inline] | ||
const fn new() -> Self { | ||
Self { | ||
buf: String::new(), | ||
state: BufStreamState::Ready, | ||
} | ||
} | ||
|
||
#[inline] | ||
fn wake(&self) { | ||
if let BufStreamState::Pending(ref waker) = self.state { | ||
waker.wake_by_ref(); | ||
} | ||
} | ||
|
||
#[inline] | ||
fn try_reserve(&mut self) { | ||
if self.buf.is_empty() { | ||
self.buf.reserve(BUF_SIZE); | ||
} | ||
} | ||
} | ||
|
||
pub(crate) struct BufWriter { | ||
inner: Rc<RefCell<BufStreamInner>>, | ||
} | ||
|
||
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(); | ||
inner.try_reserve(); | ||
|
||
inner.buf.write_str(s) | ||
} | ||
|
||
fn write_char(&mut self, c: char) -> fmt::Result { | ||
let mut inner = self.inner.borrow_mut(); | ||
|
||
inner.wake(); | ||
inner.try_reserve(); | ||
|
||
inner.buf.write_char(c) | ||
} | ||
|
||
fn write_fmt(&mut self, args: fmt::Arguments<'_>) -> fmt::Result { | ||
let mut inner = self.inner.borrow_mut(); | ||
|
||
inner.wake(); | ||
inner.try_reserve(); | ||
|
||
inner.buf.write_fmt(args) | ||
} | ||
} | ||
|
||
impl Drop for BufWriter { | ||
fn drop(&mut self) { | ||
let mut inner = self.inner.borrow_mut(); | ||
|
||
inner.wake(); | ||
inner.state = BufStreamState::Done; | ||
} | ||
} | ||
|
||
/// Creates an asynchronous buffer that operates over String. | ||
pub(crate) fn buffer() -> (BufWriter, BufReader) { | ||
let inner = Rc::new(RefCell::new(BufStreamInner::new())); | ||
|
||
let w = { | ||
let inner = inner.clone(); | ||
BufWriter { inner } | ||
}; | ||
|
||
let r = BufReader { inner }; | ||
|
||
(w, r) | ||
} | ||
|
||
pub(crate) struct BufReader { | ||
inner: Rc<RefCell<BufStreamInner>>, | ||
} | ||
|
||
impl Stream for BufReader { | ||
type Item = String; | ||
|
||
fn poll_next( | ||
self: std::pin::Pin<&mut Self>, | ||
cx: &mut std::task::Context<'_>, | ||
) -> std::task::Poll<Option<Self::Item>> { | ||
let mut inner = self.inner.borrow_mut(); | ||
|
||
if !inner.buf.is_empty() { | ||
let buf = std::mem::take(&mut inner.buf); | ||
return Poll::Ready(Some(buf)); | ||
} | ||
|
||
if let BufStreamState::Done = inner.state { | ||
return Poll::Ready(None); | ||
} | ||
|
||
inner.state = BufStreamState::Pending(cx.waker().clone()); | ||
Poll::Pending | ||
} | ||
} | ||
|
||
impl FusedStream for BufReader { | ||
fn is_terminated(&self) -> bool { | ||
let inner = self.inner.borrow(); | ||
|
||
matches!( | ||
(&inner.state, inner.buf.is_empty()), | ||
(BufStreamState::Done, true) | ||
) | ||
} | ||
} | ||
|
||
/// A buffered asynchronous string Stream. | ||
/// | ||
/// A BufStream combines a BufWriter - BufReader pair and a resolving future that writes to the | ||
/// buffer and polls the future alongside the buffer. | ||
#[pin_project] | ||
pub(crate) struct BufStream<F> | ||
where | ||
F: Future<Output = ()>, | ||
{ | ||
#[pin] | ||
resolver: MaybeDone<F>, | ||
inner: BufReader, | ||
} | ||
|
||
impl<F> BufStream<F> | ||
where | ||
F: Future<Output = ()>, | ||
{ | ||
/// Creates a `BufStream`. | ||
pub fn new<C>(f: C) -> Self | ||
where | ||
C: FnOnce(BufWriter) -> F, | ||
{ | ||
let (w, r) = buffer(); | ||
let resolver = future::maybe_done(f(w)); | ||
|
||
BufStream { inner: r, resolver } | ||
} | ||
} | ||
|
||
impl<F> Stream for BufStream<F> | ||
where | ||
F: Future<Output = ()>, | ||
{ | ||
type Item = String; | ||
|
||
#[inline] | ||
fn poll_next( | ||
self: std::pin::Pin<&mut Self>, | ||
cx: &mut std::task::Context<'_>, | ||
) -> std::task::Poll<Option<Self::Item>> { | ||
let this = self.project(); | ||
let _ = this.resolver.poll(cx); | ||
|
||
this.inner.poll_next_unpin(cx) | ||
} | ||
} | ||
|
||
impl<F> FusedStream for BufStream<F> | ||
where | ||
F: Future<Output = ()>, | ||
{ | ||
#[inline] | ||
fn is_terminated(&self) -> bool { | ||
self.inner.is_terminated() | ||
} | ||
} |
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file was deleted.
Oops, something went wrong.
This file was deleted.
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
pin-project-lite
maybe?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
See: https://docs.rs/pin-project-lite/0.2.8/pin_project_lite/#different-no-proc-macro-related-dependencies
The primary benefit of
#[pin_project]
is that rustfmt will continue to work for the struct.