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<P::Framer, ING>,
42    pub(crate) framer: P::Framer,
43    pub(crate) egress: SendRing<IOV>,
44    pub(crate) state: P::ConnState,
45}
46
47impl<P: ClientProtocol, const ING: usize, const IOV: usize> Default
48    for ClientSessionGeneric<P, ING, IOV>
49{
50    #[inline(always)]
51    fn default() -> Self {
52        Self {
53            ingress: FrameRing::new(),
54            framer: P::Framer::default(),
55            egress: SendRing::new(),
56            state: P::ConnState::default(),
57        }
58    }
59}
60
61impl<P: ClientProtocol, const ING: usize> ClientSessionGeneric<P, ING, CLIENT_IOV_CAP> {
62    #[inline(always)]
63    pub fn enqueue(&mut self, bytes: Bytes) {
64        self.egress.push(bytes);
65    }
66
67    #[inline(always)]
68    pub fn state(&self) -> &P::ConnState {
69        &self.state
70    }
71
72    #[inline(always)]
73    pub fn state_mut(&mut self) -> &mut P::ConnState {
74        &mut self.state
75    }
76
77    #[inline(always)]
78    pub fn egress_is_empty(&self) -> bool {
79        self.egress.is_empty()
80    }
81
82    pub(crate) fn on_connect(
83        &mut self,
84        proto: &mut P,
85        conn_id: SlotId,
86        codec_state: &mut <P::CodecLayer as CodecLayer>::ConnState,
87    ) {
88        let Self { state, egress, .. } = self;
89        proto.on_connect(
90            conn_id,
91            state,
92            Egress {
93                inner: egress,
94                codec_state,
95            },
96        );
97    }
98
99    pub(crate) fn on_disconnect(
100        &mut self,
101        proto: &mut P,
102        conn_id: SlotId,
103        codec_state: &mut <P::CodecLayer as CodecLayer>::ConnState,
104    ) {
105        let Self { state, egress, .. } = self;
106        proto.on_disconnect(
107            conn_id,
108            state,
109            Egress {
110                inner: egress,
111                codec_state,
112            },
113        );
114    }
115
116    pub(crate) fn ingest(
117        &mut self,
118        src: &[u8],
119        proto: &mut P,
120        conn_id: SlotId,
121        codec_state: &mut <P::CodecLayer as CodecLayer>::ConnState,
122    ) {
123        let _ = self.ingress.append(src);
124        let Self {
125            ingress,
126            framer,
127            state,
128            egress,
129            ..
130        } = self;
131        ingress.drain(framer, |head| {
132            proto.on_response(
133                conn_id,
134                state,
135                head,
136                Egress {
137                    inner: &mut *egress,
138                    codec_state: &mut *codec_state,
139                },
140            );
141        });
142    }
143
144    #[inline(always)]
145    pub(crate) fn fill_msghdr(&mut self, bytes_cap: usize) -> Option<&libc::msghdr> {
146        self.egress.fill_msghdr(bytes_cap)
147    }
148
149    #[inline(always)]
150    pub(crate) fn ack_send(&mut self, n: u32) {
151        self.egress.ack(n);
152    }
153}
dope_session/session.rs

dope_session/
session.rs
