ipc: Update to tokio 1.0.

  - In assuan:

    - tokio::io::AsyncRead::poll_read now uses a ReadBuf buffer instead
      of a &mu [u8], so use that and write to the Client's buffer only
      if a read was successful.

    - Poll::Ready does not report n_read any more, so there cannot be a
      conflict between the reported and actual number of bytes read,
      remove that case.

  - Fixes #780.

6 files changed, 27 insertions, 34 deletions

diff --git a/ipc/Cargo.toml b/ipc/Cargo.toml
index 66d7f728..9752189b 100644
--- a/ipc/Cargo.toml
+++ b/ipc/Cargo.toml
@@ -34,8 +34,9 @@ memsec = { version = ">=0.5", default-features = false }
 rand = { version = "0.7" }
 tempfile = "3.1"
 thiserror = "1.0.2"
-tokio = { version = "0.2.19", features = ["rt-core", "rt-util", "tcp", "uds", "io-util", "macros"] }
-tokio-util = { version = "0.3", features = ["compat"] }
+#tokio = { version = "0.2.19", features = ["rt-core", "rt-util", "tcp", "uds", "io-util", "macros"] }
+tokio = { version = "1", features = ["full"] }
+tokio-util = { version = "0.6", features = ["compat"] }
 socket2 = "0.3.16"
 dirs = "2.0"
 
diff --git a/ipc/examples/assuan-client.rs b/ipc/examples/assuan-client.rs
index f6d19235..aa7f3ed8 100644
--- a/ipc/examples/assuan-client.rs
+++ b/ipc/examples/assuan-client.rs
@@ -15,7 +15,7 @@ fn main() {
              .help("Commands to send to the server"))
         .get_matches();
 
