phoenix_channel_runtime/
transport.rs

1use futures::future::LocalBoxFuture;
2use thiserror::Error;
3
4/// A transport-level WebSocket message.
5#[derive(Clone, Debug, PartialEq, Eq)]
6pub enum WireMessage {
7    /// A UTF-8 WebSocket or LongPoll message.
8    Text(String),
9    /// A binary WebSocket message.
10    Binary(Vec<u8>),
11}
12
13/// Details reported when a transport connection closes.
14#[derive(Clone, Debug, Eq, PartialEq)]
15pub struct TransportClose {
16    /// WebSocket close code, if a close frame was received.
17    pub code: Option<u16>,
18    /// Human-readable close reason.
19    pub reason: String,
20    /// Whether the transport considered the close handshake clean.
21    pub was_clean: bool,
22}
23
24/// Close code and reason requested by the client.
25#[derive(Clone, Debug, Eq, PartialEq)]
26pub struct TransportCloseRequest {
27    /// WebSocket close code to send.
28    pub code: u16,
29    /// Close reason to send.
30    pub reason: String,
31}
32
33impl TransportCloseRequest {
34    /// Creates a close request.
35    pub fn new(code: u16, reason: impl Into<String>) -> Self {
36        Self {
37            code,
38            reason: reason.into(),
39        }
40    }
41}
42
43impl TransportClose {
44    /// Creates close details reported by a transport.
45    pub fn new(code: Option<u16>, reason: impl Into<String>, was_clean: bool) -> Self {
46        Self {
47            code,
48            reason: reason.into(),
49            was_clean,
50        }
51    }
52
53    /// Creates an abnormal close for a stream that ended without a close frame.
54    pub fn connection_ended() -> Self {
55        Self::new(
56            None,
57            "WebSocket connection ended without a close frame",
58            false,
59        )
60    }
61
62    /// Returns whether the managed client should reconnect this close by default.
63    pub fn should_reconnect(&self) -> bool {
64        !self.was_clean && !matches!(self.code, Some(1000 | 1008))
65    }
66}
67
68/// Message or close notification received from a transport.
69#[derive(Clone, Debug, PartialEq)]
70pub enum TransportEvent {
71    /// An incoming text or binary message.
72    Message(WireMessage),
73    /// The connection closed.
74    Closed(TransportClose),
75}
76
77/// A runtime-neutral, object-safe transport interface.
78///
79/// `LocalBoxFuture` is intentional: browser WebSocket futures are not `Send`.
80/// Native adapters can still be driven on a dedicated local executor.
81pub trait Transport {
82    /// Returns whether this transport can send Phoenix binary frames.
83    fn supports_binary(&self) -> bool {
84        true
85    }
86
87    /// Sends one complete wire message.
88    fn send<'a>(
89        &'a mut self,
90        message: WireMessage,
91    ) -> LocalBoxFuture<'a, Result<(), TransportError>>;
92
93    /// Waits for the next message or close notification.
94    fn receive<'a>(&'a mut self) -> LocalBoxFuture<'a, Result<TransportEvent, TransportError>>;
95
96    /// Closes the transport using its default close behavior.
97    fn close<'a>(&'a mut self) -> LocalBoxFuture<'a, Result<(), TransportError>>;
98
99    /// Closes the transport with an explicit WebSocket code and reason.
100    fn close_with<'a>(
101        &'a mut self,
102        request: TransportCloseRequest,
103    ) -> LocalBoxFuture<'a, Result<(), TransportError>> {
104        let _ = request;
105        self.close()
106    }
107}
108
109/// Operation in which a transport failure occurred.
110#[derive(Clone, Copy, Debug, Eq, PartialEq)]
111pub enum TransportErrorKind {
112    /// Establishing the connection failed.
113    Connect,
114    /// Sending a message failed.
115    Send,
116    /// Receiving a message failed.
117    Receive,
118    /// Closing the connection failed.
119    Close,
120    /// A transport failure without a more specific operation.
121    Other,
122}
123
124impl std::fmt::Display for TransportErrorKind {
125    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
126        formatter.write_str(match self {
127            Self::Connect => "connect",
128            Self::Send => "send",
129            Self::Receive => "receive",
130            Self::Close => "close",
131            Self::Other => "transport",
132        })
133    }
134}
135
136/// Error returned by a runtime-specific transport.
137#[derive(Clone, Debug, Error, Eq, PartialEq)]
138#[error("{kind} error: {message}")]
139pub struct TransportError {
140    kind: TransportErrorKind,
141    message: String,
142}
143
144impl TransportError {
145    /// Creates an unclassified transport error.
146    pub fn new(message: impl Into<String>) -> Self {
147        Self::with_kind(TransportErrorKind::Other, message)
148    }
149
150    /// Creates an error for a specific transport operation.
151    pub fn with_kind(kind: TransportErrorKind, message: impl Into<String>) -> Self {
152        Self {
153            kind,
154            message: message.into(),
155        }
156    }
157
158    /// Returns the operation that failed.
159    pub fn kind(&self) -> TransportErrorKind {
160        self.kind
161    }
162
163    /// Returns the transport-provided error message.
164    pub fn message(&self) -> &str {
165        &self.message
166    }
167}
168
169#[cfg(test)]
170mod tests {
171    use super::*;
172
173    #[test]
174    fn reconnects_only_after_abnormal_reconnectable_closes() {
175        assert!(!TransportClose::new(Some(1000), "normal", true).should_reconnect());
176        assert!(!TransportClose::new(Some(1008), "policy", false).should_reconnect());
177        assert!(TransportClose::new(Some(1006), "abnormal", false).should_reconnect());
178        assert!(TransportClose::connection_ended().should_reconnect());
179    }
180}
phoenix_channel_runtime/transport.rs

phoenix_channel_runtime/
transport.rs
