liminal_sdk/
remote.rs

1mod config;
2mod handles;
3mod protocol;
4#[cfg(feature = "std")]
5mod tcp;
6
7#[cfg(feature = "std")]
8pub use tcp::{
9    DeliveredMessage, OBSERVABILITY_CHANNEL, PushClient, PushWriter, PushedFrame,
10    SubscriptionStream, TcpRemoteTransport,
11};
12
13pub use config::{SdkConfig, build_channel_handle, build_conversation_handle};
14pub use handles::{
15    RemoteChannelHandle, RemoteConversationHandle, SdkChannelHandle, SdkConversationHandle,
16};
17
18#[cfg(test)]
19mod tests;
20
21use alloc::string::{String, ToString};
22use alloc::sync::Arc;
23use core::time::Duration;
24
25use crate::connection::{ConnectionPoolConfig, ReconnectConfig, ReconnectJitter};
26use crate::{ConversationId, SdkError};
27
28use self::protocol::{ProtocolRemoteTransport, RemoteTransport};
29
30/// Application-level address for a remote liminal server.
31#[derive(Clone, Debug, PartialEq, Eq)]
32pub struct ServerAddress(String);
33
34impl ServerAddress {
35    /// Creates and validates a remote server address.
36    ///
37    /// # Errors
38    ///
39    /// Returns [`SdkError`] when the supplied address is empty.
40    pub fn new(value: impl Into<String>) -> Result<Self, SdkError> {
41        let value = value.into();
42        if value.trim().is_empty() {
43            return Err(connection_error("remote mode requires a server address"));
44        }
45        Ok(Self(value))
46    }
47
48    /// Returns the server address string.
49    #[must_use]
50    pub fn as_str(&self) -> &str {
51        self.0.as_str()
52    }
53}
54
55/// Configuration for remote SDK handles.
56#[derive(Clone, Debug)]
57pub struct RemoteConfig {
58    /// Remote server address. Remote mode cannot be created without this value.
59    pub server_address: ServerAddress,
60    /// Application-visible channel name.
61    pub channel_name: String,
62    /// Application-visible conversation identifier.
63    pub conversation_id: ConversationId,
64    /// Caller/runtime-supplied connection pool configuration.
65    pub pool_config: ConnectionPoolConfig,
66    /// Reconnect policy used by the SDK-003 lifecycle state machine.
67    pub reconnect_config: ReconnectConfig,
68    transport: Arc<dyn RemoteTransport>,
69}
70
71impl RemoteConfig {
72    /// Creates remote configuration with a required server address and pool config.
73    ///
74    /// # Errors
75    ///
76    /// Returns [`SdkError`] if the address or pool configuration is invalid.
77    pub fn new(
78        server_address: impl Into<String>,
79        channel_name: impl Into<String>,
80        conversation_id: impl Into<ConversationId>,
81        pool_config: ConnectionPoolConfig,
82    ) -> Result<Self, SdkError> {
83        Ok(Self {
84            server_address: ServerAddress::new(server_address)?,
85            channel_name: channel_name.into(),
86            conversation_id: conversation_id.into(),
87            pool_config: pool_config.validate()?,
88            reconnect_config: ReconnectConfig::default(),
89            transport: Arc::new(ProtocolRemoteTransport),
90        })
91    }
92
93    /// Replaces the reconnect configuration used by remote handles.
94    #[must_use]
95    pub const fn with_reconnect_config(mut self, reconnect_config: ReconnectConfig) -> Self {
96        self.reconnect_config = reconnect_config;
97        self
98    }
99
100    /// Opens a real TCP connection to the configured server and installs the
101    /// live wire transport, replacing the in-process protocol transport.
102    ///
103    /// This performs the protocol handshake (`Connect` -> `ConnectAck`) eagerly,
104    /// so a returned configuration is already connected to the server. Subsequent
105    /// publish, subscribe, and conversation calls traverse the socket.
106    ///
107    /// # Errors
108    ///
109    /// Returns [`SdkError::Connection`] when the TCP connection cannot be
110    /// established and [`SdkError::Protocol`] when the handshake is rejected.
111    #[cfg(feature = "std")]
112    pub fn connect_tcp(mut self) -> Result<Self, SdkError> {
113        let transport = self::tcp::TcpRemoteTransport::connect(&self.server_address)?;
114        self.transport = Arc::new(transport);
115        Ok(self)
116    }
117
118    /// Opens a real TCP connection whose handshake carries `auth_token`, for a
119    /// server gated by an `[auth]` section, and installs the live wire transport.
120    ///
121    /// Additive to [`connect_tcp`]: an empty token behaves identically to it. The
122    /// server compares the token during the handshake and closes the connection on
123    /// a mismatch, which surfaces here as [`SdkError::Connection`].
124    ///
125    /// # Errors
126    ///
127    /// Returns [`SdkError::Connection`] when the TCP connection cannot be
128    /// established or the token is rejected, and [`SdkError::Protocol`] when the
129    /// handshake frames cannot be encoded or sent.
130    ///
131    /// [`connect_tcp`]: Self::connect_tcp
132    #[cfg(feature = "std")]
133    pub fn connect_tcp_with_auth(mut self, auth_token: &[u8]) -> Result<Self, SdkError> {
134        let transport =
135            self::tcp::TcpRemoteTransport::connect_with_auth(&self.server_address, auth_token)?;
136        self.transport = Arc::new(transport);
137        Ok(self)
138    }
139}
140
141/// Deterministic jitter source for lifecycle integration tests and explicit reconnect calls.
142#[derive(Clone, Copy, Debug, Default)]
143pub struct NoJitter;
144
145impl ReconnectJitter for NoJitter {
146    fn jitter(&mut self, attempt: u32, capped_delay: Duration) -> Duration {
147        core::hint::black_box((attempt, capped_delay));
148        Duration::ZERO
149    }
150}
151
152fn connection_error(description: &str) -> SdkError {
153    SdkError::Connection {
154        description: description.to_string(),
155    }
156}
liminal_sdk/remote.rs

liminal_sdk/
remote.rs
