forked/hyper
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(client): replace futures mpsc for tokio mpsc in dispatcher

Browse Source
This commit is contained in:
Sean McArthur
2020-03-25 13:04:45 -07:00
parent 5b3724eeec
commit e6a6ddef7c
1 changed files with 12 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
src/client/dispatch.rs
View File

@@ -1,6 +1,6 @@
use futures_channel::{mpsc, oneshot}; use futures_channel::oneshot;
use futures_core::Stream;
use futures_util::future; use futures_util::future;
use tokio::sync::mpsc;
use crate::common::{task, Future, Pin, Poll}; use crate::common::{task, Future, Pin, Poll};
@@ -8,7 +8,7 @@ pub type RetryPromise<T, U> = oneshot::Receiver<Result<U, (crate::Error, Option<
pub type Promise<T> = oneshot::Receiver<Result<T, crate::Error>>; pub type Promise<T> = oneshot::Receiver<Result<T, crate::Error>>;
pub fn channel<T, U>() -> (Sender<T, U>, Receiver<T, U>) { pub fn channel<T, U>() -> (Sender<T, U>, Receiver<T, U>) {
let (tx, rx) = mpsc::unbounded(); let (tx, rx) = mpsc::unbounded_channel();
let (giver, taker) = want::new(); let (giver, taker) = want::new();
let tx = Sender { let tx = Sender {
buffered_once: false, buffered_once: false,
@@ -81,9 +81,9 @@ impl<T, U> Sender<T, U> {
} }
let (tx, rx) = oneshot::channel(); let (tx, rx) = oneshot::channel();
self.inner self.inner
.unbounded_send(Envelope(Some((val, Callback::Retry(tx))))) .send(Envelope(Some((val, Callback::Retry(tx)))))
.map(move |_| rx) .map(move |_| rx)
.map_err(|e| e.into_inner().0.take().expect("envelope not dropped").0) .map_err(|mut e| (e.0).0.take().expect("envelope not dropped").0)
} }
pub fn send(&mut self, val: T) -> Result<Promise<U>, T> { pub fn send(&mut self, val: T) -> Result<Promise<U>, T> {
@@ -92,9 +92,9 @@ impl<T, U> Sender<T, U> {
} }
let (tx, rx) = oneshot::channel(); let (tx, rx) = oneshot::channel();
self.inner self.inner
.unbounded_send(Envelope(Some((val, Callback::NoRetry(tx))))) .send(Envelope(Some((val, Callback::NoRetry(tx)))))
.map(move |_| rx) .map(move |_| rx)
.map_err(|e| e.into_inner().0.take().expect("envelope not dropped").0) .map_err(|mut e| (e.0).0.take().expect("envelope not dropped").0)
} }
pub fn unbound(self) -> UnboundedSender<T, U> { pub fn unbound(self) -> UnboundedSender<T, U> {
@@ -117,9 +117,9 @@ impl<T, U> UnboundedSender<T, U> {
pub fn try_send(&mut self, val: T) -> Result<RetryPromise<T, U>, T> { pub fn try_send(&mut self, val: T) -> Result<RetryPromise<T, U>, T> {
let (tx, rx) = oneshot::channel(); let (tx, rx) = oneshot::channel();
self.inner self.inner
.unbounded_send(Envelope(Some((val, Callback::Retry(tx))))) .send(Envelope(Some((val, Callback::Retry(tx)))))
.map(move |_| rx) .map(move |_| rx)
.map_err(|e| e.into_inner().0.take().expect("envelope not dropped").0) .map_err(|mut e| (e.0).0.take().expect("envelope not dropped").0)
} }
} }
@@ -142,7 +142,7 @@ impl<T, U> Receiver<T, U> {
&mut self, &mut self,
cx: &mut task::Context<'_>, cx: &mut task::Context<'_>,
) -> Poll<Option<(T, Callback<T, U>)>> { ) -> Poll<Option<(T, Callback<T, U>)>> {
match Pin::new(&mut self.inner).poll_next(cx) { match self.inner.poll_recv(cx) {
Poll::Ready(item) => { Poll::Ready(item) => {
Poll::Ready(item.map(|mut env| env.0.take().expect("envelope not dropped"))) Poll::Ready(item.map(|mut env| env.0.take().expect("envelope not dropped")))
} }
@@ -159,9 +159,8 @@ impl<T, U> Receiver<T, U> {
} }
pub(crate) fn try_recv(&mut self) -> Option<(T, Callback<T, U>)> { pub(crate) fn try_recv(&mut self) -> Option<(T, Callback<T, U>)> {
match self.inner.try_next() { match self.inner.try_recv() {
Ok(Some(mut env)) => env.0.take(), Ok(mut env) => env.0.take(),
Ok(None) => None,
Err(_) => None, Err(_) => None,
} }
} }