dope_session/
session.rs

1use dambi::Bytes;
2
3use crate::CodecLayer;
4use crate::SlotId;
5use crate::pool::client::state::CLIENT_IOV_CAP;
6use crate::protocol::client::ClientProtocol;
7use crate::ring::frame::FrameRing;
8use crate::ring::send::SendRing;
9
10pub struct Egress<'a, L: CodecLayer> {
11    inner: &'a mut SendRing<CLIENT_IOV_CAP>,
12    codec_state: &'a mut L::ConnState,
13}
14
15impl<'a, L: CodecLayer> Egress<'a, L> {
16    #[inline(always)]
17    pub fn push(&mut self, bytes: Bytes) {
18        self.inner.push(bytes);
19    }
20
21    #[inline(always)]
22    pub fn is_empty(&self) -> bool {
23        self.inner.is_empty()
24    }
25
26    /// Mutable access to the paired codec's connection state.
27    ///
28    /// The returned borrow is reborrowed from `&mut self`, so it must be
29    /// released (use ends) before any other `Egress` method can be called.
30    /// Typical usage is sequential: trigger a codec transition, then enqueue.
31    ///
32    /// For the [`Plain`](crate::Plain) layer, `L::ConnState = ()` and this
33    /// returns a ZST `&mut ()` — zero overhead.
34    #[inline(always)]
35    pub fn codec_state(&mut self) -> &mut L::ConnState {
36        self.codec_state
37    }
38}
39
40pub struct ClientSessionGeneric<P: ClientProtocol, const ING: usize, const IOV: usize> {
41    pub(crate) ingress: FrameRing<ING>,
42    pub(crate) egress: SendRing<IOV>,
43    pub(crate) state: P::ConnState,
44}
45
46impl<P: ClientProtocol, const ING: usize, const IOV: usize> Default
47    for ClientSessionGeneric<P, ING, IOV>
48{
49    #[inline(always)]
50    fn default() -> Self {
51        Self {
52            ingress: FrameRing::new(),
53            egress: SendRing::new(),
54            state: P::ConnState::default(),
55        }
56    }
57}
58
59impl<P: ClientProtocol, const ING: usize> ClientSessionGeneric<P, ING, CLIENT_IOV_CAP> {
60    #[inline(always)]
61    pub fn enqueue(&mut self, bytes: Bytes) {
62        self.egress.push(bytes);
63    }
64
65    #[inline(always)]
66    pub fn state(&self) -> &P::ConnState {
67        &self.state
68    }
69
70    #[inline(always)]
71    pub fn state_mut(&mut self) -> &mut P::ConnState {
72        &mut self.state
73    }
74
75    #[inline(always)]
76    pub fn egress_is_empty(&self) -> bool {
77        self.egress.is_empty()
78    }
79
80    pub(crate) fn on_connect(
81        &mut self,
82        proto: &mut P,
83        conn_id: SlotId,
84        codec_state: &mut <P::CodecLayer as CodecLayer>::ConnState,
85    ) {
86        let Self { state, egress, .. } = self;
87        proto.on_connect(
88            conn_id,
89            state,
90            Egress {
91                inner: egress,
92                codec_state,
93            },
94        );
95    }
96
97    pub(crate) fn on_disconnect(
98        &mut self,
99        proto: &mut P,
100        conn_id: SlotId,
101        codec_state: &mut <P::CodecLayer as CodecLayer>::ConnState,
102    ) {
103        let Self { state, egress, .. } = self;
104        proto.on_disconnect(
105            conn_id,
106            state,
107            Egress {
108                inner: egress,
109                codec_state,
110            },
111        );
112    }
113
114    pub(crate) fn ingest(
115        &mut self,
116        src: &[u8],
117        proto: &mut P,
118        conn_id: SlotId,
119        codec_state: &mut <P::CodecLayer as CodecLayer>::ConnState,
120    ) {
121        let _ = self.ingress.append(src);
122        let Self {
123            ingress,
124            state,
125            egress,
126            ..
127        } = self;
128        ingress.drain(|buf| {
129            let (head, consumed) = proto.parse(state, buf)?;
130            proto.on_response(
131                conn_id,
132                state,
133                head,
134                Egress {
135                    inner: &mut *egress,
136                    codec_state: &mut *codec_state,
137                },
138            );
139            Some(consumed)
140        });
141    }
142
143    #[inline(always)]
144    pub(crate) fn fill_msghdr(&mut self, bytes_cap: usize) -> Option<&libc::msghdr> {
145        self.egress.fill_msghdr(bytes_cap)
146    }
147
148    #[inline(always)]
149    pub(crate) fn ack_send(&mut self, n: u32) {
150        self.egress.ack(n);
151    }
152}
dope_session/session.rs

dope_session/
session.rs
