move datagram to separate crate

Cargo.toml

@@ -3,6 +3,7 @@ members = [
     "h3",
     "h3-quinn",
     "h3-webtransport",
+    "h3-datagram",
 
     # Internal
     "examples",

examples/Cargo.toml

@@ -39,6 +39,7 @@ tracing-subscriber = { version = "0.3", default-features = false, features = [
 octets = "0.3.0"
 
 tracing-tree = { version = "0.3" }
+h3-datagram = { path = "../h3-datagram" }
 
 [features]
 tree = []

examples/webtransport_server.rs

@@ -3,9 +3,10 @@ use bytes::{BufMut, Bytes, BytesMut};
 use h3::{
     error::ErrorLevel,
     ext::Protocol,
-    quic::{self, RecvDatagramExt, SendDatagramExt, SendStreamUnframed},
+    quic::{self, SendStreamUnframed},
     server::Connection,
 };
+use h3_datagram::quic_traits::{RecvDatagramExt, SendDatagramExt};
 use h3_quinn::quinn::{self, crypto::rustls::QuicServerConfig};
 use h3_webtransport::{
     server::{self, WebTransportSession},
@@ -291,6 +292,8 @@ where
     stream::SendStream<C::SendStream, Bytes>: AsyncWrite,
     C::BidiStream: SendStreamUnframed<Bytes>,
     C::SendStream: SendStreamUnframed<Bytes>,
+    <C as RecvDatagramExt>::Error: h3::quic::Error,
+    <C as SendDatagramExt<Bytes>>::Error: h3::quic::Error,
 {
     let session_id = session.session_id();
 

h3-datagram/Cargo.toml

@@ -0,0 +1,14 @@
+[package]
+name = "h3-datagram"
+version = "0.0.1"
+edition = "2021"
+
+[dependencies]
+#h3 = { path = "../h3" }
+bytes = "1.4"
+pin-project-lite = { version = "0.2", default-features = false }
+
+[dependencies.h3]
+version = "0.0.6"
+path = "../h3"
+features = ["i-implement-a-third-party-backend-and-opt-into-breaking-changes"]

h3-datagram/readme.md

@@ -0,0 +1,8 @@
+# H3 Datagram
+
+this crate provides an implementation of the [h3-datagram](https://datatracker.ietf.org/doc/html/rfc9297) spec that works with the h3 crate.
+
+## Usage
+As stated in the [rfc](https://datatracker.ietf.org/doc/html/rfc9297#abstract) this is intended to be used for protocol extensions like [Web-Transport](https://datatracker.ietf.org/doc/draft-ietf-webtrans-http3/) and not directly by applications.
+
+> HTTP Datagrams and the Capsule Protocol are intended for use by HTTP extensions, not applications.

h3-datagram/src/datagram.rs

@@ -0,0 +1,69 @@
+use bytes::{Buf, Bytes};
+use h3::{error::Code, proto::varint::VarInt, quic::StreamId, Error};
+
+/// HTTP datagram frames
+/// See: <https://www.rfc-editor.org/rfc/rfc9297#section-2.1>
+pub struct Datagram<B = Bytes> {
+    /// Stream id divided by 4
+    stream_id: StreamId,
+    /// The data contained in the datagram
+    payload: B,
+}
+
+impl<B> Datagram<B>
+where
+    B: Buf,
+{
+    /// Creates a new datagram frame
+    pub fn new(stream_id: StreamId, payload: B) -> Self {
+        assert!(
+            stream_id.into_inner() % 4 == 0,
+            "StreamId is not divisible by 4"
+        );
+        Self { stream_id, payload }
+    }
+
+    /// Decodes a datagram frame from the QUIC datagram
+    pub fn decode(mut buf: B) -> Result<Self, Error> {
+        let q_stream_id = VarInt::decode(&mut buf)
+            .map_err(|_| Code::H3_DATAGRAM_ERROR.with_cause("Malformed datagram frame"))?;
+
+        //= https://www.rfc-editor.org/rfc/rfc9297#section-2.1
+        // Quarter Stream ID: A variable-length integer that contains the value of the client-initiated bidirectional
+        // stream that this datagram is associated with divided by four (the division by four stems
+        // from the fact that HTTP requests are sent on client-initiated bidirectional streams,
+        // which have stream IDs that are divisible by four). The largest legal QUIC stream ID
+        // value is 262-1, so the largest legal value of the Quarter Stream ID field is 260-1.
+        // Receipt of an HTTP/3 Datagram that includes a larger value MUST be treated as an HTTP/3
+        // connection error of type H3_DATAGRAM_ERROR (0x33).
+        let stream_id = StreamId::try_from(u64::from(q_stream_id) * 4)
+            .map_err(|_| Code::H3_DATAGRAM_ERROR.with_cause("Invalid stream id"))?;
+
+        let payload = buf;
+
+        Ok(Self { stream_id, payload })
+    }
+
+    #[inline]
+    /// Returns the associated stream id of the datagram
+    pub fn stream_id(&self) -> StreamId {
+        self.stream_id
+    }
+
+    #[inline]
+    /// Returns the datagram payload
+    pub fn payload(&self) -> &B {
+        &self.payload
+    }
+
+    /// Encode the datagram to wire format
+    pub fn encode<D: bytes::BufMut>(self, buf: &mut D) {
+        (VarInt::from(self.stream_id) / 4).encode(buf);
+        buf.put(self.payload);
+    }
+
+    /// Returns the datagram payload
+    pub fn into_payload(self) -> B {
+        self.payload
+    }
+}

h3-datagram/src/datagram_traits.rs

@@ -0,0 +1,21 @@
+//! Traits which define the user API for datagrams.
+//! These traits are implemented for the client and server types in the `h3` crate.
+
+use bytes::Buf;
+use h3::{
+    quic::{self, StreamId},
+    Error,
+};
+
+use crate::server::ReadDatagram;
+
+pub trait HandleDatagramsExt<C, B>
+where
+    B: Buf,
+    C: quic::Connection<B>,
+{
+    /// Sends a datagram
+    fn send_datagram(&mut self, stream_id: StreamId, data: B) -> Result<(), Error>;
+    /// Reads an incoming datagram
+    fn read_datagram(&mut self) -> ReadDatagram<C, B>;
+}

h3-datagram/src/lib.rs

@@ -0,0 +1,4 @@
+pub mod datagram;
+pub mod datagram_traits;
+pub mod quic_traits;
+pub mod server;

h3-datagram/src/quic_traits.rs

@@ -0,0 +1,38 @@
+//! QUIC Transport traits
+//!
+//! This module includes traits and types meant to allow being generic over any
+//! QUIC implementation.
+
+use core::task;
+use std::{error::Error, task::Poll};
+
+use bytes::Buf;
+
+use crate::datagram::Datagram;
+
+/// Extends the `Connection` trait for sending datagrams
+///
+/// See: <https://www.rfc-editor.org/rfc/rfc9297>
+pub trait SendDatagramExt<B: Buf> {
+    /// The error type that can occur when sending a datagram
+    type Error: Into<Box<dyn Error>>;
+
+    /// Send a datagram
+    fn send_datagram(&mut self, data: Datagram<B>) -> Result<(), Self::Error>;
+}
+
+/// Extends the `Connection` trait for receiving datagrams
+///
+/// See: <https://www.rfc-editor.org/rfc/rfc9297>
+pub trait RecvDatagramExt {
+    /// The type of `Buf` for *raw* datagrams (without the stream_id decoded)
+    type Buf: Buf;
+    /// The error type that can occur when receiving a datagram
+    type Error: Into<Box<dyn Error>>;
+
+    /// Poll the connection for incoming datagrams.
+    fn poll_accept_datagram(
+        &mut self,
+        cx: &mut task::Context<'_>,
+    ) -> Poll<Result<Option<Self::Buf>, Self::Error>>;
+}

h3-datagram/src/server.rs

@@ -0,0 +1,73 @@
+//! server API
+
+use std::{
+    future::Future,
+    marker::PhantomData,
+    task::{ready, Context, Poll},
+};
+
+use bytes::Buf;
+use h3::{
+    quic::{self, StreamId},
+    server::Connection,
+    Error,
+};
+use pin_project_lite::pin_project;
+
+use crate::{
+    datagram::Datagram,
+    datagram_traits::HandleDatagramsExt,
+    quic_traits::{self, RecvDatagramExt, SendDatagramExt},
+};
+
+impl<B, C> HandleDatagramsExt<C, B> for Connection<C, B>
+where
+    B: Buf,
+    C: quic::Connection<B> + SendDatagramExt<B> + RecvDatagramExt,
+    <C as quic_traits::RecvDatagramExt>::Error: h3::quic::Error + 'static,
+    <C as quic_traits::SendDatagramExt<B>>::Error: h3::quic::Error + 'static,
+{
+    /// Sends a datagram
+    fn send_datagram(&mut self, stream_id: StreamId, data: B) -> Result<(), Error> {
+        self.inner
+            .conn
+            .send_datagram(Datagram::new(stream_id, data))?;
+        Ok(())
+    }
+
+    /// Reads an incoming datagram
+    fn read_datagram(&mut self) -> ReadDatagram<C, B> {
+        ReadDatagram {
+            conn: self,
+            _marker: PhantomData,
+        }
+    }
+}
+
+pin_project! {
+    /// Future for [`Connection::read_datagram`]
+    pub struct ReadDatagram<'a, C, B>
+    where
+            C: quic::Connection<B>,
+            B: Buf,
+        {
+            conn: &'a mut Connection<C, B>,
+            _marker: PhantomData<B>,
+        }
+}
+
+impl<'a, C, B> Future for ReadDatagram<'a, C, B>
+where
+    C: quic::Connection<B> + RecvDatagramExt,
+    <C as quic_traits::RecvDatagramExt>::Error: h3::quic::Error + 'static,
+    B: Buf,
+{
+    type Output = Result<Option<Datagram<C::Buf>>, Error>;
+
+    fn poll(mut self: std::pin::Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        match ready!(self.conn.inner.conn.poll_accept_datagram(cx))? {
+            Some(v) => Poll::Ready(Ok(Some(Datagram::decode(v)?))),
+            None => Poll::Ready(Ok(None)),
+        }
+    }
+}

h3-quinn/Cargo.toml

@@ -21,6 +21,7 @@ quinn = { version = "0.11", default-features = false, features = [
 tokio-util = { version = "0.7.9" }
 futures = { version = "0.3.28" }
 tokio = { version = "1", features = ["io-util"], default-features = false }
+h3-datagram = { path = "../h3-datagram" }
 tracing = { version = "0.1.40", optional = true }
 
 [features]

h3-quinn/src/lib.rs

@@ -19,13 +19,11 @@ use futures::{
     stream::{self},
     Stream, StreamExt,
 };
+use h3_datagram::{datagram::Datagram, quic_traits};
 pub use quinn::{self, AcceptBi, AcceptUni, Endpoint, OpenBi, OpenUni, VarInt, WriteError};
 use quinn::{ApplicationClose, ClosedStream, ReadDatagram};
 
-use h3::{
-    ext::Datagram,
-    quic::{self, Error, StreamId, WriteBuf},
-};
+use h3::quic::{self, Error, StreamId, WriteBuf};
 use tokio_util::sync::ReusableBoxFuture;
 
 #[cfg(feature = "tracing")]
@@ -248,7 +246,7 @@ where
     }
 }
 
-impl<B> quic::SendDatagramExt<B> for Connection
+impl<B> quic_traits::SendDatagramExt<B> for Connection
 where
     B: Buf,
 {
@@ -265,7 +263,7 @@ where
     }
 }
 
-impl quic::RecvDatagramExt for Connection {
+impl quic_traits::RecvDatagramExt for Connection {
     type Buf = Bytes;
 
     type Error = ConnectionError;

h3-webtransport/Cargo.toml

@@ -12,6 +12,7 @@ http = "1"
 pin-project-lite = { version = "0.2", default-features = false }
 tracing = "0.1.37"
 tokio = { version = "1.28", default-features = false }
+h3-datagram = { path = "../h3-datagram" }
 
 [dependencies.h3]
 version = "0.0.6"

h3-webtransport/src/server.rs

@@ -12,10 +12,10 @@ use futures_util::{future::poll_fn, ready, Future};
 use h3::{
     connection::ConnectionState,
     error::{Code, ErrorLevel},
-    ext::{Datagram, Protocol},
+    ext::Protocol,
     frame::FrameStream,
     proto::frame::Frame,
-    quic::{self, OpenStreams, RecvDatagramExt, SendDatagramExt, WriteBuf},
+    quic::{self, OpenStreams, WriteBuf},
     server::Connection,
     server::RequestStream,
     Error,
@@ -24,6 +24,11 @@ use h3::{
     quic::SendStreamUnframed,
     stream::{BidiStreamHeader, BufRecvStream, UniStreamHeader},
 };
+use h3_datagram::{
+    datagram::Datagram,
+    datagram_traits::HandleDatagramsExt,
+    quic_traits::{RecvDatagramExt, SendDatagramExt},
+};
 use http::{Method, Request, Response, StatusCode};
 
 use h3::webtransport::SessionId;
@@ -39,6 +44,7 @@ use crate::stream::{BidiStream, RecvStream, SendStream};
 pub struct WebTransportSession<C, B>
 where
     C: quic::Connection<B>,
+    Connection<C, B>: HandleDatagramsExt<C, B>,
     B: Buf,
 {
     // See: https://datatracker.ietf.org/doc/html/draft-ietf-webtrans-http3/#section-2-3
@@ -51,6 +57,7 @@ where
 
 impl<C, B> WebTransportSession<C, B>
 where
+    Connection<C, B>: HandleDatagramsExt<C, B>,
     C: quic::Connection<B>,
     B: Buf,
 {
@@ -374,6 +381,7 @@ impl<'a, C, B> Future for ReadDatagram<'a, C, B>
 where
     C: quic::Connection<B> + RecvDatagramExt,
     B: Buf,
+    <C as RecvDatagramExt>::Error: h3::quic::Error + 'static,
 {
     type Output = Result<Option<(SessionId, C::Buf)>, Error>;
 

h3/Cargo.toml

@@ -54,3 +54,4 @@ tracing-subscriber = { version = "0.3", default-features = false, features = [
 ] }
 futures = { version = "0.3.28" }
 tokio-util = { version = "0.7.9" }
+h3-datagram = {path = "../h3-datagram" }

h3/src/error.rs

@@ -200,7 +200,8 @@ impl Code {
         })
     }
 
-    pub(crate) fn with_cause<E: Into<Cause>>(self, cause: E) -> Error {
+    #[doc(hidden)]
+    pub fn with_cause<E: Into<Cause>>(self, cause: E) -> Error {
         Error::from(self).with_cause(cause)
     }