Repository: tokio-rs/tokio (language: rust)

Issue:
`tokio_util::io::StreamReader` uses incorrect generic parameter for `Sink` impl
**Version**
```plaintext
tokio v1.38.0
tokio-stream v0.1.15
tokio-util v0.7.11
```


**Platform**
```plaintext
Linux 6.2.0-39-generic #40~22.04.1-Ubuntu SMP PREEMPT_DYNAMIC Thu Nov 16 10:53:04 UTC 2 x86_64 x86_64 x86_64 GNU/Linux
```

**Description**

I think that the generic parameters are used incorrectly where `tokio-util`'s `StreamReader` implements `Sink`:

https://github.com/tokio-rs/tokio/blob/8480a180e6ffdd0ec0ec213a9bebcda2445fc541/tokio-util/src/io/stream_reader.rs#L329-L330

It is written as though `StreamReader`'s second parameter is its error type, which is not correct-- the second parameter is its buffer type.

This breaks the ability to use `StreamReader` and `SinkWriter` together. In this example the error message makes it clear that the types have become confused:

```rust
use std::io::{self, Cursor};

use futures_sink::Sink;
use tokio::io::{AsyncRead, AsyncWrite};
use tokio_stream::Stream;
use tokio_util::io::{SinkWriter, StreamReader};

pub struct Foo;

impl Foo {
    pub fn into_async_rw(self) -> impl AsyncRead + AsyncWrite {
        // This is broken because StreamReader doesn't correctly pass through the Sink impl types.
        SinkWriter::new(StreamReader::new(self))
    }
}

impl Stream for Foo {
    type Item = io::Result<Cursor<Vec<u8>>>;

    fn poll_next(
        self: std::pin::Pin<&mut Self>,
        _cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Option<Self::Item>> {
        todo!()
    }
}

impl<'a> Sink<&'a [u8]> for Foo {
    type Error = io::Error;

    fn poll_ready(
        self: std::pin::Pin<&mut Self>,
        _cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Result<(), Self::Error>> {
        todo!()
    }

    fn start_send(self: std::pin::Pin<&mut Self>, _item: &'a [u8]) -> Result<(), Self::Error> {
        todo!()
    }

    fn poll_flush(
        self: std::pin::Pin<&mut Self>,
        _cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Result<(), Self::Error>> {
        todo!()
    }

    fn poll_close(
        self: std::pin::Pin<&mut Self>,
        _cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Result<(), Self::Error>> {
        todo!()
    }
}
```

```plaintext
error[E0271]: type mismatch resolving `<Foo as Sink<&[u8]>>::Error == Cursor<Vec<u8>>`
  --> src/lib.rs:11:35
   |
11 |     pub fn into_async_rw(self) -> impl AsyncRead + AsyncWrite {
   |                                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^ type mismatch resolving `<Foo as Sink<&[u8]>>::Error == Cursor<Vec<u8>>`
   |
note: expected this to be `std::io::Error`
  --> src/lib.rs:28:18
   |
28 |     type Error = io::Error;
   |                  ^^^^^^^^^
   = note: expected struct `std::io::Error`
              found struct `std::io::Cursor<Vec<u8>>`
   = note: required for `StreamReader<Foo, std::io::Cursor<Vec<u8>>>` to implement `for<'a> futures_sink::Sink<&'a [u8]>`
   = note: required for `SinkWriter<StreamReader<Foo, std::io::Cursor<Vec<u8>>>>` to implement `AsyncWrite`

error[E0277]: the trait bound `std::io::Error: From<std::io::Cursor<Vec<u8>>>` is not satisfied
  --> src/lib.rs:11:35
   |
11 |     pub fn into_async_rw(self) -> impl AsyncRead + AsyncWrite {
   |                                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^ the trait `From<std::io::Cursor<Vec<u8>>>` is not implemented for `std::io::Error`, which is required by `SinkWriter<StreamReader<Foo, std::io::Cursor<Vec<u8>>>>: AsyncWrite`
   |
   = help: the following other types implement trait `From<T>`:
             <std::io::Error as From<tokio_stream::Elapsed>>
             <std::io::Error as From<tokio::time::error::Elapsed>>
             <std::io::Error as From<TryReserveError>>
             <std::io::Error as From<NulError>>
             <std::io::Error as From<IntoInnerError<W>>>
             <std::io::Error as From<ErrorKind>>
   = note: required for `std::io::Cursor<Vec<u8>>` to implement `Into<std::io::Error>`
   = note: required for `SinkWriter<StreamReader<Foo, std::io::Cursor<Vec<u8>>>>` to implement `AsyncWrite`

```



Discussion / hints:

Fix patch:
diff --git a/tokio-util/src/io/stream_reader.rs b/tokio-util/src/io/stream_reader.rs
index 096c96f301a..7569a8b7832 100644
--- a/tokio-util/src/io/stream_reader.rs
+++ b/tokio-util/src/io/stream_reader.rs
@@ -326,7 +326,7 @@ impl<S, B> StreamReader<S, B> {
     }
 }
 
-impl<S: Sink<T, Error = E>, E, T> Sink<T> for StreamReader<S, E> {
+impl<S: Sink<T, Error = E>, B, E, T> Sink<T> for StreamReader<S, B> {
     type Error = E;
     fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         self.project().inner.poll_ready(cx)


Test patch: