Fix the build

Needed to pin the compiler to work around a compiler bug which manifests in syn.

Changes to work with futures 3.0.17.

Use latest Prost from the Git repo so we get the same version as kvproto

Use lazy_static 1.0 to avoid deprecation warnings.

Satisfy Clippy on `Err(...)?`

Signed-off-by: Nick Cameron <nrc@ncameron.org>

Cargo.toml

@@ -23,7 +23,7 @@ derive-new = "0.5"
 failure = "0.1"
 futures-preview = { version = "0.3.0-alpha.15", features = ["compat"] }
 grpcio = { version = "0.5.0-alpha", features = [ "secure", "prost-codec" ], default-features = false }
-lazy_static = "0.2.1"
+lazy_static = "1"
 log = "0.3.9"
 serde = "1.0"
 serde_derive = "1.0"
@@ -45,3 +45,6 @@ runtime = "0.3.0-alpha.3"
 runtime-tokio = "0.3.0-alpha.3"
 proptest = "0.9"
 proptest-derive = "0.1.0"
+
+[replace]
+"prost:0.5.0" = { git = "https://github.com/danburkert/prost", rev = "1944c27c3029d01ff216e7b126d846b6cf8c7d77" }

rust-toolchain

@@ -0,0 +1 @@
+nightly-2019-07-09

src/compat.rs

@@ -4,8 +4,8 @@
 //! from futures 0.1 to 1.0 easier.
 
 use futures::prelude::*;
+use futures::ready;
 use futures::task::{Context, Poll};
-use futures::{ready, try_ready};
 use std::pin::Pin;
 
 /// The status of a `loop_fn` loop.
@@ -52,7 +52,7 @@ where
         loop {
             unsafe {
                 let this = Pin::get_unchecked_mut(self);
-                match try_ready!(Pin::new_unchecked(&mut this.future).poll(cx)) {
+                match ready!(Pin::new_unchecked(&mut this.future).poll(cx))? {
                     Loop::Break(x) => return Poll::Ready(Ok(x)),
                     Loop::Continue(s) => this.future = (this.func)(s),
                 }
@@ -112,7 +112,7 @@ where
     fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<T, E>> {
         unsafe {
             let this = Pin::get_unchecked_mut(self);
-            let result = try_ready!(Pin::new_unchecked(&mut this.future).poll(cx));
+            let result = ready!(Pin::new_unchecked(&mut this.future).poll(cx))?;
             Poll::Ready((this.func)(result))
         }
     }
@@ -145,13 +145,13 @@ pub(crate) trait SinkCompat<I, E> {
     fn send_all_compat<S>(self, stream: S) -> SendAllCompat<Self, S>
     where
         S: Stream<Item = I> + Unpin,
-        Self: Sink<I, SinkError = E> + Sized + Unpin,
+        Self: Sink<I, Error = E> + Sized + Unpin,
     {
         SendAllCompat::new(self, stream)
     }
 }
 
-impl<T, E, S: Sink<T, SinkError = E>> SinkCompat<T, E> for S {}
+impl<T, E, S: Sink<T, Error = E>> SinkCompat<T, E> for S {}
 
 #[derive(Debug)]
 #[must_use = "futures do nothing unless polled"]
@@ -219,7 +219,7 @@ where
         &mut self,
         item: St::Item,
         cx: &mut Context,
-    ) -> Poll<Result<(()), Si::SinkError>> {
+    ) -> Poll<Result<(()), Si::Error>> {
         debug_assert!(self.buffered.is_none());
         match self.sink_mut().poll_ready(cx) {
             Poll::Ready(Ok(())) => Poll::Ready(self.sink_mut().start_send(item)),
@@ -237,22 +237,22 @@ where
     Si: Sink<St::Item> + Unpin,
     St: Stream + Unpin,
 {
-    type Output = Result<((Si, St)), Si::SinkError>;
+    type Output = Result<((Si, St)), Si::Error>;
 
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<((Si, St)), Si::SinkError>> {
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<((Si, St)), Si::Error>> {
         if let Some(item) = self.buffered.take() {
-            try_ready!(self.try_start_send(item, cx))
+            ready!(self.try_start_send(item, cx))?
         }
 
         loop {
             match self.stream_mut().poll_next(cx) {
-                Poll::Ready(Some(item)) => try_ready!(self.try_start_send(item, cx)),
+                Poll::Ready(Some(item)) => ready!(self.try_start_send(item, cx))?,
                 Poll::Ready(None) => {
-                    try_ready!(self.sink_mut().poll_close(cx));
+                    ready!(self.sink_mut().poll_close(cx))?;
                     return Poll::Ready(Ok(self.take_result()));
                 }
                 Poll::Pending => {
-                    try_ready!(self.sink_mut().poll_flush(cx));
+                    ready!(self.sink_mut().poll_flush(cx))?;
                     return Poll::Pending;
                 }
             }

src/rpc/pd/client.rs

@@ -141,7 +141,7 @@ impl PdClient {
             let option = CallOption::default().timeout(timeout);
             let cli = &cli.read().unwrap().client;
             executor(cli, option).unwrap().map(|r| match r {
-                Err(e) => Err(ErrorKind::Grpc(e))?,
+                Err(e) => Err(ErrorKind::Grpc(e).into()),
                 Ok(r) => {
                     {
                         let header = r.header();

src/rpc/pd/leader.rs

@@ -12,6 +12,7 @@ use futures::channel::{
     oneshot,
 };
 use futures::compat::{Compat01As03, Compat01As03Sink};
+use futures::future::TryFutureExt;
 use futures::prelude::*;
 use grpcio::{CallOption, Environment, WriteFlags};
 use kvproto::pdpb;

src/rpc/tikv/client.rs

@@ -666,7 +666,7 @@ impl KvClient {
         )
         .unwrap()
         .map(|r| match r {
-            Err(e) => Err(ErrorKind::Grpc(e))?,
+            Err(e) => Err(ErrorKind::Grpc(e).into()),
             Ok(mut r) => {
                 if let Some(e) = r.region_error() {
                     Err(e)