dope-session 0.2.3

Thin io_uring adaptor with "Manifolds"
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
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

use crate::CodecLayer;
use crate::protocol::client::ClientProtocol;
use dope_core::driver::{DriverOps, PoolDriver, ProvidedRingOps};
use dope_core::profile::Profile;
use dope_transport::Transport;

use super::super::Pool;
use super::super::core::RecvDispatch;
use super::super::slot::ConnFlags;
use super::Client;
use super::connect_source::ConnectSource;
use crate::SlotId;

impl<P, T, S, F, B> super::super::PoolRecv<B> for Pool<Client<P, T, S>, F, B>
where
    P: ClientProtocol,
    T: Transport,
    T::Addr: Clone,
    S: ConnectSource<T>,
    F: Profile,
    B: PoolDriver,
{
    fn on_recv_event(
        &mut self,
        driver: &mut B::Driver,
        token: B::Token,
        result: i32,
        more: bool,
        bid: Option<u16>,
    ) {
        match self.core.classify_recv(driver, token, result, more, bid) {
            RecvDispatch::Skipped => {}
            RecvDispatch::EofClose { slot_idx } => {
                self.disconnect_and_free(driver, slot_idx);
            }
            RecvDispatch::Errno {
                slot_idx,
                errno,
                more,
            } => {
                if errno == libc::ECANCELED {
                    if !more {
                        self.core.slots[slot_idx].io.recv.armed = false;
                    }
                    return;
                }
                self.core.slots[slot_idx].io.recv.armed = false;
                self.disconnect_and_free(driver, slot_idx);
            }
            RecvDispatch::Chunk {
                slot_idx,
                ptr,
                len,
                bid,
                more,
            } => {
                let src: &[u8] = unsafe { std::slice::from_raw_parts(ptr, len) };
                if <P::CodecLayer as CodecLayer>::IS_PASSTHROUGH {
                    self.ingest_response_bytes(slot_idx, src);
                } else {
                    self.ingest_codec_responses(driver, slot_idx, src);
                }
                driver.provided_ring().defer(bid);
                if !self.core.slots[slot_idx]
                    .conn
                    .flags
                    .contains(ConnFlags::LIVE)
                {
                    return;
                }
                self.try_submit_outbound(driver, slot_idx);
                if !more {
                    let _ = self.core.arm_recv(driver, slot_idx);
                }
            }
            RecvDispatch::DrainOnly { slot_idx, more } => {
                if !more {
                    let _ = self.core.arm_recv(driver, slot_idx);
                }
            }
        }
    }
}

impl<P, T, S, F, B> Pool<Client<P, T, S>, F, B>
where
    P: ClientProtocol,
    T: Transport,
    T::Addr: Clone,
    S: ConnectSource<T>,
    F: Profile,
    B: PoolDriver,
{
    #[cold]
    fn ingest_codec_responses(&mut self, driver: &mut B::Driver, slot_idx: usize, src: &[u8]) {
        let Some(plaintext) = self.core.process_codec_inbound(driver, slot_idx, src) else {
            return;
        };
        self.ingest_response_bytes(slot_idx, &plaintext);
        self.core.install_plaintext(slot_idx, plaintext);
    }

    fn ingest_response_bytes(&mut self, slot_idx: usize, src: &[u8]) {
        let conn_id =
            SlotId::from_parts(slot_idx as u32, self.core.slots[slot_idx].conn.generation);
        let entry = &mut self.core.slots[slot_idx];
        let user = entry.user.get_mut();
        let codec_state = entry.codec_state.get_mut();
        user.session
            .ingest(src, &mut self.direction.protocol, conn_id, codec_state);
    }
}