sark-h2 0.23.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

use dope::{core::driver::route::Identity, manifold::network::listener::conn};
use sark_core::{
    life,
    manifold::{
        schedule::{self, typed},
        task::{
            app::{Handoff, Runtime},
            response::{Outcome, Output, Pack},
        },
    },
    route,
};

use crate::{
    server::{
        self, state,
        task::{response, *},
    },
    stream,
};

pub(super) struct Completion<'a, 'state, 'd, ID: Identity + 'd, A, M>
where
    A: Runtime<'d, Target = Target<'d, ID>>,
{
    pub(super) tasks: &'a mut registry::Table<'d, ID, A::Task>,
    pub(super) connection: conn::Id<'d, ID>,
    pub(super) task_index: &'a mut registry::Index,
    pub(super) state: &'a mut Borrowed<'state, 'd, A>,
    pub(super) limits: server::Config,
    pub(super) mode: &'a mut M,
}

impl<'a, 'state, 'd, ID: Identity + 'd, A, M> Handoff<'d, Target<'d, ID>, A::Completion>
    for Completion<'a, 'state, 'd, ID, A, M>
where
    A: Runtime<'d, Target = Target<'d, ID>>,
    M: state::Transitions<conn::Id<'d, ID>>,
{
    type Complete = schedule::InsertResult<stream::Key, Next<'d, A>>;

    fn pending(self, target: Target<'d, ID>) {
        if !M::ENABLED {
            return;
        }
        let Some((conn, stream)) = self.tasks.owner(target) else {
            return;
        };
        if conn != self.connection {
            return;
        }
        self.mode
            .observe(life::Event::TaskPollPending(identify(conn, stream)));
    }

    fn complete<R, S, P>(self, target: Target<'d, ID>, output: Output<'d, R, S>) -> Self::Complete
    where
        R: route::Task<'d, S>,
        P: Pack<'d, R, S, A::Completion>,
    {
        let conn = self.connection;
        let Some(taken) = self.tasks.take_target(self.task_index, target, conn) else {
            return Ok(());
        };
        let registry::Taken { stream, .. } = taken;
        let call = identify(conn, stream);
        self.mode.observe(life::Event::TaskPollReady(call));
        let value = match output.into_parts() {
            (.., Outcome::Rejected(reject)) => response::Response::Status(reject.status()),
            (input, Outcome::Response(output)) => {
                response::Response::Completion(P::pack(Output::response(input, output)))
            }
        };
        let next = self.state.activation(self.limits);
        match self
            .state
            .responses
            .try_insert(stream, typed::Next::Control(value, next))
        {
            Ok(()) => {
                state::Transitions::ready(self.mode, call);
                Ok(())
            }
            Err(failure) => Err(failure),
        }
    }
}