Implement Encoder<BytesMut> for BytesCodec

[qbx2]

Motivation

Let's assume that we want to write an echo server (actually, I'm writing a reverse proxy). It should be great if we can write as following:

use std::io;
use futures::StreamExt;
use tokio::net::{TcpListener, TcpStream};
use tokio_util::codec::{BytesCodec, FramedRead, FramedWrite};

#[tokio::main]
async fn main() -> io::Result<()> {
    let listener = TcpListener::bind("0.0.0.0:8080").await?;

    loop {
        let (socket, _) = listener.accept().await?;
        tokio::spawn(process_socket(socket));
    }
}

async fn process_socket(socket: TcpStream) {
    let (reader, writer) = socket.into_split();
    let reader = FramedRead::new(reader, BytesCodec::new());
    let writer = FramedWrite::new(writer, BytesCodec::new());
    reader.forward(writer).await.unwrap();
}

However, rustc complains that Encoder<BytesMut> is not implemented for BytesCodec. It does not make sense because it implements Encoder<Bytes>. Therefore, I added implementation of Encoder<BytesMut> for BytesCodec.

error[E0277]: the trait bound `BytesCodec: Encoder<bytes::bytes_mut::BytesMut>` is not satisfied
    --> src/main.rs:24:20
     |
24   |     reader.forward(writer).await.unwrap();
     |            ------- ^^^^^^ the trait `Encoder<bytes::bytes_mut::BytesMut>` is not implemented for `BytesCodec`
     |            |
     |            required by a bound introduced by this call
     |
     = help: the following implementations were found:
               <BytesCodec as Encoder<bytes::bytes::Bytes>>
     = note: required because of the requirements on the impl of `futures::Sink<bytes::bytes_mut::BytesMut>` for `FramedWrite<tokio::net::tcp::OwnedWriteHalf, BytesCodec>`
note: required by a bound in `futures::StreamExt::forward`
    --> /Users/qbx2/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-util-0.3.19/src/stream/stream/mod.rs:1432:12
     |
1432 |         S: Sink<Self::Ok, Error = Self::Error>,
     |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ required by this bound in `futures::StreamExt::forward`

error[E0277]: the trait bound `BytesCodec: Encoder<bytes::bytes_mut::BytesMut>` is not satisfied
    --> src/main.rs:24:12
     |
24   |     reader.forward(writer).await.unwrap();
     |            ^^^^^^^ the trait `Encoder<bytes::bytes_mut::BytesMut>` is not implemented for `BytesCodec`
     |
     = help: the following implementations were found:
               <BytesCodec as Encoder<bytes::bytes::Bytes>>
     = note: required because of the requirements on the impl of `futures::Sink<bytes::bytes_mut::BytesMut>` for `FramedWrite<tokio::net::tcp::OwnedWriteHalf, BytesCodec>`
note: required by a bound in `futures::StreamExt::forward`
    --> /Users/qbx2/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-util-0.3.19/src/stream/stream/mod.rs:1432:27
     |
1432 |         S: Sink<Self::Ok, Error = Self::Error>,
     |                           ^^^^^^^^^^^^^^^^^^^ required by this bound in `futures::StreamExt::forward`

For more information about this error, try `rustc --explain E0277`.

Solution

Add following implementation:

impl Encoder<BytesMut> for BytesCodec {
    type Error = io::Error;

    fn encode(&mut self, data: BytesMut, buf: &mut BytesMut) -> Result<(), io::Error> {
        buf.reserve(data.len());
        buf.put(data);
        Ok(())
    }
}

[Darksonn]

Seems good to me.

[qbx2]

The CI was stuck, could you please run it again?

[Darksonn]

Sorry, our CI appears to be broken. You don't have to do anything. I'll merge your PR when CI is fixed.