fn serialize(&self, value: impl serde::Serialize) -> Vec<u8> {

rmp_serde::to_vec_named(&value).expect("expected to serialize")

}

fn deserialize<P: serde::de::DeserializeOwned>(&self, b: &[u8]) -> Result<P, AnyError> {

rmp_serde::from_slice(b).map_err(|e| InvalidRpcDataError(e.to_string()).into())

}

C: Send + Sync + 'static,

X: Clone,

S: Send + Sync + Serialization,

Read: AsyncRead + Unpin,

Write: AsyncWrite + Unpin,

dispatcher: rpc::RpcDispatcher<S, C>,

mut read: Read,

mut write: Write,

mut msg_rx: impl Receivable<Vec<u8>>,

mut shutdown_rx: Barrier<X>,

) -> io::Result<(Option<X>, Read, Write)> {

let (write_tx, mut write_rx) = mpsc::channel::<Vec<u8>>(8);

let mut decoder = MsgPackCodec::new();

let mut decoder_buf = bytes::BytesMut::new();

let shutdown_fut = shutdown_rx.wait();

pin!(shutdown_fut);

loop {

tokio::select! {

r = read.read_buf(&mut decoder_buf) => {

r?;

while let Some(frame) = decoder.decode(&mut decoder_buf)? {

match dispatcher.dispatch_with_partial(&frame.vec, frame.obj) {

MaybeSync::Sync(Some(v)) => {

let _ = write_tx.send(v).await;

},

MaybeSync::Sync(None) => continue,

MaybeSync::Future(fut) => {

let write_tx = write_tx.clone();

tokio::spawn(async move {

if let Some(v) = fut.await {

let _ = write_tx.send(v).await;

}

});

}

MaybeSync::Stream((stream, fut)) => {

if let Some(stream) = stream {

dispatcher.register_stream(write_tx.clone(), stream).await;

}

let write_tx = write_tx.clone();

tokio::spawn(async move {

if let Some(v) = fut.await {

let _ = write_tx.send(v).await;

}

});

}

}

};

},

Some(m) = write_rx.recv() => {

write.write_all(&m).await?;

},

Some(m) = msg_rx.recv_msg() => {

write.write_all(&m).await?;

},

r = &mut shutdown_fut => return Ok((r.ok(), read, write)),

}

write.flush().await?;

}

pub fn new() -> Self {

Self {

_marker: std::marker::PhantomData,

}

}

type Item = MsgPackDecoded<T>;

type Error = io::Error;

fn decode(&mut self, src: &mut bytes::BytesMut) -> Result<Option<Self::Item>, Self::Error> {

let bytes_ref = src.as_ref();

let mut cursor = Cursor::new(bytes_ref);

match rmp_serde::decode::from_read::<_, T>(&mut cursor) {

Err(

rmp_serde::decode::Error::InvalidDataRead(e)

| rmp_serde::decode::Error::InvalidMarkerRead(e),

) if e.kind() == ErrorKind::UnexpectedEof => {

src.reserve(1024);

Ok(None)

}

Err(e) => Err(std::io::Error::new(

std::io::ErrorKind::InvalidData,

e.to_string(),

)),

Ok(obj) => {

let len = cursor.position() as usize;

let vec = src[..len].to_vec();

src.advance(len);

Ok(Some(MsgPackDecoded { obj, vec }))

}

}

}

use serde::{Deserialize, Serialize};

use super::*;

#[derive(Serialize, Deserialize, PartialEq, Eq, Debug)]

pub struct Msg {

pub x: i32,

}

#[test]

fn test_protocol() {

let mut c = MsgPackCodec::<Msg>::new();

let mut buf = bytes::BytesMut::new();

assert!(c.decode(&mut buf).unwrap().is_none());

buf.extend_from_slice(rmp_serde::to_vec_named(&Msg { x: 1 }).unwrap().as_slice());

buf.extend_from_slice(rmp_serde::to_vec_named(&Msg { x: 2 }).unwrap().as_slice());

assert_eq!(

c.decode(&mut buf).unwrap().expect("expected msg1").obj,

Msg { x: 1 }

);

assert_eq!(

c.decode(&mut buf).unwrap().expect("expected msg1").obj,

Msg { x: 2 }

);

}