Bump to v0.16.0

* Use upstream sink compat shims
* Port to new Sink trait introduced in e101c891f04aba34ee29c6a8cd8321563c7e0161
* rustfmt
* Port to std::task::Context
* Add Google license header to bincode-transport/src/compat.rs
* Remove compat for it is no longer needed
* future::join as freestanding function
* Simplify dependencies
* Depend on futures-preview 0.3.0-alpha.14
* Fix infinite recursion

bincode-transport/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "tarpc-bincode-transport"
-version = "0.4.0"
+version = "0.5.0"
 authors = ["Tim Kuehn <tikue@google.com>"]
 edition = '2018'
 license = "MIT"
@@ -13,20 +13,17 @@ readme = "../README.md"
 description = "A bincode-based transport for tarpc services."
 
 [dependencies]
-bincode = { version = "1.0", features = ["i128"] }
+bincode = "1"
+futures-preview = { version = "0.3.0-alpha.14", features = ["compat"] }
 futures_legacy = { version = "0.1", package = "futures" }
 pin-utils = "0.1.0-alpha.4"
-rpc = { package = "tarpc-lib", version = "0.3", path = "../rpc", features = ["serde1"] }
+rpc = { package = "tarpc-lib", version = "0.4", path = "../rpc", features = ["serde1"] }
 serde = "1.0"
 tokio-io = "0.1"
 async-bincode = "0.4"
 tokio-tcp = "0.1"
 
-[target.'cfg(not(test))'.dependencies]
-futures-preview = { version = "0.3.0-alpha.13", features = ["compat"] }
-
 [dev-dependencies]
-futures-preview = { version = "0.3.0-alpha.13", features = ["compat"] }
 env_logger = "0.6"
 humantime = "1.0"
 libtest = "0.0.1"

bincode-transport/src/compat.rs

@@ -1,150 +0,0 @@
-use futures::{compat::Stream01CompatExt, prelude::*, ready};
-use futures_legacy::{
-    executor::{
-        self as executor01, Notify as Notify01, NotifyHandle as NotifyHandle01,
-        UnsafeNotify as UnsafeNotify01,
-    },
-    Async as Async01, AsyncSink as AsyncSink01, Sink as Sink01, Stream as Stream01,
-};
-use std::{
-    pin::Pin,
-    task::{self, Poll, Waker},
-};
-
-/// A shim to convert a 0.1 Sink + Stream to a 0.3 Sink + Stream.
-#[derive(Debug)]
-pub struct Compat<S, SinkItem> {
-    staged_item: Option<SinkItem>,
-    inner: S,
-}
-
-impl<S, SinkItem> Compat<S, SinkItem> {
-    /// Returns a new Compat.
-    pub fn new(inner: S) -> Self {
-        Compat {
-            inner,
-            staged_item: None,
-        }
-    }
-
-    /// Unwraps Compat, returning the inner value.
-    pub fn into_inner(self) -> S {
-        self.inner
-    }
-
-    /// Returns a reference to the value wrapped by Compat.
-    pub fn get_ref(&self) -> &S {
-        &self.inner
-    }
-}
-
-impl<S, SinkItem> Stream for Compat<S, SinkItem>
-where
-    S: Stream01,
-{
-    type Item = Result<S::Item, S::Error>;
-
-    fn poll_next(self: Pin<&mut Self>, waker: &Waker) -> Poll<Option<Self::Item>> {
-        unsafe {
-            let inner = &mut Pin::get_unchecked_mut(self).inner;
-            let mut compat = inner.compat();
-            let compat = Pin::new_unchecked(&mut compat);
-            match ready!(compat.poll_next(waker)) {
-                None => Poll::Ready(None),
-                Some(Ok(next)) => Poll::Ready(Some(Ok(next))),
-                Some(Err(e)) => Poll::Ready(Some(Err(e))),
-            }
-        }
-    }
-}
-
-impl<S, SinkItem> Sink for Compat<S, SinkItem>
-where
-    S: Sink01<SinkItem = SinkItem>,
-{
-    type SinkItem = SinkItem;
-    type SinkError = S::SinkError;
-
-    fn start_send(self: Pin<&mut Self>, item: SinkItem) -> Result<(), S::SinkError> {
-        let me = unsafe { Pin::get_unchecked_mut(self) };
-        assert!(me.staged_item.is_none());
-        me.staged_item = Some(item);
-        Ok(())
-    }
-
-    fn poll_ready(self: Pin<&mut Self>, waker: &Waker) -> Poll<Result<(), S::SinkError>> {
-        let notify = &WakerToHandle(waker);
-
-        executor01::with_notify(notify, 0, move || {
-            let me = unsafe { Pin::get_unchecked_mut(self) };
-            match me.staged_item.take() {
-                Some(staged_item) => match me.inner.start_send(staged_item) {
-                    Ok(AsyncSink01::Ready) => Poll::Ready(Ok(())),
-                    Ok(AsyncSink01::NotReady(item)) => {
-                        me.staged_item = Some(item);
-                        Poll::Pending
-                    }
-                    Err(e) => Poll::Ready(Err(e)),
-                },
-                None => Poll::Ready(Ok(())),
-            }
-        })
-    }
-
-    fn poll_flush(self: Pin<&mut Self>, waker: &Waker) -> Poll<Result<(), S::SinkError>> {
-        let notify = &WakerToHandle(waker);
-
-        executor01::with_notify(notify, 0, move || {
-            let me = unsafe { Pin::get_unchecked_mut(self) };
-            match me.inner.poll_complete() {
-                Ok(Async01::Ready(())) => Poll::Ready(Ok(())),
-                Ok(Async01::NotReady) => Poll::Pending,
-                Err(e) => Poll::Ready(Err(e)),
-            }
-        })
-    }
-
-    fn poll_close(self: Pin<&mut Self>, waker: &Waker) -> Poll<Result<(), S::SinkError>> {
-        let notify = &WakerToHandle(waker);
-
-        executor01::with_notify(notify, 0, move || {
-            let me = unsafe { Pin::get_unchecked_mut(self) };
-            match me.inner.close() {
-                Ok(Async01::Ready(())) => Poll::Ready(Ok(())),
-                Ok(Async01::NotReady) => Poll::Pending,
-                Err(e) => Poll::Ready(Err(e)),
-            }
-        })
-    }
-}
-
-#[derive(Clone, Debug)]
-struct WakerToHandle<'a>(&'a Waker);
-
-#[derive(Debug)]
-struct NotifyWaker(task::Waker);
-
-impl Notify01 for NotifyWaker {
-    fn notify(&self, _: usize) {
-        self.0.wake();
-    }
-}
-
-unsafe impl UnsafeNotify01 for NotifyWaker {
-    unsafe fn clone_raw(&self) -> NotifyHandle01 {
-        let ptr = Box::new(NotifyWaker(self.0.clone()));
-
-        NotifyHandle01::new(Box::into_raw(ptr))
-    }
-
-    unsafe fn drop_raw(&self) {
-        let ptr: *const dyn UnsafeNotify01 = self;
-        drop(Box::from_raw(ptr as *mut dyn UnsafeNotify01));
-    }
-}
-
-impl<'a> From<WakerToHandle<'a>> for NotifyHandle01 {
-    fn from(handle: WakerToHandle<'a>) -> NotifyHandle01 {
-        unsafe { NotifyWaker(handle.0.clone()).clone_raw() }
-    }
-}

