h3x 0.6.1-beta.2

Peer-to-peer DHTTP/3 transport over QUIC
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98

#![allow(dead_code)]

use std::{borrow::Cow, sync::Arc};

use futures::future::BoxFuture;
use snafu::Snafu;

use crate::{error::Code, quic, rpc::lifecycle::LifecycleExt, varint::VarInt};

const DRIVER_PROTOCOL_ERROR_KIND: VarInt = VarInt::from_u32(0x3f00);
const DRIVER_PROTOCOL_ERROR_FRAME_TYPE: VarInt = VarInt::from_u32(0x3f01);

#[derive(Debug, Snafu)]
#[snafu(module)]
pub(crate) enum DriverProtocolError {
    #[snafu(display("stop ack code {actual} does not match committed stop code {expected}"))]
    StopAckCodeMismatch { expected: VarInt, actual: VarInt },
    #[snafu(display("reset ack code {actual} does not match committed reset code {expected}"))]
    ResetAckCodeMismatch { expected: VarInt, actual: VarInt },
    #[snafu(display("received frame that is invalid for the current reader operation"))]
    UnexpectedReaderFrame,
    #[snafu(display("received writer ack without a matching committed operation"))]
    UnexpectedWriterAck,
    #[snafu(display("received duplicate writer credit before previous credit was consumed"))]
    DuplicateWriterCredit,
    #[snafu(display("start_send called without writer credit"))]
    StartSendWithoutCredit,
    #[snafu(display("typed frame stream ended before operation completed"))]
    FrameEof,
}

pub(crate) enum DeferredStreamError {
    Ready {
        error: quic::StreamError,
    },
    Pending {
        future: BoxFuture<'static, quic::StreamError>,
    },
}

pub(crate) fn protocol_connection_error(error: DriverProtocolError) -> quic::ConnectionError {
    quic::ConnectionError::Transport {
        source: quic::TransportError {
            kind: DRIVER_PROTOCOL_ERROR_KIND,
            frame_type: DRIVER_PROTOCOL_ERROR_FRAME_TYPE,
            // Lossy: QUIC transport error reason is a protocol string field for
            // local bridge-driver protocol failures.
            reason: Cow::Owned(error.to_string()),
        },
    }
}

pub(crate) fn latch_protocol_error<L>(
    lifecycle: &L,
    error: DriverProtocolError,
) -> quic::StreamError
where
    L: LifecycleExt,
{
    let source = lifecycle
        .latch()
        .latch_with(|| protocol_connection_error(error));
    quic::StreamError::Connection { source }
}

pub(crate) fn stream_bridge_eof() -> quic::StreamError {
    quic::StreamError::Reset {
        code: Code::H3_REQUEST_CANCELLED.into_inner(),
    }
}

pub(crate) fn latch_frame_io_error<L, E>(lifecycle: &L, error: E) -> quic::StreamError
where
    L: LifecycleExt,
    quic::ConnectionError: From<E>,
{
    let source = lifecycle
        .latch()
        .latch_with(|| quic::ConnectionError::from(error));
    quic::StreamError::Connection { source }
}

pub(crate) fn defer_err_conn<L>(lifecycle: Arc<L>) -> DeferredStreamError
where
    L: LifecycleExt + 'static,
{
    match quic::Lifecycle::check(lifecycle.as_ref()) {
        Ok(()) => DeferredStreamError::Pending {
            future: Box::pin(async move {
                let source = quic::Lifecycle::closed(lifecycle.as_ref()).await;
                quic::StreamError::Connection { source }
            }),
        },
        Err(source) => DeferredStreamError::Ready {
            error: quic::StreamError::Connection { source },
        },
    }
}