Fix poll_expired incorrectly returning Pending when there are no outstanding requests.

README.md

@@ -40,7 +40,7 @@ rather than in a separate language such as .proto. This means there's no separat
 process, and no context switching between different languages.
 
 Some other features of tarpc:
-- Pluggable transport: any type implementing `Stream<Item = Request> + Sink<Response>` can be
+- Pluggable transport: any type impling `Stream<Item = Request> + Sink<Response>` can be
   used as a transport to connect the client and server.
 - `Send + 'static` optional: if the transport doesn't require it, neither does tarpc!
 - Cascading cancellation: dropping a request will send a cancellation message to the server.
@@ -55,7 +55,7 @@ Some other features of tarpc:
   [tracing](https://github.com/tokio-rs/tracing) primitives extended with
   [OpenTelemetry](https://opentelemetry.io/) traces. Using a compatible tracing subscriber like
   [Jaeger](https://github.com/open-telemetry/opentelemetry-rust/tree/main/opentelemetry-jaeger),
-  each RPC can be traced through the client, server, and other dependencies downstream of the
+  each RPC can be traced through the client, server, amd other dependencies downstream of the
   server. Even for applications not connected to a distributed tracing collector, the
   instrumentation can also be ingested by regular loggers like
   [env_logger](https://github.com/env-logger-rs/env_logger/).
@@ -67,7 +67,7 @@ Some other features of tarpc:
 Add to your `Cargo.toml` dependencies:
 
 ```toml
-tarpc = "0.28"
+tarpc = "0.27"
 ```
 
 The `tarpc::service` attribute expands to a collection of items that form an rpc service.
@@ -82,7 +82,7 @@ your `Cargo.toml`:
 ```toml
 anyhow = "1.0"
 futures = "0.3"
-tarpc = { version = "0.28", features = ["tokio1"] }
+tarpc = { version = "0.27", features = ["tokio1"] }
 tokio = { version = "1.0", features = ["macros"] }
 ```
 
@@ -100,7 +100,7 @@ use futures::{
 };
 use tarpc::{
     client, context,
-    server::{self, incoming::Incoming, Channel},
+    server::{self, incoming::Incoming},
 };
 
 // This is the service definition. It looks a lot like a trait definition.
@@ -128,7 +128,7 @@ impl World for HelloServer {
     type HelloFut = Ready<String>;
 
     fn hello(self, _: context::Context, name: String) -> Self::HelloFut {
-        future::ready(format!("Hello, {name}!"))
+        future::ready(format!("Hello, {}!", name))
     }
 }
 ```
@@ -155,7 +155,7 @@ async fn main() -> anyhow::Result<()> {
     // specifies a deadline and trace information which can be helpful in debugging requests.
     let hello = client.hello(context::current(), "Stim".to_string()).await?;
 
-    println!("{hello}");
+    println!("{}", hello);
 
     Ok(())
 }

RELEASES.md

@@ -1,17 +1,3 @@
-## 0.28.0 (2022-04-06)
-
-### Breaking Changes
-
-- The minimum supported Rust version has increased to 1.58.0.
-- The version of opentelemetry depended on by tarpc has increased to 0.17.0.
-
-## 0.27.2 (2021-10-08)
-
-### Fixes
-
-Clients will now close their transport before dropping it. An attempt at a clean shutdown can help
-the server drop its connections more quickly.
-
 ## 0.27.1 (2021-09-22)
 
 ### Breaking Changes

example-service/Cargo.toml

@@ -1,9 +1,8 @@
 [package]
 name = "tarpc-example-service"
 version = "0.10.0"
-rust-version = "1.56"
 authors = ["Tim Kuehn <tikue@google.com>"]
-edition = "2021"
+edition = "2018"
 license = "MIT"
 documentation = "https://docs.rs/tarpc-example-service"
 homepage = "https://github.com/google/tarpc"
@@ -15,13 +14,13 @@ description = "An example server built on tarpc."
 
 [dependencies]
 anyhow = "1.0"
-clap = { version = "3.0.0-rc.9", features = ["derive"] }
+clap = "3.0.0-beta.2"
 log = "0.4"
 futures = "0.3"
 opentelemetry = { version = "0.16", features = ["rt-tokio"] }
 opentelemetry-jaeger = { version = "0.15", features = ["rt-tokio"] }
 rand = "0.8"
-tarpc = { version = "0.28", path = "../tarpc", features = ["full"] }
+tarpc = { version = "0.27", path = "../tarpc", features = ["full"] }
 tokio = { version = "1", features = ["macros", "net", "rt-multi-thread"] }
 tracing = { version = "0.1" }
 tracing-opentelemetry = "0.15"

example-service/src/client.rs

@@ -4,14 +4,14 @@
 // license that can be found in the LICENSE file or at
 // https://opensource.org/licenses/MIT.
 
-use clap::Parser;
+use clap::Clap;
 use service::{init_tracing, WorldClient};
 use std::{net::SocketAddr, time::Duration};
 use tarpc::{client, context, tokio_serde::formats::Json};
 use tokio::time::sleep;
 use tracing::Instrument;
 
-#[derive(Parser)]
+#[derive(Clap)]
 struct Flags {
     /// Sets the server address to connect to.
     #[clap(long)]

example-service/src/server.rs

@@ -4,7 +4,7 @@
 // license that can be found in the LICENSE file or at
 // https://opensource.org/licenses/MIT.
 
-use clap::Parser;
+use clap::Clap;
 use futures::{future, prelude::*};
 use rand::{
     distributions::{Distribution, Uniform},
@@ -22,7 +22,7 @@ use tarpc::{
 };
 use tokio::time;
 
-#[derive(Parser)]
+#[derive(Clap)]
 struct Flags {
     /// Sets the port number to listen on.
     #[clap(long)]
@@ -40,7 +40,7 @@ impl World for HelloServer {
         let sleep_time =
             Duration::from_millis(Uniform::new_inclusive(1, 10).sample(&mut thread_rng()));
         time::sleep(sleep_time).await;
-        format!("Hello, {name}! You are connected from {}", self.0)
+        format!("Hello, {}! You are connected from {}", name, self.0)
     }
 }
 

plugins/Cargo.toml

@@ -1,9 +1,8 @@
 [package]
 name = "tarpc-plugins"
 version = "0.12.0"
-rust-version = "1.56"
 authors = ["Adam Wright <adam.austin.wright@gmail.com>", "Tim Kuehn <timothy.j.kuehn@gmail.com>"]
-edition = "2021"
+edition = "2018"
 license = "MIT"
 documentation = "https://docs.rs/tarpc-plugins"
 homepage = "https://github.com/google/tarpc"

plugins/src/lib.rs

@@ -83,7 +83,7 @@ impl Parse for Service {
                     ident_errors,
                     syn::Error::new(
                         rpc.ident.span(),
-                        format!("method name conflicts with generated fn `{ident}::serve`")
+                        format!("method name conflicts with generated fn `{}::serve`", ident)
                     )
                 );
             }
@@ -270,7 +270,7 @@ pub fn service(attr: TokenStream, input: TokenStream) -> TokenStream {
     let methods = rpcs.iter().map(|rpc| &rpc.ident).collect::<Vec<_>>();
     let request_names = methods
         .iter()
-        .map(|m| format!("{ident}.{m}"))
+        .map(|m| format!("{}.{}", ident, m))
         .collect::<Vec<_>>();
 
     ServiceGenerator {
@@ -306,7 +306,7 @@ pub fn service(attr: TokenStream, input: TokenStream) -> TokenStream {
             .collect::<Vec<_>>(),
         future_types: &camel_case_fn_names
             .iter()
-            .map(|name| parse_str(&format!("{name}Fut")).unwrap())
+            .map(|name| parse_str(&format!("{}Fut", name)).unwrap())
             .collect::<Vec<_>>(),
         derive_serialize: derive_serialize.as_ref(),
     }
@@ -409,7 +409,7 @@ fn verify_types_were_provided(
         if !provided.iter().any(|typedecl| typedecl.ident == expected) {
             let mut e = syn::Error::new(
                 span,
-                format!("not all trait items implemented, missing: `{expected}`"),
+                format!("not all trait items implemented, missing: `{}`", expected),
             );
             let fn_span = method.sig.fn_token.span();
             e.extend(syn::Error::new(
@@ -479,7 +479,7 @@ impl<'a> ServiceGenerator<'a> {
                     ),
                     output,
                 )| {
-                    let ty_doc = format!("The response future returned by [`{service_ident}::{ident}`].");
+                    let ty_doc = format!("The response future returned by [`{}::{}`].", service_ident, ident);
                     quote! {
                         #[doc = #ty_doc]
                         type #future_type: std::future::Future<Output = #output>;

tarpc/Cargo.toml

@@ -1,12 +1,8 @@
 [package]
 name = "tarpc"
-version = "0.28.0"
+version = "0.27.1"
-rust-version = "1.58.0"
+authors = ["Adam Wright <adam.austin.wright@gmail.com>", "Tim Kuehn <timothy.j.kuehn@gmail.com>"]
-authors = [
+edition = "2018"
-    "Adam Wright <adam.austin.wright@gmail.com>",
-    "Tim Kuehn <timothy.j.kuehn@gmail.com>",
-]
-edition = "2021"
 license = "MIT"
 documentation = "https://docs.rs/tarpc"
 homepage = "https://github.com/google/tarpc"
@@ -20,20 +16,13 @@ description = "An RPC framework for Rust with a focus on ease of use."
 default = []
 
 serde1 = ["tarpc-plugins/serde1", "serde", "serde/derive"]
-tokio1 = ["tokio/rt"]
+tokio1 = ["tokio/rt-multi-thread"]
 serde-transport = ["serde1", "tokio1", "tokio-serde", "tokio-util/codec"]
 serde-transport-json = ["tokio-serde/json"]
 serde-transport-bincode = ["tokio-serde/bincode"]
 tcp = ["tokio/net"]
 
-full = [
+full = ["serde1", "tokio1", "serde-transport", "serde-transport-json", "serde-transport-bincode", "tcp"]
-    "serde1",
-    "tokio1",
-    "serde-transport",
-    "serde-transport-json",
-    "serde-transport-bincode",
-    "tcp",
-]
 
 [badges]
 travis-ci = { repository = "google/tarpc" }
@@ -50,14 +39,11 @@ static_assertions = "1.1.0"
 tarpc-plugins = { path = "../plugins", version = "0.12" }
 thiserror = "1.0"
 tokio = { version = "1", features = ["time"] }
-tokio-util = { version = "0.6.9", features = ["time"] }
+tokio-util = { version = "0.6.3", features = ["time"] }
 tokio-serde = { optional = true, version = "0.8" }
-tracing = { version = "0.1", default-features = false, features = [
+tracing = { version = "0.1", default-features = false, features = ["attributes", "log"] }
-    "attributes",
+tracing-opentelemetry = { version = "0.15", default-features = false }
-    "log",
+opentelemetry = { version = "0.16", default-features = false }
-] }
-tracing-opentelemetry = { version = "0.17.2", default-features = false }
-opentelemetry = { version = "0.17.0", default-features = false }
 
 
 [dev-dependencies]
@@ -66,13 +52,11 @@ bincode = "1.3"
 bytes = { version = "1", features = ["serde"] }
 flate2 = "1.0"
 futures-test = "0.3"
-opentelemetry = { version = "0.17.0", default-features = false, features = [
+opentelemetry = { version = "0.16", default-features = false, features = ["rt-tokio"] }
-    "rt-tokio",
+opentelemetry-jaeger = { version = "0.15", features = ["rt-tokio"] }
-] }
-opentelemetry-jaeger = { version = "0.16.0", features = ["rt-tokio"] }
 pin-utils = "0.1.0-alpha"
 serde_bytes = "0.11"
-tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+tracing-subscriber = "0.2"
 tokio = { version = "1", features = ["full", "test-util"] }
 tokio-serde = { version = "0.8", features = ["json", "bincode"] }
 trybuild = "1.0"

tarpc/examples/compression.rs

@@ -54,7 +54,7 @@ where
             if algorithm != CompressionAlgorithm::Deflate {
                 return Err(io::Error::new(
                     io::ErrorKind::InvalidData,
-                    format!("Compression algorithm {algorithm:?} not supported"),
+                    format!("Compression algorithm {:?} not supported", algorithm),
                 ));
             }
             let mut deflater = DeflateDecoder::new(payload.as_slice());
@@ -102,7 +102,7 @@ struct HelloServer;
 #[tarpc::server]
 impl World for HelloServer {
     async fn hello(self, _: context::Context, name: String) -> String {
-        format!("Hey, {name}!")
+        format!("Hey, {}!", name)
     }
 }
 

tarpc/examples/pubsub.rs

@@ -290,7 +290,7 @@ fn init_tracing(service_name: &str) -> anyhow::Result<()> {
         .install_batch(opentelemetry::runtime::Tokio)?;
 
     tracing_subscriber::registry()
-        .with(tracing_subscriber::filter::EnvFilter::from_default_env())
+        .with(tracing_subscriber::EnvFilter::from_default_env())
         .with(tracing_subscriber::fmt::layer())
         .with(tracing_opentelemetry::layer().with_tracer(tracer))
         .try_init()?;

tarpc/examples/readme.rs

@@ -29,7 +29,7 @@ impl World for HelloServer {
     type HelloFut = Ready<String>;
 
     fn hello(self, _: context::Context, name: String) -> Self::HelloFut {
-        future::ready(format!("Hello, {name}!"))
+        future::ready(format!("Hello, {}!", name))
     }
 }
 
@@ -49,7 +49,7 @@ async fn main() -> anyhow::Result<()> {
     // specifies a deadline and trace information which can be helpful in debugging requests.
     let hello = client.hello(context::current(), "Stim".to_string()).await?;
 
-    println!("{hello}");
+    println!("{}", hello);
 
     Ok(())
 }

tarpc/src/client.rs

@@ -81,10 +81,14 @@ impl<C, D> fmt::Debug for NewClient<C, D> {
     }
 }
 
-const _CHECK_USIZE: () = assert!(
+#[allow(dead_code)]
-    std::mem::size_of::<usize>() <= std::mem::size_of::<u64>(),
+#[allow(clippy::no_effect)]
-    "usize is too big to fit in u64"
+const CHECK_USIZE: () = {
-);
+    if std::mem::size_of::<usize>() > std::mem::size_of::<u64>() {
+        // TODO: replace this with panic!() as soon as RFC 2345 gets stabilized
+        ["usize is too big to fit in u64"][42];
+    }
+};
 
 /// Handles communication from the client to request dispatch.
 #[derive(Debug)]
@@ -168,8 +172,7 @@ struct ResponseGuard<'a, Resp> {
 
 /// An error that can occur in the processing of an RPC. This is not request-specific errors but
 /// rather cross-cutting errors that can always occur.
-#[derive(thiserror::Error, Clone, Debug, PartialEq, Eq, Hash)]
+#[derive(thiserror::Error, Debug)]
-#[cfg_attr(feature = "serde1", derive(serde::Serialize, serde::Deserialize))]
 pub enum RpcError {
     /// The client disconnected from the server.
     #[error("the client disconnected from the server")]

tarpc/src/lib.rs

@@ -27,7 +27,7 @@
 //! process, and no context switching between different languages.
 //!
 //! Some other features of tarpc:
-//! - Pluggable transport: any type implementing `Stream<Item = Request> + Sink<Response>` can be
+//! - Pluggable transport: any type impling `Stream<Item = Request> + Sink<Response>` can be
 //!   used as a transport to connect the client and server.
 //! - `Send + 'static` optional: if the transport doesn't require it, neither does tarpc!
 //! - Cascading cancellation: dropping a request will send a cancellation message to the server.
@@ -42,7 +42,7 @@
 //!   [tracing](https://github.com/tokio-rs/tracing) primitives extended with
 //!   [OpenTelemetry](https://opentelemetry.io/) traces. Using a compatible tracing subscriber like
 //!   [Jaeger](https://github.com/open-telemetry/opentelemetry-rust/tree/main/opentelemetry-jaeger),
-//!   each RPC can be traced through the client, server, and other dependencies downstream of the
+//!   each RPC can be traced through the client, server, amd other dependencies downstream of the
 //!   server. Even for applications not connected to a distributed tracing collector, the
 //!   instrumentation can also be ingested by regular loggers like
 //!   [env_logger](https://github.com/env-logger-rs/env_logger/).
@@ -54,7 +54,7 @@
 //! Add to your `Cargo.toml` dependencies:
 //!
 //! ```toml
-//! tarpc = "0.28"
+//! tarpc = "0.27"
 //! ```
 //!
 //! The `tarpc::service` attribute expands to a collection of items that form an rpc service.
@@ -69,7 +69,7 @@
 //! ```toml
 //! anyhow = "1.0"
 //! futures = "0.3"
-//! tarpc = { version = "0.28", features = ["tokio1"] }
+//! tarpc = { version = "0.27", features = ["tokio1"] }
 //! tokio = { version = "1.0", features = ["macros"] }
 //! ```
 //!
@@ -88,7 +88,7 @@
 //! };
 //! use tarpc::{
 //!     client, context,
-//!     server::{self, incoming::Incoming, Channel},
+//!     server::{self, incoming::Incoming},
 //! };
 //!
 //! // This is the service definition. It looks a lot like a trait definition.
@@ -132,7 +132,7 @@
 //!     type HelloFut = Ready<String>;
 //!
 //!     fn hello(self, _: context::Context, name: String) -> Self::HelloFut {
-//!         future::ready(format!("Hello, {name}!"))
+//!         future::ready(format!("Hello, {}!", name))
 //!     }
 //! }
 //! ```
@@ -168,7 +168,7 @@
 //! #     // an associated type representing the future output by the fn.
 //! #     type HelloFut = Ready<String>;
 //! #     fn hello(self, _: context::Context, name: String) -> Self::HelloFut {
-//! #         future::ready(format!("Hello, {name}!"))
+//! #         future::ready(format!("Hello, {}!", name))
 //! #     }
 //! # }
 //! # #[cfg(not(feature = "tokio1"))]
@@ -190,7 +190,7 @@
 //!     // specifies a deadline and trace information which can be helpful in debugging requests.
 //!     let hello = client.hello(context::current(), "Stim".to_string()).await?;
 //!
-//!     println!("{hello}");
+//!     println!("{}", hello);
 //!
 //!     Ok(())
 //! }
@@ -264,7 +264,7 @@ pub use tarpc_plugins::service;
 /// #[tarpc::server]
 /// impl World for HelloServer {
 ///     async fn hello(self, _: context::Context, name: String) -> String {
-///         format!("Hello, {name}! You are connected from {:?}.", self.0)
+///         format!("Hello, {}! You are connected from {:?}.", name, self.0)
 ///     }
 /// }
 /// ```
@@ -290,7 +290,7 @@ pub use tarpc_plugins::service;
 ///     fn hello(self, _: context::Context, name: String) -> Pin<Box<dyn Future<Output = String>
 ///     + Send>> {
 ///         Box::pin(async move {
-///             format!("Hello, {name}! You are connected from {:?}.", self.0)
+///             format!("Hello, {}! You are connected from {:?}.", name, self.0)
 ///         })
 ///     }
 /// }

tarpc/src/trace.rs

@@ -138,25 +138,25 @@ impl From<u64> for SpanId {
 
 impl From<opentelemetry::trace::TraceId> for TraceId {
     fn from(trace_id: opentelemetry::trace::TraceId) -> Self {
-        Self::from(u128::from_be_bytes(trace_id.to_bytes()))
+        Self::from(trace_id.to_u128())
     }
 }
 
 impl From<TraceId> for opentelemetry::trace::TraceId {
     fn from(trace_id: TraceId) -> Self {
-        Self::from_bytes(u128::from(trace_id).to_be_bytes())
+        Self::from_u128(trace_id.into())
     }
 }
 
 impl From<opentelemetry::trace::SpanId> for SpanId {
     fn from(span_id: opentelemetry::trace::SpanId) -> Self {
-        Self::from(u64::from_be_bytes(span_id.to_bytes()))
+        Self::from(span_id.to_u64())
     }
 }
 
 impl From<SpanId> for opentelemetry::trace::SpanId {
     fn from(span_id: SpanId) -> Self {
-        Self::from_bytes(u64::from(span_id).to_be_bytes())
+        Self::from_u64(span_id.0)
     }
 }
 

tarpc/src/transport/channel.rs

@@ -181,7 +181,7 @@ mod tests {
                     future::ready(request.parse::<u64>().map_err(|_| {
                         io::Error::new(
                             io::ErrorKind::InvalidInput,
-                            format!("{request:?} is not an int"),
+                            format!("{:?} is not an int", request),
                         )
                     }))
                 }),

tarpc/src/util.rs

@@ -38,34 +38,11 @@ where
     H: BuildHasher,
 {
     fn compact(&mut self, usage_ratio_threshold: f64) {
-        let usage_ratio_threshold = usage_ratio_threshold.clamp(f64::MIN_POSITIVE, 1.);
+        if self.capacity() > 1000 {
-        let cap = f64::max(1000., self.len() as f64 / usage_ratio_threshold);
+            let usage_ratio = self.len() as f64 / self.capacity() as f64;
-        self.shrink_to(cap as usize);
+            if usage_ratio < usage_ratio_threshold {
+                self.shrink_to_fit();
+            }
+        }
     }
 }
-
-#[test]
-fn test_compact() {
-    let mut map = HashMap::with_capacity(2048);
-    assert_eq!(map.capacity(), 3584);
-
-    // Make usage ratio 25%
-    for i in 0..896 {
-        map.insert(format!("k{i}"), "v");
-    }
-
-    map.compact(-1.0);
-    assert_eq!(map.capacity(), 3584);
-
-    map.compact(0.25);
-    assert_eq!(map.capacity(), 3584);
-
-    map.compact(0.50);
-    assert_eq!(map.capacity(), 1792);
-
-    map.compact(1.0);
-    assert_eq!(map.capacity(), 1792);
-
-    map.compact(2.0);
-    assert_eq!(map.capacity(), 1792);
-}

tarpc/tests/compile_fail/tarpc_server_missing_async.rs

@@ -7,8 +7,8 @@ struct HelloServer;
 
 #[tarpc::server]
 impl World for HelloServer {
-    fn hello(name: String) -> String {
+    fn hello(name: String) ->  String {
-        format!("Hello, {name}!", name)
+        format!("Hello, {}!", name)
     }
 }
 

tarpc/tests/compile_fail/tarpc_server_missing_async.stderr

@@ -7,5 +7,5 @@ error: not all trait items implemented, missing: `HelloFut`
 error: hint: `#[tarpc::server]` only rewrites async fns, and `fn hello` is not async
   --> $DIR/tarpc_server_missing_async.rs:10:5
    |
-10 |     fn hello(name: String) -> String {
+10 |     fn hello(name: String) ->  String {
    |     ^^

tarpc/tests/service_functional.rs

@@ -31,7 +31,7 @@ impl Service for Server {
     type HeyFut = Ready<String>;
 
     fn hey(self, _: context::Context, name: String) -> Self::HeyFut {
-        ready(format!("Hey, {name}."))
+        ready(format!("Hey, {}.", name))
     }
 }