sccp_protocol/server/
transport.rs

1//! Transport-neutral station admission.
2//!
3//! Listener owners establish the underlying connection and its security
4//! policy, then hand the ready byte stream to the protocol server. Session
5//! framing and lifecycle remain independent of the transport implementation.
6
7use std::fmt;
8use std::net::SocketAddr;
9
10use tokio::io::{AsyncRead, AsyncWrite};
11use tokio::sync::mpsc;
12
13use super::ServerError;
14use super::qos::StationSocketQos;
15use crate::types::{SignalingQos, StationTransport};
16
17/// Bidirectional asynchronous byte stream accepted by a station session.
18///
19/// The protocol server owns the stream after admission and applies identical
20/// framing, backpressure, registration, and shutdown behavior regardless of
21/// the underlying transport. A transport adapter may implement this trait with
22/// a plain socket, a decrypted secure stream, or an in-memory test stream.
23pub trait StationIo: AsyncRead + AsyncWrite + Unpin + Send {}
24
25impl<T> StationIo for T where T: AsyncRead + AsyncWrite + Unpin + Send {}
26
27pub(super) type BoxedStationIo = Box<dyn StationIo>;
28
29pub(super) struct AcceptedStation {
30    pub stream: BoxedStationIo,
31    pub peer: SocketAddr,
32    pub local: SocketAddr,
33    pub transport: StationTransport,
34    pub socket_qos: Option<Box<dyn StationSocketQos>>,
35}
36
37impl fmt::Debug for AcceptedStation {
38    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
39        formatter
40            .debug_struct("AcceptedStation")
41            .field("stream", &"<station I/O>")
42            .field("peer", &self.peer)
43            .field("local", &self.local)
44            .field("transport", &self.transport)
45            .field(
46                "socket_qos",
47                &self.socket_qos.as_ref().map(|_| "<socket QoS control>"),
48            )
49            .finish()
50    }
51}
52
53#[derive(Clone, Debug)]
54/// Cloneable admission endpoint returned by [`super::Server::with_ingress`].
55///
56/// A listener owner performs transport-specific setup first, then submits the
57/// ready byte stream with its actual peer address, accepted local address, and
58/// transport classification. Clones share one bounded queue, so awaiting
59/// [`Self::accept`] propagates server backpressure instead of creating
60/// unbounded session work.
61pub struct ServerIngress {
62    sender: mpsc::Sender<AcceptedStation>,
63    signaling_qos: SignalingQos,
64}
65
66impl ServerIngress {
67    pub(super) fn channel(
68        capacity: usize,
69        signaling_qos: SignalingQos,
70    ) -> (Self, mpsc::Receiver<AcceptedStation>) {
71        let (sender, receiver) = mpsc::channel(capacity);
72        (
73            Self {
74                sender,
75                signaling_qos,
76            },
77            receiver,
78        )
79    }
80
81    /// Transfer ownership of an accepted stream to the server run loop.
82    ///
83    /// `peer` identifies the remote station for events and address policy;
84    /// `local` is the concrete local endpoint used for server-list responses.
85    /// `transport` must describe the already-established stream because it is
86    /// checked against the device definition during registration. For secure
87    /// admission, complete the handshake and any certificate policy before
88    /// calling this method.
89    ///
90    /// The method waits for capacity in the ingress queue. It returns
91    /// [`ServerError::Stopped`] without starting a session if the run loop has
92    /// ended.
93    pub async fn accept<S>(
94        &self,
95        stream: S,
96        peer: SocketAddr,
97        local: SocketAddr,
98        transport: StationTransport,
99    ) -> Result<(), ServerError>
100    where
101        S: StationIo + 'static,
102    {
103        self.admit(Box::new(stream), peer, local, transport, None)
104            .await
105    }
106
107    /// Admit a stream while retaining control of its underlying TCP markings.
108    ///
109    /// The server reapplies the selected station's signaling policy after the
110    /// registration message identifies it. Marking failures are logged while
111    /// registration and subsequent protocol traffic continue normally.
112    pub async fn accept_with_socket_qos<S, Q>(
113        &self,
114        stream: S,
115        peer: SocketAddr,
116        local: SocketAddr,
117        transport: StationTransport,
118        socket_qos: Q,
119    ) -> Result<(), ServerError>
120    where
121        S: StationIo + 'static,
122        Q: StationSocketQos + 'static,
123    {
124        super::report_socket_qos(None, peer, socket_qos.apply(self.signaling_qos));
125        self.admit(
126            Box::new(stream),
127            peer,
128            local,
129            transport,
130            Some(Box::new(socket_qos)),
131        )
132        .await
133    }
134
135    async fn admit(
136        &self,
137        stream: BoxedStationIo,
138        peer: SocketAddr,
139        local: SocketAddr,
140        transport: StationTransport,
141        socket_qos: Option<Box<dyn StationSocketQos>>,
142    ) -> Result<(), ServerError> {
143        self.sender
144            .send(AcceptedStation {
145                stream,
146                peer,
147                local,
148                transport,
149                socket_qos,
150            })
151            .await
152            .map_err(|_| ServerError::Stopped)
153    }
154}
sccp_protocol/server/transport.rs

sccp_protocol/server/
transport.rs
