feat: rustls-0.21. quinn (and hence h3-quinn) do not support rustls-0.21 yet.
This commit is contained in:
Jun Kurihara
parent
380b4d055c
commit
d9868442c5
6 changed files with 595 additions and 7 deletions
21
h3-quinn/Cargo.toml
Normal file
21
h3-quinn/Cargo.toml
Normal file
|
|
@ -0,0 +1,21 @@
|
|||
[package]
|
||||
name = "h3-quinn"
|
||||
version = "0.0.1"
|
||||
rust-version = "1.59"
|
||||
authors = ["Jean-Christophe BEGUE <jc.begue@pm.me>"]
|
||||
edition = "2018"
|
||||
documentation = "https://docs.rs/h3-quinn"
|
||||
repository = "https://github.com/hyperium/h3"
|
||||
readme = "../README.md"
|
||||
description = "QUIC transport implementation based on Quinn."
|
||||
keywords = ["http3", "quic", "h3"]
|
||||
categories = ["network-programming", "web-programming"]
|
||||
license = "MIT"
|
||||
|
||||
[dependencies]
|
||||
h3 = { version = "0.0.1", path = "../h3/h3" }
|
||||
bytes = "1"
|
||||
quinn = { path = "../quinn/quinn/", default-features = false }
|
||||
quinn-proto = { path = "../quinn/quinn-proto/", default-features = false }
|
||||
tokio-util = { version = "0.7.7" }
|
||||
futures = { version = "0.3.27" }
|
||||
559
h3-quinn/src/lib.rs
Normal file
559
h3-quinn/src/lib.rs
Normal file
|
|
@ -0,0 +1,559 @@
|
|||
//! QUIC Transport implementation with Quinn
|
||||
//!
|
||||
//! This module implements QUIC traits with Quinn.
|
||||
#![deny(missing_docs)]
|
||||
|
||||
use std::{
|
||||
convert::TryInto,
|
||||
fmt::{self, Display},
|
||||
future::Future,
|
||||
sync::Arc,
|
||||
task::{self, Poll},
|
||||
};
|
||||
|
||||
use bytes::{Buf, Bytes};
|
||||
|
||||
use futures::{
|
||||
ready,
|
||||
stream::{self, BoxStream},
|
||||
StreamExt,
|
||||
};
|
||||
pub use quinn::{
|
||||
self, crypto::Session, AcceptBi, AcceptUni, Endpoint, OpenBi, OpenUni, VarInt, WriteError,
|
||||
};
|
||||
|
||||
use h3::quic::{self, Error, StreamId, WriteBuf};
|
||||
use tokio_util::sync::ReusableBoxFuture;
|
||||
|
||||
/// A QUIC connection backed by Quinn
|
||||
///
|
||||
/// Implements a [`quic::Connection`] backed by a [`quinn::Connection`].
|
||||
pub struct Connection {
|
||||
conn: quinn::Connection,
|
||||
incoming_bi: BoxStream<'static, <AcceptBi<'static> as Future>::Output>,
|
||||
opening_bi: Option<BoxStream<'static, <OpenBi<'static> as Future>::Output>>,
|
||||
incoming_uni: BoxStream<'static, <AcceptUni<'static> as Future>::Output>,
|
||||
opening_uni: Option<BoxStream<'static, <OpenUni<'static> as Future>::Output>>,
|
||||
}
|
||||
|
||||
impl Connection {
|
||||
/// Create a [`Connection`] from a [`quinn::NewConnection`]
|
||||
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, |conn| async {
|
||||
Some((conn.accept_uni().await, conn))
|
||||
})),
|
||||
opening_uni: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The error type for [`Connection`]
|
||||
///
|
||||
/// Wraps reasons a Quinn connection might be lost.
|
||||
#[derive(Debug)]
|
||||
pub struct ConnectionError(quinn::ConnectionError);
|
||||
|
||||
impl std::error::Error for ConnectionError {}
|
||||
|
||||
impl fmt::Display for ConnectionError {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
self.0.fmt(f)
|
||||
}
|
||||
}
|
||||
|
||||
impl Error for ConnectionError {
|
||||
fn is_timeout(&self) -> bool {
|
||||
matches!(self.0, quinn::ConnectionError::TimedOut)
|
||||
}
|
||||
|
||||
fn err_code(&self) -> Option<u64> {
|
||||
match self.0 {
|
||||
quinn::ConnectionError::ApplicationClosed(quinn_proto::ApplicationClose {
|
||||
error_code,
|
||||
..
|
||||
}) => Some(error_code.into_inner()),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl From<quinn::ConnectionError> for ConnectionError {
|
||||
fn from(e: quinn::ConnectionError) -> Self {
|
||||
Self(e)
|
||||
}
|
||||
}
|
||||
|
||||
impl<B> quic::Connection<B> for Connection
|
||||
where
|
||||
B: Buf,
|
||||
{
|
||||
type SendStream = SendStream<B>;
|
||||
type RecvStream = RecvStream;
|
||||
type BidiStream = BidiStream<B>;
|
||||
type OpenStreams = OpenStreams;
|
||||
type Error = ConnectionError;
|
||||
|
||||
fn poll_accept_bidi(
|
||||
&mut self,
|
||||
cx: &mut task::Context<'_>,
|
||||
) -> Poll<Result<Option<Self::BidiStream>, Self::Error>> {
|
||||
let (send, recv) = match ready!(self.incoming_bi.poll_next_unpin(cx)) {
|
||||
Some(x) => x?,
|
||||
None => return Poll::Ready(Ok(None)),
|
||||
};
|
||||
Poll::Ready(Ok(Some(Self::BidiStream {
|
||||
send: Self::SendStream::new(send),
|
||||
recv: Self::RecvStream::new(recv),
|
||||
})))
|
||||
}
|
||||
|
||||
fn poll_accept_recv(
|
||||
&mut self,
|
||||
cx: &mut task::Context<'_>,
|
||||
) -> Poll<Result<Option<Self::RecvStream>, Self::Error>> {
|
||||
let recv = match ready!(self.incoming_uni.poll_next_unpin(cx)) {
|
||||
Some(x) => x?,
|
||||
None => return Poll::Ready(Ok(None)),
|
||||
};
|
||||
Poll::Ready(Ok(Some(Self::RecvStream::new(recv))))
|
||||
}
|
||||
|
||||
fn poll_open_bidi(
|
||||
&mut self,
|
||||
cx: &mut task::Context<'_>,
|
||||
) -> Poll<Result<Self::BidiStream, Self::Error>> {
|
||||
if self.opening_bi.is_none() {
|
||||
self.opening_bi = Some(Box::pin(stream::unfold(self.conn.clone(), |conn| async {
|
||||
Some((conn.clone().open_bi().await, conn))
|
||||
})));
|
||||
}
|
||||
|
||||
let (send, recv) =
|
||||
ready!(self.opening_bi.as_mut().unwrap().poll_next_unpin(cx)).unwrap()?;
|
||||
Poll::Ready(Ok(Self::BidiStream {
|
||||
send: Self::SendStream::new(send),
|
||||
recv: Self::RecvStream::new(recv),
|
||||
}))
|
||||
}
|
||||
|
||||
fn poll_open_send(
|
||||
&mut self,
|
||||
cx: &mut task::Context<'_>,
|
||||
) -> Poll<Result<Self::SendStream, Self::Error>> {
|
||||
if self.opening_uni.is_none() {
|
||||
self.opening_uni = Some(Box::pin(stream::unfold(self.conn.clone(), |conn| async {
|
||||
Some((conn.open_uni().await, conn))
|
||||
})));
|
||||
}
|
||||
|
||||
let send = ready!(self.opening_uni.as_mut().unwrap().poll_next_unpin(cx)).unwrap()?;
|
||||
Poll::Ready(Ok(Self::SendStream::new(send)))
|
||||
}
|
||||
|
||||
fn opener(&self) -> Self::OpenStreams {
|
||||
OpenStreams {
|
||||
conn: self.conn.clone(),
|
||||
opening_bi: None,
|
||||
opening_uni: None,
|
||||
}
|
||||
}
|
||||
|
||||
fn close(&mut self, code: h3::error::Code, reason: &[u8]) {
|
||||
self.conn.close(
|
||||
VarInt::from_u64(code.value()).expect("error code VarInt"),
|
||||
reason,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// Stream opener backed by a Quinn connection
|
||||
///
|
||||
/// Implements [`quic::OpenStreams`] using [`quinn::Connection`],
|
||||
/// [`quinn::OpenBi`], [`quinn::OpenUni`].
|
||||
pub struct OpenStreams {
|
||||
conn: quinn::Connection,
|
||||
opening_bi: Option<BoxStream<'static, <OpenBi<'static> as Future>::Output>>,
|
||||
opening_uni: Option<BoxStream<'static, <OpenUni<'static> as Future>::Output>>,
|
||||
}
|
||||
|
||||
impl<B> quic::OpenStreams<B> for OpenStreams
|
||||
where
|
||||
B: Buf,
|
||||
{
|
||||
type RecvStream = RecvStream;
|
||||
type SendStream = SendStream<B>;
|
||||
type BidiStream = BidiStream<B>;
|
||||
type Error = ConnectionError;
|
||||
|
||||
fn poll_open_bidi(
|
||||
&mut self,
|
||||
cx: &mut task::Context<'_>,
|
||||
) -> Poll<Result<Self::BidiStream, Self::Error>> {
|
||||
if self.opening_bi.is_none() {
|
||||
self.opening_bi = Some(Box::pin(stream::unfold(self.conn.clone(), |conn| async {
|
||||
Some((conn.open_bi().await, conn))
|
||||
})));
|
||||
}
|
||||
|
||||
let (send, recv) =
|
||||
ready!(self.opening_bi.as_mut().unwrap().poll_next_unpin(cx)).unwrap()?;
|
||||
Poll::Ready(Ok(Self::BidiStream {
|
||||
send: Self::SendStream::new(send),
|
||||
recv: Self::RecvStream::new(recv),
|
||||
}))
|
||||
}
|
||||
|
||||
fn poll_open_send(
|
||||
&mut self,
|
||||
cx: &mut task::Context<'_>,
|
||||
) -> Poll<Result<Self::SendStream, Self::Error>> {
|
||||
if self.opening_uni.is_none() {
|
||||
self.opening_uni = Some(Box::pin(stream::unfold(self.conn.clone(), |conn| async {
|
||||
Some((conn.open_uni().await, conn))
|
||||
})));
|
||||
}
|
||||
|
||||
let send = ready!(self.opening_uni.as_mut().unwrap().poll_next_unpin(cx)).unwrap()?;
|
||||
Poll::Ready(Ok(Self::SendStream::new(send)))
|
||||
}
|
||||
|
||||
fn close(&mut self, code: h3::error::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,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Quinn-backed bidirectional stream
|
||||
///
|
||||
/// Implements [`quic::BidiStream`] which allows the stream to be split
|
||||
/// into two structs each implementing one direction.
|
||||
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> quic::RecvStream for BidiStream<B>
|
||||
where
|
||||
B: Buf,
|
||||
{
|
||||
type Buf = Bytes;
|
||||
type Error = ReadError;
|
||||
|
||||
fn poll_data(
|
||||
&mut self,
|
||||
cx: &mut task::Context<'_>,
|
||||
) -> Poll<Result<Option<Self::Buf>, Self::Error>> {
|
||||
self.recv.poll_data(cx)
|
||||
}
|
||||
|
||||
fn stop_sending(&mut self, error_code: u64) {
|
||||
self.recv.stop_sending(error_code)
|
||||
}
|
||||
}
|
||||
|
||||
impl<B> quic::SendStream<B> for BidiStream<B>
|
||||
where
|
||||
B: Buf,
|
||||
{
|
||||
type Error = SendStreamError;
|
||||
|
||||
fn poll_ready(&mut self, cx: &mut task::Context<'_>) -> Poll<Result<(), Self::Error>> {
|
||||
self.send.poll_ready(cx)
|
||||
}
|
||||
|
||||
fn poll_finish(&mut self, cx: &mut task::Context<'_>) -> Poll<Result<(), Self::Error>> {
|
||||
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<(), Self::Error> {
|
||||
self.send.send_data(data)
|
||||
}
|
||||
|
||||
fn id(&self) -> StreamId {
|
||||
self.send.id()
|
||||
}
|
||||
}
|
||||
|
||||
/// Quinn-backed receive stream
|
||||
///
|
||||
/// Implements a [`quic::RecvStream`] backed by a [`quinn::RecvStream`].
|
||||
pub struct RecvStream {
|
||||
stream: Option<quinn::RecvStream>,
|
||||
read_chunk_fut: ReadChunkFuture,
|
||||
}
|
||||
|
||||
type ReadChunkFuture = ReusableBoxFuture<
|
||||
'static,
|
||||
(
|
||||
quinn::RecvStream,
|
||||
Result<Option<quinn::Chunk>, quinn::ReadError>,
|
||||
),
|
||||
>;
|
||||
|
||||
impl RecvStream {
|
||||
fn new(stream: quinn::RecvStream) -> Self {
|
||||
Self {
|
||||
stream: Some(stream),
|
||||
// Should only allocate once the first time it's used
|
||||
read_chunk_fut: ReusableBoxFuture::new(async { unreachable!() }),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl quic::RecvStream for RecvStream {
|
||||
type Buf = Bytes;
|
||||
type Error = ReadError;
|
||||
|
||||
fn poll_data(
|
||||
&mut self,
|
||||
cx: &mut task::Context<'_>,
|
||||
) -> Poll<Result<Option<Self::Buf>, Self::Error>> {
|
||||
if let Some(mut stream) = self.stream.take() {
|
||||
self.read_chunk_fut.set(async move {
|
||||
let chunk = stream.read_chunk(usize::MAX, true).await;
|
||||
(stream, chunk)
|
||||
})
|
||||
};
|
||||
|
||||
let (stream, chunk) = ready!(self.read_chunk_fut.poll(cx));
|
||||
self.stream = Some(stream);
|
||||
Poll::Ready(Ok(chunk?.map(|c| c.bytes)))
|
||||
}
|
||||
|
||||
fn stop_sending(&mut self, error_code: u64) {
|
||||
self.stream
|
||||
.as_mut()
|
||||
.unwrap()
|
||||
.stop(VarInt::from_u64(error_code).expect("invalid error_code"))
|
||||
.ok();
|
||||
}
|
||||
}
|
||||
|
||||
/// The error type for [`RecvStream`]
|
||||
///
|
||||
/// Wraps errors that occur when reading from a receive stream.
|
||||
#[derive(Debug)]
|
||||
pub struct ReadError(quinn::ReadError);
|
||||
|
||||
impl std::error::Error for ReadError {}
|
||||
|
||||
impl fmt::Display for ReadError {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
self.0.fmt(f)
|
||||
}
|
||||
}
|
||||
|
||||
impl From<ReadError> for Arc<dyn Error> {
|
||||
fn from(e: ReadError) -> Self {
|
||||
Arc::new(e)
|
||||
}
|
||||
}
|
||||
|
||||
impl From<quinn::ReadError> for ReadError {
|
||||
fn from(e: quinn::ReadError) -> Self {
|
||||
Self(e)
|
||||
}
|
||||
}
|
||||
|
||||
impl Error for ReadError {
|
||||
fn is_timeout(&self) -> bool {
|
||||
matches!(
|
||||
self.0,
|
||||
quinn::ReadError::ConnectionLost(quinn::ConnectionError::TimedOut)
|
||||
)
|
||||
}
|
||||
|
||||
fn err_code(&self) -> Option<u64> {
|
||||
match self.0 {
|
||||
quinn::ReadError::ConnectionLost(quinn::ConnectionError::ApplicationClosed(
|
||||
quinn_proto::ApplicationClose { error_code, .. },
|
||||
)) => Some(error_code.into_inner()),
|
||||
quinn::ReadError::Reset(error_code) => Some(error_code.into_inner()),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Quinn-backed send stream
|
||||
///
|
||||
/// Implements a [`quic::SendStream`] backed by a [`quinn::SendStream`].
|
||||
pub struct SendStream<B: Buf> {
|
||||
stream: Option<quinn::SendStream>,
|
||||
writing: Option<WriteBuf<B>>,
|
||||
write_fut: WriteFuture,
|
||||
}
|
||||
|
||||
type WriteFuture =
|
||||
ReusableBoxFuture<'static, (quinn::SendStream, Result<usize, quinn::WriteError>)>;
|
||||
|
||||
impl<B> SendStream<B>
|
||||
where
|
||||
B: Buf,
|
||||
{
|
||||
fn new(stream: quinn::SendStream) -> SendStream<B> {
|
||||
Self {
|
||||
stream: Some(stream),
|
||||
writing: None,
|
||||
write_fut: ReusableBoxFuture::new(async { unreachable!() }),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<B> quic::SendStream<B> for SendStream<B>
|
||||
where
|
||||
B: Buf,
|
||||
{
|
||||
type Error = SendStreamError;
|
||||
|
||||
fn poll_ready(&mut self, cx: &mut task::Context<'_>) -> Poll<Result<(), Self::Error>> {
|
||||
if let Some(ref mut data) = self.writing {
|
||||
while data.has_remaining() {
|
||||
if let Some(mut stream) = self.stream.take() {
|
||||
let chunk = data.chunk().to_owned(); // FIXME - avoid copy
|
||||
self.write_fut.set(async move {
|
||||
let ret = stream.write(&chunk).await;
|
||||
(stream, ret)
|
||||
});
|
||||
}
|
||||
|
||||
let (stream, res) = ready!(self.write_fut.poll(cx));
|
||||
self.stream = Some(stream);
|
||||
match res {
|
||||
Ok(cnt) => data.advance(cnt),
|
||||
Err(err) => {
|
||||
return Poll::Ready(Err(SendStreamError::Write(err)));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
self.writing = None;
|
||||
Poll::Ready(Ok(()))
|
||||
}
|
||||
|
||||
fn poll_finish(&mut self, cx: &mut task::Context<'_>) -> Poll<Result<(), Self::Error>> {
|
||||
self.stream
|
||||
.as_mut()
|
||||
.unwrap()
|
||||
.poll_finish(cx)
|
||||
.map_err(Into::into)
|
||||
}
|
||||
|
||||
fn reset(&mut self, reset_code: u64) {
|
||||
let _ = self
|
||||
.stream
|
||||
.as_mut()
|
||||
.unwrap()
|
||||
.reset(VarInt::from_u64(reset_code).unwrap_or(VarInt::MAX));
|
||||
}
|
||||
|
||||
fn send_data<D: Into<WriteBuf<B>>>(&mut self, data: D) -> Result<(), Self::Error> {
|
||||
if self.writing.is_some() {
|
||||
return Err(Self::Error::NotReady);
|
||||
}
|
||||
self.writing = Some(data.into());
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn id(&self) -> StreamId {
|
||||
self.stream
|
||||
.as_ref()
|
||||
.unwrap()
|
||||
.id()
|
||||
.0
|
||||
.try_into()
|
||||
.expect("invalid stream id")
|
||||
}
|
||||
}
|
||||
|
||||
/// The error type for [`SendStream`]
|
||||
///
|
||||
/// Wraps errors that can happen writing to or polling a send stream.
|
||||
#[derive(Debug)]
|
||||
pub enum SendStreamError {
|
||||
/// Errors when writing, wrapping a [`quinn::WriteError`]
|
||||
Write(WriteError),
|
||||
/// Error when the stream is not ready, because it is still sending
|
||||
/// data from a previous call
|
||||
NotReady,
|
||||
}
|
||||
|
||||
impl std::error::Error for SendStreamError {}
|
||||
|
||||
impl Display for SendStreamError {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
write!(f, "{:?}", self)
|
||||
}
|
||||
}
|
||||
|
||||
impl From<WriteError> for SendStreamError {
|
||||
fn from(e: WriteError) -> Self {
|
||||
Self::Write(e)
|
||||
}
|
||||
}
|
||||
|
||||
impl Error for SendStreamError {
|
||||
fn is_timeout(&self) -> bool {
|
||||
matches!(
|
||||
self,
|
||||
Self::Write(quinn::WriteError::ConnectionLost(
|
||||
quinn::ConnectionError::TimedOut
|
||||
))
|
||||
)
|
||||
}
|
||||
|
||||
fn err_code(&self) -> Option<u64> {
|
||||
match self {
|
||||
Self::Write(quinn::WriteError::Stopped(error_code)) => Some(error_code.into_inner()),
|
||||
Self::Write(quinn::WriteError::ConnectionLost(
|
||||
quinn::ConnectionError::ApplicationClosed(quinn_proto::ApplicationClose {
|
||||
error_code,
|
||||
..
|
||||
}),
|
||||
)) => Some(error_code.into_inner()),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl From<SendStreamError> for Arc<dyn Error> {
|
||||
fn from(e: SendStreamError) -> Self {
|
||||
Arc::new(e)
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue