Prepare release of v0.24.1

Co-authored-by: Artem Vorotnikov <artem@vorotnikov.me>

.github/workflows/main.yml

@@ -28,6 +28,25 @@ jobs:
           profile: minimal
           toolchain: stable
           override: true
+      - uses: actions-rs/cargo@v1
+        with:
+          command: test
+      - uses: actions-rs/cargo@v1
+        with:
+          command: test
+          args: --manifest-path tarpc/Cargo.toml --features serde1
+      - uses: actions-rs/cargo@v1
+        with:
+          command: test
+          args: --manifest-path tarpc/Cargo.toml --features tokio1
+      - uses: actions-rs/cargo@v1
+        with:
+          command: test
+          args: --manifest-path tarpc/Cargo.toml --features serde-transport
+      - uses: actions-rs/cargo@v1
+        with:
+          command: test
+          args: --manifest-path tarpc/Cargo.toml --features tcp
       - uses: actions-rs/cargo@v1
         with:
           command: test

README.md

@@ -14,6 +14,8 @@
 
 # tarpc
 
+<!-- cargo-sync-readme start -->
+
 *Disclaimer*: This is not an official Google product.
 
 tarpc is an RPC framework for rust with a focus on ease of use. Defining a
@@ -22,7 +24,7 @@ writing a server is taken care of for you.
 
 [Documentation](https://docs.rs/crate/tarpc/)
 
-### What is an RPC framework?
+## What is an RPC framework?
 "RPC" stands for "Remote Procedure Call," a function call where the work of
 producing the return value is being done somewhere else. When an rpc function is
 invoked, behind the scenes the function contacts some other process somewhere
@@ -40,7 +42,7 @@ process, and no context switching between different languages.
 Some other features of tarpc:
 - Pluggable transport: any type impling `Stream<Item = Request> + Sink<Response>` can be
   used as a transport to connect the client and server.
-- `Send` optional: if the transport doesn't require it, neither does tarpc!
+- `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.
   The server will cease any unfinished work on the request, subsequently cancelling any of its
   own requests, repeating for the entire chain of transitive dependencies.
@@ -53,25 +55,26 @@ Some other features of tarpc:
   responses `Serialize + Deserialize`. It's entirely optional, though: in-memory transports can
   be used, as well, so the price of serialization doesn't have to be paid when it's not needed.
 
-### Usage
+## Usage
 Add to your `Cargo.toml` dependencies:
 
 ```toml
-tarpc = { version = "0.21.0", features = ["full"] }
+tarpc = "0.24"
 ```
 
 The `tarpc::service` attribute expands to a collection of items that form an rpc service.
 These generated types make it easy and ergonomic to write servers with less boilerplate.
 Simply implement the generated service trait, and you're off to the races!
 
-### Example
+## Example
 
-For this example, in addition to tarpc, also add two other dependencies to
+This example uses [tokio](https://tokio.rs), so add the following dependencies to
 your `Cargo.toml`:
 
 ```toml
-futures = "0.3"
-tokio = "0.2"
+futures = "1.0"
+tarpc = { version = "0.24", features = ["tokio1"] }
+tokio = { version = "1.0", features = ["macros"] }
 ```
 
 In the following example, we use an in-process channel for communication between
@@ -81,6 +84,7 @@ For a more real-world example, see [example-service](example-service).
 First, let's set up the dependencies and service definition.
 
 ```rust
+
 use futures::{
     future::{self, Ready},
     prelude::*,
@@ -109,19 +113,22 @@ implement it for our Server struct.
 #[derive(Clone)]
 struct HelloServer;
 
-#[tarpc::server]
 impl World for HelloServer {
-    async fn hello(self, _: context::Context, name: String) -> String {
-        format!("Hello, {}!", name)
+    // Each defined rpc generates two items in the trait, a fn that serves the RPC, and
+    // 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))
     }
 }
 ```
 
 Lastly let's write our `main` that will start the server. While this example uses an
-[in-process
-channel](https://docs.rs/tarpc/0.18.0/tarpc/transport/channel/struct.UnboundedChannel.html),
-tarpc also ships bincode and JSON
-tokio-net based TCP transports that are generic over all serializable types.
+[in-process channel](rpc::transport::channel), tarpc also ships a generic [`serde_transport`]
+behind the `serde-transport` feature, with additional [TCP](serde_transport::tcp) functionality
+available behind the `tcp` feature.
 
 ```rust
 #[tokio::main]
@@ -151,9 +158,11 @@ async fn main() -> io::Result<()> {
 }
 ```
 
-### Service Documentation
+## Service Documentation
 
 Use `cargo doc` as you normally would to see the documentation created for all
 items expanded by a `service!` invocation.
 
+<!-- cargo-sync-readme end -->
+
 License: MIT

RELEASES.md

@@ -1,3 +1,45 @@
+## 0.24.1 (2020-12-28)
+
+### Breaking Changes
+
+Upgrades tokio to 1.0.
+
+## 0.24.0 (2020-12-28)
+
+This release was yanked.
+
+## 0.23.0 (2020-10-19)
+
+### Breaking Changes
+
+Upgrades tokio to 0.3.
+
+## 0.22.0 (2020-08-02)
+
+This release adds some flexibility and consistency to `serde_transport`, with one new feature and
+one small breaking change.
+
+### New Features
+
+`serde_transport::tcp` now exposes framing configuration on `connect()` and `listen()`. This is
+useful if, for instance, you want to send requests or responses that are larger than the maximum
+payload allowed by default:
+
+```rust
+let mut transport = tarpc::serde_transport::tcp::connect(server_addr, Json::default);
+transport.config_mut().max_frame_length(4294967296);
+let mut client = MyClient::new(client::Config::default(), transport.await?).spawn()?;
+```
+
+### Breaking Changes
+
+The codec argument to `serde_transport::tcp::connect` changed from a Codec to impl Fn() -> Codec,
+to be consistent with `serde_transport::tcp::listen`. While only one Codec is needed, more than one
+person has been tripped up by the inconsistency between `connect` and `listen`. Unfortunately, the
+compiler errors are not much help in this case, so it was decided to simply do the more intuitive
+thing so that the compiler doesn't need to step in in the first place.
+
+
 ## 0.21.1 (2020-08-02)
 
 ### New Features
@@ -62,7 +104,7 @@ nameable futures and will just be boxing the return type anyway. This macro does
 ### Bug Fixes
 
 - https://github.com/google/tarpc/issues/304
-  
+
   A race condition in code that limits number of connections per client caused occasional panics.
 
 - https://github.com/google/tarpc/pull/295
@@ -82,7 +124,7 @@ nameable futures and will just be boxing the return type anyway. This macro does
 
 ## 0.13.0 (2018-10-16)
 
-### Breaking Changes 
+### Breaking Changes
 
 Version 0.13 marks a significant departure from previous versions of tarpc. The
 API has changed significantly. The tokio-proto crate has been torn out and

example-service/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "tarpc-example-service"
-version = "0.6.0"
+version = "0.8.0"
 authors = ["Tim Kuehn <tikue@google.com>"]
 edition = "2018"
 license = "MIT"
@@ -13,13 +13,12 @@ readme = "../README.md"
 description = "An example server built on tarpc."
 
 [dependencies]
-clap = "2.0"
+clap = "2.33"
+env_logger = "0.8"
 futures = "0.3"
 serde = { version = "1.0" }
-tarpc = { version = "0.21", path = "../tarpc", features = ["full"] }
-tokio = { version = "0.2", features = ["full"] }
-tokio-serde = { version = "0.6", features = ["json"] }
-env_logger = "0.6"
+tarpc = { version = "0.24", path = "../tarpc", features = ["full"] }
+tokio = { version = "1", features = ["full"] }
 
 [lib]
 name = "service"

example-service/src/client.rs

@@ -6,8 +6,7 @@
 
 use clap::{App, Arg};
 use std::{io, net::SocketAddr};
-use tarpc::{client, context};
-use tokio_serde::formats::Json;
+use tarpc::{client, context, tokio_serde::formats::Json};
 
 #[tokio::main]
 async fn main() -> io::Result<()> {
@@ -43,11 +42,13 @@ async fn main() -> io::Result<()> {
 
     let name = flags.value_of("name").unwrap().into();
 
-    let transport = tarpc::serde_transport::tcp::connect(server_addr, Json::default()).await?;
+    let mut transport = tarpc::serde_transport::tcp::connect(server_addr, Json::default);
+    transport.config_mut().max_frame_length(4294967296);
 
     // WorldClient is generated by the service attribute. It has a constructor `new` that takes a
     // config and any Transport as input.
-    let mut client = service::WorldClient::new(client::Config::default(), transport).spawn()?;
+    let mut client =
+        service::WorldClient::new(client::Config::default(), transport.await?).spawn()?;
 
     // The client has an RPC method for each RPC defined in the annotated trait. It takes the same
     // args as defined, with the addition of a Context, which is always the first arg. The Context

example-service/src/server.rs

@@ -14,8 +14,8 @@ use std::{
 use tarpc::{
     context,
     server::{self, Channel, Handler},
+    tokio_serde::formats::Json,
 };
-use tokio_serde::formats::Json;
 
 // This is the type that implements the generated World trait. It is the business logic
 // and is used to start the server.
@@ -57,8 +57,9 @@ async fn main() -> io::Result<()> {
 
     // JSON transport is provided by the json_transport tarpc module. It makes it easy
     // to start up a serde-powered json serialization strategy over TCP.
-    tarpc::serde_transport::tcp::listen(&server_addr, Json::default)
-        .await?
+    let mut listener = tarpc::serde_transport::tcp::listen(&server_addr, Json::default).await?;
+    listener.config_mut().max_frame_length(4294967296);
+    listener
         // Ignore accept errors.
         .filter_map(|r| future::ready(r.ok()))
         .map(server::BaseChannel::with_defaults)

hooks/pre-commit

@@ -93,7 +93,7 @@ diff=""
 for file in $(git diff --name-only --cached);
 do
 	if [ ${file: -3} == ".rs" ]; then
-        diff="$diff$(cargo fmt -- --skip-children --write-mode=diff $file)"
+        diff="$diff$(cargo fmt -- --unstable-features --skip-children --check $file)"
 	fi
 done
 if grep --quiet "^[-+]" <<< "$diff"; then

plugins/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "tarpc-plugins"
-version = "0.8.0"
+version = "0.9.0"
 authors = ["Adam Wright <adam.austin.wright@gmail.com>", "Tim Kuehn <timothy.j.kuehn@gmail.com>"]
 edition = "2018"
 license = "MIT"
@@ -19,15 +19,15 @@ serde1 = []
 travis-ci = { repository = "google/tarpc" }
 
 [dependencies]
-syn = { version = "1.0.11", features = ["full"] }
-quote = "1.0.2"
-proc-macro2 = "1.0.6"
+proc-macro2 = "1.0"
+quote = "1.0"
+syn = { version = "1.0", features = ["full"] }
 
 [lib]
 proc-macro = true
 
 [dev-dependencies]
+assert-type-eq = "0.1.0"
 futures = "0.3"
 serde = { version = "1.0", features = ["derive"] }
 tarpc = { path = "../tarpc" }
-assert-type-eq = "0.1.0"

plugins/src/lib.rs

@@ -215,6 +215,19 @@ impl Parse for DeriveSerde {
     }
 }
 
+/// Generates:
+/// - derive of Debug, serde Serialize & Deserialize
+/// - serde crate annotation
+#[proc_macro_attribute]
+pub fn derive_serde(_attr: TokenStream, item: TokenStream) -> TokenStream {
+    let mut gen: proc_macro2::TokenStream = quote! {
+        #[derive(tarpc::serde::Serialize, tarpc::serde::Deserialize)]
+        #[serde(crate = "tarpc::serde")]
+    };
+    gen.extend(proc_macro2::TokenStream::from(item));
+    proc_macro::TokenStream::from(gen)
+}
+
 /// Generates:
 /// - service trait
 /// - serve fn
@@ -240,7 +253,10 @@ pub fn service(attr: TokenStream, input: TokenStream) -> TokenStream {
     let args: &[&[PatType]] = &rpcs.iter().map(|rpc| &*rpc.args).collect::<Vec<_>>();
     let response_fut_name = &format!("{}ResponseFut", ident.unraw());
     let derive_serialize = if derive_serde.0 {
-        Some(quote!(#[derive(serde::Serialize, serde::Deserialize)]))
+        Some(
+            quote! {#[derive(tarpc::serde::Serialize, tarpc::serde::Deserialize)]
+            #[serde(crate = "tarpc::serde")]},
+        )
     } else {
         None
     };
@@ -328,60 +344,6 @@ fn transform_method(method: &mut ImplItemMethod) -> ImplItemType {
     t
 }
 
-/// Syntactic sugar to make using async functions in the server implementation
-/// easier. It does this by rewriting code like this, which would normally not
-/// compile because async functions are disallowed in trait implementations:
-///
-/// ```rust
-/// # extern crate tarpc;
-/// # use tarpc::context;
-/// # use std::net::SocketAddr;
-/// #[tarpc_plugins::service]
-/// trait World {
-///     async fn hello(name: String) -> String;
-/// }
-///
-/// #[derive(Clone)]
-/// struct HelloServer(SocketAddr);
-///
-/// #[tarpc_plugins::server]
-/// impl World for HelloServer {
-///     async fn hello(self, _: context::Context, name: String) -> String {
-///         format!("Hello, {}! You are connected from {:?}.", name, self.0)
-///     }
-/// }
-/// ```
-///
-/// Into code like this, which matches the service trait definition:
-///
-/// ```rust
-/// # extern crate tarpc;
-/// # use tarpc::context;
-/// # use std::pin::Pin;
-/// # use futures::Future;
-/// # use std::net::SocketAddr;
-/// #[tarpc_plugins::service]
-/// trait World {
-///     async fn hello(name: String) -> String;
-/// }
-///
-/// #[derive(Clone)]
-/// struct HelloServer(SocketAddr);
-///
-/// impl World for HelloServer {
-///     type HelloFut = Pin<Box<dyn Future<Output = String> + Send>>;
-///
-///     fn hello(self, _: context::Context, name: String) -> Pin<Box<dyn Future<Output = String>
-///     + Send>> {
-///         Box::pin(async move {
-///             format!("Hello, {}! You are connected from {:?}.", name, self.0)
-///         })
-///     }
-/// }
-/// ```
-///
-/// Note that this won't touch functions unless they have been annotated with
-/// `async`, meaning that this should not break existing code.
 #[proc_macro_attribute]
 pub fn server(_attr: TokenStream, input: TokenStream) -> TokenStream {
     let mut item = syn::parse_macro_input!(input as ItemImpl);

tarpc/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "tarpc"
-version = "0.21.1"
+version = "0.24.1"
 authors = ["Adam Wright <adam.austin.wright@gmail.com>", "Tim Kuehn <timothy.j.kuehn@gmail.com>"]
 edition = "2018"
 license = "MIT"
@@ -16,9 +16,9 @@ description = "An RPC framework for Rust with a focus on ease of use."
 default = []
 
 serde1 = ["tarpc-plugins/serde1", "serde", "serde/derive"]
-tokio1 = []
-serde-transport = ["tokio-serde", "tokio-util/codec"]
-tcp = ["tokio/net", "tokio/stream"]
+tokio1 = ["tokio/rt-multi-thread"]
+serde-transport = ["serde1", "tokio1", "tokio-serde/json", "tokio-util/codec"]
+tcp = ["tokio/net"]
 
 full = ["serde1", "tokio1", "serde-transport", "tcp"]
 
@@ -29,36 +29,38 @@ travis-ci = { repository = "google/tarpc" }
 anyhow = "1.0"
 fnv = "1.0"
 futures = "0.3"
-humantime = "1.0"
+humantime = "2.0"
 log = "0.4"
-pin-project = "0.4.17"
+pin-project = "1.0"
 rand = "0.7"
-tokio = { version = "0.2", features = ["time"] }
 serde = { optional = true, version = "1.0", features = ["derive"] }
 static_assertions = "1.1.0"
-tokio-util = { optional = true, version = "0.2" }
-tarpc-plugins = { path = "../plugins", version = "0.8" }
-tokio-serde = { optional = true, version = "0.6" }
+tarpc-plugins = { path = "../plugins", version = "0.9" }
+tokio = { version = "1", features = ["time"] }
+tokio-util = { optional = true, version = "0.6" }
+tokio-serde = { optional = true, version = "0.8" }
 
 [dev-dependencies]
-assert_matches = "1.0"
+assert_matches = "1.4"
 bincode = "1.3"
-bytes = { version = "0.5", features = ["serde"] }
-env_logger = "0.6"
-flate2 = "1.0.16"
-futures = "0.3"
-humantime = "1.0"
+bytes = { version = "1", features = ["serde"] }
+env_logger = "0.8"
+flate2 = "1.0"
 log = "0.4"
 pin-utils = "0.1.0-alpha"
 serde_bytes = "0.11"
-tokio = { version = "0.2", features = ["full"] }
-tokio-serde = { version = "0.6", features = ["json", "bincode"] }
+tokio = { version = "1", features = ["full"] }
+tokio-serde = { version = "0.8", features = ["json", "bincode"] }
 trybuild = "1.0"
 
 [package.metadata.docs.rs]
 all-features = true
 rustdoc-args = ["--cfg", "docsrs"]
 
+[[example]]
+name = "compression"
+required-features = ["serde-transport", "tcp"]
+
 [[example]]
 name = "server_calling_server"
 required-features = ["full"]
@@ -70,3 +72,15 @@ required-features = ["full"]
 [[example]]
 name = "pubsub"
 required-features = ["full"]
+
+[[example]]
+name = "custom_transport"
+required-features = ["serde1", "tokio1", "serde-transport"]
+
+[[test]]
+name = "service_functional"
+required-features = ["serde-transport"]
+
+[[test]]
+name = "dataservice"
+required-features = ["serde-transport", "tcp"]

tarpc/examples/compression.rs

@@ -118,7 +118,7 @@ async fn main() -> anyhow::Result<()> {
             .await;
     });
 
-    let transport = tcp::connect(addr, Bincode::default()).await?;
+    let transport = tcp::connect(addr, Bincode::default).await?;
     let mut client =
         WorldClient::new(client::Config::default(), add_compression(transport)).spawn()?;
 

tarpc/examples/custom_transport.rs

@@ -0,0 +1,52 @@
+use futures::future;
+use tarpc::context::Context;
+use tarpc::serde_transport as transport;
+use tarpc::server::{BaseChannel, Channel};
+use tokio::net::{UnixListener, UnixStream};
+use tokio_serde::formats::Bincode;
+use tokio_util::codec::length_delimited::LengthDelimitedCodec;
+
+#[tarpc::service]
+pub trait PingService {
+    async fn ping();
+}
+
+#[derive(Clone)]
+struct Service;
+
+impl PingService for Service {
+    type PingFut = future::Ready<()>;
+
+    fn ping(self, _: Context) -> Self::PingFut {
+        future::ready(())
+    }
+}
+
+#[tokio::main]
+async fn main() -> std::io::Result<()> {
+    let bind_addr = "/tmp/tarpc_on_unix_example.sock";
+
+    let _ = std::fs::remove_file(bind_addr);
+
+    let listener = UnixListener::bind(bind_addr).unwrap();
+    let codec_builder = LengthDelimitedCodec::builder();
+    tokio::spawn(async move {
+        loop {
+            let (conn, _addr) = listener.accept().await.unwrap();
+            let framed = codec_builder.new_framed(conn);
+            let transport = transport::new(framed, Bincode::default());
+
+            let fut = BaseChannel::with_defaults(transport)
+                .respond_with(Service.serve())
+                .execute();
+            tokio::spawn(fut);
+        }
+    });
+
+    let conn = UnixStream::connect(bind_addr).await?;
+    let transport = transport::new(codec_builder.new_framed(conn), Bincode::default());
+    PingServiceClient::new(Default::default(), transport)
+        .spawn()?
+        .ping(tarpc::context::current())
+        .await
+}

tarpc/examples/pubsub.rs

@@ -18,7 +18,7 @@
 ///   messages to all clients subscribed to the topic of that message.
 ///
 ///       Subscriber                        Publisher                       PubSub Server
-/// T1        |                                 |                                 |             
+/// T1        |                                 |                                 |
 /// T2        |-----Connect------------------------------------------------------>|
 /// T3        |                                 |                                 |
 /// T2        |<-------------------------------------------------------Topics-----|
@@ -103,7 +103,7 @@ impl Subscriber {
         publisher_addr: impl ToSocketAddrs,
         topics: Vec<String>,
     ) -> anyhow::Result<SubscriberHandle> {
-        let publisher = tcp::connect(publisher_addr, Json::default()).await?;
+        let publisher = tcp::connect(publisher_addr, Json::default).await?;
         let local_addr = publisher.local_addr()?;
         let mut handler = server::BaseChannel::with_defaults(publisher)
             .respond_with(Subscriber { local_addr, topics }.serve());
@@ -308,7 +308,7 @@ async fn main() -> anyhow::Result<()> {
 
     let mut publisher = publisher::PublisherClient::new(
         client::Config::default(),
-        tcp::connect(addrs.publisher, Json::default()).await?,
+        tcp::connect(addrs.publisher, Json::default).await?,
     )
     .spawn()?;
 

tarpc/examples/readme.rs

@@ -62,7 +62,7 @@ async fn main() -> io::Result<()> {
     };
     tokio::spawn(server);
 
-    let transport = tarpc::serde_transport::tcp::connect(addr, Json::default()).await?;
+    let transport = tarpc::serde_transport::tcp::connect(addr, Json::default).await?;
 
     // WorldClient is generated by the tarpc::service attribute. It has a constructor `new` that
     // takes a config and any Transport as input.

tarpc/examples/server_calling_server.rs

@@ -68,7 +68,7 @@ async fn main() -> io::Result<()> {
         .respond_with(AddServer.serve());
     tokio::spawn(add_server);
 
-    let to_add_server = tarpc::serde_transport::tcp::connect(addr, Json::default()).await?;
+    let to_add_server = tarpc::serde_transport::tcp::connect(addr, Json::default).await?;
     let add_client = add::AddClient::new(client::Config::default(), to_add_server).spawn()?;
 
     let double_listener = tarpc::serde_transport::tcp::listen("localhost:0", Json::default)
@@ -81,7 +81,7 @@ async fn main() -> io::Result<()> {
         .respond_with(DoubleServer { add_client }.serve());
     tokio::spawn(double_server);
 
-    let to_double_server = tarpc::serde_transport::tcp::connect(addr, Json::default()).await?;
+    let to_double_server = tarpc::serde_transport::tcp::connect(addr, Json::default).await?;
     let mut double_client =
         double::DoubleClient::new(client::Config::default(), to_double_server).spawn()?;
 

tarpc/src/lib.rs

@@ -3,10 +3,6 @@
 // Use of this source code is governed by an MIT-style
 // license that can be found in the LICENSE file or at
 // https://opensource.org/licenses/MIT.
-
-//! [![Latest Version](https://img.shields.io/crates/v/tarpc.svg)](https://crates.io/crates/tarpc)
-//! [![Join the chat at https://gitter.im/tarpc/Lobby](https://badges.gitter.im/tarpc/Lobby.svg)](https://gitter.im/tarpc/Lobby?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge)
-//!
 //! *Disclaimer*: This is not an official Google product.
 //!
 //! tarpc is an RPC framework for rust with a focus on ease of use. Defining a
@@ -50,7 +46,7 @@
 //! Add to your `Cargo.toml` dependencies:
 //!
 //! ```toml
-//! tarpc = "0.21.0"
+//! tarpc = "0.24"
 //! ```
 //!
 //! The `tarpc::service` attribute expands to a collection of items that form an rpc service.
@@ -59,12 +55,13 @@
 //!
 //! ## Example
 //!
-//! For this example, in addition to tarpc, also add two other dependencies to
+//! This example uses [tokio](https://tokio.rs), so add the following dependencies to
 //! your `Cargo.toml`:
 //!
 //! ```toml
-//! futures = "0.3"
-//! tokio = "0.2"
+//! futures = "1.0"
+//! tarpc = { version = "0.24", features = ["tokio1"] }
+//! tokio = { version = "1.0", features = ["macros"] }
 //! ```
 //!
 //! In the following example, we use an in-process channel for communication between
@@ -168,6 +165,9 @@
 //! #         future::ready(format!("Hello, {}!", name))
 //! #     }
 //! # }
+//! # #[cfg(not(feature = "tokio1"))]
+//! # fn main() {}
+//! # #[cfg(feature = "tokio1")]
 //! #[tokio::main]
 //! async fn main() -> io::Result<()> {
 //!     let (client_transport, server_transport) = tarpc::transport::channel::unbounded();
@@ -206,12 +206,21 @@
 pub mod rpc;
 pub use rpc::*;
 
+#[cfg(feature = "serde1")]
+pub use serde;
+
+#[cfg(feature = "serde-transport")]
+pub use tokio_serde;
+
 #[cfg(feature = "serde-transport")]
 #[cfg_attr(docsrs, doc(cfg(feature = "serde-transport")))]
 pub mod serde_transport;
 
 pub mod trace;
 
+#[cfg(feature = "serde1")]
+pub use tarpc_plugins::derive_serde;
+
 /// The main macro that creates RPC services.
 ///
 /// Rpc methods are specified, mirroring trait syntax:

tarpc/src/rpc.rs

@@ -55,7 +55,7 @@ pub enum ClientMessage<T> {
     Cancel {
         /// The trace context associates the message with a specific chain of causally-related actions,
         /// possibly orchestrated across many distributed systems.
-        #[cfg_attr(feature = "serde", serde(default))]
+        #[cfg_attr(feature = "serde1", serde(default))]
         trace_context: trace::Context,
         /// The ID of the request to cancel.
         request_id: u64,

tarpc/src/rpc/client.rs

@@ -8,6 +8,7 @@
 
 use crate::context;
 use futures::prelude::*;
+use std::fmt;
 use std::io;
 
 /// Provides a [`Client`] backed by a transport.
@@ -127,7 +128,6 @@ impl Default for Config {
 
 /// A channel and dispatch pair. The dispatch drives the sending and receiving of requests
 /// and must be polled continuously or spawned.
-#[derive(Debug)]
 pub struct NewClient<C, D> {
     /// The new client.
     pub client: C,
@@ -153,3 +153,9 @@ where
         Ok(self.client)
     }
 }
+
+impl<C, D> fmt::Debug for NewClient<C, D> {
+    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(fmt, "NewClient")
+    }
+}

tarpc/src/rpc/client/channel.rs

@@ -88,7 +88,7 @@ impl<'a, Req, Resp> Future for Call<'a, Req, Resp> {
         let resp = ready!(self.as_mut().project().fut.poll(cx));
         Poll::Ready(match resp {
             Ok(resp) => resp,
-            Err(tokio::time::Elapsed { .. }) => Err(io::Error::new(
+            Err(tokio::time::error::Elapsed { .. }) => Err(io::Error::new(
                 io::ErrorKind::TimedOut,
                 "Client dropped expired request.".to_string(),
             )),
@@ -723,7 +723,7 @@ mod tests {
     };
     use std::{pin::Pin, sync::atomic::AtomicU64, sync::Arc};
 
-    #[tokio::test(threaded_scheduler)]
+    #[tokio::test]
     async fn dispatch_response_cancels_on_drop() {
         let (cancellation, mut canceled_requests) = cancellations();
         let (_, response) = oneshot::channel();
@@ -738,7 +738,7 @@ mod tests {
         assert_eq!(canceled_requests.0.try_next().unwrap(), Some(3));
     }
 
-    #[tokio::test(threaded_scheduler)]
+    #[tokio::test]
     async fn stage_request() {
         let (mut dispatch, mut channel, _server_channel) = set_up();
         let dispatch = Pin::new(&mut dispatch);
@@ -755,7 +755,7 @@ mod tests {
     }
 
     // Regression test for  https://github.com/google/tarpc/issues/220
-    #[tokio::test(threaded_scheduler)]
+    #[tokio::test]
     async fn stage_request_channel_dropped_doesnt_panic() {
         let (mut dispatch, mut channel, mut server_channel) = set_up();
         let mut dispatch = Pin::new(&mut dispatch);
@@ -776,7 +776,7 @@ mod tests {
         dispatch.await.unwrap();
     }
 
-    #[tokio::test(threaded_scheduler)]
+    #[tokio::test]
     async fn stage_request_response_future_dropped_is_canceled_before_sending() {
         let (mut dispatch, mut channel, _server_channel) = set_up();
         let dispatch = Pin::new(&mut dispatch);
@@ -791,7 +791,7 @@ mod tests {
         assert!(dispatch.poll_next_request(cx).ready().is_none());
     }
 
-    #[tokio::test(threaded_scheduler)]
+    #[tokio::test]
     async fn stage_request_response_future_dropped_is_canceled_after_sending() {
         let (mut dispatch, mut channel, _server_channel) = set_up();
         let cx = &mut Context::from_waker(&noop_waker_ref());
@@ -813,7 +813,7 @@ mod tests {
         assert!(dispatch.project().in_flight_requests.is_empty());
     }
 
-    #[tokio::test(threaded_scheduler)]
+    #[tokio::test]
     async fn stage_request_response_closed_skipped() {
         let (mut dispatch, mut channel, _server_channel) = set_up();
         let dispatch = Pin::new(&mut dispatch);

tarpc/src/rpc/server.rs

@@ -36,7 +36,6 @@ pub use self::{
 };
 
 /// Manages clients, serving multiplexed requests over each connection.
-#[derive(Debug)]
 pub struct Server<Req, Resp> {
     config: Config,
     ghost: PhantomData<(Req, Resp)>,
@@ -48,6 +47,12 @@ impl<Req, Resp> Default for Server<Req, Resp> {
     }
 }
 
+impl<Req, Resp> fmt::Debug for Server<Req, Resp> {
+    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(fmt, "Server")
+    }
+}
+
 /// Settings that control the behavior of the server.
 #[derive(Clone, Debug)]
 pub struct Config {
@@ -167,7 +172,6 @@ where
 
 /// BaseChannel lifts a Transport to a Channel by tracking in-flight requests.
 #[pin_project]
-#[derive(Debug)]
 pub struct BaseChannel<Req, Resp, T> {
     config: Config,
     /// Writes responses to the wire and reads requests off the wire.
@@ -236,6 +240,12 @@ where
     }
 }
 
+impl<Req, Resp, T> fmt::Debug for BaseChannel<Req, Resp, T> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "BaseChannel")
+    }
+}
+
 /// The server end of an open connection with a client, streaming in requests from, and sinking
 /// responses to, the client.
 ///
@@ -384,7 +394,6 @@ where
 
 /// A running handler serving all requests coming over a channel.
 #[pin_project]
-#[derive(Debug)]
 pub struct ClientHandler<C, S>
 where
     C: Channel,
@@ -508,9 +517,17 @@ where
     }
 }
 
+impl<C, S> fmt::Debug for ClientHandler<C, S>
+where
+    C: Channel,
+{
+    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(fmt, "ClientHandler")
+    }
+}
+
 /// A future fulfilling a single client request.
 #[pin_project]
-#[derive(Debug)]
 pub struct RequestHandler<F, R> {
     #[pin]
     resp: Abortable<Resp<F, R>>,
@@ -528,8 +545,13 @@ where
     }
 }
 
+impl<F, R> fmt::Debug for RequestHandler<F, R> {
+    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(fmt, "RequestHandler")
+    }
+}
+
 #[pin_project]
-#[derive(Debug)]
 struct Resp<F, R> {
     state: RespState,
     request_id: u64,
@@ -565,7 +587,7 @@ where
                         request_id: self.request_id,
                         message: match result {
                             Ok(message) => Ok(message),
-                            Err(tokio::time::Elapsed { .. }) => {
+                            Err(tokio::time::error::Elapsed { .. }) => {
                                 debug!(
                                     "[{}] Response did not complete before deadline of {}s.",
                                     self.ctx.trace_id(),
@@ -614,6 +636,12 @@ where
     }
 }
 
+impl<F, R> fmt::Debug for Resp<F, R> {
+    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(fmt, "Resp")
+    }
+}
+
 impl<C, S> Stream for ClientHandler<C, S>
 where
     C: Channel,
@@ -624,11 +652,7 @@ where
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         loop {
             let read = self.as_mut().pump_read(cx)?;
-            let read_closed = if let Poll::Ready(None) = read {
-                true
-            } else {
-                false
-            };
+            let read_closed = matches!(read, Poll::Ready(None));
             match (read, self.as_mut().pump_write(cx, read_closed)?) {
                 (Poll::Ready(None), Poll::Ready(None)) => {
                     return Poll::Ready(None);

tarpc/src/rpc/server/testing.rs

@@ -1,3 +1,9 @@
+// Copyright 2020 Google LLC
+//
+// Use of this source code is governed by an MIT-style
+// license that can be found in the LICENSE file or at
+// https://opensource.org/licenses/MIT.
+
 use crate::server::{Channel, Config};
 use crate::{context, Request, Response};
 use fnv::FnvHashSet;
@@ -111,10 +117,7 @@ pub trait PollExt {
 
 impl<T> PollExt for Poll<Option<T>> {
     fn is_done(&self) -> bool {
-        match self {
-            Poll::Ready(None) => true,
-            _ => false,
-        }
+        matches!(self, Poll::Ready(None))
     }
 }
 

tarpc/src/rpc/server/throttle.rs

@@ -1,3 +1,9 @@
+// Copyright 2020 Google LLC
+//
+// Use of this source code is governed by an MIT-style
+// license that can be found in the LICENSE file or at
+// https://opensource.org/licenses/MIT.
+
 use super::{Channel, Config};
 use crate::{Response, ServerError};
 use futures::{future::AbortRegistration, prelude::*, ready, task::*};

tarpc/src/rpc/transport/channel.rs

@@ -78,6 +78,7 @@ impl<Item, SinkItem> Sink<SinkItem> for UnboundedChannel<Item, SinkItem> {
 }
 
 #[cfg(test)]
+#[cfg(feature = "tokio1")]
 mod tests {
     use crate::{
         client, context,
@@ -89,8 +90,7 @@ mod tests {
     use log::trace;
     use std::io;
 
-    #[cfg(feature = "tokio1")]
-    #[tokio::test(threaded_scheduler)]
+    #[tokio::test]
     async fn integration() -> io::Result<()> {
         let _ = env_logger::try_init();
 

tarpc/src/rpc/util.rs

@@ -10,8 +10,8 @@ use std::{
     time::{Duration, SystemTime},
 };
 
-#[cfg(feature = "serde")]
-#[cfg_attr(docsrs, doc(cfg(feature = "serde")))]
+#[cfg(feature = "serde1")]
+#[cfg_attr(docsrs, doc(cfg(feature = "serde1")))]
 pub mod serde;
 
 /// Extension trait for [SystemTimes](SystemTime) in the future, i.e. deadlines.

tarpc/src/serde_transport.rs

@@ -90,6 +90,22 @@ fn convert<E: Into<Box<dyn Error + Send + Sync>>>(
     poll.map(|ready| ready.map_err(|e| io::Error::new(io::ErrorKind::Other, e)))
 }
 
+/// Constructs a new transport from a framed transport and a serialization codec.
+pub fn new<S, Item, SinkItem, Codec>(
+    framed_io: Framed<S, LengthDelimitedCodec>,
+    codec: Codec,
+) -> Transport<S, Item, SinkItem, Codec>
+where
+    S: AsyncWrite + AsyncRead,
+    Item: for<'de> Deserialize<'de>,
+    SinkItem: Serialize,
+    Codec: Serializer<SinkItem> + Deserializer<Item>,
+{
+    Transport {
+        inner: SerdeFramed::new(framed_io, codec),
+    }
+}
+
 impl<S, Item, SinkItem, Codec> From<(S, Codec)> for Transport<S, Item, SinkItem, Codec>
 where
     S: AsyncWrite + AsyncRead,
@@ -97,10 +113,8 @@ where
     SinkItem: Serialize,
     Codec: Serializer<SinkItem> + Deserializer<Item>,
 {
-    fn from((inner, codec): (S, Codec)) -> Self {
-        Transport {
-            inner: SerdeFramed::new(Framed::new(inner, LengthDelimitedCodec::new()), codec),
-        }
+    fn from((io, codec): (S, Codec)) -> Self {
+        new(Framed::new(io, LengthDelimitedCodec::new()), codec)
     }
 }
 
@@ -113,6 +127,7 @@ pub mod tcp {
         futures::ready,
         std::{marker::PhantomData, net::SocketAddr},
         tokio::net::{TcpListener, TcpStream, ToSocketAddrs},
+        tokio_util::codec::length_delimited,
     };
 
     mod private {
@@ -134,34 +149,65 @@ pub mod tcp {
         }
     }
 
-    /// Returns a new JSON transport that reads from and writes to `io`.
-    pub fn new<Item, SinkItem, Codec>(
-        io: TcpStream,
-        codec: Codec,
-    ) -> Transport<TcpStream, Item, SinkItem, Codec>
+    /// A connection Future that also exposes the length-delimited framing config.
+    #[pin_project]
+    pub struct Connect<T, Item, SinkItem, CodecFn> {
+        #[pin]
+        inner: T,
+        codec_fn: CodecFn,
+        config: length_delimited::Builder,
+        ghost: PhantomData<(fn(SinkItem), fn() -> Item)>,
+    }
+
+    impl<T, Item, SinkItem, Codec, CodecFn> Future for Connect<T, Item, SinkItem, CodecFn>
     where
+        T: Future<Output = io::Result<TcpStream>>,
         Item: for<'de> Deserialize<'de>,
         SinkItem: Serialize,
         Codec: Serializer<SinkItem> + Deserializer<Item>,
+        CodecFn: Fn() -> Codec,
     {
-        Transport::from((io, codec))
+        type Output = io::Result<Transport<TcpStream, Item, SinkItem, Codec>>;
+
+        fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
+            let io = ready!(self.as_mut().project().inner.poll(cx))?;
+            Poll::Ready(Ok(new(self.config.new_framed(io), (self.codec_fn)())))
+        }
     }
 
-    /// Connects to `addr`, wrapping the connection in a JSON transport.
-    pub async fn connect<A, Item, SinkItem, Codec>(
+    impl<T, Item, SinkItem, CodecFn> Connect<T, Item, SinkItem, CodecFn> {
+        /// Returns an immutable reference to the length-delimited codec's config.
+        pub fn config(&self) -> &length_delimited::Builder {
+            &self.config
+        }
+
+        /// Returns a mutable reference to the length-delimited codec's config.
+        pub fn config_mut(&mut self) -> &mut length_delimited::Builder {
+            &mut self.config
+        }
+    }
+
+    /// Connects to `addr`, wrapping the connection in a TCP transport.
+    pub fn connect<A, Item, SinkItem, Codec, CodecFn>(
         addr: A,
-        codec: Codec,
-    ) -> io::Result<Transport<TcpStream, Item, SinkItem, Codec>>
+        codec_fn: CodecFn,
+    ) -> Connect<impl Future<Output = io::Result<TcpStream>>, Item, SinkItem, CodecFn>
     where
         A: ToSocketAddrs,
         Item: for<'de> Deserialize<'de>,
         SinkItem: Serialize,
         Codec: Serializer<SinkItem> + Deserializer<Item>,
+        CodecFn: Fn() -> Codec,
     {
-        Ok(new(TcpStream::connect(addr).await?, codec))
+        Connect {
+            inner: TcpStream::connect(addr),
+            codec_fn,
+            config: LengthDelimitedCodec::builder(),
+            ghost: PhantomData,
+        }
     }
 
-    /// Listens on `addr`, wrapping accepted connections in JSON transports.
+    /// Listens on `addr`, wrapping accepted connections in TCP transports.
     pub async fn listen<A, Item, SinkItem, Codec, CodecFn>(
         addr: A,
         codec_fn: CodecFn,
@@ -178,6 +224,7 @@ pub mod tcp {
             listener,
             codec_fn,
             local_addr,
+            config: LengthDelimitedCodec::builder(),
             ghost: PhantomData,
         })
     }
@@ -189,7 +236,8 @@ pub mod tcp {
         listener: TcpListener,
         local_addr: SocketAddr,
         codec_fn: CodecFn,
-        ghost: PhantomData<(Item, SinkItem, Codec)>,
+        config: length_delimited::Builder,
+        ghost: PhantomData<(fn() -> Item, fn(SinkItem), Codec)>,
     }
 
     impl<Item, SinkItem, Codec, CodecFn> Incoming<Item, SinkItem, Codec, CodecFn> {
@@ -197,6 +245,16 @@ pub mod tcp {
         pub fn local_addr(&self) -> SocketAddr {
             self.local_addr
         }
+
+        /// Returns an immutable reference to the length-delimited codec's config.
+        pub fn config(&self) -> &length_delimited::Builder {
+            &self.config
+        }
+
+        /// Returns a mutable reference to the length-delimited codec's config.
+        pub fn config_mut(&mut self) -> &mut length_delimited::Builder {
+            &mut self.config
+        }
     }
 
     impl<Item, SinkItem, Codec, CodecFn> Stream for Incoming<Item, SinkItem, Codec, CodecFn>
@@ -209,9 +267,12 @@ pub mod tcp {
         type Item = io::Result<Transport<TcpStream, Item, SinkItem, Codec>>;
 
         fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-            let next =
-                ready!(Pin::new(&mut self.as_mut().project().listener.incoming()).poll_next(cx)?);
-            Poll::Ready(next.map(|conn| Ok(new(conn, (self.codec_fn)()))))
+            let conn: TcpStream =
+                ready!(Pin::new(&mut self.as_mut().project().listener).poll_accept(cx)?).0;
+            Poll::Ready(Some(Ok(new(
+                self.config.new_framed(conn),
+                (self.codec_fn)(),
+            ))))
         }
     }
 }
@@ -226,7 +287,7 @@ mod tests {
         io::{self, Cursor},
         pin::Pin,
     };
-    use tokio::io::{AsyncRead, AsyncWrite};
+    use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
     use tokio_serde::formats::SymmetricalJson;
 
     fn ctx() -> Context<'static> {
@@ -241,8 +302,8 @@ mod tests {
             fn poll_read(
                 mut self: Pin<&mut Self>,
                 cx: &mut Context<'_>,
-                buf: &mut [u8],
-            ) -> Poll<io::Result<usize>> {
+                buf: &mut ReadBuf<'_>,
+            ) -> Poll<io::Result<()>> {
                 AsyncRead::poll_read(Pin::new(self.0.get_mut()), cx, buf)
             }
         }
@@ -285,8 +346,8 @@ mod tests {
             fn poll_read(
                 self: Pin<&mut Self>,
                 _cx: &mut Context<'_>,
-                _buf: &mut [u8],
-            ) -> Poll<io::Result<usize>> {
+                _buf: &mut ReadBuf<'_>,
+            ) -> Poll<io::Result<()>> {
                 unreachable!()
             }
         }

tarpc/src/trace.rs

@@ -27,7 +27,7 @@ use std::{
 /// Consists of a span identifying an event, an optional parent span identifying a causal event
 /// that triggered the current span, and a trace with which all related spans are associated.
 #[derive(Debug, Default, PartialEq, Eq, Hash, Clone, Copy)]
-#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
+#[cfg_attr(feature = "serde1", derive(serde::Serialize, serde::Deserialize))]
 pub struct Context {
     /// An identifier of the trace associated with the current context. A trace ID is typically
     /// created at a root span and passed along through all causal events.
@@ -47,12 +47,12 @@ pub struct Context {
 /// A 128-bit UUID identifying a trace. All spans caused by the same originating span share the
 /// same trace ID.
 #[derive(Debug, Default, PartialEq, Eq, Hash, Clone, Copy)]
-#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
+#[cfg_attr(feature = "serde1", derive(serde::Serialize, serde::Deserialize))]
 pub struct TraceId(u128);
 
 /// A 64-bit identifier of a span within a trace. The identifier is unique within the span's trace.
 #[derive(Debug, Default, PartialEq, Eq, Hash, Clone, Copy)]
-#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
+#[cfg_attr(feature = "serde1", derive(serde::Serialize, serde::Deserialize))]
 pub struct SpanId(u64);
 
 impl Context {

tarpc/tests/compile_fail/tarpc_server_missing_async.rs

@@ -1,4 +1,4 @@
-#[tarpc::service]
+#[tarpc::service(derive_serde = false)]
 trait World {
     async fn hello(name: String) -> String;
 }

tarpc/tests/compile_fail/tarpc_server_missing_async.stderr

@@ -9,11 +9,3 @@ error: hint: `#[tarpc::server]` only rewrites async fns, and `fn hello` is not a
    |
 10 |     fn hello(name: String) ->  String {
    |     ^^
-
-error[E0433]: failed to resolve: use of undeclared type or module `serde`
- --> $DIR/tarpc_server_missing_async.rs:1:1
-  |
-1 | #[tarpc::service]
-  | ^^^^^^^^^^^^^^^^^ use of undeclared type or module `serde`
-  |
-  = note: this error originates in an attribute macro (in Nightly builds, run with -Z macro-backtrace for more info)

tarpc/tests/dataservice.rs

@@ -0,0 +1,51 @@
+use futures::prelude::*;
+use std::io;
+use tarpc::serde_transport;
+use tarpc::{client, context, server::Handler};
+use tokio_serde::formats::Json;
+
+#[tarpc::derive_serde]
+#[derive(Debug, PartialEq)]
+pub enum TestData {
+    Black,
+    White,
+}
+
+#[tarpc::service]
+pub trait ColorProtocol {
+    async fn get_opposite_color(color: TestData) -> TestData;
+}
+
+#[derive(Clone)]
+struct ColorServer;
+
+#[tarpc::server]
+impl ColorProtocol for ColorServer {
+    async fn get_opposite_color(self, _: context::Context, color: TestData) -> TestData {
+        match color {
+            TestData::White => TestData::Black,
+            TestData::Black => TestData::White,
+        }
+    }
+}
+
+#[tokio::test]
+async fn test_call() -> io::Result<()> {
+    let transport = tarpc::serde_transport::tcp::listen("localhost:56797", Json::default).await?;
+    let addr = transport.local_addr();
+    tokio::spawn(
+        tarpc::Server::default()
+            .incoming(transport.take(1).filter_map(|r| async { r.ok() }))
+            .respond_with(ColorServer.serve()),
+    );
+
+    let transport = serde_transport::tcp::connect(addr, Json::default).await?;
+    let mut client = ColorProtocolClient::new(client::Config::default(), transport).spawn()?;
+
+    let color = client
+        .get_opposite_color(context::current(), TestData::White)
+        .await?;
+    assert_eq!(color, TestData::Black);
+
+    Ok(())
+}

tarpc/tests/service_functional.rs

@@ -6,12 +6,11 @@ use futures::{
 use std::io;
 use tarpc::{
     client::{self},
-    context, serde_transport,
+    context,
     server::{self, BaseChannel, Channel, Handler},
     transport::channel,
 };
 use tokio::join;
-use tokio_serde::formats::Json;
 
 #[tarpc_plugins::service]
 trait Service {
@@ -36,7 +35,7 @@ impl Service for Server {
     }
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test]
 async fn sequential() -> io::Result<()> {
     let _ = env_logger::try_init();
 
@@ -58,12 +57,15 @@ async fn sequential() -> io::Result<()> {
     Ok(())
 }
 
-#[cfg(feature = "serde1")]
-#[tokio::test(threaded_scheduler)]
+#[cfg(all(feature = "serde-transport", feature = "tcp"))]
+#[tokio::test]
 async fn serde() -> io::Result<()> {
+    use tarpc::serde_transport;
+    use tokio_serde::formats::Json;
+
     let _ = env_logger::try_init();
 
-    let transport = serde_transport::tcp::listen("localhost:56789", Json::default).await?;
+    let transport = tarpc::serde_transport::tcp::listen("localhost:56789", Json::default).await?;
     let addr = transport.local_addr();
     tokio::spawn(
         tarpc::Server::default()
@@ -71,7 +73,7 @@ async fn serde() -> io::Result<()> {
             .respond_with(Server.serve()),
     );
 
-    let transport = serde_transport::tcp::connect(addr, Json::default()).await?;
+    let transport = serde_transport::tcp::connect(addr, Json::default).await?;
     let mut client = ServiceClient::new(client::Config::default(), transport).spawn()?;
 
     assert_matches!(client.add(context::current(), 1, 2).await, Ok(3));
@@ -83,7 +85,7 @@ async fn serde() -> io::Result<()> {
     Ok(())
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test]
 async fn concurrent() -> io::Result<()> {
     let _ = env_logger::try_init();
 
@@ -112,7 +114,7 @@ async fn concurrent() -> io::Result<()> {
     Ok(())
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test]
 async fn concurrent_join() -> io::Result<()> {
     let _ = env_logger::try_init();
 
@@ -142,7 +144,7 @@ async fn concurrent_join() -> io::Result<()> {
     Ok(())
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test]
 async fn concurrent_join_all() -> io::Result<()> {
     let _ = env_logger::try_init();