1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
mod config;
mod handles;
mod participant;
mod protocol;
#[cfg(feature = "std")]
mod tcp;
pub mod websocket;
#[cfg(feature = "std")]
pub use tcp::{
DeliveredMessage, FlushMode, FlushOutcome, OBSERVABILITY_CHANNEL, PublishRejection, PushClient,
PushWriter, PushedFrame, SubscriptionStream, TcpRemoteTransport,
};
#[cfg(feature = "std")]
pub use websocket::{
WebSocketDeliveredMessage, WebSocketRemoteTransport, WebSocketSubscriptionStream,
};
pub use config::{SdkConfig, build_channel_handle, build_conversation_handle};
pub use handles::{
RemoteChannelHandle, RemoteConversationHandle, RemoteParticipantHandle, SdkChannelHandle,
SdkConversationHandle,
};
pub use participant::{
ParticipantResponseProvenance, ParticipantResumeStore, RemoteDetachReplayOutcome,
RemoteExpectedOperationRecovery, RemoteLostOperationResolution, RemoteLostReconnectResolution,
RemoteOperationRecordOutcome, RemoteOperationTransportFate, RemoteParticipantError,
RemoteParticipantInbound, RemoteParticipantOperation, RemoteParticipantSendOutcome,
RemoteReconnectAttemptOutcome, RemoteReconnectPermit, RemoteReconnectPermitOutcome,
RemoteReconnectPermitRecovery, RemoteReplayApplyOutcome, RemoteTransportLossOutcome,
};
#[cfg(test)]
mod tests;
use alloc::string::{String, ToString};
use alloc::sync::Arc;
use crate::connection::ConnectionPoolConfig;
use crate::{ConversationId, SdkError};
use self::protocol::{ProtocolRemoteTransport, RemoteTransport};
/// Application-level address for a remote liminal server.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ServerAddress(String);
impl ServerAddress {
/// Creates and validates a remote server address.
///
/// # Errors
///
/// Returns [`SdkError`] when the supplied address is empty.
pub fn new(value: impl Into<String>) -> Result<Self, SdkError> {
let value = value.into();
if value.trim().is_empty() {
return Err(connection_error("remote mode requires a server address"));
}
Ok(Self(value))
}
/// Returns the server address string.
#[must_use]
pub fn as_str(&self) -> &str {
self.0.as_str()
}
}
/// Configuration for remote SDK handles.
#[derive(Clone, Debug)]
pub struct RemoteConfig {
/// Remote server address. Remote mode cannot be created without this value.
pub server_address: ServerAddress,
/// Application-visible channel name.
pub channel_name: String,
/// Application-visible conversation identifier.
pub conversation_id: ConversationId,
/// Caller/runtime-supplied connection pool configuration.
pub pool_config: ConnectionPoolConfig,
transport: Arc<dyn RemoteTransport>,
/// The concretely typed WebSocket transport, retained when
/// [`connect_websocket`](Self::connect_websocket) installed it so callers
/// can drive its typed reconnect path.
#[cfg(feature = "std")]
websocket: Option<Arc<websocket::WebSocketRemoteTransport>>,
}
impl RemoteConfig {
/// Creates remote configuration with a required server address and pool config.
///
/// # Errors
///
/// Returns [`SdkError`] if the address or pool configuration is invalid.
pub fn new(
server_address: impl Into<String>,
channel_name: impl Into<String>,
conversation_id: impl Into<ConversationId>,
pool_config: ConnectionPoolConfig,
) -> Result<Self, SdkError> {
Ok(Self {
server_address: ServerAddress::new(server_address)?,
channel_name: channel_name.into(),
conversation_id: conversation_id.into(),
pool_config: pool_config.validate()?,
transport: Arc::new(ProtocolRemoteTransport),
#[cfg(feature = "std")]
websocket: None,
})
}
/// Opens a real TCP connection to the configured server and installs the
/// live wire transport, replacing the in-process protocol transport.
///
/// This performs the protocol handshake (`Connect` -> `ConnectAck`) eagerly,
/// so a returned configuration is already connected to the server. Subsequent
/// publish, subscribe, and conversation calls traverse the socket.
///
/// # Errors
///
/// Returns [`SdkError::Connection`] when the TCP connection cannot be
/// established and [`SdkError::Protocol`] when the handshake is rejected.
#[cfg(feature = "std")]
pub fn connect_tcp(mut self) -> Result<Self, SdkError> {
let transport = self::tcp::TcpRemoteTransport::connect(&self.server_address)?;
self.transport = Arc::new(transport);
self.websocket = None;
Ok(self)
}
/// Opens a real TCP connection whose handshake carries `auth_token`, for a
/// server gated by an `[auth]` section, and installs the live wire transport.
///
/// Additive to [`connect_tcp`]: an empty token behaves identically to it. The
/// server compares the token during the handshake and closes the connection on
/// a mismatch, which surfaces here as [`SdkError::Connection`].
///
/// # Errors
///
/// Returns [`SdkError::Connection`] when the TCP connection cannot be
/// established or the token is rejected, and [`SdkError::Protocol`] when the
/// handshake frames cannot be encoded or sent.
///
/// [`connect_tcp`]: Self::connect_tcp
#[cfg(feature = "std")]
pub fn connect_tcp_with_auth(mut self, auth_token: &[u8]) -> Result<Self, SdkError> {
let transport =
self::tcp::TcpRemoteTransport::connect_with_auth(&self.server_address, auth_token)?;
self.transport = Arc::new(transport);
self.websocket = None;
Ok(self)
}
/// Opens a real WebSocket connection to the configured `ws://` server
/// address and installs the live wire transport, replacing the in-process
/// protocol transport.
///
/// This performs the WebSocket upgrade and the protocol handshake
/// (`Connect` -> `ConnectAck`) eagerly through the client unit's typed
/// permit path, so a returned configuration is already connected.
/// Subsequent publish, subscribe, and conversation calls traverse the
/// socket; the concretely typed transport stays reachable through
/// [`websocket_transport`](Self::websocket_transport) for the typed
/// reconnect path.
///
/// # Errors
///
/// Returns [`SdkError::Connection`] when the address is not a usable
/// `ws://` URL, the connection cannot be established, or the handshake is
/// rejected, and [`SdkError::Protocol`] when frames cannot be encoded.
#[cfg(feature = "std")]
pub fn connect_websocket(self) -> Result<Self, SdkError> {
self.connect_websocket_with_auth(&[])
}
/// Opens a real WebSocket connection whose handshake carries
/// `auth_token`, for a server gated by an `[auth]` section, and installs
/// the live wire transport. Additive to [`connect_websocket`]: an empty
/// token behaves identically to it.
///
/// # Errors
///
/// Returns [`SdkError::Connection`] when the connection cannot be
/// established or the token is rejected, and [`SdkError::Protocol`] when
/// the handshake frames cannot be encoded or sent.
///
/// [`connect_websocket`]: Self::connect_websocket
#[cfg(feature = "std")]
pub fn connect_websocket_with_auth(mut self, auth_token: &[u8]) -> Result<Self, SdkError> {
let transport = Arc::new(websocket::WebSocketRemoteTransport::connect_with_auth(
&self.server_address,
auth_token,
)?);
self.transport = Arc::clone(&transport) as Arc<dyn RemoteTransport>;
self.websocket = Some(transport);
Ok(self)
}
/// The concretely typed WebSocket transport installed by
/// [`connect_websocket`](Self::connect_websocket), when one is installed.
#[cfg(feature = "std")]
#[must_use]
pub fn websocket_transport(&self) -> Option<Arc<websocket::WebSocketRemoteTransport>> {
self.websocket.clone()
}
}
fn connection_error(description: &str) -> SdkError {
SdkError::Connection {
description: description.to_string(),
}
}