Repository: quic-go/quic-go
Pull request #5156 (2025-05-25 07:32:06)

http3: simplify HTTP datagram handling
~~Depends on #5154.~~


Patch:
diff --git a/http3/conn.go b/http3/conn.go
index 8adfc9c5..6594766b 100644
--- a/http3/conn.go
+++ b/http3/conn.go
@@ -20,6 +20,8 @@ import (
 	"github.com/quic-go/qpack"
 )
 
+const maxQuarterStreamID = 1<<60 - 1
+
 var errGoAway = errors.New("connection in graceful shutdown")
 
 // Connection is an HTTP/3 connection.
@@ -54,7 +56,7 @@ type connection struct {
 	decoder *qpack.Decoder
 
 	streamMx     sync.Mutex
-	streams      map[protocol.StreamID]*datagrammer
+	streams      map[protocol.StreamID]*stateTrackingStream
 	lastStreamID protocol.StreamID
 	maxStreamID  protocol.StreamID
 
@@ -82,7 +84,7 @@ func newConnection(
 		enableDatagrams:  enableDatagrams,
 		decoder:          qpack.NewDecoder(func(hf qpack.HeaderField) {}),
 		receivedSettings: make(chan struct{}),
-		streams:          make(map[protocol.StreamID]*datagrammer),
+		streams:          make(map[protocol.StreamID]*stateTrackingStream),
 		maxStreamID:      protocol.InvalidStreamID,
 		lastStreamID:     protocol.InvalidStreamID,
 	}
@@ -141,23 +143,30 @@ func (c *connection) openRequestStream(
 	if err != nil {
 		return nil, err
 	}
-	datagrams := newDatagrammer(func(b []byte) error { return c.sendDatagram(str.StreamID(), b) })
+	hstr := newStateTrackingStream(str, c, func(b []byte) error { return c.sendDatagram(str.StreamID(), b) })
 	c.streamMx.Lock()
-	c.streams[str.StreamID()] = datagrams
+	c.streams[str.StreamID()] = hstr
 	c.lastStreamID = str.StreamID()
 	c.streamMx.Unlock()
-	qstr := newStateTrackingStream(str, c, datagrams)
 	rsp := &http.Response{}
-	hstr := newStream(qstr, c, datagrams, func(r io.Reader, l uint64) error {
-		hdr, err := c.decodeTrailers(r, l, maxHeaderBytes)
-		if err != nil {
-			return err
-		}
-		rsp.Trailer = hdr
-		return nil
-	})
 	trace := httptrace.ContextClientTrace(ctx)
-	return newRequestStream(hstr, requestWriter, reqDone, c.decoder, disableCompression, maxHeaderBytes, rsp, trace), nil
+	return newRequestStream(
+		newStream(hstr, c, func(r io.Reader, l uint64) error {
+			hdr, err := c.decodeTrailers(r, l, maxHeaderBytes)
+			if err != nil {
+				return err
+			}
+			rsp.Trailer = hdr
+			return nil
+		}),
+		requestWriter,
+		reqDone,
+		c.decoder,
+		disableCompression,
+		maxHeaderBytes,
+		rsp,
+		trace,
+	), nil
 }
 
 func (c *connection) decodeTrailers(r io.Reader, l, maxHeaderBytes uint64) (http.Header, error) {
@@ -176,25 +185,23 @@ func (c *connection) decodeTrailers(r io.Reader, l, maxHeaderBytes uint64) (http
 	return parseTrailers(fields)
 }
 
-func (c *connection) acceptStream(ctx context.Context) (quic.Stream, *datagrammer, error) {
+// only used by the server
+func (c *connection) acceptStream(ctx context.Context) (*stateTrackingStream, error) {
 	str, err := c.AcceptStream(ctx)
 	if err != nil {
-		return nil, nil, err
+		return nil, err
 	}
-	datagrams := newDatagrammer(func(b []byte) error { return c.sendDatagram(str.StreamID(), b) })
-	if c.perspective == protocol.PerspectiveServer {
-		strID := str.StreamID()
-		c.streamMx.Lock()
-		c.streams[strID] = datagrams
-		if c.idleTimeout > 0 {
-			if len(c.streams) == 1 {
-				c.idleTimer.Stop()
-			}
+	strID := str.StreamID()
+	hstr := newStateTrackingStream(str, c, func(b []byte) error { return c.sendDatagram(strID, b) })
+	c.streamMx.Lock()
+	c.streams[strID] = hstr
+	if c.idleTimeout > 0 {
+		if len(c.streams) == 1 {
+			c.idleTimer.Stop()
 		}
-		c.streamMx.Unlock()
-		str = newStateTrackingStream(str, c, datagrams)
 	}
-	return str, datagrams, nil
+	c.streamMx.Unlock()
+	return hstr, nil
 }
 
 func (c *connection) CloseWithError(code quic.ApplicationErrorCode, msg string) error {
@@ -397,7 +404,7 @@ func (c *connection) receiveDatagrams() error {
 		if !ok {
 			continue
 		}
-		dg.enqueue(b[n:])
+		dg.enqueueDatagram(b[n:])
 	}
 }
 
diff --git a/http3/datagram.go b/http3/datagram.go
deleted file mode 100644
index 6d570e6b..00000000
--- a/http3/datagram.go
+++ /dev/null
@@ -1,98 +0,0 @@
-package http3
-
-import (
-	"context"
-	"sync"
-)
-
-const maxQuarterStreamID = 1<<60 - 1
-
-const streamDatagramQueueLen = 32
-
-type datagrammer struct {
-	sendDatagram func([]byte) error
-
-	hasData chan struct{}
-	queue   [][]byte // TODO: use a ring buffer
-
-	mx         sync.Mutex
-	sendErr    error
-	receiveErr error
-}
-
-func newDatagrammer(sendDatagram func([]byte) error) *datagrammer {
-	return &datagrammer{
-		sendDatagram: sendDatagram,
-		hasData:      make(chan struct{}, 1),
-	}
-}
-
-func (d *datagrammer) SetReceiveError(err error) {
-	d.mx.Lock()
-	defer d.mx.Unlock()
-
-	d.receiveErr = err
-	d.signalHasData()
-}
-
-func (d *datagrammer) SetSendError(err error) {
-	d.mx.Lock()
-	defer d.mx.Unlock()
-
-	d.sendErr = err
-}
-
-func (d *datagrammer) Send(b []byte) error {
-	d.mx.Lock()
-	sendErr := d.sendErr
-	d.mx.Unlock()
-	if sendErr != nil {
-		return sendErr
-	}
-
-	return d.sendDatagram(b)
-}
-
-func (d *datagrammer) signalHasData() {
-	select {
-	case d.hasData <- struct{}{}:
-	default:
-	}
-}
-
-func (d *datagrammer) enqueue(data []byte) {
-	d.mx.Lock()
-	defer d.mx.Unlock()
-
-	if d.receiveErr != nil {
-		return
-	}
-	if len(d.queue) >= streamDatagramQueueLen {
-		return
-	}
-	d.queue = append(d.queue, data)
-	d.signalHasData()
-}
-
-func (d *datagrammer) Receive(ctx context.Context) ([]byte, error) {
-start:
-	d.mx.Lock()
-	if len(d.queue) >= 1 {
-		data := d.queue[0]
-		d.queue = d.queue[1:]
-		d.mx.Unlock()
-		return data, nil
-	}
-	if receiveErr := d.receiveErr; receiveErr != nil {
-		d.mx.Unlock()
-		return nil, receiveErr
-	}
-	d.mx.Unlock()
-
-	select {
-	case <-ctx.Done():
-		return nil, context.Cause(ctx)
-	case <-d.hasData:
-	}
-	goto start
-}
diff --git a/http3/http_stream.go b/http3/http_stream.go
index 46fb1c47..f52f5e1f 100644
--- a/http3/http_stream.go
+++ b/http3/http_stream.go
@@ -14,35 +14,38 @@ import (
 	"github.com/quic-go/qpack"
 )
 
+type datagramStream interface {
+	quic.Stream
+	SendDatagram(b []byte) error
+	ReceiveDatagram(ctx context.Context) ([]byte, error)
+}
+
 // A Stream is an HTTP/3 request stream.
 // When writing to and reading from the stream, data is framed in HTTP/3 DATA frames.
 type Stream struct {
-	quic.Stream
+	datagramStream
 	conn *connection
 
 	buf []byte // used as a temporary buffer when writing the HTTP/3 frame headers
 
 	bytesRemainingInFrame uint64
 
-	datagrams *datagrammer
-
 	parseTrailer  func(io.Reader, uint64) error
 	parsedTrailer bool
 }
 
-func newStream(str quic.Stream, conn *connection, datagrams *datagrammer, parseTrailer func(io.Reader, uint64) error) *Stream {
+func newStream(str datagramStream, conn *connection, parseTrailer func(io.Reader, uint64) error) *Stream {
 	return &Stream{
-		Stream:       str,
-		conn:         conn,
-		buf:          make([]byte, 16),
-		datagrams:    datagrams,
-		parseTrailer: parseTrailer,
+		datagramStream: str,
+		conn:           conn,
+		buf:            make([]byte, 16),
+		parseTrailer:   parseTrailer,
 	}
 }
 
 func (s *Stream) Read(b []byte) (int, error) {
 	fp := &frameParser{
-		r:    s.Stream,
+		r:    s.datagramStream,
 		conn: s.conn,
 	}
 	if s.bytesRemainingInFrame == 0 {
@@ -67,7 +70,7 @@ func (s *Stream) Read(b []byte) (int, error) {
 					return 0, errors.New("additional HEADERS frame received after trailers")
 				}
 				s.parsedTrailer = true
-				return 0, s.parseTrailer(s.Stream, f.Length)
+				return 0, s.parseTrailer(s.datagramStream, f.Length)
 			default:
 				s.conn.CloseWithError(quic.ApplicationErrorCode(ErrCodeFrameUnexpected), "")
 				// parseNextFrame skips over unknown frame types
@@ -80,9 +83,9 @@ func (s *Stream) Read(b []byte) (int, error) {
 	var n int
 	var err error
 	if s.bytesRemainingInFrame < uint64(len(b)) {
-		n, err = s.Stream.Read(b[:s.bytesRemainingInFrame])
+		n, err = s.datagramStream.Read(b[:s.bytesRemainingInFrame])
 	} else {
-		n, err = s.Stream.Read(b)
+		n, err = s.datagramStream.Read(b)
 	}
 	s.bytesRemainingInFrame -= uint64(n)
 	return n, err
@@ -95,18 +98,18 @@ func (s *Stream) hasMoreData() bool {
 func (s *Stream) Write(b []byte) (int, error) {
 	s.buf = s.buf[:0]
 	s.buf = (&dataFrame{Length: uint64(len(b))}).Append(s.buf)
-	if _, err := s.Stream.Write(s.buf); err != nil {
+	if _, err := s.datagramStream.Write(s.buf); err != nil {
 		return 0, err
 	}
-	return s.Stream.Write(b)
+	return s.datagramStream.Write(b)
 }
 
 func (s *Stream) writeUnframed(b []byte) (int, error) {
-	return s.Stream.Write(b)
+	return s.datagramStream.Write(b)
 }
 
 func (s *Stream) StreamID() protocol.StreamID {
-	return s.Stream.StreamID()
+	return s.datagramStream.StreamID()
 }
 
 // A RequestStream is a low-level abstraction representing an HTTP/3 request stream.
@@ -174,7 +177,7 @@ func (s *RequestStream) SendRequestHeader(req *http.Request) error {
 	}
 	s.isConnect = req.Method == http.MethodConnect
 	s.sentRequest = true
-	return s.requestWriter.WriteRequestHeader(s.Stream.Stream, req, s.requestedGzip)
+	return s.requestWriter.WriteRequestHeader(s.datagramStream, req, s.requestedGzip)
 }
 
 // ReadResponse reads the HTTP response from the stream.
@@ -182,7 +185,7 @@ func (s *RequestStream) SendRequestHeader(req *http.Request) error {
 // It doesn't set Response.Request and Response.TLS.
 // It is invalid to call it after Read has been called.
 func (s *RequestStream) ReadResponse() (*http.Response, error) {
-	qstr := s.Stream.Stream
+	qstr := s.datagramStream
 	fp := &frameParser{
 		conn: s.conn,
 		r: &tracingReader{
@@ -252,12 +255,12 @@ func (s *RequestStream) ReadResponse() (*http.Response, error) {
 
 func (s *Stream) SendDatagram(b []byte) error {
 	// TODO: reject if datagrams are not negotiated (yet)
-	return s.datagrams.Send(b)
+	return s.datagramStream.SendDatagram(b)
 }
 
 func (s *Stream) ReceiveDatagram(ctx context.Context) ([]byte, error) {
 	// TODO: reject if datagrams are not negotiated (yet)
-	return s.datagrams.Receive(ctx)
+	return s.datagramStream.ReceiveDatagram(ctx)
 }
 
 type tracingReader struct {
diff --git a/http3/mockgen.go b/http3/mockgen.go
index 1d790e11..1d33f54e 100644
--- a/http3/mockgen.go
+++ b/http3/mockgen.go
@@ -5,4 +5,7 @@ package http3
 //go:generate sh -c "go run go.uber.org/mock/mockgen -typed -build_flags=\"-tags=gomock\" -mock_names=TestClientConnInterface=MockClientConn  -package http3 -destination mock_clientconn_test.go github.com/quic-go/quic-go/http3 TestClientConnInterface"
 type TestClientConnInterface = clientConn
 
+//go:generate sh -c "go run go.uber.org/mock/mockgen -typed -build_flags=\"-tags=gomock\" -mock_names=DatagramStream=MockDatagramStream  -package http3 -destination mock_datagram_stream_test.go github.com/quic-go/quic-go/http3 DatagramStream"
+type DatagramStream = datagramStream
+
 //go:generate sh -c "go run go.uber.org/mock/mockgen -typed -package http3 -destination mock_quic_early_listener_test.go github.com/quic-go/quic-go/http3 QUICEarlyListener"
diff --git a/http3/server.go b/http3/server.go
index 5e405814..c5a0fda9 100644
--- a/http3/server.go
+++ b/http3/server.go
@@ -486,7 +486,7 @@ func (s *Server) handleConn(conn quic.Connection) error {
 		// * before graceful shutdown: s.graceCtx
 		// * after graceful shutdown: s.closeCtx
 		// This allows us to keep accepting (and resetting) streams after graceful shutdown has started.
-		str, datagrams, err := hconn.acceptStream(ctx)
+		str, err := hconn.acceptStream(ctx)
 		if err != nil {
 			// the underlying connection was closed (by either side)
 			if hconn.Context().Err() != nil {
@@ -536,7 +536,7 @@ func (s *Server) handleConn(conn quic.Connection) error {
 			// handleRequest will return once the request has been handled,
 			// or the underlying connection is closed
 			defer wg.Done()
-			s.handleRequest(hconn, str, datagrams, hconn.decoder)
+			s.handleRequest(hconn, str, hconn.decoder)
 		}()
 	}
 	wg.Wait()
@@ -550,7 +550,7 @@ func (s *Server) maxHeaderBytes() uint64 {
 	return uint64(s.MaxHeaderBytes)
 }
 
-func (s *Server) handleRequest(conn *connection, str quic.Stream, datagrams *datagrammer, decoder *qpack.Decoder) {
+func (s *Server) handleRequest(conn *connection, str datagramStream, decoder *qpack.Decoder) {
 	var ufh unknownFrameHandlerFunc
 	if s.StreamHijacker != nil {
 		ufh = func(ft FrameType, e error) (processed bool, err error) {
@@ -610,7 +610,7 @@ func (s *Server) handleRequest(conn *connection, str quic.Stream, datagrams *dat
 	if _, ok := req.Header["Content-Length"]; ok && req.ContentLength >= 0 {
 		contentLength = req.ContentLength
 	}
-	hstr := newStream(str, conn, datagrams, nil)
+	hstr := newStream(str, conn, nil)
 	body := newRequestBody(hstr, contentLength, conn.Context(), conn.ReceivedSettings(), conn.Settings)
 	req.Body = body
 
diff --git a/http3/state_tracking_stream.go b/http3/state_tracking_stream.go
index 6edbeedf..a5888099 100644
--- a/http3/state_tracking_stream.go
+++ b/http3/state_tracking_stream.go
@@ -9,7 +9,7 @@ import (
 	"github.com/quic-go/quic-go"
 )
 
-var _ quic.Stream = &stateTrackingStream{}
+const streamDatagramQueueLen = 32
 
 // stateTrackingStream is an implementation of quic.Stream that delegates
 // to an underlying stream
@@ -21,28 +21,32 @@ var _ quic.Stream = &stateTrackingStream{}
 type stateTrackingStream struct {
 	quic.Stream
 
+	sendDatagram func([]byte) error
+	hasData      chan struct{}
+	queue        [][]byte // TODO: use a ring buffer
+
 	mx      sync.Mutex
 	sendErr error
 	recvErr error
 
 	clearer streamClearer
-	setter  errorSetter
 }
 
+var (
+	_ datagramStream = &stateTrackingStream{}
+	_ quic.Stream    = &stateTrackingStream{}
+)
+
 type streamClearer interface {
 	clearStream(quic.StreamID)
 }
 
-type errorSetter interface {
-	SetSendError(error)
-	SetReceiveError(error)
-}
-
-func newStateTrackingStream(s quic.Stream, clearer streamClearer, setter errorSetter) *stateTrackingStream {
+func newStateTrackingStream(s quic.Stream, clearer streamClearer, sendDatagram func([]byte) error) *stateTrackingStream {
 	t := &stateTrackingStream{
-		Stream:  s,
-		clearer: clearer,
-		setter:  setter,
+		Stream:       s,
+		clearer:      clearer,
+		sendDatagram: sendDatagram,
+		hasData:      make(chan struct{}, 1),
 	}
 
 	context.AfterFunc(s.Context(), func() {
@@ -62,8 +66,6 @@ func (s *stateTrackingStream) closeSend(e error) {
 		if s.recvErr != nil {
 			s.clearer.clearStream(s.StreamID())
 		}
-
-		s.setter.SetSendError(e)
 		s.sendErr = e
 	}
 }
@@ -78,9 +80,8 @@ func (s *stateTrackingStream) closeReceive(e error) {
 		if s.sendErr != nil {
 			s.clearer.clearStream(s.StreamID())
 		}
-
-		s.setter.SetReceiveError(e)
 		s.recvErr = e
+		s.signalHasDatagram()
 	}
 }
 
@@ -114,3 +115,58 @@ func (s *stateTrackingStream) Read(b []byte) (int, error) {
 	}
 	return n, err
 }
+
+func (s *stateTrackingStream) SendDatagram(b []byte) error {
+	s.mx.Lock()
+	sendErr := s.sendErr
+	s.mx.Unlock()
+	if sendErr != nil {
+		return sendErr
+	}
+
+	return s.sendDatagram(b)
+}
+
+func (s *stateTrackingStream) signalHasDatagram() {
+	select {
+	case s.hasData <- struct{}{}:
+	default:
+	}
+}
+
+func (s *stateTrackingStream) enqueueDatagram(data []byte) {
+	s.mx.Lock()
+	defer s.mx.Unlock()
+
+	if s.recvErr != nil {
+		return
+	}
+	if len(s.queue) >= streamDatagramQueueLen {
+		return
+	}
+	s.queue = append(s.queue, data)
+	s.signalHasDatagram()
+}
+
+func (s *stateTrackingStream) ReceiveDatagram(ctx context.Context) ([]byte, error) {
+start:
+	s.mx.Lock()
+	if len(s.queue) > 0 {
+		data := s.queue[0]
+		s.queue = s.queue[1:]
+		s.mx.Unlock()
+		return data, nil
+	}
+	if receiveErr := s.recvErr; receiveErr != nil {
+		s.mx.Unlock()
+		return nil, receiveErr
+	}
+	s.mx.Unlock()
+
+	select {
+	case <-ctx.Done():
+		return nil, context.Cause(ctx)
+	case <-s.hasData:
+	}
+	goto start
+}


Test patch:
diff --git a/http3/body_test.go b/http3/body_test.go
index c96e34a1..b3aa9476 100644
--- a/http3/body_test.go
+++ b/http3/body_test.go
@@ -7,7 +7,6 @@ import (
 	"time"
 
 	"github.com/quic-go/quic-go"
-	mockquic "github.com/quic-go/quic-go/internal/mocks/quic"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -18,10 +17,10 @@ func TestResponseBodyReading(t *testing.T) {
 	mockCtrl := gomock.NewController(t)
 	var buf bytes.Buffer
 	buf.Write(getDataFrame([]byte("foobar")))
-	str := mockquic.NewMockStream(mockCtrl)
+	str := NewMockDatagramStream(mockCtrl)
 	str.EXPECT().Read(gomock.Any()).DoAndReturn(buf.Read).AnyTimes()
 	reqDone := make(chan struct{})
-	rb := newResponseBody(&Stream{Stream: str}, -1, reqDone)
+	rb := newResponseBody(&Stream{datagramStream: str}, -1, reqDone)
 
 	data, err := io.ReadAll(rb)
 	require.NoError(t, err)
@@ -30,10 +29,10 @@ func TestResponseBodyReading(t *testing.T) {
 
 func TestResponseBodyReadError(t *testing.T) {
 	mockCtrl := gomock.NewController(t)
-	str := mockquic.NewMockStream(mockCtrl)
+	str := NewMockDatagramStream(mockCtrl)
 	str.EXPECT().Read(gomock.Any()).Return(0, assert.AnError).Times(2)
 	reqDone := make(chan struct{})
-	rb := newResponseBody(&Stream{Stream: str}, -1, reqDone)
+	rb := newResponseBody(&Stream{datagramStream: str}, -1, reqDone)
 
 	_, err := rb.Read([]byte{0})
 	require.ErrorIs(t, err, assert.AnError)
@@ -49,10 +48,10 @@ func TestResponseBodyReadError(t *testing.T) {
 
 func TestResponseBodyClose(t *testing.T) {
 	mockCtrl := gomock.NewController(t)
-	str := mockquic.NewMockStream(mockCtrl)
+	str := NewMockDatagramStream(mockCtrl)
 	str.EXPECT().CancelRead(quic.StreamErrorCode(ErrCodeRequestCanceled)).Times(2)
 	reqDone := make(chan struct{})
-	rb := newResponseBody(&Stream{Stream: str}, -1, reqDone)
+	rb := newResponseBody(&Stream{datagramStream: str}, -1, reqDone)
 	require.NoError(t, rb.Close())
 	select {
 	case <-reqDone:
@@ -66,10 +65,10 @@ func TestResponseBodyClose(t *testing.T) {
 
 func TestResponseBodyConcurrentClose(t *testing.T) {
 	mockCtrl := gomock.NewController(t)
-	str := mockquic.NewMockStream(mockCtrl)
+	str := NewMockDatagramStream(mockCtrl)
 	str.EXPECT().CancelRead(quic.StreamErrorCode(ErrCodeRequestCanceled)).MaxTimes(3)
 	reqDone := make(chan struct{})
-	rb := newResponseBody(&Stream{Stream: str}, -1, reqDone)
+	rb := newResponseBody(&Stream{datagramStream: str}, -1, reqDone)
 
 	for range 3 {
 		go rb.Close()
@@ -101,11 +100,11 @@ func testResponseBodyLengthLimiting(t *testing.T, alongFrameBoundary bool) {
 		l = 3
 	}
 	mockCtrl := gomock.NewController(t)
-	str := mockquic.NewMockStream(mockCtrl)
+	str := NewMockDatagramStream(mockCtrl)
 	str.EXPECT().CancelRead(quic.StreamErrorCode(ErrCodeMessageError))
 	str.EXPECT().CancelWrite(quic.StreamErrorCode(ErrCodeMessageError))
 	str.EXPECT().Read(gomock.Any()).DoAndReturn(buf.Read).AnyTimes()
-	rb := newResponseBody(&Stream{Stream: str}, l, make(chan struct{}))
+	rb := newResponseBody(&Stream{datagramStream: str}, l, make(chan struct{}))
 	data, err := io.ReadAll(rb)
 	require.Equal(t, []byte("foobar")[:l], data)
 	require.ErrorIs(t, err, errTooMuchData)
diff --git a/http3/client_test.go b/http3/client_test.go
index 81247b89..d79ce3ae 100644
--- a/http3/client_test.go
+++ b/http3/client_test.go
@@ -81,9 +81,9 @@ func closedChan() <-chan struct{} {
 func encodeResponse(t *testing.T, status int) []byte {
 	mockCtrl := gomock.NewController(t)
 	buf := &bytes.Buffer{}
-	rstr := mockquic.NewMockStream(mockCtrl)
+	rstr := NewMockDatagramStream(mockCtrl)
 	rstr.EXPECT().Write(gomock.Any()).Do(buf.Write).AnyTimes()
-	rw := newResponseWriter(newStream(rstr, nil, nil, func(r io.Reader, u uint64) error { return nil }), nil, false, nil)
+	rw := newResponseWriter(newStream(rstr, nil, func(r io.Reader, u uint64) error { return nil }), nil, false, nil)
 	rw.WriteHeader(status)
 	rw.Flush()
 	return buf.Bytes()
@@ -380,9 +380,9 @@ func testClient1xxHandling(t *testing.T, numEarlyHints int, terminalStatus int,
 	mockCtrl := gomock.NewController(t)
 
 	var rspBuf bytes.Buffer
-	rstr := mockquic.NewMockStream(mockCtrl)
+	rstr := NewMockDatagramStream(mockCtrl)
 	rstr.EXPECT().Write(gomock.Any()).Do(rspBuf.Write).AnyTimes()
-	rw := newResponseWriter(newStream(rstr, nil, nil, func(r io.Reader, u uint64) error { return nil }), nil, false, nil)
+	rw := newResponseWriter(newStream(rstr, nil, func(r io.Reader, u uint64) error { return nil }), nil, false, nil)
 	rw.header.Add("Link", "foo")
 	rw.header.Add("Link", "bar")
 	for range numEarlyHints {
@@ -463,9 +463,9 @@ func testClientGzip(t *testing.T,
 	mockCtrl := gomock.NewController(t)
 
 	var rspBuf bytes.Buffer
-	rstr := mockquic.NewMockStream(mockCtrl)
+	rstr := NewMockDatagramStream(mockCtrl)
 	rstr.EXPECT().Write(gomock.Any()).Do(rspBuf.Write).AnyTimes()
-	rw := newResponseWriter(newStream(rstr, nil, nil, func(r io.Reader, u uint64) error { return nil }), nil, false, nil)
+	rw := newResponseWriter(newStream(rstr, nil, func(r io.Reader, u uint64) error { return nil }), nil, false, nil)
 	rw.WriteHeader(http.StatusOK)
 	if responseAddContentEncoding {
 		rw.header.Add("Content-Encoding", "gzip")
diff --git a/http3/datagram_test.go b/http3/datagram_test.go
deleted file mode 100644
index 1a79599a..00000000
--- a/http3/datagram_test.go
+++ /dev/null
@@ -1,106 +0,0 @@
-package http3
-
-import (
-	"context"
-	"net"
-	"testing"
-	"time"
-
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-)
-
-func TestDatagramReceiving(t *testing.T) {
-	dg := newDatagrammer(nil)
-
-	type result struct {
-		data []byte
-		err  error
-	}
-
-	// Receive blocks until a datagram is received
-	resultChan := make(chan result)
-	go func() {
-		defer close(resultChan)
-		data, err := dg.Receive(context.Background())
-		resultChan <- result{data: data, err: err}
-	}()
-
-	select {
-	case <-time.After(scaleDuration(10 * time.Millisecond)):
-	case <-resultChan:
-		t.Fatal("should not have received a datagram")
-	}
-	dg.enqueue([]byte("foobar"))
-
-	select {
-	case res := <-resultChan:
-		require.NoError(t, res.err)
-		require.Equal(t, []byte("foobar"), res.data)
-	case <-time.After(time.Second):
-		t.Fatal("should have received a datagram")
-	}
-
-	// up to 32 datagrams can be queued
-	for i := range streamDatagramQueueLen + 1 {
-		dg.enqueue([]byte{uint8(i)})
-	}
-	for i := range streamDatagramQueueLen {
-		data, err := dg.Receive(context.Background())
-		require.NoError(t, err)
-		require.Equal(t, []byte{uint8(i)}, data)
-	}
-
-	// Receive respects the context
-	ctx, cancel := context.WithCancel(context.Background())
-	cancel()
-	_, err := dg.Receive(ctx)
-	require.ErrorIs(t, err, context.Canceled)
-}
-
-func TestDatagramReceiveError(t *testing.T) {
-	dg := newDatagrammer(nil)
-
-	errChan := make(chan error)
-	go func() {
-		_, err := dg.Receive(context.Background())
-		errChan <- err
-	}()
-
-	select {
-	case <-time.After(scaleDuration(10 * time.Millisecond)):
-	case err := <-errChan:
-		t.Fatalf("should not have received an error: %v", err)
-	}
-
-	dg.SetReceiveError(assert.AnError)
-	select {
-	case err := <-errChan:
-		require.ErrorIs(t, err, assert.AnError)
-	case <-time.After(time.Second):
-		t.Fatal("timeout")
-	}
-
-	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
-	defer cancel()
-	_, err := dg.Receive(ctx)
-	require.ErrorIs(t, err, assert.AnError)
-}
-
-func TestDatagramSending(t *testing.T) {
-	var sendQueue [][]byte
-	errors := []error{nil, nil, assert.AnError}
-	dg := newDatagrammer(func(b []byte) error {
-		sendQueue = append(sendQueue, b)
-		err := errors[0]
-		errors = errors[1:]
-		return err
-	})
-	require.NoError(t, dg.Send([]byte("foo")))
-	require.NoError(t, dg.Send([]byte("bar")))
-	require.ErrorIs(t, dg.Send([]byte("baz")), assert.AnError)
-	require.Equal(t, [][]byte{[]byte("foo"), []byte("bar"), []byte("baz")}, sendQueue)
-
-	dg.SetSendError(net.ErrClosed)
-	require.ErrorIs(t, dg.Send([]byte("foobar")), net.ErrClosed)
-}
diff --git a/http3/http_stream_test.go b/http3/http_stream_test.go
index 39f8a175..1694289c 100644
--- a/http3/http_stream_test.go
+++ b/http3/http_stream_test.go
@@ -28,7 +28,7 @@ func getDataFrame(data []byte) []byte {
 func TestStreamReadDataFrames(t *testing.T) {
 	var buf bytes.Buffer
 	mockCtrl := gomock.NewController(t)
-	qstr := mockquic.NewMockStream(mockCtrl)
+	qstr := NewMockDatagramStream(mockCtrl)
 	qstr.EXPECT().Write(gomock.Any()).DoAndReturn(buf.Write).AnyTimes()
 	qstr.EXPECT().Read(gomock.Any()).DoAndReturn(buf.Read).AnyTimes()
 
@@ -42,7 +42,6 @@ func TestStreamReadDataFrames(t *testing.T) {
 			nil,
 			0,
 		),
-		nil,
 		func(r io.Reader, u uint64) error { return nil },
 	)
 
@@ -80,7 +79,7 @@ func TestStreamReadDataFrames(t *testing.T) {
 func TestStreamInvalidFrame(t *testing.T) {
 	var buf bytes.Buffer
 	mockCtrl := gomock.NewController(t)
-	qstr := mockquic.NewMockStream(mockCtrl)
+	qstr := NewMockDatagramStream(mockCtrl)
 	qstr.EXPECT().Write(gomock.Any()).DoAndReturn(buf.Write).AnyTimes()
 	qstr.EXPECT().Read(gomock.Any()).DoAndReturn(buf.Read).AnyTimes()
 	conn := mockquic.NewMockEarlyConnection(mockCtrl)
@@ -95,7 +94,6 @@ func TestStreamInvalidFrame(t *testing.T) {
 	str := newStream(
 		qstr,
 		newConnection(context.Background(), conn, false, protocol.PerspectiveClient, nil, 0),
-		nil,
 		func(r io.Reader, u uint64) error { return nil },
 	)
 
@@ -109,9 +107,9 @@ func TestStreamInvalidFrame(t *testing.T) {
 func TestStreamWrite(t *testing.T) {
 	var buf bytes.Buffer
 	mockCtrl := gomock.NewController(t)
-	qstr := mockquic.NewMockStream(mockCtrl)
+	qstr := NewMockDatagramStream(mockCtrl)
 	qstr.EXPECT().Write(gomock.Any()).DoAndReturn(buf.Write).AnyTimes()
-	str := newStream(qstr, nil, nil, func(r io.Reader, u uint64) error { return nil })
+	str := newStream(qstr, nil, func(r io.Reader, u uint64) error { return nil })
 	str.Write([]byte("foo"))
 	str.Write([]byte("foobar"))
 
@@ -136,11 +134,15 @@ func TestStreamWrite(t *testing.T) {
 
 func TestRequestStream(t *testing.T) {
 	mockCtrl := gomock.NewController(t)
-	qstr := mockquic.NewMockStream(mockCtrl)
+	qstr := NewMockDatagramStream(mockCtrl)
 	requestWriter := newRequestWriter()
 	conn := mockquic.NewMockEarlyConnection(mockCtrl)
 	str := newRequestStream(
-		newStream(qstr, newConnection(context.Background(), conn, false, protocol.PerspectiveClient, nil, 0), nil, func(r io.Reader, u uint64) error { return nil }),
+		newStream(
+			qstr,
+			newConnection(context.Background(), conn, false, protocol.PerspectiveClient, nil, 0),
+			func(r io.Reader, u uint64) error { return nil },
+		),
 		requestWriter,
 		make(chan struct{}),
 		qpack.NewDecoder(func(qpack.HeaderField) {}),
diff --git a/http3/mock_datagram_stream_test.go b/http3/mock_datagram_stream_test.go
new file mode 100644
index 00000000..faaab993
--- /dev/null
+++ b/http3/mock_datagram_stream_test.go
@@ -0,0 +1,498 @@
+// Code generated by MockGen. DO NOT EDIT.
+// Source: github.com/quic-go/quic-go/http3 (interfaces: DatagramStream)
+//
+// Generated by this command:
+//
+//	mockgen -typed -build_flags=-tags=gomock -mock_names=DatagramStream=MockDatagramStream -package http3 -destination mock_datagram_stream_test.go github.com/quic-go/quic-go/http3 DatagramStream
+//
+
+// Package http3 is a generated GoMock package.
+package http3
+
+import (
+	context "context"
+	reflect "reflect"
+	time "time"
+
+	quic "github.com/quic-go/quic-go"
+	gomock "go.uber.org/mock/gomock"
+)
+
+// MockDatagramStream is a mock of DatagramStream interface.
+type MockDatagramStream struct {
+	ctrl     *gomock.Controller
+	recorder *MockDatagramStreamMockRecorder
+	isgomock struct{}
+}
+
+// MockDatagramStreamMockRecorder is the mock recorder for MockDatagramStream.
+type MockDatagramStreamMockRecorder struct {
+	mock *MockDatagramStream
+}
+
+// NewMockDatagramStream creates a new mock instance.
+func NewMockDatagramStream(ctrl *gomock.Controller) *MockDatagramStream {
+	mock := &MockDatagramStream{ctrl: ctrl}
+	mock.recorder = &MockDatagramStreamMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockDatagramStream) EXPECT() *MockDatagramStreamMockRecorder {
+	return m.recorder
+}
+
+// CancelRead mocks base method.
+func (m *MockDatagramStream) CancelRead(arg0 quic.StreamErrorCode) {
+	m.ctrl.T.Helper()
+	m.ctrl.Call(m, "CancelRead", arg0)
+}
+
+// CancelRead indicates an expected call of CancelRead.
+func (mr *MockDatagramStreamMockRecorder) CancelRead(arg0 any) *MockDatagramStreamCancelReadCall {
+	mr.mock.ctrl.T.Helper()
+	call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CancelRead", reflect.TypeOf((*MockDatagramStream)(nil).CancelRead), arg0)
+	return &MockDatagramStreamCancelReadCall{Call: call}
+}
+
+// MockDatagramStreamCancelReadCall wrap *gomock.Call
+type MockDatagramStreamCancelReadCall struct {
+	*gomock.Call
+}
+
+// Return rewrite *gomock.Call.Return
+func (c *MockDatagramStreamCancelReadCall) Return() *MockDatagramStreamCancelReadCall {
+	c.Call = c.Call.Return()
+	return c
+}
+
+// Do rewrite *gomock.Call.Do
+func (c *MockDatagramStreamCancelReadCall) Do(f func(quic.StreamErrorCode)) *MockDatagramStreamCancelReadCall {
+	c.Call = c.Call.Do(f)
+	return c
+}
+
+// DoAndReturn rewrite *gomock.Call.DoAndReturn
+func (c *MockDatagramStreamCancelReadCall) DoAndReturn(f func(quic.StreamErrorCode)) *MockDatagramStreamCancelReadCall {
+	c.Call = c.Call.DoAndReturn(f)
+	return c
+}
+
+// CancelWrite mocks base method.
+func (m *MockDatagramStream) CancelWrite(arg0 quic.StreamErrorCode) {
+	m.ctrl.T.Helper()
+	m.ctrl.Call(m, "CancelWrite", arg0)
+}
+
+// CancelWrite indicates an expected call of CancelWrite.
+func (mr *MockDatagramStreamMockRecorder) CancelWrite(arg0 any) *MockDatagramStreamCancelWriteCall {
+	mr.mock.ctrl.T.Helper()
+	call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CancelWrite", reflect.TypeOf((*MockDatagramStream)(nil).CancelWrite), arg0)
+	return &MockDatagramStreamCancelWriteCall{Call: call}
+}
+
+// MockDatagramStreamCancelWriteCall wrap *gomock.Call
+type MockDatagramStreamCancelWriteCall struct {
+	*gomock.Call
+}
+
+// Return rewrite *gomock.Call.Return
+func (c *MockDatagramStreamCancelWriteCall) Return() *MockDatagramStreamCancelWriteCall {
+	c.Call = c.Call.Return()
+	return c
+}
+
+// Do rewrite *gomock.Call.Do
+func (c *MockDatagramStreamCancelWriteCall) Do(f func(quic.StreamErrorCode)) *MockDatagramStreamCancelWriteCall {
+	c.Call = c.Call.Do(f)
+	return c
+}
+
+// DoAndReturn rewrite *gomock.Call.DoAndReturn
+func (c *MockDatagramStreamCancelWriteCall) DoAndReturn(f func(quic.StreamErrorCode)) *MockDatagramStreamCancelWriteCall {
+	c.Call = c.Call.DoAndReturn(f)
+	return c
+}
+
+// Close mocks base method.
+func (m *MockDatagramStream) Close() error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Close")
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Close indicates an expected call of Close.
+func (mr *MockDatagramStreamMockRecorder) Close() *MockDatagramStreamCloseCall {
+	mr.mock.ctrl.T.Helper()
+	call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Close", reflect.TypeOf((*MockDatagramStream)(nil).Close))
+	return &MockDatagramStreamCloseCall{Call: call}
+}
+
+// MockDatagramStreamCloseCall wrap *gomock.Call
+type MockDatagramStreamCloseCall struct {
+	*gomock.Call
+}
+
+// Return rewrite *gomock.Call.Return
+func (c *MockDatagramStreamCloseCall) Return(arg0 error) *MockDatagramStreamCloseCall {
+	c.Call = c.Call.Return(arg0)
+	return c
+}
+
+// Do rewrite *gomock.Call.Do
+func (c *MockDatagramStreamCloseCall) Do(f func() error) *MockDatagramStreamCloseCall {
+	c.Call = c.Call.Do(f)
+	return c
+}
+
+// DoAndReturn rewrite *gomock.Call.DoAndReturn
+func (c *MockDatagramStreamCloseCall) DoAndReturn(f func() error) *MockDatagramStreamCloseCall {
+	c.Call = c.Call.DoAndReturn(f)
+	return c
+}
+
+// Context mocks base method.
+func (m *MockDatagramStream) Context() context.Context {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Context")
+	ret0, _ := ret[0].(context.Context)
+	return ret0
+}
+
+// Context indicates an expected call of Context.
+func (mr *MockDatagramStreamMockRecorder) Context() *MockDatagramStreamContextCall {
+	mr.mock.ctrl.T.Helper()
+	call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Context", reflect.TypeOf((*MockDatagramStream)(nil).Context))
+	return &MockDatagramStreamContextCall{Call: call}
+}
+
+// MockDatagramStreamContextCall wrap *gomock.Call
+type MockDatagramStreamContextCall struct {
+	*gomock.Call
+}
+
+// Return rewrite *gomock.Call.Return
+func (c *MockDatagramStreamContextCall) Return(arg0 context.Context) *MockDatagramStreamContextCall {
+	c.Call = c.Call.Return(arg0)
+	return c
+}
+
+// Do rewrite *gomock.Call.Do
+func (c *MockDatagramStreamContextCall) Do(f func() context.Context) *MockDatagramStreamContextCall {
+	c.Call = c.Call.Do(f)
+	return c
+}
+
+// DoAndReturn rewrite *gomock.Call.DoAndReturn
+func (c *MockDatagramStreamContextCall) DoAndReturn(f func() context.Context) *MockDatagramStreamContextCall {
+	c.Call = c.Call.DoAndReturn(f)
+	return c
+}
+
+// Read mocks base method.
+func (m *MockDatagramStream) Read(p []byte) (int, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Read", p)
+	ret0, _ := ret[0].(int)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// Read indicates an expected call of Read.
+func (mr *MockDatagramStreamMockRecorder) Read(p any) *MockDatagramStreamReadCall {
+	mr.mock.ctrl.T.Helper()
+	call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Read", reflect.TypeOf((*MockDatagramStream)(nil).Read), p)
+	return &MockDatagramStreamReadCall{Call: call}
+}
+
+// MockDatagramStreamReadCall wrap *gomock.Call
+type MockDatagramStreamReadCall struct {
+	*gomock.Call
+}
+
+// Return rewrite *gomock.Call.Return
+func (c *MockDatagramStreamReadCall) Return(n int, err error) *MockDatagramStreamReadCall {
+	c.Call = c.Call.Return(n, err)
+	return c
+}
+
+// Do rewrite *gomock.Call.Do
+func (c *MockDatagramStreamReadCall) Do(f func([]byte) (int, error)) *MockDatagramStreamReadCall {
+	c.Call = c.Call.Do(f)
+	return c
+}
+
+// DoAndReturn rewrite *gomock.Call.DoAndReturn
+func (c *MockDatagramStreamReadCall) DoAndReturn(f func([]byte) (int, error)) *MockDatagramStreamReadCall {
+	c.Call = c.Call.DoAndReturn(f)
+	return c
+}
+
+// ReceiveDatagram mocks base method.
+func (m *MockDatagramStream) ReceiveDatagram(ctx context.Context) ([]byte, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "ReceiveDatagram", ctx)
+	ret0, _ := ret[0].([]byte)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// ReceiveDatagram indicates an expected call of ReceiveDatagram.
+func (mr *MockDatagramStreamMockRecorder) ReceiveDatagram(ctx any) *MockDatagramStreamReceiveDatagramCall {
+	mr.mock.ctrl.T.Helper()
+	call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ReceiveDatagram", reflect.TypeOf((*MockDatagramStream)(nil).ReceiveDatagram), ctx)
+	return &MockDatagramStreamReceiveDatagramCall{Call: call}
+}
+
+// MockDatagramStreamReceiveDatagramCall wrap *gomock.Call
+type MockDatagramStreamReceiveDatagramCall struct {
+	*gomock.Call
+}
+
+// Return rewrite *gomock.Call.Return
+func (c *MockDatagramStreamReceiveDatagramCall) Return(arg0 []byte, arg1 error) *MockDatagramStreamReceiveDatagramCall {
+	c.Call = c.Call.Return(arg0, arg1)
+	return c
+}
+
+// Do rewrite *gomock.Call.Do
+func (c *MockDatagramStreamReceiveDatagramCall) Do(f func(context.Context) ([]byte, error)) *MockDatagramStreamReceiveDatagramCall {
+	c.Call = c.Call.Do(f)
+	return c
+}
+
+// DoAndReturn rewrite *gomock.Call.DoAndReturn
+func (c *MockDatagramStreamReceiveDatagramCall) DoAndReturn(f func(context.Context) ([]byte, error)) *MockDatagramStreamReceiveDatagramCall {
+	c.Call = c.Call.DoAndReturn(f)
+	return c
+}
+
+// SendDatagram mocks base method.
+func (m *MockDatagramStream) SendDatagram(b []byte) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "SendDatagram", b)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// SendDatagram indicates an expected call of SendDatagram.
+func (mr *MockDatagramStreamMockRecorder) SendDatagram(b any) *MockDatagramStreamSendDatagramCall {
+	mr.mock.ctrl.T.Helper()
+	call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SendDatagram", reflect.TypeOf((*MockDatagramStream)(nil).SendDatagram), b)
+	return &MockDatagramStreamSendDatagramCall{Call: call}
+}
+
+// MockDatagramStreamSendDatagramCall wrap *gomock.Call
+type MockDatagramStreamSendDatagramCall struct {
+	*gomock.Call
+}
+
+// Return rewrite *gomock.Call.Return
+func (c *MockDatagramStreamSendDatagramCall) Return(arg0 error) *MockDatagramStreamSendDatagramCall {
+	c.Call = c.Call.Return(arg0)
+	return c
+}
+
+// Do rewrite *gomock.Call.Do
+func (c *MockDatagramStreamSendDatagramCall) Do(f func([]byte) error) *MockDatagramStreamSendDatagramCall {
+	c.Call = c.Call.Do(f)
+	return c
+}
+
+// DoAndReturn rewrite *gomock.Call.DoAndReturn
+func (c *MockDatagramStreamSendDatagramCall) DoAndReturn(f func([]byte) error) *MockDatagramStreamSendDatagramCall {
+	c.Call = c.Call.DoAndReturn(f)
+	return c
+}
+
+// SetDeadline mocks base method.
+func (m *MockDatagramStream) SetDeadline(t time.Time) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "SetDeadline", t)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// SetDeadline indicates an expected call of SetDeadline.
+func (mr *MockDatagramStreamMockRecorder) SetDeadline(t any) *MockDatagramStreamSetDeadlineCall {
+	mr.mock.ctrl.T.Helper()
+	call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SetDeadline", reflect.TypeOf((*MockDatagramStream)(nil).SetDeadline), t)
+	return &MockDatagramStreamSetDeadlineCall{Call: call}
+}
+
+// MockDatagramStreamSetDeadlineCall wrap *gomock.Call
+type MockDatagramStreamSetDeadlineCall struct {
+	*gomock.Call
+}
+
+// Return rewrite *gomock.Call.Return
+func (c *MockDatagramStreamSetDeadlineCall) Return(arg0 error) *MockDatagramStreamSetDeadlineCall {
+	c.Call = c.Call.Return(arg0)
+	return c
+}
+
+// Do rewrite *gomock.Call.Do
+func (c *MockDatagramStreamSetDeadlineCall) Do(f func(time.Time) error) *MockDatagramStreamSetDeadlineCall {
+	c.Call = c.Call.Do(f)
+	return c
+}
+
+// DoAndReturn rewrite *gomock.Call.DoAndReturn
+func (c *MockDatagramStreamSetDeadlineCall) DoAndReturn(f func(time.Time) error) *MockDatagramStreamSetDeadlineCall {
+	c.Call = c.Call.DoAndReturn(f)
+	return c
+}
+
+// SetReadDeadline mocks base method.
+func (m *MockDatagramStream) SetReadDeadline(t time.Time) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "SetReadDeadline", t)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// SetReadDeadline indicates an expected call of SetReadDeadline.
+func (mr *MockDatagramStreamMockRecorder) SetReadDeadline(t any) *MockDatagramStreamSetReadDeadlineCall {
+	mr.mock.ctrl.T.Helper()
+	call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SetReadDeadline", reflect.TypeOf((*MockDatagramStream)(nil).SetReadDeadline), t)
+	return &MockDatagramStreamSetReadDeadlineCall{Call: call}
+}
+
+// MockDatagramStreamSetReadDeadlineCall wrap *gomock.Call
+type MockDatagramStreamSetReadDeadlineCall struct {
+	*gomock.Call
+}
+
+// Return rewrite *gomock.Call.Return
+func (c *MockDatagramStreamSetReadDeadlineCall) Return(arg0 error) *MockDatagramStreamSetReadDeadlineCall {
+	c.Call = c.Call.Return(arg0)
+	return c
+}
+
+// Do rewrite *gomock.Call.Do
+func (c *MockDatagramStreamSetReadDeadlineCall) Do(f func(time.Time) error) *MockDatagramStreamSetReadDeadlineCall {
+	c.Call = c.Call.Do(f)
+	return c
+}
+
+// DoAndReturn rewrite *gomock.Call.DoAndReturn
+func (c *MockDatagramStreamSetReadDeadlineCall) DoAndReturn(f func(time.Time) error) *MockDatagramStreamSetReadDeadlineCall {
+	c.Call = c.Call.DoAndReturn(f)
+	return c
+}
+
+// SetWriteDeadline mocks base method.
+func (m *MockDatagramStream) SetWriteDeadline(t time.Time) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "SetWriteDeadline", t)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// SetWriteDeadline indicates an expected call of SetWriteDeadline.
+func (mr *MockDatagramStreamMockRecorder) SetWriteDeadline(t any) *MockDatagramStreamSetWriteDeadlineCall {
+	mr.mock.ctrl.T.Helper()
+	call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SetWriteDeadline", reflect.TypeOf((*MockDatagramStream)(nil).SetWriteDeadline), t)
+	return &MockDatagramStreamSetWriteDeadlineCall{Call: call}
+}
+
+// MockDatagramStreamSetWriteDeadlineCall wrap *gomock.Call
+type MockDatagramStreamSetWriteDeadlineCall struct {
+	*gomock.Call
+}
+
+// Return rewrite *gomock.Call.Return
+func (c *MockDatagramStreamSetWriteDeadlineCall) Return(arg0 error) *MockDatagramStreamSetWriteDeadlineCall {
+	c.Call = c.Call.Return(arg0)
+	return c
+}
+
+// Do rewrite *gomock.Call.Do
+func (c *MockDatagramStreamSetWriteDeadlineCall) Do(f func(time.Time) error) *MockDatagramStreamSetWriteDeadlineCall {
+	c.Call = c.Call.Do(f)
+	return c
+}
+
+// DoAndReturn rewrite *gomock.Call.DoAndReturn
+func (c *MockDatagramStreamSetWriteDeadlineCall) DoAndReturn(f func(time.Time) error) *MockDatagramStreamSetWriteDeadlineCall {
+	c.Call = c.Call.DoAndReturn(f)
+	return c
+}
+
+// StreamID mocks base method.
+func (m *MockDatagramStream) StreamID() quic.StreamID {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "StreamID")
+	ret0, _ := ret[0].(quic.StreamID)
+	return ret0
+}
+
+// StreamID indicates an expected call of StreamID.
+func (mr *MockDatagramStreamMockRecorder) StreamID() *MockDatagramStreamStreamIDCall {
+	mr.mock.ctrl.T.Helper()
+	call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "StreamID", reflect.TypeOf((*MockDatagramStream)(nil).StreamID))
+	return &MockDatagramStreamStreamIDCall{Call: call}
+}
+
+// MockDatagramStreamStreamIDCall wrap *gomock.Call
+type MockDatagramStreamStreamIDCall struct {
+	*gomock.Call
+}
+
+// Return rewrite *gomock.Call.Return
+func (c *MockDatagramStreamStreamIDCall) Return(arg0 quic.StreamID) *MockDatagramStreamStreamIDCall {
+	c.Call = c.Call.Return(arg0)
+	return c
+}
+
+// Do rewrite *gomock.Call.Do
+func (c *MockDatagramStreamStreamIDCall) Do(f func() quic.StreamID) *MockDatagramStreamStreamIDCall {
+	c.Call = c.Call.Do(f)
+	return c
+}
+
+// DoAndReturn rewrite *gomock.Call.DoAndReturn
+func (c *MockDatagramStreamStreamIDCall) DoAndReturn(f func() quic.StreamID) *MockDatagramStreamStreamIDCall {
+	c.Call = c.Call.DoAndReturn(f)
+	return c
+}
+
+// Write mocks base method.
+func (m *MockDatagramStream) Write(p []byte) (int, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Write", p)
+	ret0, _ := ret[0].(int)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// Write indicates an expected call of Write.
+func (mr *MockDatagramStreamMockRecorder) Write(p any) *MockDatagramStreamWriteCall {
+	mr.mock.ctrl.T.Helper()
+	call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Write", reflect.TypeOf((*MockDatagramStream)(nil).Write), p)
+	return &MockDatagramStreamWriteCall{Call: call}
+}
+
+// MockDatagramStreamWriteCall wrap *gomock.Call
+type MockDatagramStreamWriteCall struct {
+	*gomock.Call
+}
+
+// Return rewrite *gomock.Call.Return
+func (c *MockDatagramStreamWriteCall) Return(n int, err error) *MockDatagramStreamWriteCall {
+	c.Call = c.Call.Return(n, err)
+	return c
+}
+
+// Do rewrite *gomock.Call.Do
+func (c *MockDatagramStreamWriteCall) Do(f func([]byte) (int, error)) *MockDatagramStreamWriteCall {
+	c.Call = c.Call.Do(f)
+	return c
+}
+
+// DoAndReturn rewrite *gomock.Call.DoAndReturn
+func (c *MockDatagramStreamWriteCall) DoAndReturn(f func([]byte) (int, error)) *MockDatagramStreamWriteCall {
+	c.Call = c.Call.DoAndReturn(f)
+	return c
+}
diff --git a/http3/response_writer_test.go b/http3/response_writer_test.go
index 64989e2d..07ebaad6 100644
--- a/http3/response_writer_test.go
+++ b/http3/response_writer_test.go
@@ -6,8 +6,6 @@ import (
 	"net/http"
 	"testing"
 
-	mockquic "github.com/quic-go/quic-go/internal/mocks/quic"
-
 	"github.com/quic-go/qpack"
 
 	"github.com/stretchr/testify/require"
@@ -59,11 +57,11 @@ func (rw *testResponseWriter) DecodeBody(t *testing.T) []byte {
 func newTestResponseWriter(t *testing.T) *testResponseWriter {
 	buf := &bytes.Buffer{}
 	mockCtrl := gomock.NewController(t)
-	str := mockquic.NewMockStream(mockCtrl)
+	str := NewMockDatagramStream(mockCtrl)
 	str.EXPECT().Write(gomock.Any()).DoAndReturn(buf.Write).AnyTimes()
 	str.EXPECT().SetReadDeadline(gomock.Any()).Return(nil).AnyTimes()
 	str.EXPECT().SetWriteDeadline(gomock.Any()).Return(nil).AnyTimes()
-	rw := newResponseWriter(newStream(str, nil, nil, func(r io.Reader, u uint64) error { return nil }), nil, false, nil)
+	rw := newResponseWriter(newStream(str, nil, func(r io.Reader, u uint64) error { return nil }), nil, false, nil)
 	return &testResponseWriter{responseWriter: rw, buf: buf}
 }
 
diff --git a/http3/server_test.go b/http3/server_test.go
index 02df9a0a..237d9ae4 100644
--- a/http3/server_test.go
+++ b/http3/server_test.go
@@ -227,7 +227,7 @@ func testServerRequestHandling(t *testing.T,
 ) (responseHeaders map[string][]string, body []byte) {
 	responseBuf := &bytes.Buffer{}
 	mockCtrl := gomock.NewController(t)
-	str := mockquic.NewMockStream(mockCtrl)
+	str := NewMockDatagramStream(mockCtrl)
 	str.EXPECT().Context().Return(context.Background()).AnyTimes()
 	str.EXPECT().Write(gomock.Any()).DoAndReturn(responseBuf.Write).AnyTimes()
 	str.EXPECT().CancelRead(gomock.Any())
@@ -246,7 +246,7 @@ func testServerRequestHandling(t *testing.T,
 	qconn.EXPECT().Context().Return(context.Background()).AnyTimes()
 	conn := newConnection(context.Background(), qconn, false, protocol.PerspectiveServer, nil, 0)
 
-	s.handleRequest(conn, str, nil, qpack.NewDecoder(nil))
+	s.handleRequest(conn, str, qpack.NewDecoder(nil))
 	hfs := decodeHeader(t, responseBuf)
 	fp := frameParser{r: responseBuf}
 	var content []byte
@@ -295,7 +295,7 @@ func TestServerHandlerBodyNotRead(t *testing.T) {
 
 func TestServerFirstFrameNotHeaders(t *testing.T) {
 	mockCtrl := gomock.NewController(t)
-	str := mockquic.NewMockStream(mockCtrl)
+	str := NewMockDatagramStream(mockCtrl)
 	str.EXPECT().Write(gomock.Any()).AnyTimes()
 	str.EXPECT().Context().Return(context.Background()).AnyTimes()
 	var buf bytes.Buffer
@@ -311,12 +311,12 @@ func TestServerFirstFrameNotHeaders(t *testing.T) {
 	qconn.EXPECT().CloseWithError(quic.ApplicationErrorCode(ErrCodeFrameUnexpected), gomock.Any())
 	conn := newConnection(context.Background(), qconn, false, protocol.PerspectiveServer, nil, 0)
 
-	s.handleRequest(conn, str, nil, qpack.NewDecoder(nil))
+	s.handleRequest(conn, str, qpack.NewDecoder(nil))
 }
 
 func testServerHandlerBodyNotRead(t *testing.T, req *http.Request, handler http.HandlerFunc) {
 	mockCtrl := gomock.NewController(t)
-	str := mockquic.NewMockStream(mockCtrl)
+	str := NewMockDatagramStream(mockCtrl)
 	str.EXPECT().Write(gomock.Any()).AnyTimes()
 	str.EXPECT().Context().Return(context.Background()).AnyTimes()
 	str.EXPECT().CancelRead(quic.StreamErrorCode(ErrCodeNoError))
@@ -339,13 +339,13 @@ func testServerHandlerBodyNotRead(t *testing.T, req *http.Request, handler http.
 	qconn.EXPECT().Context().Return(context.Background()).AnyTimes()
 	conn := newConnection(context.Background(), qconn, false, protocol.PerspectiveServer, nil, 0)
 
-	s.handleRequest(conn, str, nil, qpack.NewDecoder(nil))
+	s.handleRequest(conn, str, qpack.NewDecoder(nil))
 	require.True(t, called)
 }
 
 func TestServerStreamResetByClient(t *testing.T) {
 	mockCtrl := gomock.NewController(t)
-	str := mockquic.NewMockStream(mockCtrl)
+	str := NewMockDatagramStream(mockCtrl)
 	done := make(chan struct{})
 	str.EXPECT().CancelRead(quic.StreamErrorCode(ErrCodeRequestIncomplete))
 	str.EXPECT().CancelWrite(quic.StreamErrorCode(ErrCodeRequestIncomplete)).Do(func(quic.StreamErrorCode) { close(done) })
@@ -364,7 +364,7 @@ func TestServerStreamResetByClient(t *testing.T) {
 	qconn.EXPECT().LocalAddr().AnyTimes()
 	conn := newConnection(context.Background(), qconn, false, protocol.PerspectiveServer, nil, 0)
 
-	s.handleRequest(conn, str, nil, qpack.NewDecoder(nil))
+	s.handleRequest(conn, str, qpack.NewDecoder(nil))
 	select {
 	case <-done:
 	case <-time.After(time.Second):
@@ -393,7 +393,7 @@ func TestServerPanickingHandler(t *testing.T) {
 
 func testServerPanickingHandler(t *testing.T, handler http.HandlerFunc) (logOutput string) {
 	mockCtrl := gomock.NewController(t)
-	str := mockquic.NewMockStream(mockCtrl)
+	str := NewMockDatagramStream(mockCtrl)
 	str.EXPECT().Context().Return(context.Background()).AnyTimes()
 	str.EXPECT().CancelRead(quic.StreamErrorCode(ErrCodeInternalError))
 	str.EXPECT().CancelWrite(quic.StreamErrorCode(ErrCodeInternalError))
@@ -415,7 +415,7 @@ func testServerPanickingHandler(t *testing.T, handler http.HandlerFunc) (logOutp
 	qconn.EXPECT().Context().Return(context.Background()).AnyTimes()
 	conn := newConnection(context.Background(), qconn, false, protocol.PerspectiveServer, nil, 0)
 
-	s.handleRequest(conn, str, nil, qpack.NewDecoder(nil))
+	s.handleRequest(conn, str, qpack.NewDecoder(nil))
 	return logBuf.String()
 }
 
@@ -556,7 +556,7 @@ func TestServerRequestContext(t *testing.T) {
 func TestServerHTTPStreamHijacking(t *testing.T) {
 	responseBuf := &bytes.Buffer{}
 	mockCtrl := gomock.NewController(t)
-	str := mockquic.NewMockStream(mockCtrl)
+	str := NewMockDatagramStream(mockCtrl)
 	str.EXPECT().Context().Return(context.Background()).AnyTimes()
 	str.EXPECT().Write(gomock.Any()).DoAndReturn(responseBuf.Write).AnyTimes()
 	str.EXPECT().Read(gomock.Any()).DoAndReturn(
@@ -578,7 +578,7 @@ func TestServerHTTPStreamHijacking(t *testing.T) {
 	qconn.EXPECT().Context().Return(context.Background()).AnyTimes()
 	conn := newConnection(context.Background(), qconn, false, protocol.PerspectiveServer, nil, 0)
 
-	s.handleRequest(conn, str, nil, qpack.NewDecoder(nil))
+	s.handleRequest(conn, str, qpack.NewDecoder(nil))
 	hfs := decodeHeader(t, responseBuf)
 	require.Equal(t, hfs[":status"], []string{"200"})
 	require.Equal(t, []byte("foobar"), responseBuf.Bytes())
diff --git a/http3/state_tracking_stream_test.go b/http3/state_tracking_stream_test.go
index a256d906..03f2558a 100644
--- a/http3/state_tracking_stream_test.go
+++ b/http3/state_tracking_stream_test.go
@@ -5,6 +5,7 @@ import (
 	"context"
 	"errors"
 	"io"
+	"net"
 	"os"
 	"testing"
 	"time"
@@ -17,31 +18,29 @@ import (
 	"go.uber.org/mock/gomock"
 )
 
-type mockStreamClearer struct {
-	cleared *quic.StreamID
+func canceledCtx() context.Context {
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	return ctx
 }
 
-func (s *mockStreamClearer) clearStream(id quic.StreamID) {
-	s.cleared = &id
+func checkDatagramReceive(t *testing.T, str *stateTrackingStream) {
+	t.Helper()
+	_, err := str.ReceiveDatagram(canceledCtx())
+	require.ErrorIs(t, err, context.Canceled)
 }
 
-type mockErrorSetter struct {
-	sendErrs []error
-	recvErrs []error
-
-	sendSent chan struct{}
+func checkDatagramSend(t *testing.T, str *stateTrackingStream) {
+	t.Helper()
+	require.NoError(t, str.SendDatagram([]byte("test")))
 }
 
-func (e *mockErrorSetter) SetSendError(err error) {
-	e.sendErrs = append(e.sendErrs, err)
-
-	if e.sendSent != nil {
-		close(e.sendSent)
-	}
+type mockStreamClearer struct {
+	cleared *quic.StreamID
 }
 
-func (e *mockErrorSetter) SetReceiveError(err error) {
-	e.recvErrs = append(e.recvErrs, err)
+func (s *mockStreamClearer) clearStream(id quic.StreamID) {
+	s.cleared = &id
 }
 
 func TestStateTrackingStreamRead(t *testing.T) {
@@ -59,30 +58,24 @@ func testStateTrackingStreamRead(t *testing.T, expectedErr error) {
 	qstr.EXPECT().StreamID().AnyTimes().Return(quic.StreamID(1337))
 	qstr.EXPECT().Context().Return(context.Background()).AnyTimes()
 
-	var (
-		clearer mockStreamClearer
-		setter  mockErrorSetter
-		str     = newStateTrackingStream(qstr, &clearer, &setter)
-	)
+	var clearer mockStreamClearer
+	str := newStateTrackingStream(qstr, &clearer, func(b []byte) error { return nil })
 
 	// deadline errors are ignored
 	qstr.EXPECT().Read(gomock.Any()).Return(0, os.ErrDeadlineExceeded)
 	_, err := str.Read(make([]byte, 3))
 	require.ErrorIs(t, err, os.ErrDeadlineExceeded)
 	require.Nil(t, clearer.cleared)
-	require.Empty(t, setter.recvErrs)
-	require.Empty(t, setter.sendErrs)
 
 	if expectedErr == io.EOF {
 		buf := bytes.NewBuffer([]byte("foobar"))
 		qstr.EXPECT().Read(gomock.Any()).DoAndReturn(buf.Read).AnyTimes()
 
-		for i := 0; i < 3; i++ {
+		for range 3 {
 			_, err := str.Read([]byte{0})
 			require.NoError(t, err)
 			require.Nil(t, clearer.cleared)
-			require.Empty(t, setter.recvErrs)
-			require.Empty(t, setter.sendErrs)
+			checkDatagramReceive(t, str)
 		}
 	} else {
 		qstr.EXPECT().Read(gomock.Any()).Return(0, expectedErr).AnyTimes()
@@ -95,9 +88,11 @@ func testStateTrackingStreamRead(t *testing.T, expectedErr error) {
 		require.ErrorIs(t, err, expectedErr)
 	}
 	require.Nil(t, clearer.cleared)
-	require.Len(t, setter.recvErrs, 1)
-	require.Equal(t, expectedErr, setter.recvErrs[0])
-	require.Empty(t, setter.sendErrs)
+	// the receive side registered the error
+	_, err = str.ReceiveDatagram(canceledCtx())
+	require.ErrorIs(t, err, expectedErr)
+	// the send side is still open
+	require.NoError(t, str.SendDatagram([]byte("foo")))
 }
 
 func TestStateTrackingStreamWrite(t *testing.T) {
@@ -106,11 +101,8 @@ func TestStateTrackingStreamWrite(t *testing.T) {
 	qstr.EXPECT().StreamID().AnyTimes().Return(quic.StreamID(1337))
 	qstr.EXPECT().Context().Return(context.Background()).AnyTimes()
 
-	var (
-		clearer mockStreamClearer
-		setter  mockErrorSetter
-		str     = newStateTrackingStream(qstr, &clearer, &setter)
-	)
+	var clearer mockStreamClearer
+	str := newStateTrackingStream(qstr, &clearer, func(b []byte) error { return nil })
 
 	qstr.EXPECT().Write([]byte("foo")).Return(3, nil)
 	qstr.EXPECT().Write([]byte("baz")).Return(0, os.ErrDeadlineExceeded)
@@ -119,22 +111,21 @@ func TestStateTrackingStreamWrite(t *testing.T) {
 	_, err := str.Write([]byte("foo"))
 	require.NoError(t, err)
 	require.Nil(t, clearer.cleared)
-	require.Empty(t, setter.recvErrs)
-	require.Empty(t, setter.sendErrs)
+	checkDatagramReceive(t, str)
+	checkDatagramSend(t, str)
 
 	// deadline errors are ignored
 	_, err = str.Write([]byte("baz"))
 	require.ErrorIs(t, err, os.ErrDeadlineExceeded)
 	require.Nil(t, clearer.cleared)
-	require.Empty(t, setter.recvErrs)
-	require.Empty(t, setter.sendErrs)
+	checkDatagramReceive(t, str)
+	checkDatagramSend(t, str)
 
 	_, err = str.Write([]byte("bar"))
 	require.ErrorIs(t, err, assert.AnError)
 	require.Nil(t, clearer.cleared)
-	require.Empty(t, setter.recvErrs)
-	require.Len(t, setter.sendErrs, 1)
-	require.Equal(t, assert.AnError, setter.sendErrs[0])
+	checkDatagramReceive(t, str)
+	require.ErrorIs(t, str.SendDatagram([]byte("test")), assert.AnError)
 }
 
 func TestStateTrackingStreamCancelRead(t *testing.T) {
@@ -144,11 +135,8 @@ func TestStateTrackingStreamCancelRead(t *testing.T) {
 	qstr.EXPECT().StreamID().AnyTimes().Return(streamID)
 	qstr.EXPECT().Context().Return(context.Background()).AnyTimes()
 
-	var (
-		clearer mockStreamClearer
-		setter  mockErrorSetter
-		str     = newStateTrackingStream(qstr, &clearer, &setter)
-	)
+	var clearer mockStreamClearer
+	str := newStateTrackingStream(qstr, &clearer, func(b []byte) error { return nil })
 
 	buf := bytes.NewBuffer([]byte("foobar"))
 	qstr.EXPECT().Read(gomock.Any()).DoAndReturn(buf.Read).AnyTimes()
@@ -156,14 +144,14 @@ func TestStateTrackingStreamCancelRead(t *testing.T) {
 	_, err := str.Read(make([]byte, 3))
 	require.NoError(t, err)
 	require.Nil(t, clearer.cleared)
-	require.Empty(t, setter.recvErrs)
-	require.Empty(t, setter.sendErrs)
+	checkDatagramReceive(t, str)
+	checkDatagramSend(t, str)
 
 	str.CancelRead(1337)
 	require.Nil(t, clearer.cleared)
-	require.Len(t, setter.recvErrs, 1)
-	require.Equal(t, &quic.StreamError{StreamID: streamID, ErrorCode: 1337}, setter.recvErrs[0])
-	require.Empty(t, setter.sendErrs)
+	_, err = str.ReceiveDatagram(canceledCtx())
+	require.ErrorIs(t, err, &quic.StreamError{StreamID: streamID, ErrorCode: 1337})
+	checkDatagramSend(t, str)
 }
 
 func TestStateTrackingStreamCancelWrite(t *testing.T) {
@@ -173,25 +161,21 @@ func TestStateTrackingStreamCancelWrite(t *testing.T) {
 	qstr.EXPECT().StreamID().AnyTimes().Return(streamID)
 	qstr.EXPECT().Context().Return(context.Background()).AnyTimes()
 
-	var (
-		clearer mockStreamClearer
-		setter  mockErrorSetter
-		str     = newStateTrackingStream(qstr, &clearer, &setter)
-	)
+	var clearer mockStreamClearer
+	str := newStateTrackingStream(qstr, &clearer, func(b []byte) error { return nil })
 
 	qstr.EXPECT().Write(gomock.Any())
 	qstr.EXPECT().CancelWrite(quic.StreamErrorCode(1337))
 	_, err := str.Write([]byte("foobar"))
 	require.NoError(t, err)
 	require.Nil(t, clearer.cleared)
-	require.Empty(t, setter.recvErrs)
-	require.Empty(t, setter.sendErrs)
+	checkDatagramReceive(t, str)
+	checkDatagramSend(t, str)
 
 	str.CancelWrite(1337)
 	require.Nil(t, clearer.cleared)
-	require.Empty(t, setter.recvErrs)
-	require.Len(t, setter.sendErrs, 1)
-	require.Equal(t, &quic.StreamError{StreamID: streamID, ErrorCode: 1337}, setter.sendErrs[0])
+	checkDatagramReceive(t, str)
+	require.ErrorIs(t, str.SendDatagram([]byte("test")), &quic.StreamError{StreamID: streamID, ErrorCode: 1337})
 }
 
 func TestStateTrackingStreamContext(t *testing.T) {
@@ -201,29 +185,25 @@ func TestStateTrackingStreamContext(t *testing.T) {
 	ctx, cancel := context.WithCancelCause(context.Background())
 	qstr.EXPECT().Context().Return(ctx).AnyTimes()
 
-	var (
-		clearer mockStreamClearer
-		setter  = mockErrorSetter{
-			sendSent: make(chan struct{}),
-		}
-	)
+	var clearer mockStreamClearer
+	str := newStateTrackingStream(qstr, &clearer, func(b []byte) error { return nil })
 
-	_ = newStateTrackingStream(qstr, &clearer, &setter)
 	require.Nil(t, clearer.cleared)
-	require.Empty(t, setter.recvErrs)
-	require.Empty(t, setter.sendErrs)
+	checkDatagramReceive(t, str)
+	checkDatagramSend(t, str)
+
+	cancel(assert.AnError)
+	require.Eventually(t, func() bool {
+		err := str.SendDatagram([]byte("test"))
+		if err == nil {
+			return false
+		}
+		require.ErrorIs(t, err, assert.AnError)
+		return true
+	}, time.Second, scaleDuration(5*time.Millisecond))
 
-	testErr := errors.New("test error")
-	cancel(testErr)
-	select {
-	case <-setter.sendSent:
-	case <-time.After(time.Second):
-		t.Fatal("timeout")
-	}
+	checkDatagramReceive(t, str)
 	require.Nil(t, clearer.cleared)
-	require.Empty(t, setter.recvErrs)
-	require.Len(t, setter.sendErrs, 1)
-	require.Equal(t, testErr, setter.sendErrs[0])
 }
 
 func TestStateTrackingStreamReceiveThenSend(t *testing.T) {
@@ -233,11 +213,8 @@ func TestStateTrackingStreamReceiveThenSend(t *testing.T) {
 	qstr.EXPECT().StreamID().AnyTimes().Return(streamID)
 	qstr.EXPECT().Context().Return(context.Background()).AnyTimes()
 
-	var (
-		clearer mockStreamClearer
-		setter  mockErrorSetter
-		str     = newStateTrackingStream(qstr, &clearer, &setter)
-	)
+	var clearer mockStreamClearer
+	str := newStateTrackingStream(qstr, &clearer, func(b []byte) error { return nil })
 
 	buf := bytes.NewBuffer([]byte("foobar"))
 	qstr.EXPECT().Read(gomock.Any()).DoAndReturn(buf.Read).AnyTimes()
@@ -245,16 +222,15 @@ func TestStateTrackingStreamReceiveThenSend(t *testing.T) {
 	require.NoError(t, err)
 
 	require.Nil(t, clearer.cleared)
-	require.Len(t, setter.recvErrs, 1)
-	require.Equal(t, io.EOF, setter.recvErrs[0])
+	_, err = str.ReceiveDatagram(canceledCtx())
+	require.ErrorIs(t, err, io.EOF)
 
 	testErr := errors.New("test error")
 	qstr.EXPECT().Write([]byte("bar")).Return(0, testErr)
 
 	_, err = str.Write([]byte("bar"))
 	require.ErrorIs(t, err, testErr)
-	require.Len(t, setter.sendErrs, 1)
-	require.Equal(t, testErr, setter.sendErrs[0])
+	require.ErrorIs(t, str.SendDatagram([]byte("test")), testErr)
 
 	require.Equal(t, &streamID, clearer.cleared)
 }
@@ -266,11 +242,8 @@ func TestStateTrackingStreamSendThenReceive(t *testing.T) {
 	qstr.EXPECT().StreamID().AnyTimes().Return(streamID)
 	qstr.EXPECT().Context().Return(context.Background()).AnyTimes()
 
-	var (
-		clearer mockStreamClearer
-		setter  mockErrorSetter
-		str     = newStateTrackingStream(qstr, &clearer, &setter)
-	)
+	var clearer mockStreamClearer
+	str := newStateTrackingStream(qstr, &clearer, func(b []byte) error { return nil })
 
 	testErr := errors.New("test error")
 	qstr.EXPECT().Write([]byte("bar")).Return(0, testErr)
@@ -278,16 +251,90 @@ func TestStateTrackingStreamSendThenReceive(t *testing.T) {
 	_, err := str.Write([]byte("bar"))
 	require.ErrorIs(t, err, testErr)
 	require.Nil(t, clearer.cleared)
-	require.Len(t, setter.sendErrs, 1)
-	require.Equal(t, testErr, setter.sendErrs[0])
+	require.ErrorIs(t, str.SendDatagram([]byte("test")), testErr)
 
 	buf := bytes.NewBuffer([]byte("foobar"))
 	qstr.EXPECT().Read(gomock.Any()).DoAndReturn(buf.Read).AnyTimes()
 
 	_, err = io.ReadAll(str)
 	require.NoError(t, err)
-	require.Len(t, setter.recvErrs, 1)
-	require.Equal(t, io.EOF, setter.recvErrs[0])
+	_, err = str.ReceiveDatagram(canceledCtx())
+	require.ErrorIs(t, err, io.EOF)
 
 	require.Equal(t, &streamID, clearer.cleared)
 }
+
+func TestDatagramReceiving(t *testing.T) {
+	mockCtrl := gomock.NewController(t)
+	qstr := mockquic.NewMockStream(mockCtrl)
+	qstr.EXPECT().StreamID().AnyTimes().Return(quic.StreamID(1337))
+	qstr.EXPECT().Context().Return(context.Background()).AnyTimes()
+
+	str := newStateTrackingStream(qstr, nil, func(b []byte) error { return nil })
+	type result struct {
+		data []byte
+		err  error
+	}
+
+	// Receive blocks until a datagram is received
+	resultChan := make(chan result)
+	go func() {
+		defer close(resultChan)
+		data, err := str.ReceiveDatagram(context.Background())
+		resultChan <- result{data: data, err: err}
+	}()
+
+	select {
+	case <-time.After(scaleDuration(10 * time.Millisecond)):
+	case <-resultChan:
+		t.Fatal("should not have received a datagram")
+	}
+	str.enqueueDatagram([]byte("foobar"))
+
+	select {
+	case res := <-resultChan:
+		require.NoError(t, res.err)
+		require.Equal(t, []byte("foobar"), res.data)
+	case <-time.After(time.Second):
+		t.Fatal("should have received a datagram")
+	}
+
+	// up to 32 datagrams can be queued
+	for i := range streamDatagramQueueLen + 1 {
+		str.enqueueDatagram([]byte{uint8(i)})
+	}
+	for i := range streamDatagramQueueLen {
+		data, err := str.ReceiveDatagram(context.Background())
+		require.NoError(t, err)
+		require.Equal(t, []byte{uint8(i)}, data)
+	}
+
+	// Receive respects the context
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	_, err := str.ReceiveDatagram(ctx)
+	require.ErrorIs(t, err, context.Canceled)
+}
+
+func TestDatagramSending(t *testing.T) {
+	var sendQueue [][]byte
+	errors := []error{nil, nil, assert.AnError}
+	mockCtrl := gomock.NewController(t)
+	qstr := mockquic.NewMockStream(mockCtrl)
+	qstr.EXPECT().StreamID().AnyTimes().Return(quic.StreamID(1337))
+	qstr.EXPECT().Context().Return(context.Background()).AnyTimes()
+
+	str := newStateTrackingStream(qstr, nil, func(b []byte) error {
+		sendQueue = append(sendQueue, b)
+		err := errors[0]
+		errors = errors[1:]
+		return err
+	})
+	require.NoError(t, str.SendDatagram([]byte("foo")))
+	require.NoError(t, str.SendDatagram([]byte("bar")))
+	require.ErrorIs(t, str.SendDatagram([]byte("baz")), assert.AnError)
+	require.Equal(t, [][]byte{[]byte("foo"), []byte("bar"), []byte("baz")}, sendQueue)
+
+	str.closeSend(net.ErrClosed)
+	require.ErrorIs(t, str.SendDatagram([]byte("foobar")), net.ErrClosed)
+}
diff --git a/integrationtests/self/http_datagram_test.go b/integrationtests/self/http_datagram_test.go
index 6c8e6ff9..ecd3db13 100644
--- a/integrationtests/self/http_datagram_test.go
+++ b/integrationtests/self/http_datagram_test.go
@@ -162,7 +162,7 @@ func TestHTTPDatagrams(t *testing.T) {
 	port := startHTTPServer(t, mux, func(s *http3.Server) { s.EnableDatagrams = true })
 	str := dialAndOpenHTTPDatagramStream(t, fmt.Sprintf("https://localhost:%d/datagrams", port))
 
-	for i := 0; i < num; i++ {
+	for i := range num {
 		b := make([]byte, 8)
 		binary.BigEndian.PutUint64(b, uint64(i))
 		require.NoError(t, str.SendDatagram(bytes.Repeat(b, 100)))
@@ -178,6 +178,8 @@ loop:
 			}
 		case err := <-errChan:
 			t.Fatalf("receiving datagrams failed: %s", err)
+		case <-time.After(time.Second):
+			t.Fatal("timeout")
 		}
 	}
 	str.CancelWrite(42)