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, each frame's text beside
3//! it when asked. Subscriptions made while down are sent on the next open; a post made while
4//! down is refused, and one cut off 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        self.next_with_text().await.map(|(event, _)| event)
101    }
102
103    /// The next event, and for a frame, read or not, the text it arrived as: for a program that
104    /// still reads some channels its own way. `None` once the feed has stopped.
105    pub async fn next_with_text(&mut self) -> Option<(Event, Option<String>)> {
106        poll_fn(|cx| {
107            let Ok(mut driver) = self.driver.try_borrow_mut() else {
108                // Never: the driver is only held inside a call, never across one.
109                cx.waker().wake_by_ref();
110                return Poll::Pending;
111            };
112            if let Some(delivered) = driver.events.pop_front() {
113                return Poll::Ready(Some(delivered));
114            }
115            if driver.stopped {
116                return Poll::Ready(None);
117            }
118            driver.waker = Some(cx.waker().clone());
119            Poll::Pending
120        })
121        .await
122    }
123
124    /// Closes the socket and stops.
125    pub fn close(self) {
126        drop(self);
127    }
128
129    /// Runs `act` on a feed that has not stopped. An event it leaves needs no
130    /// wake: [`Feed::next`] takes the feed mutably, so nothing waits on it.
131    fn with<T>(&self, act: impl FnOnce(&mut Driver) -> Result<T, FeedError>) -> Result<T, FeedError> {
132        let mut driver = self.driver.try_borrow_mut().map_err(|_| FeedError::Stopped)?;
133        if driver.stopped {
134            return Err(FeedError::Stopped);
135        }
136        act(&mut driver)
137    }
138}
139
140impl Drop for Feed {
141    fn drop(&mut self) {
142        if let Ok(mut driver) = self.driver.try_borrow_mut() {
143            driver.stop();
144        }
145    }
146}
147
148/// The feed's state. The socket's handlers and the timer hold it weakly, so
149/// the feed alone keeps it, and with it every closure JS can call.
150struct Driver {
151    url: String,
152    session: Session,
153    global: JsValue,
154    /// The global `setTimeout` and `clearTimeout`.
155    timers: Option<(Function, Function)>,
156    /// The socket dialling or open; none while waiting to dial again.
157    socket: Option<WebSocket>,
158    /// The pending timeout: the next tick while open, the next dial while
159    /// closed, none while dialling. A browser gives a number, Node an object.
160    timer: Option<JsValue>,
161    calls: Calls,
162    /// Each event, and for a frame the text it arrived as.
163    events: VecDeque<(Event, Option<String>)>,
164    waker: Option<Waker>,
165    stopped: bool,
166}
167
168impl Driver {
169    /// A new socket. While it dials, the session still answers: a subscription
170    /// is kept for the open, a post is refused. Nothing waits on a handshake.
171    fn dial(&mut self) {
172        match connect(&self.global, &self.url) {
173            Ok(socket) => {
174                socket.set_onopen(Some(self.calls.open.as_ref().unchecked_ref()));
175                socket.set_onmessage(Some(self.calls.message.as_ref().unchecked_ref()));
176                socket.set_onclose(Some(self.calls.close.as_ref().unchecked_ref()));
177                socket.set_onerror(Some(self.calls.error.as_ref().unchecked_ref()));
178                self.socket = Some(socket);
179            }
180            Err(reason) => self.lost(reason),
181        }
182    }
183
184    fn opened(&mut self) {
185        for frame in self.session.opened(clock()) {
186            if let Err(reason) = self.send(&frame) {
187                return self.lost(reason);
188            }
189        }
190        self.events.push_back((Event::Opened, None));
191        self.arm_tick();
192    }
193
194    fn received(&mut self, message: &MessageEvent) {
195        // The venue sends text only; anything else is not its.
196        let Some(text) = message.data().as_string() else { return };
197        let event = match self.session.received(clock(), &text) {
198            Ok(frame) => Event::Frame(Box::new(frame)),
199            Err(bad) => Event::Unreadable(bad),
200        };
201        self.events.push_back((event, Some(text)));
202    }
203
204    /// The timer fired: a tick while open, the next dial while closed.
205    fn woke(&mut self) {
206        self.timer = None;
207        if self.session.is_open() {
208            match self.session.tick(clock()) {
209                Tick::Wait => {}
210                Tick::Ping => {
211                    if let Err(reason) = self.send(PING) {
212                        return self.lost(reason);
213                    }
214                }
215                Tick::Reconnect => return self.lost("nothing arrived for too long".to_owned()),
216            }
217            self.arm_tick();
218        } else if self.socket.is_none() {
219            self.dial();
220        }
221    }
222
223    /// Sends `frame`, or drops a socket that will not take it.
224    fn carry(&mut self, frame: &str) {
225        if let Err(reason) = self.send(frame) {
226            self.lost(reason);
227        }
228    }
229
230    fn send(&self, frame: &str) -> Result<(), String> {
231        let Some(socket) = &self.socket else { return Err("no socket to send on".to_owned()) };
232        socket.send_with_str(frame).map_err(|error| said(&error).unwrap_or_else(|| "the send failed".to_owned()))
233    }
234
235    /// The socket is gone, or never came: report it with the session's delay,
236    /// and dial again after it.
237    fn lost(&mut self, reason: String) {
238        self.hang_up();
239        let retry = self.session.closed(clock(), random());
240        let after = retry.after_ms;
241        self.events.push_back((Event::Closed { reason, retry }, None));
242        self.arm(after);
243    }
244
245    /// Closes the socket after taking its handlers off, so nothing it still
246    /// has queued reaches the feed. A half-open socket's close may never
247    /// finish; nothing waits for it.
248    fn hang_up(&mut self) {
249        if let Some(socket) = self.socket.take() {
250            socket.set_onopen(None);
251            socket.set_onmessage(None);
252            socket.set_onclose(None);
253            socket.set_onerror(None);
254            let _ = socket.close();
255        }
256    }
257
258    fn stop(&mut self) {
259        self.hang_up();
260        self.disarm();
261        self.stopped = true;
262    }
263
264    /// The timer for the session's next tick, while open.
265    fn arm_tick(&mut self) {
266        if let Some(at) = self.session.due() {
267            self.arm(at.saturating_sub(clock()));
268        }
269    }
270
271    /// One timeout for `ms`, in place of any pending. Without a timer the feed
272    /// can neither tell a dead socket nor dial again, so it stops.
273    fn arm(&mut self, ms: u64) {
274        self.disarm();
275        let Some((set_timeout, _)) = &self.timers else { return self.stop() };
276        let delay = JsValue::from(u32::try_from(ms).map_or(LONGEST_DELAY, |ms| ms.min(LONGEST_DELAY)));
277        match set_timeout.call2(&self.global, self.calls.wake.as_ref(), &delay) {
278            Ok(timer) => self.timer = Some(timer),
279            Err(_) => self.stop(),
280        }
281    }
282
283    fn disarm(&mut self) {
284        if let (Some(timer), Some((_, clear_timeout))) = (self.timer.take(), &self.timers) {
285            let _ = clear_timeout.call1(&self.global, &timer);
286        }
287    }
288
289    /// The waker to call once the driver is let go, when `next` has something.
290    fn wakeable(&mut self) -> Option<Waker> {
291        if self.events.is_empty() && !self.stopped { None } else { self.waker.take() }
292    }
293}
294
295/// The longest delay `setTimeout` keeps; a longer one fires at once.
296const LONGEST_DELAY: u32 = 0x7fff_ffff;
297
298#[cfg(test)]
299#[cfg(target_arch = "wasm32")]
300#[path = "tests/feed.rs"]
301mod tests;
df_hyperliquid_core/client/web/socket/mod.rs

df_hyperliquid_core/client/web/socket/
mod.rs
