diff options
author | Nora Widdecke <nora@sequoia-pgp.org> | 2021-11-04 15:44:29 +0100 |
---|---|---|
committer | Nora Widdecke <nora@sequoia-pgp.org> | 2021-12-01 12:38:57 +0100 |
commit | 7c1eb5cb0a86fd8198681d7ec53db092876de0cd (patch) | |
tree | ffab83255afa2a49e148b8eb660d6b102d138764 /ipc | |
parent | 6c161b47eed93e1fe98f53c2961d32c5712a3ee2 (diff) |
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.
Diffstat (limited to 'ipc')
-rw-r--r-- | ipc/Cargo.toml | 5 | ||||
-rw-r--r-- | ipc/examples/assuan-client.rs | 2 | ||||
-rw-r--r-- | ipc/examples/gpg-agent-client.rs | 2 | ||||
-rw-r--r-- | ipc/src/assuan/mod.rs | 36 | ||||
-rw-r--r-- | ipc/src/gnupg.rs | 4 | ||||
-rw-r--r-- | ipc/src/lib.rs | 12 |
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) } } |