Fix bug in RequestStream

examples/Cargo.toml

@@ -52,6 +52,6 @@ path = "client.rs"
 name = "server"
 path = "server.rs"
 
-[[example]]
-name = "webtransport_server"
-path = "webtransport_server.rs"
+#[[example]]
+#name = "webtransport_server"
+#path = "webtransport_server.rs"

examples/server.rs

@@ -1,13 +1,13 @@
 use std::{net::SocketAddr, path::PathBuf, sync::Arc};
 
 use bytes::{Bytes, BytesMut};
-use http::{Request, StatusCode};
+use http::StatusCode;
 use rustls::pki_types::{CertificateDer, PrivateKeyDer};
 use structopt::StructOpt;
 use tokio::{fs::File, io::AsyncReadExt};
 use tracing::{error, info, trace_span};
 
-use h3::{error::ErrorLevel, quic::BidiStream, server::RequestStream};
+use h3::{error::ErrorLevel, server::RequestResolver};
 use h3_quinn::quinn::{self, crypto::rustls::QuicServerConfig};
 
 #[derive(StructOpt, Debug)]
@@ -118,13 +118,11 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
                     loop {
                         match h3_conn.accept().await {
-                            Ok(Some((req, stream))) => {
-                                info!("new request: {:#?}", req);
-
+                            Ok(Some(resolver)) => {
                                 let root = root.clone();
 
                                 tokio::spawn(async {
-                                    if let Err(e) = handle_request(req, stream, root).await {
+                                    if let Err(e) = handle_request(resolver, root).await {
                                         error!("handling request failed: {}", e);
                                     }
                                 });
@@ -159,14 +157,21 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     Ok(())
 }
 
-async fn handle_request<T>(
-    req: Request<()>,
-    mut stream: RequestStream<T, Bytes>,
+async fn handle_request<C>(
+    resolver: RequestResolver<C, Bytes>,
     serve_root: Arc<Option<PathBuf>>,
 ) -> Result<(), Box<dyn std::error::Error>>
 where
-    T: BidiStream<Bytes>,
+    C: h3::quic::Connection<Bytes>,
 {
+    let (req, mut stream) = if let Some((req, stream)) = resolver.resolve_request().await? {
+        info!("received request: {:?}", req);
+        (req, stream)
+    } else {
+        info!("no request to resolve");
+        return Ok(());
+    };
+
     let (status, to_serve) = match serve_root.as_deref() {
         None => (StatusCode::OK, None),
         Some(_) if req.uri().path().contains("..") => (StatusCode::NOT_FOUND, None),

h3-webtransport/src/server.rs

@@ -167,7 +167,8 @@ where
 
     /// Accepts an incoming bidirectional stream or request
     pub async fn accept_bi(&self) -> Result<Option<AcceptedBi<C, B>>, Error> {
-        // Get the next stream
+        todo!("fix webtransport later")
+        /*    // Get the next stream
         // Accept the incoming stream
         let stream = poll_fn(|cx| {
             let mut conn = self.server_conn.lock().unwrap();
@@ -230,6 +231,7 @@ where
                 }
             }
         }
+        */
     }
 
     /// Open a new bidirectional stream

h3/src/client/builder.rs

@@ -11,9 +11,10 @@
 
 use crate::{
     config::Config,
     connection::{ConnectionInner, SharedStateRef},
     error::Error,
     quic::{self},
+    shared_state::SharedState2,
 };
 
 use super::connection::{Connection, SendRequest};
@@ -107,25 +108,30 @@
         B: Buf,
     {
         let open = quic.opener();
-        let conn_state = SharedStateRef::default();
+        let conn_state = Arc::new(SharedState2::default());
 
         let conn_waker = Some(future::poll_fn(|cx| Poll::Ready(cx.waker().clone())).await);
 
+        let inner = ConnectionInner::new(quic, conn_state.clone(), self.config).await?;
+        let send_request = SendRequest {
+            open,
+            conn_state: todo!(),
+            conn_state2: conn_state,
+            conn_waker,
+            max_field_section_size: self.config.settings.max_field_section_size,
+            sender_count: Arc::new(AtomicUsize::new(1)),
+            send_grease_frame: self.config.send_grease,
+            _buf: PhantomData,
+            error_sender: inner.error_sender.clone(),
+        };
+
         Ok((
             Connection {
-                inner: ConnectionInner::new(quic, conn_state.clone(), self.config).await?,
+                inner,
                 sent_closing: None,
                 recv_closing: None,
             },
-            SendRequest {
-                open,
-                conn_state,
-                conn_waker,
-                max_field_section_size: self.config.settings.max_field_section_size,
-                sender_count: Arc::new(AtomicUsize::new(1)),
-                send_grease_frame: self.config.send_grease,
-                _buf: PhantomData,
-            },
+            send_request,
         ))
     }
 }

h3/src/client/connection.rs

@@ -10,6 +10,7 @@
 use futures_util::future;
 use http::request;
 
+use tokio::sync::mpsc::UnboundedSender;
 #[cfg(feature = "tracing")]
 use tracing::{info, instrument, trace};
 
@@ -20,6 +21,7 @@
     proto::{frame::Frame, headers::Header, push::PushId},
     qpack,
     quic::{self, StreamId},
+    shared_state::SharedState2,
     stream::{self, BufRecvStream},
 };
 
@@ -109,12 +111,14 @@
 {
     pub(super) open: T,
     pub(super) conn_state: SharedStateRef,
+    pub(super) conn_state2: Arc<SharedState2>,
     pub(super) max_field_section_size: u64, // maximum size for a header we receive
     // counts instances of SendRequest to close the connection when the last is dropped.
     pub(super) sender_count: Arc<AtomicUsize>,
     pub(super) conn_waker: Option<Waker>,
     pub(super) _buf: PhantomData<fn(B)>,
     pub(super) send_grease_frame: bool,
+    pub(super) error_sender: UnboundedSender<(Code, &'static str)>,
 }
 
 impl<T, B> SendRequest<T, B>
@@ -153,7 +157,7 @@
         //# client MUST send only a single request on a given stream.
         let mut stream = future::poll_fn(|cx| self.open.poll_open_bidi(cx))
             .await
             .map_err(|e| self.maybe_conn_err(e))?;
 
         //= https://www.rfc-editor.org/rfc/rfc9114#section-4.2
         //= type=TODO
@@ -188,6 +192,7 @@
                 self.max_field_section_size,
                 self.conn_state.clone(),
                 self.send_grease_frame,
+                self.error_sender.clone(),
             ),
         };
         // send the grease frame only once
@@ -216,13 +221,15 @@
             .fetch_add(1, std::sync::atomic::Ordering::Release);
 
         Self {
+            conn_state2: self.conn_state2.clone(),
             open: self.open.clone(),
             conn_state: self.conn_state.clone(),
             max_field_section_size: self.max_field_section_size,
             sender_count: self.sender_count.clone(),
             conn_waker: self.conn_waker.clone(),
             _buf: PhantomData,
             send_grease_frame: self.send_grease_frame,
+            error_sender: self.error_sender.clone(),
         }
     }
 }
@@ -388,7 +395,7 @@
                 Ok(Frame::Settings(_)) => {
                     #[cfg(feature = "tracing")]
                     trace!("Got settings");
                     ()
                 }
 
                 Ok(Frame::Goaway(id)) => {

h3/src/client/stream.rs

@@ -92,7 +92,7 @@
     pub async fn recv_response(&mut self) -> Result<Response<()>, Error> {
         let mut frame = future::poll_fn(|cx| self.inner.stream.poll_next(cx))
             .await
             .map_err(|e| self.maybe_conn_err(e))?
             .ok_or_else(|| {
                 Code::H3_GENERAL_PROTOCOL_ERROR.with_reason(
                     "Did not receive response headers",
@@ -150,7 +150,7 @@
     // TODO what if called before recv_response ?
     #[cfg_attr(feature = "tracing", instrument(skip_all, level = "trace"))]
     pub async fn recv_data(&mut self) -> Result<Option<impl Buf>, Error> {
-        self.inner.recv_data().await
+        future::poll_fn(|cx| self.poll_recv_data(cx)).await
     }
 
     /// Receive request body