h3x 0.2.0

High-performance zero-copy DHTTP/3 implementation
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
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116

use std::pin::pin;

use http_body::Body;
use http_body_util::BodyExt;
use snafu::{ResultExt, Snafu};

use super::{MessageStreamError, WriteStream};
use crate::qpack::field::hyper::{
    header_map_to_field_lines, hyper_request_parts_to_field_lines,
    hyper_response_parts_to_field_lines,
};

#[derive(Debug, Snafu)]
#[snafu(module, visibility(pub(crate)))]
pub enum SendMessageError<E: std::error::Error + 'static> {
    #[snafu(display("failed to send message on stream"))]
    Stream { source: MessageStreamError },
    #[snafu(display("failed to read body frame"))]
    Body { source: E },
}

impl<E: std::error::Error + 'static> SendMessageError<E> {
    pub fn map_body_error<E1: std::error::Error + 'static>(
        self,
        f: impl FnOnce(E) -> E1,
    ) -> SendMessageError<E1> {
        match self {
            SendMessageError::Stream { source } => SendMessageError::Stream { source },
            SendMessageError::Body { source } => SendMessageError::Body { source: f(source) },
        }
    }
}

impl WriteStream {
    pub(crate) async fn send_hyper_body<B: Body>(
        &mut self,
        body: B,
    ) -> Result<(), SendMessageError<B::Error>>
    where
        B::Data: Send,
        B::Error: std::error::Error + 'static,
    {
        let mut body = pin!(body);
        while let Some(frame) = body.frame().await {
            let frame = frame.context(send_message_error::BodySnafu)?;
            let frame = match frame.into_data() {
                Ok(data) => {
                    self.send_data(data)
                        .await
                        .context(send_message_error::StreamSnafu)?;
                    continue;
                }
                Err(frame) => frame,
            };
            let frame = match frame.into_trailers() {
                Ok(trailers) => {
                    self.send_header(header_map_to_field_lines(trailers))
                        .await
                        .context(send_message_error::StreamSnafu)?;
                    break;
                }
                Err(frame) => frame,
            };

            tracing::warn!("ignore unknown http body frame");
            _ = frame;
        }
        Ok(())
    }

    pub async fn send_hyper_request_parts(
        &mut self,
        parts: http::request::Parts,
    ) -> Result<(), MessageStreamError> {
        self.send_header(hyper_request_parts_to_field_lines(parts))
            .await
    }

    pub async fn send_hyper_request<B: Body>(
        &mut self,
        request: http::Request<B>,
    ) -> Result<(), SendMessageError<B::Error>>
    where
        B::Data: Send,
        B::Error: std::error::Error + 'static,
    {
        let (parts, body) = request.into_parts();
        self.send_header(hyper_request_parts_to_field_lines(parts))
            .await
            .context(send_message_error::StreamSnafu)?;
        self.send_hyper_body(body).await
    }

    pub async fn send_hyper_response_parts(
        &mut self,
        parts: http::response::Parts,
    ) -> Result<(), MessageStreamError> {
        self.send_header(hyper_response_parts_to_field_lines(parts))
            .await
    }

    pub async fn send_hyper_response<B: Body>(
        &mut self,
        response: http::Response<B>,
    ) -> Result<(), SendMessageError<B::Error>>
    where
        B::Data: Send,
        B::Error: std::error::Error + 'static,
    {
        let (parts, body) = response.into_parts();
        self.send_header(hyper_response_parts_to_field_lines(parts))
            .await
            .context(send_message_error::StreamSnafu)?;
        self.send_hyper_body(body).await
    }
}