Repository: rusoto/rusoto (language: rust)

Issue:
Why does ByteStream require Sync?
This is a question rather than feature request/bug report. It seems that since rusoto 0.43, the [ByteStream](https://docs.rs/rusoto_core/0.46.0/rusoto_core/struct.ByteStream.html) structure requires that the underlying stream is `Sync`.

This is quite inconvenient as I happen to have a structure that interacts with C library via FFI that's marked as `Send + !Sync`.

As is my understanding, `Sync` is only necessary if you want to use the structure from multiple threads at once (rather than one at a time, i.e. `Send`).

What's the reason why Sync became needed? I'm new to Rust and async stuff so I might just be missing something obvious.


Discussion / hints:
That does seem odd. I suspect it's just fallout from the massive PR to convert everything to async/await, but I'm new around here. If you'd be up for submitting a PR that removes the `Sync` bound, I'd be up for trying to get it merged.

Fix patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index c9ee1195748..36b3c6fad16 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -11,7 +11,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   match `rusoto_credential`
 - Swap the non-RustCrypto `md5` crate for the RustCrypto `md-5` crate, to match
   usage of RustCrypto `sha2` crate
-
+- Remove `Sync` constraint on `ByteStream`-related functions.
 ## [0.46.0] - 2021-01-05
 
 - Display `rusoto_core::Client` in docs
diff --git a/rusoto/signature/src/stream.rs b/rusoto/signature/src/stream.rs
index 09925767eee..3eff6013632 100644
--- a/rusoto/signature/src/stream.rs
+++ b/rusoto/signature/src/stream.rs
@@ -13,7 +13,7 @@ pin_project! {
     pub struct ByteStream {
         size_hint: Option<usize>,
         #[pin]
-        inner: Pin<Box<dyn Stream<Item = Result<Bytes, io::Error>> + Send + Sync + 'static>>,
+        inner: Pin<Box<dyn Stream<Item = Result<Bytes, io::Error>> + Send + 'static>>,
     }
 }
 
@@ -21,7 +21,7 @@ impl ByteStream {
     /// Create a new `ByteStream` by wrapping a `futures` stream.
     pub fn new<S>(stream: S) -> ByteStream
     where
-        S: Stream<Item = Result<Bytes, io::Error>> + Send + Sync + 'static,
+        S: Stream<Item = Result<Bytes, io::Error>> + Send + 'static,
     {
         ByteStream {
             size_hint: None,
@@ -33,7 +33,7 @@ impl ByteStream {
     /// size_hint to satisy S3's `PutObject` API.
     pub fn new_with_size<S>(stream: S, size_hint: usize) -> ByteStream
     where
-        S: Stream<Item = Result<Bytes, io::Error>> + Send + Sync + 'static,
+        S: Stream<Item = Result<Bytes, io::Error>> + Send + 'static,
     {
         ByteStream {
             size_hint: Some(size_hint),
@@ -46,12 +46,12 @@ impl ByteStream {
     }
 
     /// Return an implementation of `AsyncRead` that uses async i/o to consume the stream.
-    pub fn into_async_read(self) -> impl AsyncRead + Send + Sync {
+    pub fn into_async_read(self) -> impl AsyncRead + Send {
         ImplAsyncRead::new(self.inner)
     }
 
     /// Return an implementation of `Read` that uses blocking i/o to consume the stream.
-    pub fn into_blocking_read(self) -> impl io::Read + Send + Sync {
+    pub fn into_blocking_read(self) -> impl io::Read + Send {
         ImplBlockingRead::new(self.inner)
     }
 }
@@ -84,12 +84,12 @@ pin_project! {
     struct ImplAsyncRead {
         buffer: BytesMut,
         #[pin]
-        stream: futures::stream::Fuse<Pin<Box<dyn Stream<Item = Result<Bytes, io::Error>> + Send + Sync>>>,
+        stream: futures::stream::Fuse<Pin<Box<dyn Stream<Item = Result<Bytes, io::Error>> + Send>>>,
     }
 }
 
 impl ImplAsyncRead {
-    fn new(stream: Pin<Box<dyn Stream<Item = Result<Bytes, io::Error>> + Send + Sync>>) -> Self {
+    fn new(stream: Pin<Box<dyn Stream<Item = Result<Bytes, io::Error>> + Send>>) -> Self {
         ImplAsyncRead {
             buffer: BytesMut::new(),
             stream: stream.fuse(),
@@ -128,7 +128,7 @@ pin_project! {
 }
 
 impl ImplBlockingRead {
-    fn new(stream: Pin<Box<dyn Stream<Item = Result<Bytes, io::Error>> + Send + Sync>>) -> Self {
+    fn new(stream: Pin<Box<dyn Stream<Item = Result<Bytes, io::Error>> + Send>>) -> Self {
         ImplBlockingRead {
             inner: ImplAsyncRead::new(stream),
         }
@@ -140,7 +140,11 @@ impl io::Read for ImplBlockingRead {
         let rt = tokio::runtime::Runtime::new()?;
         rt.block_on(future::poll_fn(|cx| {
             let mut buf = ReadBuf::new(buf);
-            futures::ready!(AsyncRead::poll_read(Pin::new(&mut self.inner), cx, &mut buf))?;
+            futures::ready!(AsyncRead::poll_read(
+                Pin::new(&mut self.inner),
+                cx,
+                &mut buf
+            ))?;
             Poll::Ready(Ok(buf.filled().len()))
         }))
     }


Test patch: