splittable RequestStream (#97)

Closes #83

Author: yuyuyureka

h3/src/client.rs

@@ -261,17 +261,11 @@ impl Builder {
     }
 }
 
-pub struct RequestStream<S, B>
-where
-    S: quic::RecvStream,
-{
-    inner: connection::RequestStream<FrameStream<S>, B>,
+pub struct RequestStream<S, B> {
+    inner: connection::RequestStream<S, B>,
 }
 
-impl<S, B> ConnectionState for RequestStream<S, B>
-where
-    S: quic::RecvStream,
-{
+impl<S, B> ConnectionState for RequestStream<S, B> {
     fn shared_state(&self) -> &SharedStateRef {
         &self.inner.conn_state
     }
@@ -339,7 +333,7 @@ where
 
 impl<S, B> RequestStream<S, B>
 where
-    S: quic::RecvStream + quic::SendStream<B>,
+    S: quic::SendStream<B>,
     B: Buf,
 {
     pub async fn send_data(&mut self, buf: B) -> Result<(), Error> {
@@ -354,3 +348,19 @@ where
         self.inner.finish().await
     }
 }
+
+impl<S, B> RequestStream<S, B>
+where
+    S: quic::BidiStream<B>,
+    B: Buf,
+{
+    pub fn split(
+        self,
+    ) -> (
+        RequestStream<S::SendStream, B>,
+        RequestStream<S::RecvStream, B>,
+    ) {
+        let (send, recv) = self.inner.split();
+        (RequestStream { inner: send }, RequestStream { inner: recv })
+    }
+}

h3/src/connection.rs

@@ -1,6 +1,5 @@
 use std::{
     convert::TryFrom,
-    marker::PhantomData,
     sync::{Arc, RwLock, RwLockReadGuard, RwLockWriteGuard},
     task::{Context, Poll},
 };
@@ -79,9 +78,9 @@ where
     pub(super) shared: SharedStateRef,
     conn: C,
     control_send: C::SendStream,
-    control_recv: Option<FrameStream<C::RecvStream>>,
-    decoder_recv: Option<AcceptedRecvStream<C::RecvStream>>,
-    encoder_recv: Option<AcceptedRecvStream<C::RecvStream>>,
+    control_recv: Option<FrameStream<C::RecvStream, B>>,
+    decoder_recv: Option<AcceptedRecvStream<C::RecvStream, B>>,
+    encoder_recv: Option<AcceptedRecvStream<C::RecvStream, B>>,
     pending_recv_streams: Vec<AcceptRecvStream<C::RecvStream>>,
     // The id of the last stream received by this connection:
     // request and push stream for server and clients respectively.
@@ -307,21 +306,23 @@ where
 }
 
 pub struct RequestStream<S, B> {
-    pub(super) stream: S,
+    pub(super) stream: FrameStream<S, B>,
     pub(super) trailers: Option<Bytes>,
     pub(super) conn_state: SharedStateRef,
     pub(super) max_field_section_size: u64,
-    _phantom_buffer: PhantomData<B>,
 }
 
 impl<S, B> RequestStream<S, B> {
-    pub fn new(stream: S, max_field_section_size: u64, conn_state: SharedStateRef) -> Self {
+    pub fn new(
+        stream: FrameStream<S, B>,
+        max_field_section_size: u64,
+        conn_state: SharedStateRef,
+    ) -> Self {
         Self {
             stream,
             conn_state,
             max_field_section_size,
             trailers: None,
-            _phantom_buffer: PhantomData,
         }
     }
 }
@@ -332,7 +333,7 @@ impl<S, B> ConnectionState for RequestStream<S, B> {
     }
 }
 
-impl<S, B> RequestStream<FrameStream<S>, B>
+impl<S, B> RequestStream<S, B>
 where
     S: quic::RecvStream,
 {
@@ -406,9 +407,9 @@ where
     }
 }
 
-impl<S, B> RequestStream<FrameStream<S>, B>
+impl<S, B> RequestStream<S, B>
 where
-    S: quic::SendStream<B> + quic::RecvStream,
+    S: quic::SendStream<B>,
     B: Buf,
 {
     /// Send some data on the response body.
@@ -449,3 +450,33 @@ where
             .map_err(|e| self.maybe_conn_err(e))
     }
 }
+
+impl<S, B> RequestStream<S, B>
+where
+    S: quic::BidiStream<B>,
+    B: Buf,
+{
+    pub(crate) fn split(
+        self,
+    ) -> (
+        RequestStream<S::SendStream, B>,
+        RequestStream<S::RecvStream, B>,
+    ) {
+        let (send, recv) = self.stream.split();
+
+        (
+            RequestStream {
+                stream: send,
+                trailers: None,
+                conn_state: self.conn_state.clone(),
+                max_field_section_size: 0,
+            },
+            RequestStream {
+                stream: recv,
+                trailers: self.trailers,
+                conn_state: self.conn_state,
+                max_field_section_size: self.max_field_section_size,
+            },
+        )
+    }
+}

h3/src/frame.rs

@@ -1,3 +1,4 @@
+use std::marker::PhantomData;
 use std::task::{Context, Poll};
 
 use bytes::{Buf, Bytes};
@@ -12,26 +13,21 @@ use crate::{
         frame::{self, Frame, PayloadLen},
         stream::StreamId,
     },
