Skip to content

fix: Fuse oneshot channel #147

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.

Sign up for GitHub

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

Merged
merged 4 commits into from
Mar 11, 2025
Merged

fix: Fuse oneshot channel #147

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
95ce856
Do not use tokio::select! for race2
rklaehn Mar 10, 2025
01588c0
fuse the oneshot channel
rklaehn Mar 10, 2025
2c0e602
Revert "Do not use tokio::select! for race2"
rklaehn Mar 11, 2025
17e56a5
Add remark about is_terminated
rklaehn Mar 11, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions src/server.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -329,7 +329,7 @@ where
{
pub(crate) fn new(recv: C::RecvStream) -> (Self, UnwrapToPending<RpcServerError<C>>) {
let (error_send, error_recv) = oneshot::channel();
let error_recv = UnwrapToPending(error_recv);
let error_recv = UnwrapToPending(futures_lite::future::fuse(error_recv));
(Self(recv, Some(error_send), PhantomData), error_recv)
}
}
Expand Down Expand Up @@ -449,12 +449,13 @@ impl<C: ConnectionErrors> fmt::Display for RpcServerError<C> {
impl<C: ConnectionErrors> error::Error for RpcServerError<C> {}

/// Take an oneshot receiver and just return Pending the underlying future returns `Err(oneshot::Canceled)`
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
/// Take an oneshot receiver and just return Pending the underlying future returns `Err(oneshot::Canceled)`
/// Take an oneshot receiver and just return Pending when the underlying future returns `Err(oneshot::Canceled)`

pub(crate) struct UnwrapToPending<T>(oneshot::Receiver<T>);
pub(crate) struct UnwrapToPending<T>(futures_lite::future::Fuse<oneshot::Receiver<T>>);

impl<T> Future for UnwrapToPending<T> {
type Output = T;

fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
// todo: use is_terminated from tokio 1.44 here to avoid the fused wrapper
match Pin::new(&mut self.0).poll(cx) {
Poll::Ready(Ok(x)) => Poll::Ready(x),
Poll::Ready(Err(_)) => Poll::Pending,
Expand Down
Loading