splittable RequestStream

Author: Your Name

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>,

h3/src/server.rs

@@ -1,6 +1,7 @@
 use std::{
     collections::HashSet,
     convert::TryFrom,
+    sync::Arc,
     task::{Context, Poll},
 };
 
@@ -107,8 +108,10 @@ where
         };
 
         let mut request_stream = RequestStream {
-            stream_id: stream.id(),
-            request_end: self.request_end_send.clone(),
+            request_end: Arc::new(RequestEnd {
+                request_end: self.request_end_send.clone(),
+                stream_id: stream.id(),
+            }),
             inner: connection::RequestStream::new(
                 stream,
                 self.max_field_section_size,
@@ -296,19 +299,23 @@ impl Builder {
     }
 }
 
-pub struct RequestStream<S, B>
-where
-    S: quic::RecvStream,
-{
-    inner: connection::RequestStream<FrameStream<S>, B>,
-    stream_id: StreamId,
+pub struct RequestEnd {
     request_end: mpsc::UnboundedSender<StreamId>,
+    stream_id: StreamId,
 }
 
-impl<S, B> ConnectionState for RequestStream<S, B>
-where
-    S: quic::RecvStream,
-{
+pub struct RequestStream<S, B> {
+    inner: connection::RequestStream<S, B>,
+    request_end: Arc<RequestEnd>,
+}
+
+impl<S, B> AsMut<connection::RequestStream<S, B>> for RequestStream<S, B> {
+    fn as_mut(&mut self) -> &mut connection::RequestStream<S, B> {
+        &mut self.inner
+    }
+}
+
+impl<S, B> ConnectionState for RequestStream<S, B> {
     fn shared_state(&self) -> &SharedStateRef {
         &self.inner.conn_state
     }
@@ -329,7 +336,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_response(&mut self, resp: Response<()>) -> Result<(), Error> {
@@ -393,10 +400,32 @@ where
     }
 }
 
-impl<S, B> Drop for RequestStream<S, B>
+impl<S, B> RequestStream<S, B>
 where
-    S: quic::RecvStream,
+    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,
+                request_end: self.request_end.clone(),
+            },
+            RequestStream {
+                inner: recv,
+                request_end: self.request_end,
+            },
+        )
+    }
+}
+
+impl Drop for RequestEnd {
     fn drop(&mut self) {
         if let Err(e) = self.request_end.send(self.stream_id) {
             error!(

h3/src/stream.rs

@@ -159,12 +159,12 @@ where
     }
 }
 
-pub(super) enum AcceptedRecvStream<S>
+pub(super) enum AcceptedRecvStream<S, B>
 where
     S: quic::RecvStream,
 {
-    Control(FrameStream<S>),
-    Push(u64, FrameStream<S>),
+    Control(FrameStream<S, B>),
+    Push(u64, FrameStream<S, B>),
     Encoder(S),
     Decoder(S),
     Reserved,
@@ -195,7 +195,7 @@ where
         }
     }
 
-    pub fn into_stream(self) -> Result<AcceptedRecvStream<S>, Error> {
+    pub fn into_stream<B>(self) -> Result<AcceptedRecvStream<S, B>, Error> {
         Ok(match self.ty.expect("Stream type not resolved yet") {
             StreamType::CONTROL => {
                 AcceptedRecvStream::Control(FrameStream::with_bufs(self.stream, self.buf))