-    let mut rt = tokio::runtime::Runtime::new().unwrap();
+    let rt = tokio::runtime::Runtime::new().unwrap();
     rt.block_on(async {
         let mut c = Client::connect(matches.value_of("server").unwrap()).await.unwrap();
         for command in matches.values_of("commands").unwrap() {
diff --git a/ipc/examples/gpg-agent-client.rs b/ipc/examples/gpg-agent-client.rs
index f993b8c2..8f2d7415 100644
--- a/ipc/examples/gpg-agent-client.rs
+++ b/ipc/examples/gpg-agent-client.rs
@@ -23,7 +23,7 @@ fn main() {
         Context::new().unwrap()
     };
 
-    let mut rt = tokio::runtime::Runtime::new().unwrap();
+    let rt = tokio::runtime::Runtime::new().unwrap();
     rt.block_on(async {
         let mut agent = Agent::connect(&ctx).await.unwrap();
 
diff --git a/ipc/src/assuan/mod.rs b/ipc/src/assuan/mod.rs
index 5a4d9f20..5d6fd3d7 100644
--- a/ipc/src/assuan/mod.rs
+++ b/ipc/src/assuan/mod.rs
@@ -299,32 +299,24 @@ impl Stream for Client {
             }
 
             // No more linebreaks in the buffer.  We need to get more.
-            // First, grow the buffer.
-            let buffer_len = buffer.len();
-            buffer.resize(buffer_len + MAX_LINE_LENGTH, 0);
-
-            match reader.as_mut().poll_read(cx, &mut buffer[buffer_len..])? {
-                Poll::Ready(n_read) if n_read == 0 => {
-                    // EOF.
-                    buffer.resize(buffer_len, 0);
-                    if ! buffer.is_empty() {
-                        // Incomplete server response.
-                        return Poll::Ready(Some(Err(Error::ConnectionClosed(
-                            buffer.clone()).into())));
-
+            // First, get a new read buffer.
+            // Later, append the read data to the Client's buffer
+
+            let mut vec = vec![0u8; MAX_LINE_LENGTH];
+            let mut read_buf = tokio::io::ReadBuf::new(&mut vec);
+
+            match reader.as_mut().poll_read(cx, &mut read_buf)? {
+                Poll::Ready(()) => {
+                    if read_buf.filled().is_empty() {
+                        // End of stream.
+                        return Poll::Ready(None)
+                    } else {
+                        buffer.extend_from_slice(read_buf.filled());
+                        continue;
                     }
-
-                    // End of stream.
-                    return Poll::Ready(None);
-                },
-
-                Poll::Ready(n_read) => {
-                    buffer.resize(buffer_len + n_read, 0);
-                    continue;
                 },
 
                 Poll::Pending => {
-                    buffer.resize(buffer_len, 0);
                     return Poll::Pending;
                 },
             }
diff --git a/ipc/src/gnupg.rs b/ipc/src/gnupg.rs
index dc3ed078..8fb2c490 100644
--- a/ipc/src/gnupg.rs
+++ b/ipc/src/gnupg.rs
@@ -785,7 +785,7 @@ impl<'a> crypto::Signer for KeyPair<'a> {
                 | (DSA, PublicKey::DSA { .. })
                 | (EdDSA, PublicKey::EdDSA { .. })
                 | (ECDSA, PublicKey::ECDSA { .. }) => {
-                    let mut rt = tokio::runtime::Runtime::new()?;
+                    let rt = tokio::runtime::Runtime::new()?;
 
                     rt.block_on(async move {
                         let mut a = Agent::connect_to(&self.agent_socket).await?;
@@ -816,7 +816,7 @@ impl<'a> crypto::Decryptor for KeyPair<'a> {
             (PublicKey::RSA { .. }, Ciphertext::RSA { .. })
                 | (PublicKey::ElGamal { .. }, Ciphertext::ElGamal { .. })
                 | (PublicKey::ECDH { .. }, Ciphertext::ECDH { .. }) => {
-                    let mut rt = tokio::runtime::Runtime::new()?;
+                    let rt = tokio::runtime::Runtime::new()?;
 
                     rt.block_on(async move {
                         let mut a = Agent::connect_to(&self.agent_socket).await?;
diff --git a/ipc/src/lib.rs b/ipc/src/lib.rs
index 734725bb..c64d536e 100644
--- a/ipc/src/lib.rs
+++ b/ipc/src/lib.rs
@@ -162,8 +162,8 @@ impl Descriptor {
             stream.set_nodelay(true)?;
 
             let (reader, writer) = stream.into_split();
-            use tokio_util::compat::Tokio02AsyncReadCompatExt;
-            use tokio_util::compat::Tokio02AsyncWriteCompatExt;
+            use tokio_util::compat::TokioAsyncReadCompatExt;
+            use tokio_util::compat::TokioAsyncWriteCompatExt;
             let (reader, writer) = (reader.compat(), writer.compat_write());
 
             let network =
@@ -388,7 +388,7 @@ impl Server {
         let handler = (self.descriptor.factory)(self.descriptor.clone(), &local)?;
 
         let server = async move {
-            let mut socket = tokio::net::TcpListener::from_std(l).unwrap();
+            let socket = tokio::net::TcpListener::from_std(l).unwrap();
 
             loop {
                 let (mut socket, _) = socket.accept().await?;
@@ -401,8 +401,8 @@ impl Server {
 
                 let (reader, writer) = socket.into_split();
 
-                use tokio_util::compat::Tokio02AsyncReadCompatExt;
-                use tokio_util::compat::Tokio02AsyncWriteCompatExt;
+                use tokio_util::compat::TokioAsyncReadCompatExt;
+                use tokio_util::compat::TokioAsyncWriteCompatExt;
                 let (reader, writer) = (reader.compat(), writer.compat_write());
 
                 let network =
@@ -414,7 +414,7 @@ impl Server {
             }
         };
 
-        local.block_on(&mut self.runtime, server)
+        local.block_on(&self.runtime, server)
     }
 }