fix(tonic): make Streaming Sync again

The boxed `Decoder` inside `Streaming` need not be `Sync` since
#804.  Unfortunately, that makes
`Streaming` non-`Sync`, meaning that all the generated `tonic` futures
cannot be awaited in `Sync` futures.

In fact, the only times we use the `Decoder`, we have a `&mut` unique
reference to it, so we are guaranteed not to require synchronization.

The `sync_wrapper` crate encodes this reasoning, allowing us to safely
make the `Streaming` type `Sync` regardless of whether the contained
`Decoder` is `Sync` or not.

Author: Twey

tonic/Cargo.toml

@@ -94,6 +94,7 @@ zstd = { version = "0.13.0", optional = true }
 
 # channel
 hyper-timeout = {version = "0.5", optional = true}
+sync_wrapper = "1.0.2"
 
 [dev-dependencies]
 bencher = "0.1.5"

tonic/src/codec/decode.rs

@@ -11,6 +11,7 @@ use std::{
     task::ready,
     task::{Context, Poll},
 };
+use sync_wrapper::SyncWrapper;
 use tokio_stream::Stream;
 use tracing::{debug, trace};
 
@@ -19,7 +20,7 @@ use tracing::{debug, trace};
 /// This will wrap some inner [`Body`] and [`Decoder`] and provide an interface
 /// to fetch the message stream and trailing metadata
 pub struct Streaming<T> {
-    decoder: Box<dyn Decoder<Item = T, Error = Status> + Send + 'static>,
+    decoder: SyncWrapper<Box<dyn Decoder<Item = T, Error = Status> + Send + 'static>>,
     inner: StreamingInner,
 }
 
@@ -367,8 +368,8 @@ impl<T> Streaming<T> {
     }
 
     fn decode_chunk(&mut self) -> Result<Option<T>, Status> {
-        match self.inner.decode_chunk(self.decoder.buffer_settings())? {
-            Some(mut decode_buf) => match self.decoder.decode(&mut decode_buf)? {
+        match self.inner.decode_chunk(self.decoder.get_mut().buffer_settings())? {
+            Some(mut decode_buf) => match self.decoder.get_mut().decode(&mut decode_buf)? {
                 Some(msg) => {
                     self.inner.state = State::ReadHeader;
                     Ok(Some(msg))