sark-h2 0.22.0

Fiber-free HTTP/2 protocol primitives and manifold server for sark
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
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143

use std::ops;

use dope::{core::driver::route::Identity, manifold::network::listener::conn::Id};
use o3::buffer::hold;
use sark_core::{
    decode::request,
    duplex, life,
    manifold::{schedule, schedule::typed, stream::Boundary, task::app::Runtime},
};
use sark_http::payload::response::status::Code;

use crate::{
    conn,
    server::{
        response::{Head, Receipt, Status},
        state,
        task::{
            self, response,
            response::{Progress, input},
        },
    },
    stream,
};

pub(super) struct Opening<'a, 'b, 'd, ID: Identity, R, M>
where
    R: Runtime<'d>,
{
    pub(super) driver: &'a mut response::Driver<'b, 'd, ID, R, M>,
}

impl<'d, ID: Identity + 'd, R, M> Opening<'_, '_, 'd, ID, R, M>
where
    R: Runtime<'d, Target = task::Target<'d, ID>>,
    M: state::Transitions<Id<'d, ID>>,
{
    pub(super) fn open(
        self,
        stream: stream::Key,
        request: input::Request<'d, R::Prepared>,
    ) -> Progress<'d, R> {
        let app = self.driver.router.as_ref().get_ref();
        let input = match request.into_body() {
            request::Body::Open(request) => match duplex::Application::try_open(app, request) {
                Ok((session, body)) => input::Input::open(session, body),
                Err((error, ..)) => return self.reject(stream, error),
            },
            request::Body::Finished(request) => match duplex::Application::try_open(app, request) {
                Ok((session, body)) => input::Input::finished(session, body),
                Err((error, ..)) => return self.reject(stream, error),
            },
        };
        let head = Status::new(Code::OK)
            .prepare(self.driver.connection, stream, Boundary::More)
            .map_err(conn::Error::code)?;
        self.resume(stream, head, input)
    }

    pub(super) fn resume(
        mut self,
        stream: stream::Key,
        head: Head<'d>,
        input: input::Input<'d, R::Duplex>,
    ) -> Progress<'d, R> {
        match head
            .drive(self.driver.connection)
            .map_err(conn::Error::code)?
        {
            ops::ControlFlow::Break(head) => Ok(typed::Control::Control {
                value: response::Response::DuplexHead { head, input },
                turn: Self::blocked(),
            }),
            ops::ControlFlow::Continue(receipt) => {
                self.committed(stream, receipt);
                Ok(typed::Control::Data {
                    value: response::Data::Duplex { input },
                    turn: Self::progress(),
                })
            }
        }
    }

    fn reject(
        mut self,
        stream: stream::Key,
        error: duplex::OpenError<R::Error<'d>>,
    ) -> Progress<'d, R> {
        let status = match error {
            duplex::OpenError::InvalidRequest(error) => R::reject(error).status(),
            duplex::OpenError::Length => Code::BAD_REQUEST,
            duplex::OpenError::Unsupported => Code::INTERNAL_SERVER_ERROR,
        };
        let head = Status::new(status)
            .prepare(self.driver.connection, stream, Boundary::End)
            .map_err(conn::Error::code)?;
        match head
            .drive(self.driver.connection)
            .map_err(conn::Error::code)?
        {
            ops::ControlFlow::Break(head) => Ok(typed::Control::Control {
                value: response::Response::Head {
                    head,
                    outcome: life::Outcome::Rejected,
                    count: response::Count::Pending,
                },
                turn: Self::blocked(),
            }),
            ops::ControlFlow::Continue(receipt) => {
                self.committed(stream, receipt);
                task::identify(self.driver.connection_id, stream)
                    .retire(self.driver.mode, life::Outcome::Rejected);
                Ok(typed::Control::Complete)
            }
        }
    }

    fn committed(&mut self, stream: stream::Key, receipt: Receipt<'d>) {
        let (_, linked, copied) = receipt.into_parts();
        let exchange = task::identify(self.driver.connection_id, stream);
        if let Some(copied) = copied {
            self.driver.mode.observe(life::Event::ResponseBodyCopied {
                exchange,
                read: copied.read,
                written: copied.written,
                cause: copied.cause,
            });
        }
        self.driver.mode.observe(life::Event::ResponseCommitted {
            exchange,
            body_bytes: 0,
            charge: hold::Charge::default(),
            linked,
        });
    }

    fn blocked() -> schedule::Turn<conn::DataBlock> {
        schedule::Turn::Blocked(conn::DataBlock::Egress)
    }

    fn progress() -> schedule::Turn<conn::DataBlock> {
        schedule::Turn::Progress
    }
}