df_hyperliquid_core/live/session/
mod.rs

1//! One connection's state with no socket and no clock: the socket's owner hands in what happened
2//! and when, and sends what comes back. Of the venue's per-IP limits, subscriptions, users and
3//! posts in flight are refused here; connections and message rates are the owner's to keep.
4
5pub mod limits;
6pub mod timing;
7
8pub use limits::{
9    Limits, MAX_CONNECTIONS, MAX_INFLIGHT_POSTS, MAX_MESSAGES_PER_MINUTE, MAX_NEW_CONNECTIONS_PER_MINUTE,
10    MAX_SUBSCRIPTIONS, MAX_USERS, Refused,
11};
12pub use timing::{Retry, Tick, Timing};
13
14use crate::client::Network;
15use crate::live::{BadFrame, Frame, Post, Subscription, decode, post, subscribe, unsubscribe};
16
17#[derive(Debug, Clone, Copy, PartialEq, Eq)]
18enum Link {
19    Closed,
20    Open {
21        /// When it opened.
22        since: u64,
23        /// When anything last arrived; `since` until something does.
24        heard: u64,
25        /// When the venue last sent what keeps the socket open on its side, a pong or a
26        /// subscription's data; `since` until it does. Its acks, refusals and post answers do
27        /// not count: a socket that gets only acks is closed at 60 s all the same.
28        kept: u64,
29        /// Whether anything has arrived at all.
30        talked: bool,
31        /// When the last ping went out; `since` until one does.
32        pinged: u64,
33    },
34}
35
36/// One connection to one network's websocket.
37#[derive(Debug, Clone)]
38pub struct Session {
39    network: Network,
40    limits: Limits,
41    timing: Timing,
42    wanted: Vec<Subscription>,
43    link: Link,
44    /// Reconnects since a socket last worked.
45    failures: u32,
46    next_id: u64,
47    waiting: Vec<u64>,
48}
49
50impl Session {
51    /// A closed session on `network`, with the venue's limits and the default timing.
52    #[must_use]
53    pub fn new(network: Network) -> Self {
54        Session {
55            network,
56            limits: Limits::VENUE,
57            timing: Timing::DEFAULT,
58            wanted: Vec::new(),
59            link: Link::Closed,
60            failures: 0,
61            next_id: 1,
62            waiting: Vec::new(),
63        }
64    }
65
66    /// The limits are per IP: sessions sharing one split them.
67    #[must_use]
68    pub const fn with_limits(mut self, limits: Limits) -> Self {
69        self.limits = limits;
70        self
71    }
72
73    #[must_use]
74    pub const fn with_timing(mut self, timing: Timing) -> Self {
75        self.timing = timing;
76        self
77    }
78
79    #[must_use]
80    pub const fn network(&self) -> Network {
81        self.network
82    }
83
84    /// Where to connect: the network's public websocket.
85    #[must_use]
86    pub const fn url(&self) -> &'static str {
87        self.network.ws_url()
88    }
89
90    #[must_use]
91    pub const fn is_open(&self) -> bool {
92        matches!(self.link, Link::Open { .. })
93    }
94
95    /// What this session subscribes to on every socket, in the order it was asked for.
96    #[must_use]
97    pub fn wanted(&self) -> &[Subscription] {
98        &self.wanted
99    }
100
101    /// The distinct users the wanted subscriptions follow.
102    #[must_use]
103    pub fn users(&self) -> usize {
104        let mut users: Vec<_> = self.wanted.iter().filter_map(Subscription::user).collect();
105        users.sort_unstable();
106        users.dedup();
107        users.len()
108    }
109
110    /// The ids of posts sent on this socket and not yet answered.
111    #[must_use]
112    pub fn waiting(&self) -> &[u64] {
113        &self.waiting
114    }
115
116    /// Wants `subscription` from now on, on this socket and every later one. Gives the frame to
117    /// send when the socket is open and it is new; nothing when it was already wanted. Past the
118    /// limits, or when the venue would not take it, it is refused and not kept.
119    pub fn subscribe(&mut self, subscription: Subscription) -> Result<Option<String>, Refused> {
120        if self.wanted.contains(&subscription) {
121            return Ok(None);
122        }
123        subscription.check()?;
124        if self.wanted.len() >= self.limits.subscriptions {
125            return Err(Refused::Subscriptions { limit: self.limits.subscriptions });
126        }
127        if let Some(user) = subscription.user()
128            && !self.wanted.iter().any(|wanted| wanted.user() == Some(user))
129            && self.users() >= self.limits.users
130        {
131            return Err(Refused::Users { limit: self.limits.users });
132        }
133        let frame = self.is_open().then(|| subscribe(&subscription));
134        self.wanted.push(subscription);
135        Ok(frame)
136    }
137
138    /// Stops wanting `subscription`. Gives the frame to send when the socket is open and it was
139    /// wanted.
140    pub fn unsubscribe(&mut self, subscription: &Subscription) -> Option<String> {
141        let at = self.wanted.iter().position(|wanted| wanted == subscription)?;
142        let gone = self.wanted.remove(at);
143        self.is_open().then(|| unsubscribe(&gone))
144    }
145
146    /// Gives the next post's id and its frame. Only on an open socket, and only while fewer
147    /// than the limit are waiting for an answer.
148    pub fn post(&mut self, request: &Post) -> Result<(u64, String), Refused> {
149        if !self.is_open() {
150            return Err(Refused::Closed);
151        }
152        if self.waiting.len() >= self.limits.posts {
153            return Err(Refused::Posts { limit: self.limits.posts });
154        }
155        let id = self.next_id;
156        self.next_id = self.next_id.saturating_add(1);
157        self.waiting.push(id);
158        Ok((id, post(id, request)))
159    }
160
161    /// The socket opened at `now`. Gives the frames that subscribe to everything wanted, in
162    /// order.
163    pub fn opened(&mut self, now: u64) -> Vec<String> {
164        self.link = Link::Open { since: now, heard: now, kept: now, talked: false, pinged: now };
165        self.wanted.iter().map(subscribe).collect()
166    }
167
168    /// A text frame arrived at `now`. Anything that arrives shows the socket is alive, whether it
169    /// reads or not; only a pong or a subscription's data puts off the next ping. A post's answer
170    /// or refusal stops it waiting.
171    pub fn received(&mut self, now: u64, text: &str) -> Result<Frame, BadFrame> {
172        let frame = decode(text);
173        if let Link::Open { heard, kept, talked, .. } = &mut self.link {
174            *heard = (*heard).max(now);
175            *talked = true;
176            if frame.as_ref().is_ok_and(|frame| !matches!(frame, Frame::Ack(_) | Frame::Error(_) | Frame::Post(_))) {
177                *kept = (*kept).max(now);
178            }
179        }
180        let answered = match &frame {
181            Ok(Frame::Post(reply)) => Some(reply.id),
182            Ok(Frame::Error(refusal)) => refusal.post_id(),
183            _ => None,
184        };
185        if let Some(id) = answered {
186            self.waiting.retain(|waiting| *waiting != id);
187        }
188        frame
189    }
190}
191
192#[cfg(test)]
193#[path = "tests/session.rs"]
194mod tests;
df_hyperliquid_core/live/session/mod.rs

df_hyperliquid_core/live/session/
mod.rs
