df_hyperliquid_core/client/web/socket/
mod.rs

1//! The websocket in the browser and Workers: a [`Feed`] carries one [`crate::live::Session`]
2//! over the global `WebSocket` and hands back [`crate::live::Event`]s. Subscriptions made
3//! while down are sent on the next open; a post made while down is refused, and one cut off
4//! by a drop is reported, never resent.
5//!
6//! Everything runs on the one JS thread: the socket's events and the timer call into the
7//! feed as tasks of their own, and [`Feed::next`] waits for the events they leave.
8
9mod js;
10
11use std::{
12    cell::RefCell,
13    collections::VecDeque,
14    future::poll_fn,
15    rc::Rc,
16    task::{Poll, Waker},
17};
18
19use crate::client::web::clock;
20use crate::client::web::socket::js::{Calls, connect, function, random, said};
21use crate::live::{Event, FeedError, PING, Post, Session, Subscription, Tick};
22use js_sys::Function;
23use wasm_bindgen::{JsCast, JsValue};
24use web_sys::{MessageEvent, WebSocket};
25
26/// One websocket connection, kept open. Dropping the feed closes it.
27pub struct Feed {
28    driver: Rc<RefCell<Driver>>,
29}
30
31impl Feed {
32    /// Connects to the session's network.
33    #[must_use]
34    pub fn start(session: Session) -> Feed {
35        let url = session.url();
36        Feed::start_at(url, session)
37    }
38
39    /// Connects to `url` instead of the session's network: a proxy, or a test.
40    #[must_use]
41    pub fn start_at(url: impl Into<String>, session: Session) -> Feed {
42        let global: JsValue = js_sys::global().into();
43        let timers = function(&global, "setTimeout").zip(function(&global, "clearTimeout"));
44        let driver = Rc::new_cyclic(|driver| {
45            RefCell::new(Driver {
46                url: url.into(),
47                session,
48                global,
49                timers,
50                socket: None,
51                timer: None,
52                calls: Calls::new(driver),
53                events: VecDeque::new(),
54                waker: None,
55                stopped: false,
56            })
57        });
58        if let Ok(mut driver) = driver.try_borrow_mut() {
59            if driver.timers.is_some() {
60                driver.dial();
61            } else {
62                driver.stop();
63            }
64        }
65        Feed { driver }
66    }
67
68    /// Wants `subscription` from now on, across reconnects. Refused past the
69    /// session's limits.
70    pub async fn subscribe(&self, subscription: Subscription) -> Result<(), FeedError> {
71        self.with(|driver| {
72            if let Some(frame) = driver.session.subscribe(subscription)? {
73                driver.carry(&frame);
74            }
75            Ok(())
76        })
77    }
78
79    pub fn unsubscribe(&self, subscription: Subscription) -> Result<(), FeedError> {
80        self.with(|driver| {
81            if let Some(frame) = driver.session.unsubscribe(&subscription) {
82                driver.carry(&frame);
83            }
84            Ok(())
85        })
86    }
87
88    /// Sends an info request or a signed action over the socket, and gives
89    /// its id; the answer arrives as [`crate::live::Frame::Post`] with the same id.
90    pub async fn post(&self, request: Post) -> Result<u64, FeedError> {
91        self.with(|driver| {
92            let (id, frame) = driver.session.post(&request)?;
93            driver.carry(&frame);
94            Ok(id)
95        })
96    }
97
98    /// The next event; `None` once the feed has stopped.
99    pub async fn next(&mut self) -> Option<Event> {
100        poll_fn(|cx| {
101            let Ok(mut driver) = self.driver.try_borrow_mut() else {
102                // Never: the driver is only held inside a call, never across one.
103                cx.waker().wake_by_ref();
104                return Poll::Pending;
105            };
106            if let Some(event) = driver.events.pop_front() {
107                return Poll::Ready(Some(event));
108            }
109            if driver.stopped {
110                return Poll::Ready(None);
111            }
112            driver.waker = Some(cx.waker().clone());
113            Poll::Pending
114        })
115        .await
116    }
117
118    /// Closes the socket and stops.
119    pub fn close(self) {
120        drop(self);
121    }
122
123    /// Runs `act` on a feed that has not stopped. An event it leaves needs no
124    /// wake: [`Feed::next`] takes the feed mutably, so nothing waits on it.
125    fn with<T>(&self, act: impl FnOnce(&mut Driver) -> Result<T, FeedError>) -> Result<T, FeedError> {
126        let mut driver = self.driver.try_borrow_mut().map_err(|_| FeedError::Stopped)?;
127        if driver.stopped {
128            return Err(FeedError::Stopped);
129        }
130        act(&mut driver)
131    }
132}
133
134impl Drop for Feed {
135    fn drop(&mut self) {
136        if let Ok(mut driver) = self.driver.try_borrow_mut() {
137            driver.stop();
138        }
139    }
140}
141
142/// The feed's state. The socket's handlers and the timer hold it weakly, so
143/// the feed alone keeps it, and with it every closure JS can call.
144struct Driver {
145    url: String,
146    session: Session,
147    global: JsValue,
148    /// The global `setTimeout` and `clearTimeout`.
149    timers: Option<(Function, Function)>,
150    /// The socket dialling or open; none while waiting to dial again.
151    socket: Option<WebSocket>,
152    /// The pending timeout: the next tick while open, the next dial while
153    /// closed, none while dialling. A browser gives a number, Node an object.
154    timer: Option<JsValue>,
155    calls: Calls,
156    events: VecDeque<Event>,
157    waker: Option<Waker>,
158    stopped: bool,
159}
160
161impl Driver {
162    /// A new socket. While it dials, the session still answers: a subscription
163    /// is kept for the open, a post is refused. Nothing waits on a handshake.
164    fn dial(&mut self) {
165        match connect(&self.global, &self.url) {
166            Ok(socket) => {
167                socket.set_onopen(Some(self.calls.open.as_ref().unchecked_ref()));
168                socket.set_onmessage(Some(self.calls.message.as_ref().unchecked_ref()));
169                socket.set_onclose(Some(self.calls.close.as_ref().unchecked_ref()));
170                socket.set_onerror(Some(self.calls.error.as_ref().unchecked_ref()));
171                self.socket = Some(socket);
172            }
173            Err(reason) => self.lost(reason),
174        }
175    }
176
177    fn opened(&mut self) {
178        for frame in self.session.opened(clock()) {
179            if let Err(reason) = self.send(&frame) {
180                return self.lost(reason);
181            }
182        }
183        self.events.push_back(Event::Opened);
184        self.arm_tick();
185    }
186
187    fn received(&mut self, message: &MessageEvent) {
188        // The venue sends text only; anything else is not its.
189        let Some(text) = message.data().as_string() else { return };
190        let event = match self.session.received(clock(), &text) {
191            Ok(frame) => Event::Frame(Box::new(frame)),
192            Err(bad) => Event::Unreadable(bad),
193        };
194        self.events.push_back(event);
195    }
196
197    /// The timer fired: a tick while open, the next dial while closed.
198    fn woke(&mut self) {
199        self.timer = None;
200        if self.session.is_open() {
201            match self.session.tick(clock()) {
202                Tick::Wait => {}
203                Tick::Ping => {
204                    if let Err(reason) = self.send(PING) {
205                        return self.lost(reason);
206                    }
207                }
208                Tick::Reconnect => return self.lost("nothing arrived for too long".to_owned()),
209            }
210            self.arm_tick();
211        } else if self.socket.is_none() {
212            self.dial();
213        }
214    }
215
216    /// Sends `frame`, or drops a socket that will not take it.
217    fn carry(&mut self, frame: &str) {
218        if let Err(reason) = self.send(frame) {
219            self.lost(reason);
220        }
221    }
222
223    fn send(&self, frame: &str) -> Result<(), String> {
224        let Some(socket) = &self.socket else { return Err("no socket to send on".to_owned()) };
225        socket.send_with_str(frame).map_err(|error| said(&error).unwrap_or_else(|| "the send failed".to_owned()))
226    }
227
228    /// The socket is gone, or never came: report it with the session's delay,
229    /// and dial again after it.
230    fn lost(&mut self, reason: String) {
231        self.hang_up();
232        let retry = self.session.closed(clock(), random());
233        let after = retry.after_ms;
234        self.events.push_back(Event::Closed { reason, retry });
235        self.arm(after);
236    }
237
238    /// Closes the socket after taking its handlers off, so nothing it still
239    /// has queued reaches the feed. A half-open socket's close may never
240    /// finish; nothing waits for it.
241    fn hang_up(&mut self) {
242        if let Some(socket) = self.socket.take() {
243            socket.set_onopen(None);
244            socket.set_onmessage(None);
245            socket.set_onclose(None);
246            socket.set_onerror(None);
247            let _ = socket.close();
248        }
249    }
250
251    fn stop(&mut self) {
252        self.hang_up();
253        self.disarm();
254        self.stopped = true;
255    }
256
257    /// The timer for the session's next tick, while open.
258    fn arm_tick(&mut self) {
259        if let Some(at) = self.session.due() {
260            self.arm(at.saturating_sub(clock()));
261        }
262    }
263
264    /// One timeout for `ms`, in place of any pending. Without a timer the feed
265    /// can neither tell a dead socket nor dial again, so it stops.
266    fn arm(&mut self, ms: u64) {
267        self.disarm();
268        let Some((set_timeout, _)) = &self.timers else { return self.stop() };
269        let delay = JsValue::from(u32::try_from(ms).map_or(LONGEST_DELAY, |ms| ms.min(LONGEST_DELAY)));
270        match set_timeout.call2(&self.global, self.calls.wake.as_ref(), &delay) {
271            Ok(timer) => self.timer = Some(timer),
272            Err(_) => self.stop(),
273        }
274    }
275
276    fn disarm(&mut self) {
277        if let (Some(timer), Some((_, clear_timeout))) = (self.timer.take(), &self.timers) {
278            let _ = clear_timeout.call1(&self.global, &timer);
279        }
280    }
281
282    /// The waker to call once the driver is let go, when `next` has something.
283    fn wakeable(&mut self) -> Option<Waker> {
284        if self.events.is_empty() && !self.stopped { None } else { self.waker.take() }
285    }
286}
287
288/// The longest delay `setTimeout` keeps; a longer one fires at once.
289const LONGEST_DELAY: u32 = 0x7fff_ffff;
290
291#[cfg(test)]
292#[cfg(target_arch = "wasm32")]
293#[path = "tests/feed.rs"]
294mod tests;
df_hyperliquid_core/client/web/socket/mod.rs

df_hyperliquid_core/client/web/socket/
mod.rs
