Update tower-discover to std::future (#320)

This bumps tower-discover to 0.3.0-alpha.1

Cargo.toml

@@ -4,7 +4,7 @@ members = [
   # "tower",
   # "tower-balance",
   # "tower-buffer",
-  # "tower-discover",
+  "tower-discover",
   # "tower-filter",
   # "tower-hedge",
   # "tower-layer",

tower-discover/CHANGELOG.md

@@ -1,3 +1,7 @@
+# 0.3.0-alpha.1
+
+- Move to `std::future`
+
 # 0.1.0 (April 26, 2019)
 
 - Initial release

tower-discover/Cargo.toml

@@ -8,13 +8,13 @@ name = "tower-discover"
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v0.1.x" git tag.
-version = "0.1.0"
+version = "0.3.0-alpha.1"
 authors = ["Tower Maintainers <team@tower-rs.com>"]
 license = "MIT"
 readme = "README.md"
 repository = "https://github.com/tower-rs/tower"
 homepage = "https://github.com/tower-rs/tower"
-documentation = "https://docs.rs/tower-discover/0.1.0"
+documentation = "https://docs.rs/tower-discover/0.3.0-alpha.1"
 description = """
 Abstracts over service discovery strategies.
 """
@@ -22,5 +22,6 @@ categories = ["asynchronous", "network-programming"]
 edition = "2018"
 
 [dependencies]
-futures = "0.1.26"
-tower-service = "0.2.0"
+futures-core-preview = "0.3.0-alpha.18"
+tower-service = "0.3.0-alpha.1"
+pin-project = "0.4.0-alpha.9"

tower-discover/src/lib.rs

@@ -1,4 +1,4 @@
-#![doc(html_root_url = "https://docs.rs/tower-discover/0.1.0")]
+#![doc(html_root_url = "https://docs.rs/tower-discover/0.3.0-alpha.1")]
 #![deny(rust_2018_idioms)]
 #![allow(elided_lifetimes_in_paths)]
 
@@ -15,8 +15,11 @@ mod stream;
 
 pub use crate::{list::ServiceList, stream::ServiceStream};
 
-use futures::Poll;
 use std::hash::Hash;
+use std::{
+    pin::Pin,
+    task::{Context, Poll},
+};
 
 /// Provide a uniform set of services able to satisfy a request.
 ///
@@ -34,7 +37,10 @@ pub trait Discover {
     type Error;
 
     /// Yields the next discovery change set.
-    fn poll(&mut self) -> Poll<Change<Self::Key, Self::Service>, Self::Error>;
+    fn poll_discover(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Result<Change<Self::Key, Self::Service>, Self::Error>>;
 }
 
 /// A change in the service set

tower-discover/src/list.rs

@@ -1,12 +1,17 @@
 use crate::{error::Never, Change, Discover};
-use futures::{Async, Poll};
+use pin_project::pin_project;
 use std::iter::{Enumerate, IntoIterator};
+use std::{
+    pin::Pin,
+    task::{Context, Poll},
+};
 use tower_service::Service;
 
 /// Static service discovery based on a predetermined list of services.
 ///
 /// `ServiceList` is created with an initial list of services. The discovery
 /// process will yield this list once and do nothing after.
+#[pin_project]
 pub struct ServiceList<T>
 where
     T: IntoIterator,
@@ -36,10 +41,13 @@ where
     type Service = U;
     type Error = Never;
 
-    fn poll(&mut self) -> Poll<Change<Self::Key, Self::Service>, Self::Error> {
-        match self.inner.next() {
-            Some((i, service)) => Ok(Change::Insert(i, service).into()),
-            None => Ok(Async::NotReady),
+    fn poll_discover(
+        mut self: Pin<&mut Self>,
+        _: &mut Context<'_>,
+    ) -> Poll<Result<Change<Self::Key, Self::Service>, Self::Error>> {
+        match self.project().inner.next() {
+            Some((i, service)) => Poll::Ready(Ok(Change::Insert(i, service))),
+            None => Poll::Pending,
         }
     }
 }

tower-discover/src/stream.rs

@@ -1,41 +1,49 @@
 use crate::{Change, Discover};
-use futures::{try_ready, Async, Poll, Stream};
+use futures_core::{ready, TryStream};
+use pin_project::pin_project;
 use std::hash::Hash;
+use std::{
+    pin::Pin,
+    task::{Context, Poll},
+};
 use tower_service::Service;
 
 /// Dynamic service discovery based on a stream of service changes.
+#[pin_project]
 pub struct ServiceStream<S> {
-    inner: futures::stream::Fuse<S>,
+    #[pin]
+    inner: S,
 }
 
 impl<S> ServiceStream<S> {
     pub fn new<K, Svc, Request>(services: S) -> Self
     where
-        S: Stream<Item = Change<K, Svc>>,
+        S: TryStream<Ok = Change<K, Svc>>,
         K: Hash + Eq,
         Svc: Service<Request>,
     {
-        ServiceStream {
-            inner: services.fuse(),
-        }
+        ServiceStream { inner: services }
     }
 }
 
 impl<S, K, Svc> Discover for ServiceStream<S>
 where
     K: Hash + Eq,
-    S: Stream<Item = Change<K, Svc>>,
+    S: TryStream<Ok = Change<K, Svc>>,
 {
     type Key = K;
     type Service = Svc;
     type Error = S::Error;
 
-    fn poll(&mut self) -> Poll<Change<Self::Key, Self::Service>, Self::Error> {
-        match try_ready!(self.inner.poll()) {
-            Some(c) => Ok(Async::Ready(c)),
+    fn poll_discover(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Result<Change<Self::Key, Self::Service>, Self::Error>> {
+        match ready!(self.project().inner.try_poll_next(cx)).transpose()? {
+            Some(c) => Poll::Ready(Ok(c)),
             None => {
                 // there are no more service changes coming
-                Ok(Async::NotReady)
+                Poll::Pending
             }
         }
     }