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

use std::pin;

use dope::{core::driver::route::Identity, manifold::network::listener::conn};
use sark_core::{deadline, life, manifold::task::app::Runtime};
use sark_http::{field::metadata::section::rules, owner::cause};

use crate::{
    frame::numbers,
    server::{
        state::Transitions,
        task::{
            self, Borrowed, Target, ingress, registry,
            response::{self, deadline::Deadline},
            timers::Due,
        },
    },
};

pub(in crate::server::task) fn apply<'d, ID: Identity + 'd, R, Q, F>(
    due: deadline::Poll<Due>,
    router: pin::Pin<&mut R>,
    tasks: &mut registry::Table<'d, ID, R::Task>,
    conn: conn::Id<'d, ID>,
    index: &mut registry::Index,
    state: &mut Borrowed<'_, 'd, R, F>,
    mode: &mut Q,
) -> Deadline
where
    R: Runtime<'d, Target = Target<'d, ID>>,
    Q: Transitions<conn::Id<'d, ID>>,
    F: rules::Policy,
{
    let due = match due {
        deadline::Poll::Idle => return Deadline::Ready,
        deadline::Poll::Renewed => return Deadline::Wake,
        deadline::Poll::Expired(due) => due,
    };
    match due {
        Due::Exchange { key, phase } => {
            let xchg = task::identify(conn, key);
            let mut expiry = None;
            if let Some(taken) = tasks.take_stream(index, key) {
                expiry = Some(xchg.begin_expiry(mode, phase));
                Runtime::cancel(router, taken.task);
                if taken.unreported {
                    mode.observe(life::Event::TaskCancelled(xchg));
                }
            }
            let discarded = {
                let expiry = &mut expiry;
                let mode = &mut *mode;
                state.data().discard(xchg, move || {
                    expiry.get_or_insert_with(|| xchg.begin_expiry(mode, phase));
                    mode.body(xchg)
                })
            };
            let code = match discarded {
                ingress::Discard::Missing(_) | ingress::Discard::Removed(_) => {
                    numbers::ErrorCode::Cancel
                }
                ingress::Discard::Failed { error, .. } => error.code(),
            };
            let reply = state.responses.remove(key);
            if reply.is_some() {
                expiry.get_or_insert_with(|| xchg.begin_expiry(mode, phase));
            }
            let done = if expiry.is_some() {
                state.connection.streams().retire(key, code)
            } else {
                mode.cancel(key);
                state.connection.cancel_ingress(key, code)
            };
            if let Some(write) = done {
                task::release_retired(mode, conn, write, cause::Staging::Reset);
            }
            if let Some(reply) = reply {
                response::discard(reply, state.connection, xchg, mode, cause::Staging::Reset);
            }
            if let Some(expiry) = expiry {
                expiry.finish(mode);
            }
            Deadline::Expired
        }
        Due::Critical => Deadline::Critical,
    }
}