sark-h2 0.25.0

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

use std::pin;

use dope::{
    core::driver::{route::Identity, schedule::ready::task::owner},
    manifold::network::listener::conn::Id,
    net::wire,
};
use sark_core::{
    life,
    manifold::{server::app as semantic, task::app::Runtime},
};

use crate::server::{
    runtime::output::flush,
    task::{Context, Retired, *},
};

pub(in crate::server::task::app) struct Close<'a, 'd, ID: Identity + 'd, R, M>
where
    R: Runtime<'d, Target = Target<'d, ID>>,
{
    pub(super) router: pin::Pin<&'a mut R>,
    pub(super) tasks: &'a mut registry::Table<'d, ID, R::Task>,
    pub(super) mode: &'a mut M,
}

impl<'d, ID: Identity + 'd, R, M> Close<'_, 'd, ID, R, M>
where
    R: Runtime<'d, Target = Target<'d, ID>>,
{
    pub(in crate::server::task::app) fn new<'a>(
        router: pin::Pin<&'a mut R>,
        tasks: &'a mut registry::Table<'d, ID, R::Task>,
        mode: &'a mut M,
    ) -> Close<'a, 'd, ID, R, M> {
        Close {
            router,
            tasks,
            mode,
        }
    }

    pub(in crate::server::task::app) fn apply<W, D, A>(
        self,
        conn: &mut Context<'_, '_, '_, '_, '_, 'd, ID, R, W, M, D, A>,
    ) where
        W: wire::Spec,
        M: Transport<'d, ID>,
        M::Output: flush::Evidence<'d>,
        D: timers::Policy<'d>,
        A: service::Select<R> + semantic::Application<'d, service::Connection<'d, A>, Id<'d, ID>>,
    {
        let Self {
            mut router,
            tasks,
            mode,
        } = self;
        let id = conn.id();
        mode.observe(life::Event::SessionClosing(id));
        let state = conn.state_mut();
        let index = &mut state.task_index;
        if !index.is_empty() {
            for at in 0..index.capacity() {
                let Some(stream) = index.key_at(at) else {
                    continue;
                };
                let Some(taken) = tasks.take_stream(index, stream) else {
                    continue;
                };
                let (_, stream, task, unreported) = taken.into_parts();
                Runtime::cancel(router.as_mut(), task);
                if unreported {
                    mode.observe(life::Event::TaskCancelled(identify(id, stream)));
                }
                if state.book.responses.entry(stream).is_none() {
                    identify(id, stream).retire(mode, life::Outcome::Cancelled);
                }
            }
            index.clear();
        }
        if let Err(cause) = Runtime::try_retire(router.as_ref(), owner::Owner::from(id)) {
            life::Mode::diagnostic(mode, &cause);
        }
        state.turn().close(id, mode);
    }

    pub(in crate::server::task::app) fn audit<W, D, A>(
        self,
        conn: &mut Retired<'_, '_, 'd, ID, R, W, M, D, A>,
    ) where
        W: wire::Spec,
        M: Transport<'d, ID>,
        M::Output: flush::Evidence<'d>,
        D: timers::Policy<'d>,
        A: service::Select<R> + semantic::Application<'d, service::Connection<'d, A>, Id<'d, ID>>,
    {
        let id = conn.id();
        let bytes = conn.response_len();
        let state = conn.state_mut();
        let count = state.task_index.len();
        let empty = state.task_index.is_empty();
        life::Mode::close(self.mode, id, || {
            state.turn().audit(count, empty).wire_bytes(bytes)
        });
    }
}