-    quic::{RecvStream, SendStream},
+    quic::{BidiStream, RecvStream, SendStream},
     stream::WriteBuf,
 };
 
-pub struct FrameStream<S>
-where
-    S: RecvStream,
-{
+pub struct FrameStream<S, B> {
     stream: S,
     bufs: BufList<Bytes>,
     decoder: FrameDecoder,
     remaining_data: usize,
     /// Set to true when `stream` reaches the end.
     is_eos: bool,
+    _phantom_buffer: PhantomData<B>,
 }
 
-impl<S> FrameStream<S>
-where
-    S: RecvStream,
-{
+impl<S, B> FrameStream<S, B> {
     pub fn new(stream: S) -> Self {
         Self::with_bufs(stream, BufList::new())
     }
@@ -43,9 +39,15 @@ where
             decoder: FrameDecoder::default(),
             remaining_data: 0,
             is_eos: false,
+            _phantom_buffer: PhantomData,
         }
     }
+}
 
+impl<S, B> FrameStream<S, B>
+where
+    S: RecvStream,
+{
     pub fn poll_next(
         &mut self,
         cx: &mut Context<'_>,
@@ -136,9 +138,9 @@ where
     }
 }
 
-impl<T, B> SendStream<B> for FrameStream<T>
+impl<T, B> SendStream<B> for FrameStream<T, B>
 where
-    T: SendStream<B> + RecvStream,
+    T: SendStream<B>,
     B: Buf,
 {
     type Error = <T as SendStream<B>>::Error;
@@ -164,6 +166,34 @@ where
     }
 }
 
+impl<S, B> FrameStream<S, B>
+where
+    S: BidiStream<B>,
+    B: Buf,
+{
+    pub(crate) fn split(self) -> (FrameStream<S::SendStream, B>, FrameStream<S::RecvStream, B>) {
+        let (send, recv) = self.stream.split();
+        (
+            FrameStream {
+                stream: send,
+                bufs: BufList::new(),
+                decoder: FrameDecoder::default(),
+                remaining_data: 0,
+                is_eos: false,
+                _phantom_buffer: PhantomData,
+            },
+            FrameStream {
+                stream: recv,
+                bufs: self.bufs,
+                decoder: self.decoder,
+                remaining_data: self.remaining_data,
+                is_eos: self.is_eos,
+                _phantom_buffer: PhantomData,
+            },
+        )
+    }
+}
+
 #[derive(Default)]
 pub struct FrameDecoder {
     expected: Option<usize>,
@@ -338,7 +368,7 @@ mod tests {
         Frame::headers(&b"trailer"[..]).encode_with_payload(&mut buf);
         recv.chunk(buf.freeze());
 
-        let mut stream = FrameStream::new(recv);
+        let mut stream: FrameStream<_, ()> = FrameStream::new(recv);
 
         assert_poll_matches!(
             |mut cx| stream.poll_next(&mut cx),
@@ -366,7 +396,7 @@ mod tests {
         Frame::headers(&b"header"[..]).encode_with_payload(&mut buf);
         let mut buf = buf.freeze();
         recv.chunk(buf.split_to(buf.len() - 1));
-        let mut stream = FrameStream::new(recv);
+        let mut stream: FrameStream<_, ()> = FrameStream::new(recv);
 
         assert_poll_matches!(
             |mut cx| stream.poll_next(&mut cx),
@@ -385,7 +415,7 @@ mod tests {
         FrameType::DATA.encode(&mut buf);
         VarInt::from(4u32).encode(&mut buf);
         recv.chunk(buf.freeze());
-        let mut stream = FrameStream::new(recv);
+        let mut stream: FrameStream<_, ()> = FrameStream::new(recv);
 
         assert_poll_matches!(
             |mut cx| stream.poll_next(&mut cx),
@@ -407,7 +437,7 @@ mod tests {
         let mut buf = buf.freeze();
         recv.chunk(buf.split_to(buf.len() - 2));
         recv.chunk(buf);
-        let mut stream = FrameStream::new(recv);
+        let mut stream: FrameStream<_, ()> = FrameStream::new(recv);
 
         // We get the total size of data about to be received
         assert_poll_matches!(
@@ -436,7 +466,7 @@ mod tests {
         VarInt::from(4u32).encode(&mut buf);
         buf.put_slice(&b"b"[..]);
         recv.chunk(buf.freeze());
-        let mut stream = FrameStream::new(recv);
+        let mut stream: FrameStream<_, ()> = FrameStream::new(recv);
 
         assert_poll_matches!(
             |mut cx| stream.poll_next(&mut cx),
@@ -468,7 +498,7 @@ mod tests {
         Frame::Data(Bytes::from("body")).encode_with_payload(&mut buf);
 
         recv.chunk(buf.freeze());
-        let mut stream = FrameStream::new(recv);
+        let mut stream: FrameStream<_, ()> = FrameStream::new(recv);
 
         assert_poll_matches!(
             |mut cx| stream.poll_next(&mut cx),
@@ -490,7 +520,7 @@ mod tests {
         buf.put_slice(&b"bo"[..]);
         recv.chunk(buf.clone().freeze());
 
-        let mut stream = FrameStream::new(recv);
+        let mut stream: FrameStream<_, ()> = FrameStream::new(recv);
 
         assert_poll_matches!(
             |mut cx| stream.poll_next(&mut cx),