bincode-transport/src/lib.rs

@@ -9,13 +9,8 @@
 #![feature(futures_api, arbitrary_self_types, await_macro, async_await)]
 #![deny(missing_docs, missing_debug_implementations)]
 
-use self::compat::Compat;
 use async_bincode::{AsyncBincodeStream, AsyncDestination};
-use futures::{
+use futures::{compat::*, prelude::*, ready};
-    compat::{Compat01As03, Future01CompatExt, Stream01CompatExt},
-    prelude::*,
-    ready,
-};
 use pin_utils::unsafe_pinned;
 use serde::{Deserialize, Serialize};
 use std::{
@@ -24,29 +19,20 @@ use std::{
     marker::PhantomData,
     net::SocketAddr,
     pin::Pin,
-    task::{Poll, Waker},
+    task::{Context, Poll},
 };
 use tokio_io::{AsyncRead, AsyncWrite};
 use tokio_tcp::{TcpListener, TcpStream};
 
-mod compat;
-
 /// A transport that serializes to, and deserializes from, a [`TcpStream`].
 #[derive(Debug)]
 pub struct Transport<S, Item, SinkItem> {
-    inner: Compat<AsyncBincodeStream<S, Item, SinkItem, AsyncDestination>, SinkItem>,
+    inner: Compat01As03Sink<AsyncBincodeStream<S, Item, SinkItem, AsyncDestination>, SinkItem>,
-}
-
-impl<S, Item, SinkItem> Transport<S, Item, SinkItem> {
-    /// Returns the transport underlying the bincode transport.
-    pub fn into_inner(self) -> S {
-        self.inner.into_inner().into_inner()
-    }
 }
 
 impl<S, Item, SinkItem> Transport<S, Item, SinkItem> {
     unsafe_pinned!(
-        inner: Compat<AsyncBincodeStream<S, Item, SinkItem, AsyncDestination>, SinkItem>
+        inner: Compat01As03Sink<AsyncBincodeStream<S, Item, SinkItem, AsyncDestination>, SinkItem>
     );
 }
 
@@ -57,8 +43,8 @@ where
 {
     type Item = io::Result<Item>;
 
-    fn poll_next(self: Pin<&mut Self>, waker: &Waker) -> Poll<Option<io::Result<Item>>> {
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<io::Result<Item>>> {
-        match self.inner().poll_next(waker) {
+        match self.inner().poll_next(cx) {
             Poll::Pending => Poll::Pending,
             Poll::Ready(None) => Poll::Ready(None),
             Poll::Ready(Some(Ok(next))) => Poll::Ready(Some(Ok(next))),
@@ -69,12 +55,11 @@ where
     }
 }
 
-impl<S, Item, SinkItem> Sink for Transport<S, Item, SinkItem>
+impl<S, Item, SinkItem> Sink<SinkItem> for Transport<S, Item, SinkItem>
 where
     S: AsyncWrite,
     SinkItem: Serialize,
 {
-    type SinkItem = SinkItem;
     type SinkError = io::Error;
 
     fn start_send(self: Pin<&mut Self>, item: SinkItem) -> io::Result<()> {
@@ -83,16 +68,16 @@ where
             .map_err(|e| io::Error::new(io::ErrorKind::Other, e))
     }
 
-    fn poll_ready(self: Pin<&mut Self>, waker: &Waker) -> Poll<io::Result<()>> {
+    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
-        convert(self.inner().poll_ready(waker))
+        convert(self.inner().poll_ready(cx))
     }
 
-    fn poll_flush(self: Pin<&mut Self>, waker: &Waker) -> Poll<io::Result<()>> {
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
-        convert(self.inner().poll_flush(waker))
+        convert(self.inner().poll_flush(cx))
     }
 
-    fn poll_close(self: Pin<&mut Self>, waker: &Waker) -> Poll<io::Result<()>> {
+    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
-        convert(self.inner().poll_close(waker))
+        convert(self.inner().poll_close(cx))
     }
 }
 
@@ -133,7 +118,7 @@ where
 impl<S, Item, SinkItem> From<S> for Transport<S, Item, SinkItem> {
     fn from(inner: S) -> Self {
         Transport {
-            inner: Compat::new(AsyncBincodeStream::from(inner).for_async()),
+            inner: Compat01As03Sink::new(AsyncBincodeStream::from(inner).for_async()),
         }
     }
 }
@@ -189,8 +174,8 @@ where
 {
     type Item = io::Result<Transport<TcpStream, Item, SinkItem>>;
 
-    fn poll_next(self: Pin<&mut Self>, waker: &Waker) -> Poll<Option<Self::Item>> {
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        let next = ready!(self.incoming().poll_next(waker)?);
+        let next = ready!(self.incoming().poll_next(cx)?);
         Poll::Ready(next.map(|conn| Ok(new(conn))))
     }
 }

bincode-transport/tests/cancel.rs

@@ -11,7 +11,7 @@
 use futures::{
     compat::{Executor01CompatExt, Future01CompatExt},
     prelude::*,
-    stream,
+    stream::FuturesUnordered,
 };
 use log::{info, trace};
 use rand::distributions::{Distribution, Normal};
@@ -126,7 +126,10 @@ async fn run() -> io::Result<()> {
         let response = client.call(ctx, "ping".into());
         requests.push(response.map(move |r| (trace_id, r)));
     }
-    let (fastest_response, _) = await!(stream::futures_unordered(requests).into_future());
+    let (fastest_response, _) = await!(requests
+        .into_iter()
+        .collect::<FuturesUnordered<_>>()
+        .into_future());
     let (trace_id, resp) = fastest_response.unwrap();
     info!("[{}] fastest_response = {:?}", trace_id, resp);
 

example-service/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "tarpc-example-service"
-version = "0.3.0"
+version = "0.4.0"
 authors = ["Tim Kuehn <tikue@google.com>"]
 edition = "2018"
 license = "MIT"
@@ -13,11 +13,11 @@ readme = "../README.md"
 description = "An example server built on tarpc."
 
 [dependencies]
-bincode-transport = { package = "tarpc-bincode-transport", version = "0.4", path = "../bincode-transport" }
+bincode-transport = { package = "tarpc-bincode-transport", version = "0.5", path = "../bincode-transport" }
 clap = "2.0"
-futures-preview = { version = "0.3.0-alpha.13", features = ["compat"] }
+futures-preview = { version = "0.3.0-alpha.14", features = ["compat"] }
 serde = { version = "1.0" }
-tarpc = { version = "0.15", path = "../tarpc", features = ["serde1"] }
+tarpc = { version = "0.16", path = "../tarpc", features = ["serde1"] }
 tokio = "0.1"
 tokio-executor = "0.1"
 

rpc/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "tarpc-lib"
-version = "0.3.0"
+version = "0.4.0"
 authors = ["Tim Kuehn <tikue@google.com>"]
 edition = '2018'
 license = "MIT"
@@ -18,6 +18,7 @@ serde1 = ["trace/serde", "serde", "serde/derive"]
 
 [dependencies]
 fnv = "1.0"
+futures-preview = { version = "0.3.0-alpha.14", features = ["compat"] }
 humantime = "1.0"
 log = "0.4"
 pin-utils = "0.1.0-alpha.4"
@@ -26,11 +27,7 @@ tokio-timer = "0.2"
 trace = { package = "tarpc-trace", version = "0.2", path = "../trace" }
 serde = { optional = true, version = "1.0" }
 
-[target.'cfg(not(test))'.dependencies]
-futures-preview = { version = "0.3.0-alpha.13", features = ["compat"] }
-
 [dev-dependencies]
-futures-preview = { version = "0.3.0-alpha.13", features = ["compat"] }
+futures-test-preview = { version = "0.3.0-alpha.14" }
-futures-test-preview = { version = "0.3.0-alpha.13" }
 env_logger = "0.6"
 tokio = "0.1"

rpc/src/client/channel.rs

@@ -15,7 +15,7 @@ use futures::{
     prelude::*,
     ready,
     stream::Fuse,
-    task::Waker,
+    task::Context,
     Poll,
 };
 use humantime::format_rfc3339;
@@ -65,14 +65,19 @@ struct Send<'a, Req, Resp> {
     fut: MapOkDispatchResponse<SendMapErrConnectionReset<'a, Req, Resp>, Resp>,
 }
 
-type SendMapErrConnectionReset<'a, Req, Resp> =
+type SendMapErrConnectionReset<'a, Req, Resp> = MapErrConnectionReset<
-    MapErrConnectionReset<futures::sink::Send<'a, mpsc::Sender<DispatchRequest<Req, Resp>>>>;
+    futures::sink::Send<'a, mpsc::Sender<DispatchRequest<Req, Resp>>, DispatchRequest<Req, Resp>>,
+>;
 
 impl<'a, Req, Resp> Send<'a, Req, Resp> {
     unsafe_pinned!(
         fut: MapOkDispatchResponse<
             MapErrConnectionReset<
-                futures::sink::Send<'a, mpsc::Sender<DispatchRequest<Req, Resp>>>,
+                futures::sink::Send<
+                    'a,
+                    mpsc::Sender<DispatchRequest<Req, Resp>>,
+                    DispatchRequest<Req, Resp>,
+                >,
             >,
             Resp,
         >
@@ -82,8 +87,8 @@ impl<'a, Req, Resp> Send<'a, Req, Resp> {
 impl<'a, Req, Resp> Future for Send<'a, Req, Resp> {
     type Output = io::Result<DispatchResponse<Resp>>;
 
-    fn poll(mut self: Pin<&mut Self>, waker: &Waker) -> Poll<Self::Output> {
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        self.as_mut().fut().poll(waker)
+        self.as_mut().fut().poll(cx)
     }
 }
 
@@ -101,8 +106,8 @@ impl<'a, Req, Resp> Call<'a, Req, Resp> {
 impl<'a, Req, Resp> Future for Call<'a, Req, Resp> {
     type Output = io::Result<Resp>;
 
-    fn poll(mut self: Pin<&mut Self>, waker: &Waker) -> Poll<Self::Output> {
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        self.as_mut().fut().poll(waker)
+        self.as_mut().fut().poll(cx)
     }
 }
 
@@ -177,8 +182,8 @@ impl<Resp> DispatchResponse<Resp> {
 impl<Resp> Future for DispatchResponse<Resp> {
     type Output = io::Result<Resp>;
 
-    fn poll(mut self: Pin<&mut Self>, waker: &Waker) -> Poll<io::Result<Resp>> {
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<Resp>> {
-        let resp = ready!(self.response.poll_unpin(waker));
+        let resp = ready!(self.response.poll_unpin(cx));
 
         self.complete = true;
 
@@ -258,6 +263,7 @@ where
 {
     let (to_dispatch, pending_requests) = mpsc::channel(config.pending_request_buffer);
     let (cancellation, canceled_requests) = cancellations();
+    let canceled_requests = canceled_requests.fuse();
 
     crate::spawn(
         RequestDispatch {
@@ -296,7 +302,7 @@ struct RequestDispatch<Req, Resp, C> {
     /// Requests waiting to be written to the wire.
     pending_requests: Fuse<mpsc::Receiver<DispatchRequest<Req, Resp>>>,
     /// Requests that were dropped.
-    canceled_requests: CanceledRequests,
+    canceled_requests: Fuse<CanceledRequests>,
     /// Requests already written to the wire that haven't yet received responses.
     in_flight_requests: FnvHashMap<u64, InFlightData<Resp>>,
     /// Configures limits to prevent unlimited resource usage.
@@ -313,12 +319,12 @@ where
 {
     unsafe_pinned!(server_addr: SocketAddr);
     unsafe_pinned!(in_flight_requests: FnvHashMap<u64, InFlightData<Resp>>);
-    unsafe_pinned!(canceled_requests: CanceledRequests);
+    unsafe_pinned!(canceled_requests: Fuse<CanceledRequests>);
     unsafe_pinned!(pending_requests: Fuse<mpsc::Receiver<DispatchRequest<Req, Resp>>>);
     unsafe_pinned!(transport: Fuse<C>);
 
-    fn pump_read(self: &mut Pin<&mut Self>, waker: &Waker) -> PollIo<()> {
+    fn pump_read(self: &mut Pin<&mut Self>, cx: &mut Context<'_>) -> PollIo<()> {
-        Poll::Ready(match ready!(self.as_mut().transport().poll_next(waker)?) {
+        Poll::Ready(match ready!(self.as_mut().transport().poll_next(cx)?) {
             Some(response) => {
                 self.complete(response);
                 Some(Ok(()))
@@ -330,13 +336,13 @@ where
         })
     }
 
-    fn pump_write(self: &mut Pin<&mut Self>, waker: &Waker) -> PollIo<()> {
+    fn pump_write(self: &mut Pin<&mut Self>, cx: &mut Context<'_>) -> PollIo<()> {
         enum ReceiverStatus {
             NotReady,
             Closed,
         }
 
-        let pending_requests_status = match self.poll_next_request(waker)? {
+        let pending_requests_status = match self.poll_next_request(cx)? {
             Poll::Ready(Some(dispatch_request)) => {
                 self.write_request(dispatch_request)?;
                 return Poll::Ready(Some(Ok(())));
@@ -345,7 +351,7 @@ where
             Poll::Pending => ReceiverStatus::NotReady,
         };
 
-        let canceled_requests_status = match self.poll_next_cancellation(waker)? {
+        let canceled_requests_status = match self.poll_next_cancellation(cx)? {
             Poll::Ready(Some((context, request_id))) => {
                 self.write_cancel(context, request_id)?;
                 return Poll::Ready(Some(Ok(())));
@@ -356,12 +362,12 @@ where
 
         match (pending_requests_status, canceled_requests_status) {
             (ReceiverStatus::Closed, ReceiverStatus::Closed) => {
-                ready!(self.as_mut().transport().poll_flush(waker)?);
+                ready!(self.as_mut().transport().poll_flush(cx)?);
                 Poll::Ready(None)
             }
             (ReceiverStatus::NotReady, _) | (_, ReceiverStatus::NotReady) => {
                 // No more messages to process, so flush any messages buffered in the transport.
-                ready!(self.as_mut().transport().poll_flush(waker)?);
+                ready!(self.as_mut().transport().poll_flush(cx)?);
 
                 // Even if we fully-flush, we return Pending, because we have no more requests
                 // or cancellations right now.
@@ -373,7 +379,7 @@ where
     /// Yields the next pending request, if one is ready to be sent.
     fn poll_next_request(
         self: &mut Pin<&mut Self>,
-        waker: &Waker,
+        cx: &mut Context<'_>,
     ) -> PollIo<DispatchRequest<Req, Resp>> {
         if self.as_mut().in_flight_requests().len() >= self.config.max_in_flight_requests {
             info!(
@@ -387,13 +393,13 @@ where
             return Poll::Pending;
         }
 
-        while let Poll::Pending = self.as_mut().transport().poll_ready(waker)? {
+        while let Poll::Pending = self.as_mut().transport().poll_ready(cx)? {
             // We can't yield a request-to-be-sent before the transport is capable of buffering it.
-            ready!(self.as_mut().transport().poll_flush(waker)?);
+            ready!(self.as_mut().transport().poll_flush(cx)?);
         }
 
         loop {
-            match ready!(self.as_mut().pending_requests().poll_next_unpin(waker)) {
+            match ready!(self.as_mut().pending_requests().poll_next_unpin(cx)) {
                 Some(request) => {
                     if request.response_completion.is_canceled() {
                         trace!(
@@ -416,14 +422,15 @@ where
     /// Yields the next pending cancellation, and, if one is ready, cancels the associated request.
     fn poll_next_cancellation(
         self: &mut Pin<&mut Self>,
-        waker: &Waker,
+        cx: &mut Context<'_>,
     ) -> PollIo<(context::Context, u64)> {
-        while let Poll::Pending = self.as_mut().transport().poll_ready(waker)? {
+        while let Poll::Pending = self.as_mut().transport().poll_ready(cx)? {
-            ready!(self.as_mut().transport().poll_flush(waker)?);
+            ready!(self.as_mut().transport().poll_flush(cx)?);
         }
 
         loop {
-            match ready!(self.as_mut().canceled_requests().poll_next_unpin(waker)) {
+            let cancellation = self.as_mut().canceled_requests().poll_next_unpin(cx);
+            match ready!(cancellation) {
                 Some(request_id) => {
                     if let Some(in_flight_data) =
                         self.as_mut().in_flight_requests().remove(&request_id)
@@ -530,10 +537,10 @@ where
 {
     type Output = io::Result<()>;
 
-    fn poll(mut self: Pin<&mut Self>, waker: &Waker) -> Poll<io::Result<()>> {
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
         trace!("[{}] RequestDispatch::poll", self.as_mut().server_addr());
         loop {
-            match (self.pump_read(waker)?, self.pump_write(waker)?) {
+            match (self.pump_read(cx)?, self.pump_write(cx)?) {
                 (read, write @ Poll::Ready(None)) => {
                     if self.as_mut().in_flight_requests().is_empty() {
                         info!(
@@ -620,8 +627,8 @@ impl RequestCancellation {
 impl Stream for CanceledRequests {
     type Item = u64;
 
-    fn poll_next(mut self: Pin<&mut Self>, waker: &Waker) -> Poll<Option<u64>> {
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<u64>> {
-        self.0.poll_next_unpin(waker)
+        self.0.poll_next_unpin(cx)
     }
 }
 
@@ -652,8 +659,8 @@ where
 {
     type Output = io::Result<Fut::Ok>;
 
-    fn poll(mut self: Pin<&mut Self>, waker: &Waker) -> Poll<Self::Output> {
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        match self.as_mut().future().try_poll(waker) {
+        match self.as_mut().future().try_poll(cx) {
             Poll::Pending => Poll::Pending,
             Poll::Ready(result) => {
                 self.finished().take().expect(
@@ -692,8 +699,8 @@ where
 {
     type Output = Result<DispatchResponse<Resp>, Fut::Error>;
 
-    fn poll(mut self: Pin<&mut Self>, waker: &Waker) -> Poll<Self::Output> {
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        match self.as_mut().future().try_poll(waker) {
+        match self.as_mut().future().try_poll(cx) {
             Poll::Pending => Poll::Pending,
             Poll::Ready(result) => {
                 let response = self
@@ -735,8 +742,8 @@ where
 {
     type Output = Result<Fut2::Ok, Fut2::Error>;
 
-    fn poll(self: Pin<&mut Self>, waker: &Waker) -> Poll<Self::Output> {
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        self.try_chain().poll(waker, |result| match result {
+        self.try_chain().poll(cx, |result| match result {
             Ok(ok) => TryChainAction::Future(ok),
             Err(err) => TryChainAction::Output(Err(err)),
         })
@@ -768,7 +775,11 @@ where
         TryChain::First(fut1)
     }
 
-    fn poll<F>(self: Pin<&mut Self>, waker: &Waker, f: F) -> Poll<Result<Fut2::Ok, Fut2::Error>>
+    fn poll<F>(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        f: F,
+    ) -> Poll<Result<Fut2::Ok, Fut2::Error>>
     where
         F: FnOnce(Result<Fut1::Ok, Fut1::Error>) -> TryChainAction<Fut2>,
     {
@@ -781,14 +792,14 @@ where
             let output = match this {
                 TryChain::First(fut1) => {
                     // Poll the first future
-                    match unsafe { Pin::new_unchecked(fut1) }.try_poll(waker) {
+                    match unsafe { Pin::new_unchecked(fut1) }.try_poll(cx) {
                         Poll::Pending => return Poll::Pending,
                         Poll::Ready(output) => output,
                     }
                 }
                 TryChain::Second(fut2) => {
                     // Poll the second future
-                    return unsafe { Pin::new_unchecked(fut2) }.try_poll(waker);
+                    return unsafe { Pin::new_unchecked(fut2) }.try_poll(cx);
                 }
                 TryChain::Empty => {
                     panic!("future must not be polled after it returned `Poll::Ready`");
@@ -807,7 +818,9 @@ where
 
 #[cfg(test)]
 mod tests {
-    use super::{CanceledRequests, Channel, RequestCancellation, RequestDispatch};
+    use super::{
+        CanceledRequests, Channel, DispatchResponse, RequestCancellation, RequestDispatch,
+    };
     use crate::{
         client::Config,
         context,
@@ -815,7 +828,7 @@ mod tests {
         ClientMessage, Response,
     };
     use fnv::FnvHashMap;
-    use futures::{channel::mpsc, prelude::*, Poll};
+    use futures::{channel::mpsc, prelude::*, task::Context, Poll};
     use futures_test::task::noop_waker_ref;
     use std::{
         marker,
@@ -828,20 +841,12 @@ mod tests {
     #[test]
     fn stage_request() {
         let (mut dispatch, mut channel, _server_channel) = set_up();
-
-        // Test that a request future dropped before it's processed by dispatch will cause the request
-        // to not be added to the in-flight request map.
-        let _resp = tokio::runtime::current_thread::block_on_all(
-            channel
-                .send(context::current(), "hi".to_string())
-                .boxed()
-                .compat(),
-        );
-
         let mut dispatch = Pin::new(&mut dispatch);
-        let waker = &noop_waker_ref();
+        let cx = &mut Context::from_waker(&noop_waker_ref());
 
-        let req = dispatch.poll_next_request(waker).ready();
+        let _resp = send_request(&mut channel, "hi");
+
+        let req = dispatch.poll_next_request(cx).ready();
         assert!(req.is_some());
 
         let req = req.unwrap();
@@ -849,49 +854,77 @@ mod tests {
         assert_eq!(req.request, "hi".to_string());
     }
 
+    // Regression test for  https://github.com/google/tarpc/issues/220
     #[test]
-    fn stage_request_response_future_dropped() {
+    fn stage_request_channel_dropped_doesnt_panic() {
-        let (mut dispatch, mut channel, _server_channel) = set_up();
+        let (mut dispatch, mut channel, mut server_channel) = set_up();
+        let mut dispatch = Pin::new(&mut dispatch);
+        let cx = &mut Context::from_waker(&noop_waker_ref());
 
-        // Test that a request future dropped before it's processed by dispatch will cause the request
+        let _ = send_request(&mut channel, "hi");
-        // to not be added to the in-flight request map.
-        let resp = tokio::runtime::current_thread::block_on_all(
-            channel
-                .send(context::current(), "hi".into())
-                .boxed()
-                .compat(),
-        )
-        .unwrap();
-        drop(resp);
         drop(channel);
 
-        let mut dispatch = Pin::new(&mut dispatch);
+        assert!(dispatch.as_mut().poll(cx).is_ready());
-        let waker = &noop_waker_ref();
+        send_response(
-
+            &mut server_channel,
-        dispatch.poll_next_cancellation(waker).unwrap();
+            Response {
-        assert!(dispatch.poll_next_request(waker).ready().is_none());
+                request_id: 0,
+                message: Ok("hello".into()),
+            },
+        );
+        tokio::runtime::current_thread::block_on_all(dispatch.boxed().compat()).unwrap();
     }
 
     #[test]
-    fn stage_request_response_future_closed() {
+    fn stage_request_response_future_dropped_is_canceled_before_sending() {
         let (mut dispatch, mut channel, _server_channel) = set_up();
+        let mut dispatch = Pin::new(&mut dispatch);
+        let cx = &mut Context::from_waker(&noop_waker_ref());
+
+        let _ = send_request(&mut channel, "hi");
+
+        // Drop the channel so polling returns none if no requests are currently ready.
+        drop(channel);
+        // Test that a request future dropped before it's processed by dispatch will cause the request
+        // to not be added to the in-flight request map.
+        assert!(dispatch.poll_next_request(cx).ready().is_none());
+    }
+
+    #[test]
+    fn stage_request_response_future_dropped_is_canceled_after_sending() {
+        let (mut dispatch, mut channel, _server_channel) = set_up();
+        let cx = &mut Context::from_waker(&noop_waker_ref());
+        let mut dispatch = Pin::new(&mut dispatch);
+
+        let req = send_request(&mut channel, "hi");
+
+        assert!(dispatch.as_mut().pump_write(cx).ready().is_some());
+        assert!(!dispatch.as_mut().in_flight_requests().is_empty());
+
+        // Test that a request future dropped after it's processed by dispatch will cause the request
+        // to be removed from the in-flight request map.
+        drop(req);
+        if let Poll::Ready(Some(_)) = dispatch.as_mut().poll_next_cancellation(cx).unwrap() {
+            // ok
+        } else {
+            panic!("Expected request to be cancelled")
+        };
+        assert!(dispatch.in_flight_requests().is_empty());
+    }
+
+    #[test]
+    fn stage_request_response_closed_skipped() {
+        let (mut dispatch, mut channel, _server_channel) = set_up();
+        let mut dispatch = Pin::new(&mut dispatch);
+        let cx = &mut Context::from_waker(&noop_waker_ref());
 
         // Test that a request future that's closed its receiver but not yet canceled its request --
         // i.e. still in `drop fn` -- will cause the request to not be added to the in-flight request
         // map.
-        let resp = tokio::runtime::current_thread::block_on_all(
+        let mut resp = send_request(&mut channel, "hi");
-            channel
+        resp.response.get_mut().close();
-                .send(context::current(), "hi".into())
-                .boxed()
-                .compat(),
-        )
-        .unwrap();
-        drop(resp);
-        drop(channel);
 
-        let mut dispatch = Pin::new(&mut dispatch);
+        assert!(dispatch.poll_next_request(cx).is_pending());
-        let waker = &noop_waker_ref();
-        assert!(dispatch.poll_next_request(waker).ready().is_none());
     }
 
     fn set_up() -> (
@@ -908,7 +941,7 @@ mod tests {
         let dispatch = RequestDispatch::<String, String, _> {
             transport: client_channel.fuse(),
             pending_requests: pending_requests.fuse(),
-            canceled_requests: CanceledRequests(canceled_requests),
+            canceled_requests: CanceledRequests(canceled_requests).fuse(),
             in_flight_requests: FnvHashMap::default(),
             config: Config::default(),
             server_addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0),
@@ -925,6 +958,27 @@ mod tests {
         (dispatch, channel, server_channel)
     }
 
+    fn send_request(
+        channel: &mut Channel<String, String>,
+        request: &str,
+    ) -> DispatchResponse<String> {
+        tokio::runtime::current_thread::block_on_all(
+            channel
+                .send(context::current(), request.to_string())
+                .boxed()
+                .compat(),
+        )
+        .unwrap()
+    }
+
+    fn send_response(
+        channel: &mut UnboundedChannel<ClientMessage<String>, Response<String>>,
+        response: Response<String>,
+    ) {
+        tokio::runtime::current_thread::block_on_all(channel.send(response).boxed().compat())
+            .unwrap();
+    }
+
     trait PollTest {
         type T;
         fn unwrap(self) -> Poll<Self::T>;
@@ -955,5 +1009,4 @@ mod tests {
             }
         }
     }
-
 }

rpc/src/lib.rs

@@ -156,7 +156,7 @@ thread_local! {
             // INIT must always be called before accessing SPAWN.
             // Otherwise, accessing SPAWN can trigger undefined behavior due to race conditions.
             INIT.call_once(|| {});
-            RefCell::new(SEED_SPAWN.clone().expect("init() must be called."))
+            RefCell::new(SEED_SPAWN.as_ref().expect("init() must be called.").box_clone())
         }
     };
 }
@@ -182,12 +182,6 @@ trait CloneSpawn: Spawn {
     fn box_clone(&self) -> Box<dyn CloneSpawn>;
 }
 
-impl Clone for Box<dyn CloneSpawn> {
-    fn clone(&self) -> Self {
-        self.box_clone()
-    }
-}
-
 impl<S: Spawn + Clone + 'static> CloneSpawn for S {
     fn box_clone(&self) -> Box<dyn CloneSpawn> {
         Box::new(self.clone())

rpc/src/server/filter.rs

@@ -15,7 +15,7 @@ use futures::{
     prelude::*,
     ready,
     stream::Fuse,
-    task::{Poll, Waker},
+    task::{Context, Poll},
 };
 use log::{debug, error, info, trace, warn};
 use pin_utils::unsafe_pinned;
@@ -197,7 +197,10 @@ impl<S, Req, Resp> ConnectionFilter<S, Req, Resp> {
         }
     }
 
-    fn poll_listener<C>(mut self: Pin<&mut Self>, cx: &Waker) -> PollIo<NewConnection<Req, Resp, C>>
+    fn poll_listener<C>(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> PollIo<NewConnection<Req, Resp, C>>
     where
         S: Stream<Item = Result<C, io::Error>>,
         C: Transport<Item = ClientMessage<Req>, SinkItem = Response<Resp>> + Send,
@@ -208,7 +211,10 @@ impl<S, Req, Resp> ConnectionFilter<S, Req, Resp> {
         }
     }
 
-    fn poll_closed_connections(self: &mut Pin<&mut Self>, cx: &Waker) -> Poll<io::Result<()>> {
+    fn poll_closed_connections(
+        self: &mut Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<io::Result<()>> {
         match ready!(self.as_mut().closed_connections_rx().poll_next_unpin(cx)) {
             Some(addr) => {
                 self.handle_closed_connection(&addr);
@@ -226,7 +232,7 @@ where
 {
     type Item = io::Result<Channel<Req, Resp, T>>;
 
-    fn poll_next(mut self: Pin<&mut Self>, cx: &Waker) -> PollIo<Channel<Req, Resp, T>> {
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> PollIo<Channel<Req, Resp, T>> {
         loop {
             match (
                 self.as_mut().poll_listener(cx)?,

rpc/src/server/mod.rs

@@ -7,7 +7,7 @@
 //! Provides a server that concurrently handles many connections sending multiplexed requests.
 
 use crate::{
-    context::Context, util::deadline_compat, util::AsDuration, util::Compact, ClientMessage,
+    context, util::deadline_compat, util::AsDuration, util::Compact, ClientMessage,
     ClientMessageKind, PollIo, Request, Response, ServerError, Transport,
 };
 use fnv::FnvHashMap;
@@ -17,7 +17,7 @@ use futures::{
     prelude::*,
     ready,
     stream::Fuse,
-    task::{Poll, Waker},
+    task::{Context, Poll},
     try_ready,
 };
 use humantime::format_rfc3339;
@@ -128,12 +128,12 @@ where
     Req: Send + 'static,
     Resp: Send + 'static,
     T: Transport<Item = ClientMessage<Req>, SinkItem = Response<Resp>> + Send + 'static,
-    F: FnOnce(Context, Req) -> Fut + Send + 'static + Clone,
+    F: FnOnce(context::Context, Req) -> Fut + Send + 'static + Clone,
     Fut: Future<Output = io::Result<Resp>> + Send + 'static,
 {
     type Output = ();
 
-    fn poll(mut self: Pin<&mut Self>, cx: &Waker) -> Poll<()> {
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
         while let Some(channel) = ready!(self.as_mut().incoming().poll_next(cx)) {
             match channel {
                 Ok(channel) => {
@@ -165,7 +165,7 @@ where
     /// Responds to all requests with `request_handler`.
     fn respond_with<F, Fut>(self, request_handler: F) -> Running<Self, F>
     where
-        F: FnOnce(Context, Req) -> Fut + Send + 'static + Clone,
+        F: FnOnce(context::Context, Req) -> Fut + Send + 'static + Clone,
         Fut: Future<Output = io::Result<Resp>> + Send + 'static,
     {
         Running {
@@ -234,15 +234,24 @@ where
         self.as_mut().transport().start_send(response)
     }
 
-    pub(crate) fn poll_ready(mut self: Pin<&mut Self>, cx: &Waker) -> Poll<io::Result<()>> {
+    pub(crate) fn poll_ready(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<io::Result<()>> {
         self.as_mut().transport().poll_ready(cx)
     }
 
-    pub(crate) fn poll_flush(mut self: Pin<&mut Self>, cx: &Waker) -> Poll<io::Result<()>> {
+    pub(crate) fn poll_flush(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<io::Result<()>> {
         self.as_mut().transport().poll_flush(cx)
     }
 
-    pub(crate) fn poll_next(mut self: Pin<&mut Self>, cx: &Waker) -> PollIo<ClientMessage<Req>> {
+    pub(crate) fn poll_next(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> PollIo<ClientMessage<Req>> {
         self.as_mut().transport().poll_next(cx)
     }
 
@@ -255,7 +264,7 @@ where
     /// responses and resolves when the connection is closed.
     pub fn respond_with<F, Fut>(self, f: F) -> impl Future<Output = ()>
     where
-        F: FnOnce(Context, Req) -> Fut + Send + 'static + Clone,
+        F: FnOnce(context::Context, Req) -> Fut + Send + 'static + Clone,
         Fut: Future<Output = io::Result<Resp>> + Send + 'static,
         Req: 'static,
         Resp: 'static,
@@ -281,9 +290,9 @@ where
 struct ClientHandler<Req, Resp, T, F> {
     channel: Channel<Req, Resp, T>,
     /// Responses waiting to be written to the wire.
-    pending_responses: Fuse<mpsc::Receiver<(Context, Response<Resp>)>>,
+    pending_responses: Fuse<mpsc::Receiver<(context::Context, Response<Resp>)>>,
     /// Handed out to request handlers to fan in responses.
-    responses_tx: mpsc::Sender<(Context, Response<Resp>)>,
+    responses_tx: mpsc::Sender<(context::Context, Response<Resp>)>,
     /// Number of requests currently being responded to.
     in_flight_requests: FnvHashMap<u64, AbortHandle>,
     /// Request handler.
@@ -293,8 +302,8 @@ struct ClientHandler<Req, Resp, T, F> {
 impl<Req, Resp, T, F> ClientHandler<Req, Resp, T, F> {
     unsafe_pinned!(channel: Channel<Req, Resp, T>);
     unsafe_pinned!(in_flight_requests: FnvHashMap<u64, AbortHandle>);
-    unsafe_pinned!(pending_responses: Fuse<mpsc::Receiver<(Context, Response<Resp>)>>);
+    unsafe_pinned!(pending_responses: Fuse<mpsc::Receiver<(context::Context, Response<Resp>)>>);
-    unsafe_pinned!(responses_tx: mpsc::Sender<(Context, Response<Resp>)>);
+    unsafe_pinned!(responses_tx: mpsc::Sender<(context::Context, Response<Resp>)>);
     // For this to be safe, field f must be private, and code in this module must never
     // construct PinMut<F>.
     unsafe_unpinned!(f: F);
@@ -305,12 +314,15 @@ where
     Req: Send + 'static,
     Resp: Send + 'static,
     T: Transport<Item = ClientMessage<Req>, SinkItem = Response<Resp>> + Send,
-    F: FnOnce(Context, Req) -> Fut + Send + 'static + Clone,
+    F: FnOnce(context::Context, Req) -> Fut + Send + 'static + Clone,
     Fut: Future<Output = io::Result<Resp>> + Send + 'static,
 {
     /// If at max in-flight requests, check that there's room to immediately write a throttled
     /// response.
-    fn poll_ready_if_throttling(mut self: Pin<&mut Self>, cx: &Waker) -> Poll<io::Result<()>> {
+    fn poll_ready_if_throttling(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<io::Result<()>> {
         if self.in_flight_requests.len()
             >= self.channel.config.max_in_flight_requests_per_connection
         {
@@ -328,7 +340,7 @@ where
         Poll::Ready(Ok(()))
     }
 
-    fn pump_read(mut self: Pin<&mut Self>, cx: &Waker) -> PollIo<()> {
+    fn pump_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> PollIo<()> {
         ready!(self.as_mut().poll_ready_if_throttling(cx)?);
 
         Poll::Ready(match ready!(self.as_mut().channel().poll_next(cx)?) {
@@ -350,7 +362,11 @@ where
         })
     }
 
-    fn pump_write(mut self: Pin<&mut Self>, cx: &Waker, read_half_closed: bool) -> PollIo<()> {
+    fn pump_write(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        read_half_closed: bool,
+    ) -> PollIo<()> {
         match self.as_mut().poll_next_response(cx)? {
             Poll::Ready(Some((_, response))) => {
                 self.as_mut().channel().start_send(response)?;
@@ -379,8 +395,8 @@ where
 
     fn poll_next_response(
         mut self: Pin<&mut Self>,
-        cx: &Waker,
+        cx: &mut Context<'_>,
-    ) -> PollIo<(Context, Response<Resp>)> {
+    ) -> PollIo<(context::Context, Response<Resp>)> {
         // Ensure there's room to write a response.
         while let Poll::Pending = self.as_mut().channel().poll_ready(cx)? {
             ready!(self.as_mut().channel().poll_flush(cx)?);
@@ -421,7 +437,7 @@ where
     ) -> io::Result<()> {
         let request_id = request.id;
         let peer = self.as_mut().channel().client_addr;
-        let ctx = Context {
+        let ctx = context::Context {
             deadline: request.deadline,
             trace_context,
         };
@@ -527,12 +543,12 @@ where
     Req: Send + 'static,
     Resp: Send + 'static,
     T: Transport<Item = ClientMessage<Req>, SinkItem = Response<Resp>> + Send,
-    F: FnOnce(Context, Req) -> Fut + Send + 'static + Clone,
+    F: FnOnce(context::Context, Req) -> Fut + Send + 'static + Clone,
     Fut: Future<Output = io::Result<Resp>> + Send + 'static,
 {
     type Output = io::Result<()>;
 
-    fn poll(mut self: Pin<&mut Self>, cx: &Waker) -> Poll<io::Result<()>> {
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
         trace!("[{}] ClientHandler::poll", self.channel.client_addr);
         loop {
             let read = self.as_mut().pump_read(cx)?;

rpc/src/transport/channel.rs

@@ -7,7 +7,7 @@
 //! Transports backed by in-memory channels.
 
 use crate::{PollIo, Transport};
-use futures::{channel::mpsc, task::Waker, Poll, Sink, Stream};
+use futures::{channel::mpsc, task::Context, Poll, Sink, Stream};
 use pin_utils::unsafe_pinned;
 use std::pin::Pin;
 use std::{
@@ -45,16 +45,15 @@ impl<Item, SinkItem> UnboundedChannel<Item, SinkItem> {
 impl<Item, SinkItem> Stream for UnboundedChannel<Item, SinkItem> {
     type Item = Result<Item, io::Error>;
 
-    fn poll_next(self: Pin<&mut Self>, cx: &Waker) -> PollIo<Item> {
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> PollIo<Item> {
         self.rx().poll_next(cx).map(|option| option.map(Ok))
     }
 }
 
-impl<Item, SinkItem> Sink for UnboundedChannel<Item, SinkItem> {
+impl<Item, SinkItem> Sink<SinkItem> for UnboundedChannel<Item, SinkItem> {
-    type SinkItem = SinkItem;
     type SinkError = io::Error;
 
-    fn poll_ready(self: Pin<&mut Self>, cx: &Waker) -> Poll<io::Result<()>> {
+    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
         self.tx()
             .poll_ready(cx)
             .map_err(|_| io::Error::from(io::ErrorKind::NotConnected))
@@ -66,13 +65,13 @@ impl<Item, SinkItem> Sink for UnboundedChannel<Item, SinkItem> {
             .map_err(|_| io::Error::from(io::ErrorKind::NotConnected))
     }
 
-    fn poll_flush(self: Pin<&mut Self>, cx: &Waker) -> Poll<Result<(), Self::SinkError>> {
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::SinkError>> {
         self.tx()
             .poll_flush(cx)
             .map_err(|_| io::Error::from(io::ErrorKind::NotConnected))
     }
 
-    fn poll_close(self: Pin<&mut Self>, cx: &Waker) -> Poll<io::Result<()>> {
+    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
         self.tx()
             .poll_close(cx)
             .map_err(|_| io::Error::from(io::ErrorKind::NotConnected))
@@ -80,8 +79,8 @@ impl<Item, SinkItem> Sink for UnboundedChannel<Item, SinkItem> {
 }
 
 impl<Item, SinkItem> Transport for UnboundedChannel<Item, SinkItem> {
-    type Item = Item;
     type SinkItem = SinkItem;
+    type Item = Item;
 
     fn peer_addr(&self) -> io::Result<SocketAddr> {
         Ok(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0))
@@ -130,8 +129,11 @@ mod tests {
             Ok::<_, io::Error>((response1, response2))
         };
 
-        let (response1, response2) =
+        let (response1, response2) = run_future(future::join(
-            run_future(server.join(responses.unwrap_or_else(|e| panic!(e)))).1;
+            server,
+            responses.unwrap_or_else(|e| panic!(e)),
+        ))
+        .1;
 
         trace!("response1: {:?}, response2: {:?}", response1, response2);
 

rpc/src/transport/mod.rs

@@ -12,9 +12,10 @@
 use futures::prelude::*;
 use std::{
     io,
+    marker::PhantomData,
     net::SocketAddr,
     pin::Pin,
-    task::{Poll, Waker},
+    task::{Context, Poll},
 };
 
 pub mod channel;
@@ -23,7 +24,7 @@ pub mod channel;
 pub trait Transport
 where
     Self: Stream<Item = io::Result<<Self as Transport>::Item>>,
-    Self: Sink<SinkItem = <Self as Transport>::SinkItem, SinkError = io::Error>,
+    Self: Sink<<Self as Transport>::SinkItem, SinkError = io::Error>,
 {
     /// The type read off the transport.
     type Item;
@@ -37,77 +38,78 @@ where
 }
 
 /// Returns a new Transport backed by the given Stream + Sink and connecting addresses.
-pub fn new<S, Item>(
+pub fn new<S, SinkItem, Item>(
     inner: S,
     peer_addr: SocketAddr,
     local_addr: SocketAddr,
-) -> impl Transport<Item = Item, SinkItem = S::SinkItem>
+) -> impl Transport<Item = Item, SinkItem = SinkItem>
 where
     S: Stream<Item = io::Result<Item>>,
-    S: Sink<SinkError = io::Error>,
+    S: Sink<SinkItem, SinkError = io::Error>,
 {
     TransportShim {
         inner,
         peer_addr,
         local_addr,
+        _marker: PhantomData,
     }
 }
 
 /// A transport created by adding peers to a Stream + Sink.
 #[derive(Debug)]
-struct TransportShim<S> {
+struct TransportShim<S, SinkItem> {
     peer_addr: SocketAddr,
     local_addr: SocketAddr,
     inner: S,
+    _marker: PhantomData<SinkItem>,
 }
 
-impl<S> TransportShim<S> {
+impl<S, SinkItem> TransportShim<S, SinkItem> {
     pin_utils::unsafe_pinned!(inner: S);
 }
 
-impl<S> Stream for TransportShim<S>
+impl<S, SinkItem> Stream for TransportShim<S, SinkItem>
 where
     S: Stream,
 {
     type Item = S::Item;
 
-    fn poll_next(self: Pin<&mut Self>, waker: &Waker) -> Poll<Option<S::Item>> {
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<S::Item>> {
-        self.inner().poll_next(waker)
+        self.inner().poll_next(cx)
     }
 }
 
-impl<S> Sink for TransportShim<S>
+impl<S, Item> Sink<Item> for TransportShim<S, Item>
 where
-    S: Sink,
+    S: Sink<Item>,
 {
-    type SinkItem = S::SinkItem;
     type SinkError = S::SinkError;
 
-    fn start_send(self: Pin<&mut Self>, item: S::SinkItem) -> Result<(), S::SinkError> {
+    fn start_send(self: Pin<&mut Self>, item: Item) -> Result<(), S::SinkError> {
         self.inner().start_send(item)
     }
 
-    fn poll_ready(self: Pin<&mut Self>, waker: &Waker) -> Poll<Result<(), S::SinkError>> {
+    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), S::SinkError>> {
-        self.inner().poll_ready(waker)
+        self.inner().poll_ready(cx)
     }
 
-    fn poll_flush(self: Pin<&mut Self>, waker: &Waker) -> Poll<Result<(), S::SinkError>> {
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), S::SinkError>> {
-        self.inner().poll_flush(waker)
+        self.inner().poll_flush(cx)
     }
 
-    fn poll_close(self: Pin<&mut Self>, waker: &Waker) -> Poll<Result<(), S::SinkError>> {
+    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), S::SinkError>> {
-        self.inner().poll_close(waker)
+        self.inner().poll_close(cx)
     }
 }
 
-impl<S, Item> Transport for TransportShim<S>
+impl<S, SinkItem, Item> Transport for TransportShim<S, SinkItem>
 where
-    S: Stream + Sink,
+    S: Stream + Sink<SinkItem>,
     Self: Stream<Item = io::Result<Item>>,
-    Self: Sink<SinkItem = S::SinkItem, SinkError = io::Error>,
+    Self: Sink<SinkItem, SinkError = io::Error>,
 {
     type Item = Item;
-    type SinkItem = S::SinkItem;
+    type SinkItem = SinkItem;
 
     /// The address of the remote peer this transport is in communication with.
     fn peer_addr(&self) -> io::Result<SocketAddr> {

rpc/src/util/deadline_compat.rs

@@ -5,10 +5,10 @@
 // https://opensource.org/licenses/MIT.
 
 use futures::{
-    compat::{Compat01As03, Future01CompatExt},
+    compat::*,
     prelude::*,
     ready,
-    task::{Poll, Waker},
+    task::{Context, Poll},
 };
 use pin_utils::unsafe_pinned;
 use std::pin::Pin;
@@ -50,15 +50,15 @@ where
 {
     type Output = Result<T::Ok, timeout::Error<T::Error>>;
 
-    fn poll(mut self: Pin<&mut Self>, waker: &Waker) -> Poll<Self::Output> {
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         // First, try polling the future
-        match self.as_mut().future().try_poll(waker) {
+        match self.as_mut().future().try_poll(cx) {
             Poll::Ready(Ok(v)) => return Poll::Ready(Ok(v)),
             Poll::Pending => {}
             Poll::Ready(Err(e)) => return Poll::Ready(Err(timeout::Error::inner(e))),
         }
 
-        let delay = self.delay().poll_unpin(waker);
+        let delay = self.delay().poll_unpin(cx);
 
         // Now check the timer
         match ready!(delay) {

tarpc/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "tarpc"
-version = "0.15.0"
+version = "0.16.0"
 authors = ["Adam Wright <adam.austin.wright@gmail.com>", "Tim Kuehn <timothy.j.kuehn@gmail.com>"]
 edition = "2018"
 license = "MIT"
@@ -19,20 +19,17 @@ serde1 = ["rpc/serde1", "serde", "serde/derive"]
 travis-ci = { repository = "google/tarpc" }
 
 [dependencies]
+futures-preview = { version = "0.3.0-alpha.14", features = ["compat"] }
 log = "0.4"
 serde = { optional = true, version = "1.0" }
+rpc = { package = "tarpc-lib", path = "../rpc", version = "0.4" }
 tarpc-plugins = { path = "../plugins", version = "0.5.0" }
-rpc = { package = "tarpc-lib", path = "../rpc", version = "0.3" }
-
-[target.'cfg(not(test))'.dependencies]
-futures-preview = "0.3.0-alpha.13"
 
 [dev-dependencies]
-bincode = "1.0"
+bincode = "1"
 bytes = { version = "0.4", features = ["serde"] }
 humantime = "1.0"
-futures-preview = { version = "0.3.0-alpha.13", features = ["compat"] }
+bincode-transport = { package = "tarpc-bincode-transport", version = "0.5", path = "../bincode-transport" }
-bincode-transport = { package = "tarpc-bincode-transport", version = "0.4", path = "../bincode-transport" }
 env_logger = "0.6"
 libtest = "0.0.1"
 tokio = "0.1"

tarpc/examples/service_registry.rs

@@ -18,7 +18,7 @@ mod registry {
         io,
         pin::Pin,
         sync::Arc,
-        task::{Poll, Waker},
+        task::{Context, Poll},
     };
     use tarpc::{
         client::{self, Client},
@@ -213,11 +213,11 @@ mod registry {
     {
         type Output = Output;
 
-        fn poll(self: Pin<&mut Self>, waker: &Waker) -> Poll<Output> {
+        fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Output> {
             unsafe {
                 match Pin::get_unchecked_mut(self) {
-                    Either::Left(car) => Pin::new_unchecked(car).poll(waker),
+                    Either::Left(car) => Pin::new_unchecked(car).poll(cx),
-                    Either::Right(cdr) => Pin::new_unchecked(cdr).poll(waker),
+                    Either::Right(cdr) => Pin::new_unchecked(cdr).poll(cx),
                 }
             }
         }

tarpc/src/macros.rs

@@ -177,7 +177,7 @@ macro_rules! service {
         impl<S: Service> ::std::future::Future for ResponseFut<S> {
             type Output = ::std::io::Result<Response>;
 
-            fn poll(self: ::std::pin::Pin<&mut Self>, waker: &::std::task::Waker)
+            fn poll(self: ::std::pin::Pin<&mut Self>, cx: &mut ::std::task::Context<'_>)
                 -> ::std::task::Poll<::std::io::Result<Response>>
             {
                 unsafe {
@@ -185,7 +185,7 @@ macro_rules! service {
                         $(
                             ResponseFut::$fn_name(resp) =>
                                 ::std::pin::Pin::new_unchecked(resp)
-                                    .poll(waker)
+                                    .poll(cx)
                                     .map(Response::$fn_name)
                                     .map(Ok),
                         )*