sark-h2 0.29.5

Fiber-free HTTP/2 protocol primitives and manifold server for sark
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
117
118
119

use core::ops::ControlFlow;

use o3::buffer::storage;
use sark_core::{life::session, manifold::stream};
use sark_http::{field, field::metadata::section::rules};

use crate::{
    ServerRole,
    conn::{self, event, machine},
    stream::Key,
};

/// Exclusive write turn for one H2 server stream.
pub struct Writer<'a, F, C, E>
where
    F: rules::Policy,
    C: session::Storage,
    E: event::Evidence,
{
    connection: &'a mut machine::Connection<ServerRole, F, C, rules::RejectPush, E>,
    key: Key,
}

impl<'a, F, C, E> Writer<'a, F, C, E>
where
    F: rules::Policy,
    C: session::Storage,
    E: event::Evidence,
{
    pub(in crate::conn::machine) fn new(
        conn: &'a mut machine::Connection<ServerRole, F, C, rules::RejectPush, E>,
        key: Key,
    ) -> Self {
        Self {
            connection: conn,
            key,
        }
    }
}

impl<F, C, E> stream::Write<storage::Shared> for Writer<'_, F, C, E>
where
    F: rules::Policy,
    C: session::Storage,
    E: event::Evidence,
{
    type Block = conn::DataBlock;
    type Error = conn::Error;

    fn headers<S: field::Source>(
        &mut self,
        source: S,
        boundary: stream::Boundary,
    ) -> Result<stream::Control<Self::Block>, Self::Error> {
        match self
            .connection
            .streams()
            .headers()
            .try_send(self.key, source, boundary)
        {
            Ok(ControlFlow::Continue(())) => Ok(stream::Control::Written),
            Ok(ControlFlow::Break(_)) => Ok(stream::Control::Prepared(conn::DataBlock::Egress)),
            Err(error) if error.is_pressure() => {
                Ok(stream::Control::Blocked(conn::DataBlock::Egress))
            }
            Err(error) => Err(error),
        }
    }

    fn data(
        &mut self,
        prefix: &[u8],
        payload: &storage::Shared,
        offset: usize,
        boundary: stream::Boundary,
    ) -> Result<stream::Data<Self::Block>, Self::Error> {
        let head_len = prefix.len();
        let data_len = payload.len();
        let (prefix, body) = if offset < prefix.len() {
            (&prefix[offset..], 0)
        } else {
            (&[][..], offset - prefix.len())
        };
        if prefix.is_empty() && body >= payload.len() {
            return Err(conn::Error::DataRange {
                offset,
                prefix: head_len,
                body: data_len,
            });
        }
        let outcome = self
            .connection
            .streams()
            .retained()
            .try_parts(self.key, prefix, payload, body, boundary)?;
        match outcome {
            conn::DataOutcome::Sent(written) => Ok(stream::Data::Written(written)),
            conn::DataOutcome::Ended => Err(conn::Error::DataRange {
                offset,
                prefix: head_len,
                body: data_len,
            }),
            conn::DataOutcome::Blocked(block) => Ok(stream::Data::Blocked(block)),
        }
    }

    fn commit(&mut self) -> Result<ControlFlow<Self::Block>, Self::Error> {
        Ok(
            match self.connection.streams().headers().try_commit(self.key)? {
                ControlFlow::Continue(()) => ControlFlow::Continue(()),
                ControlFlow::Break(_) => ControlFlow::Break(conn::DataBlock::Egress),
            },
        )
    }

    fn discard(&mut self) {
        self.connection.streams().headers().discard(self.key);
    }
}