dope_session/pool/client/
recv.rs

1use crate::CodecLayer;
2use crate::protocol::client::ClientProtocol;
3use dope_core::driver::{DriverOps, PoolDriver, ProvidedRingOps};
4use dope_core::profile::Profile;
5use dope_transport::Transport;
6
7use super::super::Pool;
8use super::super::core::RecvDispatch;
9use super::super::slot::ConnFlags;
10use super::Client;
11use super::connect_source::ConnectSource;
12use crate::SlotId;
13
14impl<P, T, S, F, B> super::super::PoolRecv<B> for Pool<Client<P, T, S>, F, B>
15where
16    P: ClientProtocol,
17    T: Transport,
18    T::Addr: Clone,
19    S: ConnectSource<T>,
20    F: Profile,
21    B: PoolDriver,
22{
23    fn on_recv_event(
24        &mut self,
25        driver: &mut B::Driver,
26        token: B::Token,
27        result: i32,
28        more: bool,
29        bid: Option<u16>,
30    ) {
31        match self.core.classify_recv(driver, token, result, more, bid) {
32            RecvDispatch::Skipped => {}
33            RecvDispatch::EofClose { slot_idx } => {
34                self.disconnect_and_free(driver, slot_idx);
35            }
36            RecvDispatch::Errno {
37                slot_idx,
38                errno,
39                more,
40            } => {
41                if errno == libc::ECANCELED {
42                    if !more {
43                        self.core.slots[slot_idx].io.recv.armed = false;
44                    }
45                    return;
46                }
47                self.core.slots[slot_idx].io.recv.armed = false;
48                self.disconnect_and_free(driver, slot_idx);
49            }
50            RecvDispatch::Chunk {
51                slot_idx,
52                ptr,
53                len,
54                bid,
55                more,
56            } => {
57                let src: &[u8] = unsafe { std::slice::from_raw_parts(ptr, len) };
58                if <P::CodecLayer as CodecLayer>::IS_PASSTHROUGH {
59                    self.ingest_response_bytes(slot_idx, src);
60                } else {
61                    self.ingest_codec_responses(driver, slot_idx, src);
62                }
63                driver.provided_ring().defer(bid);
64                if !self.core.slots[slot_idx]
65                    .conn
66                    .flags
67                    .contains(ConnFlags::LIVE)
68                {
69                    return;
70                }
71                self.try_submit_outbound(driver, slot_idx);
72                if !more {
73                    let _ = self.core.arm_recv(driver, slot_idx);
74                }
75            }
76            RecvDispatch::DrainOnly { slot_idx, more } => {
77                if !more {
78                    let _ = self.core.arm_recv(driver, slot_idx);
79                }
80            }
81        }
82    }
83}
84
85impl<P, T, S, F, B> Pool<Client<P, T, S>, F, B>
86where
87    P: ClientProtocol,
88    T: Transport,
89    T::Addr: Clone,
90    S: ConnectSource<T>,
91    F: Profile,
92    B: PoolDriver,
93{
94    #[cold]
95    fn ingest_codec_responses(&mut self, driver: &mut B::Driver, slot_idx: usize, src: &[u8]) {
96        let Some(plaintext) = self.core.process_codec_inbound(driver, slot_idx, src) else {
97            return;
98        };
99        self.ingest_response_bytes(slot_idx, &plaintext);
100        self.core.install_plaintext(slot_idx, plaintext);
101    }
102
103    fn ingest_response_bytes(&mut self, slot_idx: usize, src: &[u8]) {
104        let conn_id =
105            SlotId::from_parts(slot_idx as u32, self.core.slots[slot_idx].conn.generation);
106        let entry = &mut self.core.slots[slot_idx];
107        let user = entry.user.get_mut();
108        let codec_state = entry.codec_state.get_mut();
109        user.session
110            .ingest(src, &mut self.direction.protocol, conn_id, codec_state);
111    }
112}
dope_session/pool/client/recv.rs

dope_session/pool/client/
recv.rs
