add streaming body to requests and responses #3

Merged
lx merged 64 commits from stream-body into main 2022-09-13 10:56:54 +00:00
6 changed files with 66 additions and 39 deletions
Showing only changes of commit 4745e7c4ba - Show all commits

View file

@ -227,9 +227,8 @@ impl ClientConn {
let code = resp[0];
if code == 0 {
let mut deser = rmp_serde::decode::Deserializer::from_read_ref(&resp[1..]);
let res = T::Response::deserialize_msg(&mut deser, stream).await?;
Ok(res)
let ser_resp = rmp_serde::decode::from_read_ref(&resp[1..])?;
Ok(T::Response::deserialize_msg(ser_resp, stream).await)
} else {
let msg = String::from_utf8(resp[1..].to_vec()).unwrap_or_default();
Err(Error::Remote(code, msg))

View file

@ -5,8 +5,7 @@ use std::sync::Arc;
use arc_swap::ArcSwapOption;
use async_trait::async_trait;
use serde::de::Error as DeError;
use serde::{Deserialize, Deserializer, Serialize, Serializer};
use serde::{Deserialize, Serialize};
use crate::error::Error;
use crate::netapp::*;
@ -22,42 +21,61 @@ pub trait Message: SerializeMessage + Send + Sync {
/// A trait for de/serializing messages, with possible associated stream.
#[async_trait]
pub trait SerializeMessage: Sized {
fn serialize_msg<S: Serializer>(
&self,
serializer: S,
) -> Result<(S::Ok, Option<AssociatedStream>), S::Error>;
type SerializableSelf: Serialize + for<'de> Deserialize<'de> + Send;
async fn deserialize_msg<'de, D: Deserializer<'de> + Send>(
deserializer: D,
stream: AssociatedStream,
) -> Result<Self, D::Error>;
// TODO should return Result
fn serialize_msg(&self) -> (Self::SerializableSelf, Option<AssociatedStream>);
// TODO should return Result
async fn deserialize_msg(ser_self: Self::SerializableSelf, stream: AssociatedStream) -> Self;
}
pub trait AutoSerialize: Serialize + for<'de> Deserialize<'de> + Clone + Send + Sync {}
#[async_trait]
impl<T> SerializeMessage for T
where
T: Serialize + for<'de> Deserialize<'de> + Send + Sync,
T: AutoSerialize,
{
fn serialize_msg<S: Serializer>(
&self,
serializer: S,
) -> Result<(S::Ok, Option<AssociatedStream>), S::Error> {
self.serialize(serializer).map(|r| (r, None))
type SerializableSelf = Self;
fn serialize_msg(&self) -> (Self::SerializableSelf, Option<AssociatedStream>) {
(self.clone(), None)
}
async fn deserialize_msg<'de, D: Deserializer<'de> + Send>(
deserializer: D,
mut stream: AssociatedStream,
) -> Result<Self, D::Error> {
use futures::StreamExt;
let res = Self::deserialize(deserializer)?;
if stream.next().await.is_some() {
return Err(D::Error::custom(
"failed to deserialize: found associated stream when none expected",
));
async fn deserialize_msg(ser_self: Self::SerializableSelf, stream: AssociatedStream) -> Self {
// TODO verify no stream
ser_self
}
}
impl AutoSerialize for () {}
#[async_trait]
impl<T, E> SerializeMessage for Result<T, E>
where
T: SerializeMessage + Send,
E: SerializeMessage + Send,
{
type SerializableSelf = Result<T::SerializableSelf, E::SerializableSelf>;
fn serialize_msg(&self) -> (Self::SerializableSelf, Option<AssociatedStream>) {
match self {
Ok(ok) => {
let (msg, stream) = ok.serialize_msg();
(Ok(msg), stream)
}
Err(err) => {
let (msg, stream) = err.serialize_msg();
(Err(msg), stream)
}
}
}
async fn deserialize_msg(ser_self: Self::SerializableSelf, stream: AssociatedStream) -> Self {
match ser_self {
Ok(ok) => Ok(T::deserialize_msg(ok, stream).await),
Err(err) => Err(E::deserialize_msg(err, stream).await),
}
Ok(res)
}
}
@ -139,7 +157,7 @@ where
prio: RequestPriority,
) -> Result<<M as Message>::Response, Error>
where
B: Borrow<M>,
B: Borrow<M> + Send + Sync,
{
if *target == self.netapp.id {
match self.handler.load_full() {
@ -202,8 +220,8 @@ where
match self.0.handler.load_full() {
None => Err(Error::NoHandler),
Some(h) => {
let mut deser = rmp_serde::decode::Deserializer::from_read_ref(buf);
let req = M::deserialize_msg(&mut deser, stream).await?;
let req = rmp_serde::decode::from_read_ref(buf)?;
let req = M::deserialize_msg(req, stream).await;
let res = h.handle(&req, from).await;
let res_bytes = rmp_to_vec_all_named(&res)?;
Ok(res_bytes)

View file

@ -32,12 +32,14 @@ pub(crate) type VersionTag = [u8; 16];
/// Value of the Netapp version used in the version tag
pub(crate) const NETAPP_VERSION_TAG: u64 = 0x6e65746170700004; // netapp 0x0004
#[derive(Serialize, Deserialize, Debug)]
#[derive(Serialize, Deserialize, Debug, Clone)]
pub(crate) struct HelloMessage {
pub server_addr: Option<IpAddr>,
pub server_port: u16,
}
impl AutoSerialize for HelloMessage {}
impl Message for HelloMessage {
type Response = ();
}

View file

@ -29,7 +29,7 @@ const FAILED_PING_THRESHOLD: usize = 3;
// -- Protocol messages --
#[derive(Serialize, Deserialize)]
#[derive(Serialize, Deserialize, Clone)]
struct PingMessage {
pub id: u64,
pub peer_list_hash: hash::Digest,
@ -39,7 +39,9 @@ impl Message for PingMessage {
type Response = PingMessage;
}
#[derive(Serialize, Deserialize)]
impl AutoSerialize for PingMessage {}
#[derive(Serialize, Deserialize, Clone)]
struct PeerListMessage {
pub list: Vec<(NodeID, SocketAddr)>,
}
@ -48,6 +50,8 @@ impl Message for PeerListMessage {
type Response = PeerListMessage;
}
impl AutoSerialize for PeerListMessage {}
// -- Algorithm data structures --
#[derive(Debug)]

View file

@ -151,9 +151,10 @@ impl Stream for DataReader {
}
let mut body = [0; MAX_CHUNK_LENGTH as usize];
body[..buf.len()].copy_from_slice(&buf);
let len = buf.len();
body[..len].copy_from_slice(buf);
buf.clear();
Poll::Ready(Some((body, MAX_CHUNK_LENGTH as usize)))
Poll::Ready(Some((body, len)))
}
}
}

View file

@ -8,6 +8,8 @@ use futures::Stream;
use log::info;
use serde::Serialize;
use tokio::sync::watch;
/// A node's identifier, which is also its public cryptographic key
@ -34,7 +36,8 @@ where
let mut se = rmp_serde::Serializer::new(&mut wr)
.with_struct_map()
.with_string_variants();
let (_, stream) = val.serialize_msg(&mut se)?;
let (val, stream) = val.serialize_msg();
val.serialize(&mut se)?;
Ok((wr, stream))
}