#![deny(missing_docs)]
use std::{
convert::TryInto,
future::Future,
pin::Pin,
sync::Arc,
task::{self, Poll, ready},
};
use bytes::{Buf, Bytes};
use futures_util::{
Stream, StreamExt,
stream::{self},
};
use http3::{
error::Code,
quic::{self, ConnectionErrorIncoming, StreamErrorIncoming, StreamId, WriteBuf},
};
use quinn::ReadError;
pub use quinn::{self, AcceptBi, AcceptUni, Endpoint, OpenBi, OpenUni, VarInt};
#[cfg(feature = "tracing")]
use tracing::instrument;
type BoxStreamSync<'a, T> = Pin<Box<dyn Stream<Item = T> + Sync + Send + 'a>>;
pub struct Connection {
conn: quinn::Connection,
incoming_bi: BoxStreamSync<'static, <AcceptBi<'static> as Future>::Output>,
opening_bi: Option<BoxStreamSync<'static, <OpenBi<'static> as Future>::Output>>,
incoming_uni: BoxStreamSync<'static, <AcceptUni<'static> as Future>::Output>,
opening_uni: Option<BoxStreamSync<'static, <OpenUni<'static> as Future>::Output>>,
}
impl Connection {
pub fn new(conn: quinn::Connection) -> Self {
Self {
conn: conn.clone(),
incoming_bi: Box::pin(stream::unfold(conn.clone(), |conn| async {
Some((conn.accept_bi().await, conn))
})),
opening_bi: None,
incoming_uni: Box::pin(stream::unfold(conn.clone(), |conn| async {
Some((conn.accept_uni().await, conn))
})),
opening_uni: None,
}
}
}
impl<B> quic::Connection<B> for Connection
where
B: Buf,
{
type RecvStream = RecvStream;
type OpenStreams = OpenStreams;
#[cfg_attr(feature = "tracing", instrument(skip_all, level = "trace"))]
fn poll_accept_bidi(
&mut self,
cx: &mut task::Context<'_>,
) -> Poll<Result<Self::BidiStream, ConnectionErrorIncoming>> {
let (send, recv) = ready!(self.incoming_bi.poll_next_unpin(cx))
.expect("self.incoming_bi BoxStream never returns None")
.map_err(convert_connection_error)?;
Poll::Ready(Ok(Self::BidiStream {
send: Self::SendStream::new(send),
recv: Self::RecvStream::new(recv),
}))
}
#[cfg_attr(feature = "tracing", instrument(skip_all, level = "trace"))]
fn poll_accept_recv(
&mut self,
cx: &mut task::Context<'_>,
) -> Poll<Result<Self::RecvStream, ConnectionErrorIncoming>> {
let recv = ready!(self.incoming_uni.poll_next_unpin(cx))
.expect("self.incoming_uni BoxStream never returns None")
.map_err(convert_connection_error)?;
Poll::Ready(Ok(Self::RecvStream::new(recv)))
}
fn opener(&self) -> Self::OpenStreams {
OpenStreams {
conn: self.conn.clone(),
opening_bi: None,
opening_uni: None,
}
}
}
fn convert_connection_error(e: quinn::ConnectionError) -> http3::quic::ConnectionErrorIncoming {
match e {
quinn::ConnectionError::ApplicationClosed(application_close) => {
ConnectionErrorIncoming::ApplicationClose {
error_code: application_close.error_code.into(),
}
}
quinn::ConnectionError::TimedOut => ConnectionErrorIncoming::Timeout,
error @ quinn::ConnectionError::VersionMismatch
| error @ quinn::ConnectionError::Reset
| error @ quinn::ConnectionError::LocallyClosed
| error @ quinn::ConnectionError::CidsExhausted
| error @ quinn::ConnectionError::TransportError(_)
| error @ quinn::ConnectionError::ConnectionClosed(_) => {
ConnectionErrorIncoming::Undefined(Arc::new(error))
}
}
}
impl<B> quic::OpenStreams<B> for Connection
where
B: Buf,
{
type SendStream = SendStream<B>;
type BidiStream = BidiStream<B>;
#[cfg_attr(feature = "tracing", instrument(skip_all, level = "trace"))]
fn poll_open_bidi(
&mut self,
cx: &mut task::Context<'_>,
) -> Poll<Result<Self::BidiStream, StreamErrorIncoming>> {
let bi = self.opening_bi.get_or_insert_with(|| {
Box::pin(stream::unfold(self.conn.clone(), |conn| async {
Some((conn.open_bi().await, conn))
}))
});
let (send, recv) = ready!(bi.poll_next_unpin(cx))
.expect("BoxStream does not return None")
.map_err(|e| StreamErrorIncoming::ConnectionErrorIncoming {
connection_error: convert_connection_error(e),
})?;
Poll::Ready(Ok(Self::BidiStream {
send: Self::SendStream::new(send),
recv: RecvStream::new(recv),
}))
}
#[cfg_attr(feature = "tracing", instrument(skip_all, level = "trace"))]
fn poll_open_send(
&mut self,
cx: &mut task::Context<'_>,
) -> Poll<Result<Self::SendStream, StreamErrorIncoming>> {
let uni = self.opening_uni.get_or_insert_with(|| {
Box::pin(stream::unfold(self.conn.clone(), |conn| async {
Some((conn.open_uni().await, conn))
}))
});
let send = ready!(uni.poll_next_unpin(cx))
.expect("BoxStream does not return None")
.map_err(|e| StreamErrorIncoming::ConnectionErrorIncoming {
connection_error: convert_connection_error(e),
})?;
Poll::Ready(Ok(Self::SendStream::new(send)))
}
#[cfg_attr(feature = "tracing", instrument(skip_all, level = "trace"))]
fn close(&mut self, code: Code, reason: &[u8]) {
self.conn.close(
VarInt::from_u64(code.value()).expect("error code VarInt"),
reason,
);
}
}
pub struct OpenStreams {
conn: quinn::Connection,
opening_bi: Option<BoxStreamSync<'static, <OpenBi<'static> as Future>::Output>>,
opening_uni: Option<BoxStreamSync<'static, <OpenUni<'static> as Future>::Output>>,
}
impl<B> quic::OpenStreams<B> for OpenStreams
where
B: Buf,
{
type SendStream = SendStream<B>;
type BidiStream = BidiStream<B>;
#[cfg_attr(feature = "tracing", instrument(skip_all, level = "trace"))]
fn poll_open_bidi(
&mut self,
cx: &mut task::Context<'_>,
) -> Poll<Result<Self::BidiStream, StreamErrorIncoming>> {
let bi = self.opening_bi.get_or_insert_with(|| {
Box::pin(stream::unfold(self.conn.clone(), |conn| async {
Some((conn.open_bi().await, conn))
}))
});
let (send, recv) = ready!(bi.poll_next_unpin(cx))
.expect("BoxStream does not return None")
.map_err(|e| StreamErrorIncoming::ConnectionErrorIncoming {
connection_error: convert_connection_error(e),
})?;
Poll::Ready(Ok(Self::BidiStream {
send: Self::SendStream::new(send),
recv: RecvStream::new(recv),
}))
}
#[cfg_attr(feature = "tracing", instrument(skip_all, level = "trace"))]
fn poll_open_send(
&mut self,
cx: &mut task::Context<'_>,
) -> Poll<Result<Self::SendStream, StreamErrorIncoming>> {
let uni = self.opening_uni.get_or_insert_with(|| {
Box::pin(stream::unfold(self.conn.clone(), |conn| async {
Some((conn.open_uni().await, conn))
}))
});
let send = ready!(uni.poll_next_unpin(cx))
.expect("BoxStream does not return None")
.map_err(|e| StreamErrorIncoming::ConnectionErrorIncoming {
connection_error: convert_connection_error(e),
})?;
Poll::Ready(Ok(Self::SendStream::new(send)))
}
#[cfg_attr(feature = "tracing", instrument(skip_all, level = "trace"))]
fn close(&mut self, code: Code, reason: &[u8]) {
self.conn.close(
VarInt::from_u64(code.value()).expect("error code VarInt"),
reason,
);
}
}
impl Clone for OpenStreams {
fn clone(&self) -> Self {
Self {
conn: self.conn.clone(),
opening_bi: None,
opening_uni: None,
}
}
}
pub struct BidiStream<B>
where
B: Buf,
{
send: SendStream<B>,
recv: RecvStream,
}
impl<B> quic::BidiStream<B> for BidiStream<B>
where
B: Buf,
{
type SendStream = SendStream<B>;
type RecvStream = RecvStream;
fn split(self) -> (Self::SendStream, Self::RecvStream) {
(self.send, self.recv)
}
}
impl<B: Buf> quic::RecvStream for BidiStream<B> {
type Buf = Bytes;
fn poll_data(
&mut self,
cx: &mut task::Context<'_>,
) -> Poll<Result<Option<Self::Buf>, StreamErrorIncoming>> {
self.recv.poll_data(cx)
}
fn stop_sending(&mut self, error_code: u64) {
self.recv.stop_sending(error_code)
}
fn recv_id(&self) -> StreamId {
self.recv.recv_id()
}
}
impl<B> quic::SendStream<B> for BidiStream<B>
where
B: Buf,
{
fn poll_ready(&mut self, cx: &mut task::Context<'_>) -> Poll<Result<(), StreamErrorIncoming>> {
self.send.poll_ready(cx)
}
fn poll_finish(&mut self, cx: &mut task::Context<'_>) -> Poll<Result<(), StreamErrorIncoming>> {
self.send.poll_finish(cx)
}
fn reset(&mut self, reset_code: u64) {
self.send.reset(reset_code)
}
fn send_data<D: Into<WriteBuf<B>>>(&mut self, data: D) -> Result<(), StreamErrorIncoming> {
self.send.send_data(data)
}
fn send_id(&self) -> StreamId {
self.send.send_id()
}
}
impl<B> quic::SendStreamUnframed<B> for BidiStream<B>
where
B: Buf,
{
fn poll_send<D: Buf>(
&mut self,
cx: &mut task::Context<'_>,
buf: &mut D,
) -> Poll<Result<usize, StreamErrorIncoming>> {
self.send.poll_send(cx, buf)
}
}
impl<B> quic::Is0rtt for BidiStream<B>
where
B: Buf,
{
fn is_0rtt(&self) -> bool {
self.recv.is_0rtt()
}
}
pub struct RecvStream {
stream: quinn::RecvStream,
is_0rtt: bool,
}
impl RecvStream {
fn new(stream: quinn::RecvStream) -> Self {
let is_0rtt = stream.is_0rtt();
Self { stream, is_0rtt }
}
}
impl quic::RecvStream for RecvStream {
type Buf = Bytes;
#[cfg_attr(feature = "tracing", instrument(skip_all, level = "trace"))]
fn poll_data(
&mut self,
cx: &mut task::Context<'_>,
) -> Poll<Result<Option<Self::Buf>, StreamErrorIncoming>> {
let mut read_chunk = std::pin::pin!(self.stream.read_chunk(usize::MAX, true));
let chunk = ready!(read_chunk.as_mut().poll(cx));
Poll::Ready(Ok(chunk
.map_err(convert_read_error_to_stream_error)?
.map(|c| c.bytes)))
}
#[cfg_attr(feature = "tracing", instrument(skip_all, level = "trace"))]
fn stop_sending(&mut self, error_code: u64) {
let error_code = VarInt::from_u64(error_code).expect("invalid error_code");
let _ = self.stream.stop(error_code);
}
#[cfg_attr(feature = "tracing", instrument(skip_all, level = "trace"))]
fn recv_id(&self) -> StreamId {
let num: u64 = self.stream.id().into();
num.try_into().expect("invalid stream id")
}
}
impl quic::Is0rtt for RecvStream {
fn is_0rtt(&self) -> bool {
self.is_0rtt
}
}
fn convert_read_error_to_stream_error(error: ReadError) -> StreamErrorIncoming {
match error {
ReadError::Reset(var_int) => StreamErrorIncoming::StreamTerminated {
error_code: var_int.into_inner(),
},
ReadError::ConnectionLost(connection_error) => {
StreamErrorIncoming::ConnectionErrorIncoming {
connection_error: convert_connection_error(connection_error),
}
}
error @ ReadError::ClosedStream => StreamErrorIncoming::Unknown(Box::new(error)),
ReadError::IllegalOrderedRead => panic!("http3-quinn-rs only performs ordered reads"),
error @ ReadError::ZeroRttRejected => StreamErrorIncoming::Unknown(Box::new(error)),
}
}
fn convert_write_error_to_stream_error(error: quinn::WriteError) -> StreamErrorIncoming {
match error {
quinn::WriteError::Stopped(var_int) => StreamErrorIncoming::StreamTerminated {
error_code: var_int.into_inner(),
},
quinn::WriteError::ConnectionLost(connection_error) => {
StreamErrorIncoming::ConnectionErrorIncoming {
connection_error: convert_connection_error(connection_error),
}
}
error @ quinn::WriteError::ClosedStream | error @ quinn::WriteError::ZeroRttRejected => {
StreamErrorIncoming::Unknown(Box::new(error))
}
}
}
pub struct SendStream<B: Buf> {
stream: quinn::SendStream,
writing: Option<WriteBuf<B>>,
}
impl<B> SendStream<B>
where
B: Buf,
{
fn new(stream: quinn::SendStream) -> SendStream<B> {
Self {
stream,
writing: None,
}
}
}
impl<B> quic::SendStream<B> for SendStream<B>
where
B: Buf,
{
#[cfg_attr(feature = "tracing", instrument(skip_all, level = "trace"))]
fn poll_ready(&mut self, cx: &mut task::Context<'_>) -> Poll<Result<(), StreamErrorIncoming>> {
if let Some(ref mut data) = self.writing {
while data.has_remaining() {
let stream = Pin::new(&mut self.stream);
let written = ready!(stream.poll_write(cx, data.chunk()))
.map_err(convert_write_error_to_stream_error)?;
data.advance(written);
}
}
self.writing = None;
Poll::Ready(Ok(()))
}
#[cfg_attr(feature = "tracing", instrument(skip_all, level = "trace"))]
fn poll_finish(
&mut self,
_cx: &mut task::Context<'_>,
) -> Poll<Result<(), StreamErrorIncoming>> {
Poll::Ready(
self.stream
.finish()
.map_err(|e| StreamErrorIncoming::Unknown(Box::new(e))),
)
}
#[cfg_attr(feature = "tracing", instrument(skip_all, level = "trace"))]
fn reset(&mut self, reset_code: u64) {
let _ = self
.stream
.reset(VarInt::from_u64(reset_code).unwrap_or(VarInt::MAX));
}
#[cfg_attr(feature = "tracing", instrument(skip_all, level = "trace"))]
fn send_data<D: Into<WriteBuf<B>>>(&mut self, data: D) -> Result<(), StreamErrorIncoming> {
if self.writing.is_some() {
#[cfg(feature = "tracing")]
tracing::error!("send_data called while send stream is not ready");
return Err(StreamErrorIncoming::ConnectionErrorIncoming {
connection_error: ConnectionErrorIncoming::InternalError(
"internal error in the http stack".to_string(),
),
});
}
self.writing = Some(data.into());
Ok(())
}
#[cfg_attr(feature = "tracing", instrument(skip_all, level = "trace"))]
fn send_id(&self) -> StreamId {
let num: u64 = self.stream.id().into();
num.try_into().expect("invalid stream id")
}
}
impl<B> quic::SendStreamUnframed<B> for SendStream<B>
where
B: Buf,
{
#[cfg_attr(feature = "tracing", instrument(skip_all, level = "trace"))]
fn poll_send<D: Buf>(
&mut self,
cx: &mut task::Context<'_>,
buf: &mut D,
) -> Poll<Result<usize, StreamErrorIncoming>> {
if self.writing.is_some() {
panic!("poll_send called while send stream is not ready")
}
let s = Pin::new(&mut self.stream);
let res = ready!(s.poll_write(cx, buf.chunk()));
match res {
Ok(written) => {
buf.advance(written);
Poll::Ready(Ok(written))
}
Err(err) => Poll::Ready(Err(convert_write_error_to_stream_error(err))),
}
}
}
#[cfg(test)]
mod tests {
use std::time::Duration;
use futures_util::future::poll_fn;
use super::*;
use crate::{quic::RecvStream as _, tests::Pair};
async fn connected_recv_stream() -> (
RecvStream,
quinn::SendStream,
quinn::Connection,
quinn::Connection,
) {
let mut pair = Pair::default();
let server = pair.server();
let (client, server) = tokio::join!(pair.client_inner(), async {
server.endpoint.accept().await.unwrap().await.unwrap()
});
let (mut send, _) = client.open_bi().await.unwrap();
send.write_all(b"seed").await.unwrap();
let (_, recv) = server.accept_bi().await.unwrap();
let mut recv = RecvStream::new(recv);
let seed = poll_fn(|cx| recv.poll_data(cx)).await.unwrap().unwrap();
assert_eq!(seed, Bytes::from_static(b"seed"));
(recv, send, client, server)
}
#[tokio::test]
async fn recv_stream_wakes_after_pending_read() {
let (mut recv, mut send, _client, _server) = connected_recv_stream().await;
let read = poll_fn(|cx| recv.poll_data(cx));
let write = async {
tokio::task::yield_now().await;
send.write_all(b"payload").await.unwrap();
send.finish().unwrap();
};
let (data, ()) =
tokio::time::timeout(Duration::from_secs(5), async { tokio::join!(read, write) })
.await
.expect("pending read was not woken");
assert_eq!(data.unwrap().unwrap(), Bytes::from_static(b"payload"));
assert!(
tokio::time::timeout(Duration::from_secs(5), poll_fn(|cx| recv.poll_data(cx)))
.await
.expect("receive stream did not reach EOF")
.unwrap()
.is_none()
);
}
#[tokio::test]
async fn recv_stream_stop_after_pending_preserves_error_code() {
const ERROR_CODE: u64 = 0x10c;
let (mut recv, send, _client, _server) = connected_recv_stream().await;
poll_fn(|cx| match recv.poll_data(cx) {
Poll::Pending => Poll::Ready(()),
Poll::Ready(result) => panic!("read unexpectedly completed: {result:?}"),
})
.await;
recv.stop_sending(ERROR_CODE);
assert!(poll_fn(|cx| recv.poll_data(cx)).await.unwrap().is_none());
drop(recv);
let stopped = tokio::time::timeout(Duration::from_secs(5), send.stopped())
.await
.expect("peer did not receive STOP_SENDING")
.unwrap();
assert_eq!(stopped, Some(VarInt::from_u64(ERROR_CODE).unwrap()));
}
}