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
use std::{
fmt::{
Debug as FmtDebug,
Formatter,
Result as FmtResult,
},
pin::Pin,
sync::Arc,
time::Duration,
};
use anyhow::Result;
use tokio_tungstenite::tungstenite::{
http::Request,
protocol::WebSocketConfig,
};
use crate::{
core::{
packet::codecs::WsIoPacketCodec,
types::{
ArcAsyncUnaryResultHandler,
BoxAsyncUnaryResultHandler,
},
},
session::WsIoClientSession,
};
// Types
type InitHandler = Box<
dyn for<'a> Fn(
Arc<WsIoClientSession>,
Option<&'a [u8]>,
&'a WsIoPacketCodec,
) -> Pin<Box<dyn Future<Output = Result<Option<Vec<u8>>>> + Send + 'a>>
+ Send
+ Sync
+ 'static,
>;
type RequestModifier =
Box<dyn Fn(Request<()>) -> Pin<Box<dyn Future<Output = Result<Request<()>>> + Send>> + Send + Sync + 'static>;
// Structs
pub(crate) struct WsIoClientConfig {
/// Maximum duration to wait for the WebSocket connection attempt.
///
/// This timeout covers the transport connection and WebSocket HTTP upgrade
/// handshake performed by `connect_async_with_config`. When unset, the
/// connection attempt is allowed to wait indefinitely.
pub(crate) connect_timeout: Option<Duration>,
/// Maximum duration to wait for graceful WebSocket shutdown after
/// `disconnect` is requested.
///
/// If the read/write tasks do not finish before this timeout, they are
/// aborted so `disconnect().await` can complete.
pub(crate) disconnect_timeout: Duration,
/// Optional client-side init handler used during the server handshake.
///
/// When the server sends an init packet, this handler receives the optional
/// decoded payload and may return optional response data to encode and send
/// back to the server.
pub(crate) init_handler: Option<InitHandler>,
/// Maximum duration allowed for `init_handler` to execute.
pub(crate) init_handler_timeout: Duration,
/// Maximum duration to wait for the server to send the init packet.
///
/// This timeout starts after the WebSocket connection is established. If the
/// init packet is not received in time, the session setup fails.
pub(crate) init_packet_timeout: Duration,
/// Optional handler invoked when a session closes.
///
/// The handler is awaited with `on_session_close_handler_timeout`.
pub(crate) on_session_close_handler: Option<BoxAsyncUnaryResultHandler<WsIoClientSession>>,
/// Maximum duration allowed for `on_session_close_handler` to execute.
pub(crate) on_session_close_handler_timeout: Duration,
/// Optional handler invoked after the session has become ready.
///
/// This handler is spawned asynchronously after the ready state is reached; it
/// is not part of the blocking handshake path.
pub(crate) on_session_ready_handler: Option<ArcAsyncUnaryResultHandler<WsIoClientSession>>,
/// Packet codec used to encode and decode ws.io protocol packets.
///
/// It must match the server namespace codec and controls whether encoded
/// packets use WebSocket text or binary messages.
pub(crate) packet_codec: WsIoPacketCodec,
/// Interval between client heartbeat frames sent after the WebSocket session
/// is created.
///
/// The heartbeat is a one-byte binary WebSocket frame; the server ignores
/// single-byte binary frames before protocol packet decoding.
pub(crate) ping_interval: Duration,
/// Maximum duration to wait for the ready packet from the server.
///
/// The client sends its init response first, then waits for the server to mark
/// the session ready. If the ready packet is not received in time, setup fails.
pub(crate) ready_packet_timeout: Duration,
/// Delay before attempting to reconnect after a disconnected session.
pub(crate) reconnect_delay: Duration,
/// Optional async modifier for the WebSocket HTTP request.
///
/// Use this to adjust headers or other request metadata before
/// `connect_async_with_config` is called.
pub(crate) request_modifier: Option<RequestModifier>,
/// Tungstenite WebSocket transport limits and buffer sizes.
///
/// This config is passed into the client connection and is also used to size
/// internal session channels from the configured max-write/write-buffer ratio.
pub(crate) websocket_config: WebSocketConfig,
}
impl FmtDebug for WsIoClientConfig {
fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
f.debug_struct("WsIoClientConfig")
.field("connect_timeout", &self.connect_timeout)
.field("disconnect_timeout", &self.disconnect_timeout)
.field("init_handler", &self.init_handler.as_ref().map(|_| "<handler>"))
.field("init_handler_timeout", &self.init_handler_timeout)
.field("init_packet_timeout", &self.init_packet_timeout)
.field(
"on_session_close_handler",
&self.on_session_close_handler.as_ref().map(|_| "<handler>"),
)
.field(
"on_session_close_handler_timeout",
&self.on_session_close_handler_timeout,
)
.field(
"on_session_ready_handler",
&self.on_session_ready_handler.as_ref().map(|_| "<handler>"),
)
.field("packet_codec", &self.packet_codec)
.field("ping_interval", &self.ping_interval)
.field("ready_packet_timeout", &self.ready_packet_timeout)
.field("reconnect_delay", &self.reconnect_delay)
.field(
"request_modifier",
&self.request_modifier.as_ref().map(|_| "<modifier>"),
)
.field("websocket_config", &self.websocket_config)
.finish()
}
}