brokapi_bybit_protocol/ws/
models.rs

1//! The messages of the WebSocket streams: the requests we send and the frames that come back.
2//!
3//! Every request is `{"op": ..., "req_id": ..., "args": [...]}`. The exchange answers a subscription once with
4//! `{"success": true/false, "ret_msg": ..., "op": "subscribe", "req_id": ...}`, then data frames arrive with the
5//! `topic` of the subscription until it is cancelled.
6
7use serde::Serialize;
8use serde_json::{Value, json};
9
10/// The path of the public stream of a category.
11pub fn public_path(category: &str) -> String {
12    format!("/v5/public/{category}")
13}
14
15/// The path of the private stream.
16pub const PRIVATE_PATH: &str = "/v5/private";
17
18/// A request to the stream.
19#[derive(Debug, Clone, PartialEq, Serialize)]
20pub struct WsRequest {
21    /// `subscribe`, `unsubscribe`, `auth` or `ping`.
22    pub op: String,
23    /// An id of ours, repeated in the answer. Empty when none.
24    #[serde(skip_serializing_if = "String::is_empty")]
25    pub req_id: String,
26    /// The topics (or the login arguments).
27    #[serde(skip_serializing_if = "Vec::is_empty")]
28    pub args: Vec<Value>,
29}
30
31impl WsRequest {
32    /// Subscribes to one topic, for example `orderbook.50.BTCUSDT`.
33    ///
34    /// # Examples
35    ///
36    /// ```
37    /// use brokapi_bybit_protocol::ws::models::WsRequest;
38    ///
39    /// let r = WsRequest::subscribe("7", "orderbook.1.BTCUSDT");
40    /// assert_eq!(r.to_json(), r#"{"op":"subscribe","req_id":"7","args":["orderbook.1.BTCUSDT"]}"#);
41    /// ```
42    pub fn subscribe(req_id: &str, topic: &str) -> Self {
43        Self {
44            op: "subscribe".into(),
45            req_id: req_id.into(),
46            args: vec![json!(topic)],
47        }
48    }
49
50    /// Cancels the subscription to one topic.
51    pub fn unsubscribe(req_id: &str, topic: &str) -> Self {
52        Self {
53            op: "unsubscribe".into(),
54            req_id: req_id.into(),
55            args: vec![json!(topic)],
56        }
57    }
58
59    /// The heartbeat the exchange asks for every 20 seconds.
60    pub fn ping() -> Self {
61        Self {
62            op: "ping".into(),
63            req_id: String::new(),
64            args: Vec::new(),
65        }
66    }
67
68    /// The login of the private stream: the key, the `expires` time (milliseconds) and the signature.
69    pub fn auth(api_key: &str, api_secret: &str, expires_ms: i64) -> Self {
70        let [key, expires, signature] = crate::sign::ws_auth_args(api_key, api_secret, expires_ms);
71        Self {
72            op: "auth".into(),
73            req_id: String::new(),
74            // The expires time is a number in the login, the key and the signature are texts.
75            args: vec![
76                json!(key),
77                json!(expires.parse::<i64>().unwrap_or(expires_ms)),
78                json!(signature),
79            ],
80        }
81    }
82
83    /// The text that goes on the wire.
84    pub fn to_json(&self) -> String {
85        serde_json::to_string(self).unwrap_or_default()
86    }
87}
88
89/// How a data frame relates to the earlier ones.
90#[derive(Debug, Clone, Copy, PartialEq, Eq)]
91pub enum DataKind {
92    /// The whole state: the order book is replaced.
93    Snapshot,
94    /// A change to apply to the state held.
95    Delta,
96    /// A frame of a private stream: it has no `type`.
97    Event,
98}
99
100/// A frame from the stream.
101#[derive(Debug, Clone, PartialEq)]
102pub enum WsIncoming {
103    /// A connection was made (and, for the private stream, logged in). Sent by the session itself, before the data of
104    /// the restored subscriptions.
105    Connected,
106    /// The answer to a request.
107    Ack {
108        /// `subscribe`, `unsubscribe` or `auth`.
109        op: String,
110        /// The `req_id` of the request, when it had one.
111        req_id: Option<String>,
112        /// Whether the exchange accepted it.
113        success: bool,
114        /// The reason when it did not.
115        message: String,
116    },
117    /// Data of a topic.
118    Data {
119        /// The topic, `orderbook.50.BTCUSDT` or `order`.
120        topic: String,
121        /// Snapshot, delta or event.
122        kind: DataKind,
123        /// The time the exchange made the frame, milliseconds (`ts` of a public frame, `creationTime` of a private one).
124        ts: i64,
125        /// The time of the matching engine, milliseconds, when the frame has it.
126        cts: i64,
127        /// The `data` of the frame.
128        data: Value,
129    },
130    /// The answer to a heartbeat.
131    Pong,
132    /// Something else: kept as text.
133    Unknown(String),
134}
135
136/// Reads one text frame.
137///
138/// # Examples
139///
140/// ```
141/// use brokapi_bybit_protocol::ws::models::{DataKind, WsIncoming, parse_incoming};
142///
143/// let ok = parse_incoming(r#"{"success":true,"ret_msg":"","conn_id":"x","req_id":"q1","op":"subscribe"}"#);
144/// assert!(matches!(ok, WsIncoming::Ack { success: true, .. }));
145///
146/// let data = parse_incoming(r#"{"topic":"orderbook.1.BTCUSDT","ts":1,"type":"snapshot","data":{"s":"BTCUSDT"},"cts":2}"#);
147/// assert!(matches!(data, WsIncoming::Data { kind: DataKind::Snapshot, cts: 2, .. }));
148///
149/// assert_eq!(parse_incoming(r#"{"op":"pong","args":["1"]}"#), WsIncoming::Pong);
150/// ```
151pub fn parse_incoming(text: &str) -> WsIncoming {
152    let Ok(v) = serde_json::from_str::<Value>(text) else {
153        return WsIncoming::Unknown(text.to_string());
154    };
155    let op = v.get("op").and_then(Value::as_str).unwrap_or("");
156    let ret_msg = v.get("ret_msg").and_then(Value::as_str).unwrap_or("");
157    if op == "pong" || ret_msg == "pong" {
158        return WsIncoming::Pong;
159    }
160    if let (Some(topic), Some(data)) = (v.get("topic").and_then(Value::as_str), v.get("data")) {
161        let kind = match v.get("type").and_then(Value::as_str) {
162            Some("snapshot") => DataKind::Snapshot,
163            Some("delta") => DataKind::Delta,
164            _ => DataKind::Event,
165        };
166        let ts = v
167            .get("ts")
168            .or_else(|| v.get("creationTime"))
169            .and_then(Value::as_i64)
170            .unwrap_or(0);
171        return WsIncoming::Data {
172            topic: topic.to_string(),
173            kind,
174            ts,
175            cts: v.get("cts").and_then(Value::as_i64).unwrap_or(0),
176            data: data.clone(),
177        };
178    }
179    if !op.is_empty()
180        && let Some(success) = v.get("success").and_then(Value::as_bool)
181    {
182        return WsIncoming::Ack {
183            op: op.to_string(),
184            req_id: v
185                .get("req_id")
186                .and_then(Value::as_str)
187                .filter(|s| !s.is_empty())
188                .map(str::to_string),
189            success,
190            message: ret_msg.to_string(),
191        };
192    }
193    WsIncoming::Unknown(text.to_string())
194}
195
196#[cfg(test)]
197mod tests {
198    use super::*;
199
200    #[test]
201    fn the_login_carries_the_expires_time_as_a_number() {
202        let r = WsRequest::auth("key", "secret", 1_700_000_000_000);
203        let v: Value = serde_json::from_str(&r.to_json()).unwrap();
204        assert_eq!(v["op"], "auth");
205        assert_eq!(v["args"][0], "key");
206        assert_eq!(v["args"][1], 1_700_000_000_000_i64);
207        assert_eq!(v["args"][2].as_str().unwrap().len(), 64);
208        assert!(v.get("req_id").is_none());
209    }
210
211    #[test]
212    fn a_failed_answer_keeps_the_reason() {
213        let f = parse_incoming(
214            r#"{"success":false,"ret_msg":"error:handler not found,topic:orderbook.7.BTCUSDT","req_id":"b","op":"subscribe"}"#,
215        );
216        match f {
217            WsIncoming::Ack {
218                success,
219                message,
220                req_id,
221                ..
222            } => {
223                assert!(!success);
224                assert!(message.contains("handler not found"));
225                assert_eq!(req_id.as_deref(), Some("b"));
226            }
227            other => panic!("{other:?}"),
228        }
229    }
230
231    #[test]
232    fn a_private_frame_is_an_event_and_a_pong_of_the_public_stream_is_a_pong() {
233        let f = parse_incoming(r#"{"id":"1","topic":"order","creationTime":5,"data":[]}"#);
234        assert!(matches!(
235            f,
236            WsIncoming::Data {
237                kind: DataKind::Event,
238                ts: 5,
239                ..
240            }
241        ));
242        assert_eq!(
243            parse_incoming(
244                r#"{"success":true,"ret_msg":"pong","conn_id":"c","req_id":"","op":"ping"}"#
245            ),
246            WsIncoming::Pong
247        );
248        assert!(matches!(parse_incoming("not json"), WsIncoming::Unknown(_)));
249    }
250}
brokapi_bybit_protocol/ws/models.rs

brokapi_bybit_protocol/ws/
models.rs
