sark-h2 0.29.1

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 std::ops;

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

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

impl<'d, ID: Identity + 'd, R, M, F, P> response::Driver<'_, 'd, ID, R, M, F, P>
where
    R: Runtime<'d, Target = task::Target<'d, ID>>,
    M: state::Transitions<Id<'d, ID>>,
    M::Retention: Handoff,
    F: rules::Policy,
    P: FnMut(Id<'d, ID>, stream::Key, &dyn std::fmt::Debug),
{
    pub(super) fn open(
        &mut self,
        stream: stream::Key,
        request: input::Request<'d, R::Prepared>,
    ) -> Progress<'d, R> {
        let app = self.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, input::Buf::new(body)),
                Err((error, ..)) => return self.reject_open(stream, error),
            },
            request::Body::Finished(request) => match duplex::Application::try_open(app, request) {
                Ok((session, body)) => input::Input::finished(session, input::Buf::new(body)),
                Err((error, ..)) => return self.reject_open(stream, error),
            },
        };
        let head = Status::new(Code::OK)
            .prepare(self.connection, stream, Boundary::More)
            .map_err(|error| self.error(stream, error))?;
        self.connection.hold_duplex(stream);
        self.resume_open(stream, head, input)
    }

    pub(super) fn resume_open(
        &mut self,
        stream: stream::Key,
        head: Head<'d>,
        input: input::Input<'d, R::Duplex>,
    ) -> Progress<'d, R> {
        let progress = match head.drive(self.connection) {
            Ok(progress) => progress,
            Err(error) => {
                drop(input);
                self.connection.release_duplex(stream);
                return Err(self.error(stream, error));
            }
        };
        match progress {
            ops::ControlFlow::Break(head) => Ok(typed::Control::Control {
                value: response::Response::DuplexHead { head, input },
                turn: schedule::Turn::Blocked(conn::DataBlock::Egress),
            }),
            ops::ControlFlow::Continue(receipt) => {
                drop(receipt.observe(self.mode, task::identify(self.connection_id, stream)));
                Ok(typed::Control::Data {
                    value: response::Data::Duplex { input },
                    turn: schedule::Turn::Progress,
                })
            }
        }
    }

    fn reject_open(
        &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.connection, stream, Boundary::End)
            .map_err(|error| self.error(stream, error))?;
        match head
            .drive(self.connection)
            .map_err(|error| self.error(stream, error))?
        {
            ops::ControlFlow::Break(head) => Ok(typed::Control::Control {
                value: response::Response::Head {
                    head,
                    state: sark_core::manifold::task::response::State::Complete(
                        life::Outcome::Rejected,
                    ),
                    count: response::Count::Pending,
                },
                turn: schedule::Turn::Blocked(conn::DataBlock::Egress),
            }),
            ops::ControlFlow::Continue(receipt) => {
                let exchange = task::identify(self.connection_id, stream);
                drop(receipt.observe(self.mode, exchange));
                exchange.retire(self.mode, life::Outcome::Rejected);
                Ok(typed::Control::Complete)
            }
        }
    }
}