Tokio - Serde - Bincode

[Bakuta]

I'm attempting to create an asynchronous stream that sends/receives serialized/deserialized data. I've been pouring through various crates and have come to the conclusion that I'll need the following crates: tokio_serde, tokio_util, tokio, and futures.

However, where I look I can't seem to find how to piece all of these together.
I have essentially an initial stream (Tcp/Unix Stream) and want one task to write data to the stream, and another to read data from the stream.

let stream = // 
let (rd, wr) = stream.into_split();

// reader
tokio::spawn( async move {
    let ld = FramedRead::new(rd, LengthDelimitedCodec::new());
    let bincode = SymmetricalBincode::<MyType>::default();
    let deserialized = SymmetricallyFramed::new(ld, bincode); 

    // I would like to do something like
    while let Some(my_type) = deserialized.next().await? {

    }
});

// I want to do something similar to the writer here

However, I can't quite seem to get all of this working together and haven't found a good example/resources/docs on how to actually compose all of these pieces.

Any help/advice would be greatly appreciated

Edit: I've also been looking at the async_bincode crate but I'm not quite sure if this is the direction I want to take.

[alice]

You don't need all those things. You can just do this:

let mut ld = FramedRead::new(rd, LengthDelimitedCodec::new());
while let Some(bytes) = ld.next().await {
    let bytes = bytes?;
    let my_type: T = bincode::deserialize(&bytes)?;
    // .. use my_type
}

[Bakuta]

And I presume that when writing from the other end of the stream (on the server side persay) I need to use FramedWrite correct?
e.g.

let serialized = bincode::serialize::<MyType>(&bytes)?;
let mut ld = FramedWrite::new(wr, LengthDelimitedCodec::new());
ld.send(serialized).await?;

Thanks again for the help!!

[alice]

Yep. The return value of bincode::serialize is an Vec<u8>, but you can turn that into the type accepted by FramedWrite<LengthDelimitedCodec> with an Bytes::from(the_vec) call.