phoenix_channel_client/client/
config.rs

1use std::{rc::Rc, time::Duration};
2
3use futures::future::LocalBoxFuture;
4use phoenix_channel_runtime::{Transport, TransportError};
5use serde_json::Value;
6
7use super::{DisconnectReason, TelemetryHook};
8
9/// Information supplied to a [`Connector`] for one connection attempt.
10#[derive(Clone, Copy, Debug, Eq, PartialEq)]
11pub struct ConnectContext {
12    /// Zero-based connection-attempt number for the current connection cycle.
13    pub attempt: u32,
14}
15
16/// Creates target-specific WebSocket transports for the driver.
17pub trait Connector {
18    /// Starts one transport connection attempt.
19    fn connect(
20        &self,
21        context: ConnectContext,
22    ) -> LocalBoxFuture<'static, Result<Box<dyn Transport>, TransportError>>;
23}
24
25/// Supplies sleeps without choosing an executor.
26pub trait Timer {
27    /// Completes after `duration` according to the runtime clock.
28    fn sleep(&self, duration: Duration) -> LocalBoxFuture<'static, ()>;
29    /// Returns a monotonic duration from an implementation-defined origin.
30    fn now(&self) -> Duration;
31}
32
33/// Information supplied when loading a channel's join payload.
34#[derive(Clone, Copy, Debug, Eq, PartialEq)]
35pub struct JoinContext {
36    /// Zero-based join-attempt number for the current join cycle.
37    pub attempt: u32,
38    /// Whether this join follows a prior successful join.
39    pub is_rejoin: bool,
40}
41
42/// Information supplied to a custom reconnect policy.
43#[derive(Clone, Debug, Eq, PartialEq)]
44pub struct ReconnectContext {
45    /// Reconnect attempt number after the disconnect.
46    pub attempt: u32,
47    /// Transport or client condition that caused the disconnect.
48    pub reason: DisconnectReason,
49}
50
51/// Decision returned by a custom reconnect policy.
52#[derive(Clone, Copy, Debug, Eq, PartialEq)]
53pub enum ReconnectAction {
54    /// Retry after the specified delay.
55    RetryAfter(Duration),
56    /// Remain disconnected until an explicit connect or reconnect request.
57    Stop,
58}
59
60/// Callback that decides whether and when to reconnect.
61pub type ReconnectPolicy = Rc<dyn Fn(ReconnectContext) -> ReconnectAction>;
62
63/// Async callback that produces a fresh JSON payload for each join attempt.
64pub type JoinPayloadLoader =
65    Rc<dyn Fn(JoinContext) -> LocalBoxFuture<'static, Result<Value, String>>>;
66
67/// Creates a join payload loader that clones the same value for every attempt.
68pub fn static_join_payload(payload: Value) -> JoinPayloadLoader {
69    Rc::new(move |_| {
70        let payload = payload.clone();
71        Box::pin(async move { Ok(payload) })
72    })
73}
74
75/// Heartbeat, retry, timeout, buffering, and telemetry settings.
76#[derive(Clone)]
77pub struct Options {
78    pub(crate) heartbeat_interval: Duration,
79    pub(crate) heartbeat_timeout: Duration,
80    pub(crate) connect_timeout: Duration,
81    pub(crate) join_timeout: Duration,
82    pub(crate) call_timeout: Duration,
83    pub(crate) leave_timeout: Duration,
84    pub(crate) reconnect_delay: Rc<dyn Fn(u32) -> Duration>,
85    pub(crate) reconnect_policy: Option<ReconnectPolicy>,
86    pub(crate) rejoin_delay: Rc<dyn Fn(u32) -> Duration>,
87    pub(crate) command_capacity: usize,
88    pub(crate) push_buffer_capacity: usize,
89    pub(crate) event_capacity: usize,
90    pub(crate) connect_on_start: bool,
91    pub(crate) telemetry: Option<TelemetryHook>,
92}
93
94impl Default for Options {
95    fn default() -> Self {
96        Self {
97            heartbeat_interval: Duration::from_secs(30),
98            heartbeat_timeout: Duration::from_secs(30),
99            connect_timeout: Duration::from_secs(10),
100            join_timeout: Duration::from_secs(10),
101            call_timeout: Duration::from_secs(10),
102            leave_timeout: Duration::from_secs(10),
103            reconnect_delay: Rc::new(default_retry_delay),
104            reconnect_policy: None,
105            rejoin_delay: Rc::new(default_retry_delay),
106            command_capacity: 64,
107            push_buffer_capacity: 64,
108            event_capacity: 256,
109            connect_on_start: true,
110            telemetry: None,
111        }
112    }
113}
114
115impl Options {
116    /// Sets how often the connected driver sends automatic heartbeats.
117    pub fn heartbeat_interval(mut self, interval: Duration) -> Self {
118        self.heartbeat_interval = interval;
119        self
120    }
121
122    /// Sets how long an automatic heartbeat acknowledgement may take.
123    pub fn heartbeat_timeout(mut self, timeout: Duration) -> Self {
124        self.heartbeat_timeout = timeout;
125        self
126    }
127
128    /// Sets the maximum duration of one transport connection attempt.
129    pub fn connect_timeout(mut self, timeout: Duration) -> Self {
130        self.connect_timeout = timeout;
131        self
132    }
133
134    /// Sets the join, call, and leave timeout to one common value.
135    pub fn request_timeout(mut self, timeout: Duration) -> Self {
136        self.join_timeout = timeout;
137        self.call_timeout = timeout;
138        self.leave_timeout = timeout;
139        self
140    }
141
142    /// Sets the default channel join timeout.
143    pub fn join_timeout(mut self, timeout: Duration) -> Self {
144        self.join_timeout = timeout;
145        self
146    }
147
148    /// Sets the default correlated call timeout.
149    pub fn call_timeout(mut self, timeout: Duration) -> Self {
150        self.call_timeout = timeout;
151        self
152    }
153
154    /// Sets the default channel leave timeout.
155    pub fn leave_timeout(mut self, timeout: Duration) -> Self {
156        self.leave_timeout = timeout;
157        self
158    }
159
160    /// Sets the delay function used by the default reconnect policy.
161    pub fn reconnect_delay(mut self, delay: impl Fn(u32) -> Duration + 'static) -> Self {
162        self.reconnect_delay = Rc::new(delay);
163        self
164    }
165
166    /// Installs a policy that classifies every disconnect and selects a retry.
167    pub fn reconnect_policy(
168        mut self,
169        policy: impl Fn(ReconnectContext) -> ReconnectAction + 'static,
170    ) -> Self {
171        self.reconnect_policy = Some(Rc::new(policy));
172        self
173    }
174
175    /// Sets the delay before each automatic channel rejoin attempt.
176    pub fn rejoin_delay(mut self, delay: impl Fn(u32) -> Duration + 'static) -> Self {
177        self.rejoin_delay = Rc::new(delay);
178        self
179    }
180
181    /// Sets the bounded capacity of the driver command queue.
182    pub fn command_capacity(mut self, capacity: usize) -> Self {
183        self.command_capacity = capacity.max(1);
184        self
185    }
186
187    /// Sets how many calls may wait for a socket connection or channel join.
188    pub fn push_buffer_capacity(mut self, capacity: usize) -> Self {
189        self.push_buffer_capacity = capacity;
190        self
191    }
192
193    /// Sets the capacity of each socket or channel event subscriber.
194    pub fn event_capacity(mut self, capacity: usize) -> Self {
195        self.event_capacity = capacity.max(1);
196        self
197    }
198
199    /// Selects whether the driver connects immediately when it starts.
200    pub fn connect_on_start(mut self, enabled: bool) -> Self {
201        self.connect_on_start = enabled;
202        self
203    }
204
205    /// Installs a callback for structured client telemetry.
206    pub fn telemetry(mut self, hook: TelemetryHook) -> Self {
207        self.telemetry = Some(hook);
208        self
209    }
210}
211
212fn default_retry_delay(attempt: u32) -> Duration {
213    match attempt {
214        0 => Duration::ZERO,
215        1 => Duration::from_secs(1),
216        2 => Duration::from_secs(2),
217        3 => Duration::from_secs(5),
218        _ => Duration::from_secs(10),
219    }
220}
phoenix_channel_client/client/config.rs

phoenix_channel_client/client/
config.rs
