df_hyperliquid_core/client/web/
socket.rs

1//! The websocket in the browser and Workers: a [`Feed`] carries one [`crate::live::Session`]
2//! over the global `WebSocket`. Subscriptions made while down are sent on the next open; a
3//! post made while down is refused, and one cut off by a drop is reported, never resent.
4//!
5//! Everything runs on the one JS thread: the socket's events and the timer call into the
6//! feed as tasks of their own, and [`Feed::next`] waits for the events they leave.
7
8use std::{
9    cell::RefCell,
10    collections::VecDeque,
11    future::poll_fn,
12    rc::{Rc, Weak},
13    task::{Poll, Waker},
14};
15
16use crate::client::web::clock;
17use crate::live::{BadFrame, Frame, PING, Post, Refused, Retry, Session, Subscription, Tick};
18use js_sys::{Array, Function, Math, Reflect};
19use wasm_bindgen::{JsCast, JsValue, closure::Closure};
20use web_sys::{CloseEvent, MessageEvent, WebSocket};
21
22/// What a feed hands back, in the order it happened.
23#[derive(Debug, Clone, PartialEq, Eq)]
24pub enum Event {
25    /// The socket opened and everything wanted was subscribed again.
26    Opened,
27    /// Boxed: a frame is large and most events are not frames.
28    Frame(Box<Frame>),
29    /// A frame that did not read. The socket is still alive.
30    Unreadable(BadFrame),
31    /// The socket closed, or could not open. The feed dials again after
32    /// `retry.after_ms`; the posts in `retry.unanswered` may or may not have
33    /// landed, so read the account to find out.
34    Closed { reason: String, retry: Retry },
35}
36
37#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
38pub enum FeedError {
39    #[error(transparent)]
40    Refused(#[from] Refused),
41    /// The runtime gave no timer (no `setTimeout`, or one that threw), so the
42    /// feed could not keep its time and closed for good.
43    #[error("the feed has stopped")]
44    Stopped,
45}
46
47/// One websocket connection, kept open. Dropping the feed closes it.
48pub struct Feed {
49    driver: Rc<RefCell<Driver>>,
50}
51
52impl Feed {
53    /// Connects to the session's network.
54    #[must_use]
55    pub fn start(session: Session) -> Feed {
56        let url = session.url();
57        Feed::start_at(url, session)
58    }
59
60    /// Connects to `url` instead of the session's network: a proxy, or a test.
61    #[must_use]
62    pub fn start_at(url: impl Into<String>, session: Session) -> Feed {
63        let global: JsValue = js_sys::global().into();
64        let timers = function(&global, "setTimeout").zip(function(&global, "clearTimeout"));
65        let driver = Rc::new_cyclic(|driver| {
66            RefCell::new(Driver {
67                url: url.into(),
68                session,
69                global,
70                timers,
71                socket: None,
72                timer: None,
73                calls: Calls::new(driver),
74                events: VecDeque::new(),
75                waker: None,
76                stopped: false,
77            })
78        });
79        if let Ok(mut driver) = driver.try_borrow_mut() {
80            if driver.timers.is_some() {
81                driver.dial();
82            } else {
83                driver.stop();
84            }
85        }
86        Feed { driver }
87    }
88
89    /// Wants `subscription` from now on, across reconnects. Refused past the
90    /// session's limits.
91    pub async fn subscribe(&self, subscription: Subscription) -> Result<(), FeedError> {
92        self.with(|driver| {
93            if let Some(frame) = driver.session.subscribe(subscription)? {
94                driver.carry(&frame);
95            }
96            Ok(())
97        })
98    }
99
100    pub fn unsubscribe(&self, subscription: Subscription) -> Result<(), FeedError> {
101        self.with(|driver| {
102            if let Some(frame) = driver.session.unsubscribe(&subscription) {
103                driver.carry(&frame);
104            }
105            Ok(())
106        })
107    }
108
109    /// Sends an info request or a signed action over the socket, and gives
110    /// its id; the answer arrives as [`Frame::Post`] with the same id.
111    pub async fn post(&self, request: Post) -> Result<u64, FeedError> {
112        self.with(|driver| {
113            let (id, frame) = driver.session.post(&request)?;
114            driver.carry(&frame);
115            Ok(id)
116        })
117    }
118
119    /// The next event; `None` once the feed has stopped.
120    pub async fn next(&mut self) -> Option<Event> {
121        poll_fn(|cx| {
122            let Ok(mut driver) = self.driver.try_borrow_mut() else {
123                // Never: the driver is only held inside a call, never across one.
124                cx.waker().wake_by_ref();
125                return Poll::Pending;
126            };
127            if let Some(event) = driver.events.pop_front() {
128                return Poll::Ready(Some(event));
129            }
130            if driver.stopped {
131                return Poll::Ready(None);
132            }
133            driver.waker = Some(cx.waker().clone());
134            Poll::Pending
135        })
136        .await
137    }
138
139    /// Closes the socket and stops.
140    pub fn close(self) {
141        drop(self);
142    }
143
144    /// Runs `act` on a feed that has not stopped. An event it leaves needs no
145    /// wake: [`Feed::next`] takes the feed mutably, so nothing waits on it.
146    fn with<T>(&self, act: impl FnOnce(&mut Driver) -> Result<T, FeedError>) -> Result<T, FeedError> {
147        let mut driver = self.driver.try_borrow_mut().map_err(|_| FeedError::Stopped)?;
148        if driver.stopped {
149            return Err(FeedError::Stopped);
150        }
151        act(&mut driver)
152    }
153}
154
155impl Drop for Feed {
156    fn drop(&mut self) {
157        if let Ok(mut driver) = self.driver.try_borrow_mut() {
158            driver.stop();
159        }
160    }
161}
162
163/// The feed's state. The socket's handlers and the timer hold it weakly, so
164/// the feed alone keeps it, and with it every closure JS can call.
165struct Driver {
166    url: String,
167    session: Session,
168    global: JsValue,
169    /// The global `setTimeout` and `clearTimeout`.
170    timers: Option<(Function, Function)>,
171    /// The socket dialling or open; none while waiting to dial again.
172    socket: Option<WebSocket>,
173    /// The pending timeout: the next tick while open, the next dial while
174    /// closed, none while dialling. A browser gives a number, Node an object.
175    timer: Option<JsValue>,
176    calls: Calls,
177    events: VecDeque<Event>,
178    waker: Option<Waker>,
179    stopped: bool,
180}
181
182impl Driver {
183    /// A new socket. While it dials, the session still answers: a subscription
184    /// is kept for the open, a post is refused. Nothing waits on a handshake.
185    fn dial(&mut self) {
186        match connect(&self.global, &self.url) {
187            Ok(socket) => {
188                socket.set_onopen(Some(self.calls.open.as_ref().unchecked_ref()));
189                socket.set_onmessage(Some(self.calls.message.as_ref().unchecked_ref()));
190                socket.set_onclose(Some(self.calls.close.as_ref().unchecked_ref()));
191                socket.set_onerror(Some(self.calls.error.as_ref().unchecked_ref()));
192                self.socket = Some(socket);
193            }
194            Err(reason) => self.lost(reason),
195        }
196    }
197
198    fn opened(&mut self) {
199        for frame in self.session.opened(clock()) {
200            if let Err(reason) = self.send(&frame) {
201                return self.lost(reason);
202            }
203        }
204        self.events.push_back(Event::Opened);
205        self.arm_tick();
206    }
207
208    fn received(&mut self, message: &MessageEvent) {
209        // The venue sends text only; anything else is not its.
210        let Some(text) = message.data().as_string() else { return };
211        let event = match self.session.received(clock(), &text) {
212            Ok(frame) => Event::Frame(Box::new(frame)),
213            Err(bad) => Event::Unreadable(bad),
214        };
215        self.events.push_back(event);
216    }
217
218    /// The timer fired: a tick while open, the next dial while closed.
219    fn woke(&mut self) {
220        self.timer = None;
221        if self.session.is_open() {
222            match self.session.tick(clock()) {
223                Tick::Wait => {}
224                Tick::Ping => {
225                    if let Err(reason) = self.send(PING) {
226                        return self.lost(reason);
227                    }
228                }
229                Tick::Reconnect => return self.lost("nothing arrived for too long".to_owned()),
230            }
231            self.arm_tick();
232        } else if self.socket.is_none() {
233            self.dial();
234        }
235    }
236
237    /// Sends `frame`, or drops a socket that will not take it.
238    fn carry(&mut self, frame: &str) {
239        if let Err(reason) = self.send(frame) {
240            self.lost(reason);
241        }
242    }
243
244    fn send(&self, frame: &str) -> Result<(), String> {
245        let Some(socket) = &self.socket else { return Err("no socket to send on".to_owned()) };
246        socket.send_with_str(frame).map_err(|error| said(&error).unwrap_or_else(|| "the send failed".to_owned()))
247    }
248
249    /// The socket is gone, or never came: report it with the session's delay,
250    /// and dial again after it.
251    fn lost(&mut self, reason: String) {
252        self.hang_up();
253        let retry = self.session.closed(clock(), random());
254        let after = retry.after_ms;
255        self.events.push_back(Event::Closed { reason, retry });
256        self.arm(after);
257    }
258
259    /// Closes the socket after taking its handlers off, so nothing it still
260    /// has queued reaches the feed. A half-open socket's close may never
261    /// finish; nothing waits for it.
262    fn hang_up(&mut self) {
263        if let Some(socket) = self.socket.take() {
264            socket.set_onopen(None);
265            socket.set_onmessage(None);
266            socket.set_onclose(None);
267            socket.set_onerror(None);
268            let _ = socket.close();
269        }
270    }
271
272    fn stop(&mut self) {
273        self.hang_up();
274        self.disarm();
275        self.stopped = true;
276    }
277
278    /// The timer for the session's next tick, while open.
279    fn arm_tick(&mut self) {
280        if let Some(at) = self.session.due() {
281            self.arm(at.saturating_sub(clock()));
282        }
283    }
284
285    /// One timeout for `ms`, in place of any pending. Without a timer the feed
286    /// can neither tell a dead socket nor dial again, so it stops.
287    fn arm(&mut self, ms: u64) {
288        self.disarm();
289        let Some((set_timeout, _)) = &self.timers else { return self.stop() };
290        let delay = JsValue::from(u32::try_from(ms).map_or(LONGEST_DELAY, |ms| ms.min(LONGEST_DELAY)));
291        match set_timeout.call2(&self.global, self.calls.wake.as_ref(), &delay) {
292            Ok(timer) => self.timer = Some(timer),
293            Err(_) => self.stop(),
294        }
295    }
296
297    fn disarm(&mut self) {
298        if let (Some(timer), Some((_, clear_timeout))) = (self.timer.take(), &self.timers) {
299            let _ = clear_timeout.call1(&self.global, &timer);
300        }
301    }
302
303    /// The waker to call once the driver is let go, when `next` has something.
304    fn wakeable(&mut self) -> Option<Waker> {
305        if self.events.is_empty() && !self.stopped { None } else { self.waker.take() }
306    }
307}
308
309/// The longest delay `setTimeout` keeps; a longer one fires at once.
310const LONGEST_DELAY: u32 = 0x7fff_ffff;
311
312/// The closures JS calls, made once a feed and dropped with it. Each holds the
313/// driver weakly and is handed to every socket the feed dials in turn.
314struct Calls {
315    open: Closure<dyn FnMut()>,
316    message: Closure<dyn FnMut(MessageEvent)>,
317    close: Closure<dyn FnMut(CloseEvent)>,
318    error: Closure<dyn FnMut(JsValue)>,
319    wake: Closure<dyn FnMut()>,
320}
321
322impl Calls {
323    fn new(driver: &Weak<RefCell<Driver>>) -> Calls {
324        let (open, message, close, error, wake) =
325            (driver.clone(), driver.clone(), driver.clone(), driver.clone(), driver.clone());
326        Calls {
327            open: Closure::new(move || on(&open, Driver::opened)),
328            message: Closure::new(move |event: MessageEvent| on(&message, |driver| driver.received(&event))),
329            close: Closure::new(move |event: CloseEvent| {
330                let reason = match event.reason() {
331                    reason if reason.is_empty() => format!("the socket closed ({})", event.code()),
332                    reason => reason,
333                };
334                on(&close, |driver| driver.lost(reason));
335            }),
336            // An error always ends the socket; the close that follows is not
337            // waited for.
338            error: Closure::new(move |event: JsValue| {
339                let reason = said(&event).unwrap_or_else(|| "the socket failed".to_owned());
340                on(&error, |driver| driver.lost(reason));
341            }),
342            wake: Closure::new(move || on(&wake, Driver::woke)),
343        }
344    }
345}
346
347/// Runs `act` on the driver for JS, then wakes `next` if it has something. The
348/// waker is called once the driver is let go, so one that polls at once finds
349/// it free.
350fn on(driver: &Weak<RefCell<Driver>>, act: impl FnOnce(&mut Driver)) {
351    let Some(driver) = driver.upgrade() else { return };
352    let waker = {
353        let Ok(mut driver) = driver.try_borrow_mut() else { return };
354        act(&mut driver);
355        driver.wakeable()
356    };
357    if let Some(waker) = waker {
358        waker.wake();
359    }
360}
361
362/// A new socket to `url`, from the global `WebSocket`.
363fn connect(global: &JsValue, url: &str) -> Result<WebSocket, String> {
364    let class = function(global, "WebSocket").ok_or_else(|| "this runtime has no WebSocket".to_owned())?;
365    let socket = Reflect::construct(&class, &Array::of1(&JsValue::from_str(url)))
366        .map_err(|error| said(&error).unwrap_or_else(|| "the socket could not be made".to_owned()))?;
367    Ok(socket.unchecked_into())
368}
369
370/// The global function `name`, if the runtime has it.
371fn function(global: &JsValue, name: &str) -> Option<Function> {
372    Reflect::get(global, &JsValue::from_str(name)).ok()?.dyn_into().ok()
373}
374
375/// The `message` an error or an error event carries, if any. A browser's
376/// socket error carries none.
377fn said(error: &JsValue) -> Option<String> {
378    let message = Reflect::get(error, &JsValue::from_str("message")).ok()?.as_string()?;
379    (!message.is_empty()).then_some(message)
380}
381
382/// Enough to spread reconnects: many clients dropped at once must not return
383/// at once. Not for anything secret. `Math.random()`'s bits, mixed so the low
384/// ones the session's jitter reads are not a small float's trailing zeros.
385fn random() -> u64 {
386    let bits = Math::random().to_bits().wrapping_mul(0x9e37_79b9_7f4a_7c15);
387    bits ^ (bits >> 29) ^ clock().rotate_left(29)
388}
389
390#[cfg(test)]
391#[cfg(target_arch = "wasm32")]
392#[path = "tests/socket.rs"]
393mod tests;
df_hyperliquid_core/client/web/socket.rs

df_hyperliquid_core/client/web/
socket.rs
