sark-h2 0.28.2

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

use o3::depot::raw::RingEdit;
use sark_core::{
    life::session,
    manifold::{server::drain, stream::Boundary},
};
use sark_http::field::metadata::section::{rules, state::Selected};

use crate::conn::{event, machine::ready::Incoming, registry::store::Notice};

impl<R, F, C, A, E> AsRef<event::Event<Selected<F>, F::Block, C, E>> for Incoming<'_, R, F, C, A, E>
where
    R: crate::Role,
    F: rules::Policy,
    C: session::Storage,
    A: rules::Push,
    E: event::Evidence,
{
    fn as_ref(&self) -> &event::Event<Selected<F>, F::Block, C, E> {
        let queue = &self.0.ingress.events.queue;
        // SAFETY: construction admitted the nonempty queue; this exclusive
        // connection loan prevents any mutation of its initialized front.
        unsafe { queue.as_slices().0.get_unchecked(0) }
    }
}

impl<R, F, C, A, E> drain::Ready for Incoming<'_, R, F, C, A, E>
where
    R: crate::Role,
    F: rules::Policy,
    C: session::Storage,
    A: rules::Push,
    E: event::Evidence,
{
    type Item = event::Event<Selected<F>, F::Block, C, E>;

    fn take(self) -> Self::Item {
        // SAFETY: this consumes the same exclusive nonempty connection loan
        // admitted by Incoming::new; no operation can invalidate its front.
        let event = unsafe { self.0.ingress.events.queue.front_raw() }.remove();
        if self.0.state.lifecycle.keeps_events() {
            let terminal = match &event {
                event::Event::Headers {
                    stream_id,
                    boundary: Boundary::End,
                    ..
                }
                | event::Event::Data {
                    stream_id,
                    boundary: Boundary::End,
                    ..
                } => Some((*stream_id, Notice::End)),
                event::Event::StreamReset { stream_id, .. } => Some((*stream_id, Notice::Reset)),
                _ => None,
            };
            if let Some((key, notice)) = terminal {
                self.0.state.active_streams.store.release(key, notice);
            }
        }
        event
    }
}