rust_okx/ws/
event.rs

1//! WebSocket event and response decoding.
2
3use bytes::Bytes;
4use serde::de::DeserializeOwned;
5use serde_json::Value;
6
7use crate::{Error, NumberString};
8
9use super::arg::Arg;
10use super::error::WsError;
11
12/// A WebSocket event surfaced by [`OkxWs::next_event`](crate::ws::OkxWs::next_event).
13///
14/// OKX docs: <https://www.okx.com/docs-v5/en/#overview-websocket>
15#[derive(Debug, Clone)]
16#[non_exhaustive]
17pub enum WsEvent {
18    /// Subscription acknowledgement.
19    Subscribed(Arg),
20    /// Unsubscription acknowledgement.
21    Unsubscribed(Arg),
22    /// Private login acknowledgement.
23    Login,
24    /// OKX WebSocket error event.
25    Error {
26        /// OKX error code.
27        code: String,
28        /// OKX error message.
29        msg: String,
30    },
31    /// WebSocket service notice, usually for service upgrades.
32    Notice(WsNotice),
33    /// Connection-count notification for channels with connection limits.
34    ChannelConnectionCount(WsChannelConnectionCount),
35    /// Connection-count limit error. The server terminated this channel subscription.
36    ChannelConnectionCountError(WsChannelConnectionCount),
37    /// WebSocket operation response, e.g. order placement or cancellation.
38    Operation(WsOperation),
39    /// Channel data push.
40    Push(Push),
41    /// The client reconnected and started subscription recovery.
42    Reconnected,
43    /// The remote side closed the connection.
44    Disconnected,
45}
46
47/// A service notice pushed by OKX.
48///
49/// OKX docs: <https://www.okx.com/docs-v5/en/#overview-websocket-notice>
50#[derive(Debug, Clone, Default)]
51#[non_exhaustive]
52pub struct WsNotice {
53    /// OKX notice code, if supplied.
54    pub code: String,
55    /// OKX notice message.
56    pub msg: String,
57}
58
59/// A channel connection-count notification.
60///
61/// OKX docs: <https://www.okx.com/docs-v5/en/#overview-websocket-connection-count-limit>
62#[derive(Debug, Clone)]
63#[non_exhaustive]
64pub struct WsChannelConnectionCount {
65    /// Channel name associated with the notification.
66    pub channel: String,
67    /// Current connection count for the channel.
68    pub conn_count: NumberString,
69    /// WebSocket connection ID.
70    pub conn_id: String,
71}
72
73/// A WebSocket operation response.
74///
75/// OKX docs: <https://www.okx.com/docs-v5/en/#overview-websocket>
76#[derive(Debug, Clone)]
77#[non_exhaustive]
78pub struct WsOperation {
79    /// Client-provided message ID.
80    pub id: Option<String>,
81    /// OKX operation name.
82    pub op: String,
83    /// Top-level OKX response code.
84    pub code: String,
85    /// Top-level OKX response message.
86    pub msg: String,
87    /// Gateway receive timestamp, when supplied.
88    pub in_time: Option<NumberString>,
89    /// Gateway send timestamp, when supplied.
90    pub out_time: Option<NumberString>,
91    raw: Bytes,
92}
93
94impl WsOperation {
95    /// Deserialize the response `data` array into typed rows.
96    pub fn parse<T: DeserializeOwned>(&self) -> Result<Vec<T>, Error> {
97        serde_json::from_slice(&self.raw)
98            .map_err(|source| WsError::Decode {
99                context: "operation".to_owned(),
100                source,
101            })
102            .map_err(Error::from)
103    }
104
105    /// Return the raw JSON `data` bytes.
106    pub fn raw_data(&self) -> &Bytes {
107        &self.raw
108    }
109}
110
111/// A channel data push.
112///
113/// The concrete rows are available in [`crate::ws::model`] and can be decoded
114/// with [`Push::parse`].
115///
116/// OKX docs: <https://www.okx.com/docs-v5/en/#overview-websocket-subscribe>
117#[derive(Debug, Clone)]
118#[non_exhaustive]
119pub struct Push {
120    /// Channel argument associated with this push.
121    pub arg: Arg,
122    /// OKX push action, usually `snapshot` or `update`.
123    pub action: Option<String>,
124    raw: Bytes,
125}
126
127impl Push {
128    pub(crate) fn new(arg: Arg, action: Option<String>, raw: Bytes) -> Self {
129        Self { arg, action, raw }
130    }
131
132    /// Deserialize the push `data` array into typed rows.
133    pub fn parse<T: DeserializeOwned>(&self) -> Result<Vec<T>, Error> {
134        serde_json::from_slice(&self.raw)
135            .map_err(|source| WsError::Decode {
136                context: "push".to_owned(),
137                source,
138            })
139            .map_err(Error::from)
140    }
141
142    /// Return the raw JSON `data` bytes.
143    pub fn raw_data(&self) -> &Bytes {
144        &self.raw
145    }
146}
147
148pub(crate) fn parse_text_event(text: &str) -> Result<Option<WsEvent>, Error> {
149    let value: Value = serde_json::from_str(text).map_err(|source| WsError::Decode {
150        context: text.chars().take(40).collect(),
151        source,
152    })?;
153    match value.get("event").and_then(Value::as_str) {
154        Some("subscribe") => {
155            let arg = parse_arg(&value)?;
156            Ok(Some(WsEvent::Subscribed(arg)))
157        }
158        Some("unsubscribe") => {
159            let arg = parse_arg(&value)?;
160            Ok(Some(WsEvent::Unsubscribed(arg)))
161        }
162        Some("login") => Ok(Some(WsEvent::Login)),
163        Some("error") => Ok(Some(WsEvent::Error {
164            code: string_field(&value, "code"),
165            msg: string_field(&value, "msg"),
166        })),
167        Some("notice") => Ok(Some(WsEvent::Notice(WsNotice {
168            code: string_field(&value, "code"),
169            msg: string_field(&value, "msg"),
170        }))),
171        Some("channel-conn-count") => Ok(Some(WsEvent::ChannelConnectionCount(
172            parse_channel_connection_count(&value),
173        ))),
174        Some("channel-conn-count-error") => Ok(Some(WsEvent::ChannelConnectionCountError(
175            parse_channel_connection_count(&value),
176        ))),
177        _ if value.get("arg").is_some() && value.get("data").is_some() => {
178            let arg = parse_arg(&value)?;
179            let raw = data_bytes(&value)?;
180            Ok(Some(WsEvent::Push(Push::new(
181                arg,
182                value
183                    .get("action")
184                    .or_else(|| value.get("eventType"))
185                    .and_then(Value::as_str)
186                    .map(ToOwned::to_owned),
187                raw,
188            ))))
189        }
190        _ if value.get("op").is_some() => Ok(Some(WsEvent::Operation(parse_operation(&value)?))),
191        _ => Ok(None),
192    }
193}
194
195fn parse_channel_connection_count(value: &Value) -> WsChannelConnectionCount {
196    WsChannelConnectionCount {
197        channel: string_field(value, "channel"),
198        conn_count: string_field(value, "connCount").into(),
199        conn_id: string_field(value, "connId"),
200    }
201}
202
203fn parse_operation(value: &Value) -> Result<WsOperation, Error> {
204    Ok(WsOperation {
205        id: value
206            .get("id")
207            .and_then(Value::as_str)
208            .map(ToOwned::to_owned),
209        op: string_field(value, "op"),
210        code: string_field(value, "code"),
211        msg: string_field(value, "msg"),
212        in_time: optional_number_string(value, "inTime"),
213        out_time: optional_number_string(value, "outTime"),
214        raw: data_bytes(value)?,
215    })
216}
217
218fn parse_arg(value: &Value) -> Result<Arg, Error> {
219    let arg = value.get("arg").ok_or_else(|| WsError::MissingArg {
220        raw: value_snippet(value),
221    })?;
222    let channel_name = channel_name(value);
223    serde_json::from_value(arg.clone())
224        .map_err(|source| WsError::Decode {
225            context: format!("channel {channel_name}"),
226            source,
227        })
228        .map_err(Error::from)
229}
230
231fn data_bytes(value: &Value) -> Result<Bytes, Error> {
232    let data = value
233        .get("data")
234        .cloned()
235        .unwrap_or_else(|| Value::Array(vec![]));
236    let raw = serde_json::to_vec(&data)
237        .map_err(|source| WsError::Encode { source })
238        .map_err(Error::from)?;
239    Ok(Bytes::from(raw))
240}
241
242fn channel_name(value: &Value) -> String {
243    value
244        .get("arg")
245        .and_then(|arg| arg.get("channel"))
246        .and_then(Value::as_str)
247        .unwrap_or("<unknown>")
248        .to_owned()
249}
250
251fn value_snippet(value: &Value) -> String {
252    raw_snippet(&value.to_string())
253}
254
255fn raw_snippet(raw: &str) -> String {
256    const MAX_RAW_SNIPPET_CHARS: usize = 200;
257    let mut snippet: String = raw.chars().take(MAX_RAW_SNIPPET_CHARS).collect();
258    if raw.chars().count() > MAX_RAW_SNIPPET_CHARS {
259        snippet.push('…');
260    }
261    snippet
262}
263
264fn string_field(value: &Value, field: &str) -> String {
265    value
266        .get(field)
267        .and_then(Value::as_str)
268        .unwrap_or_default()
269        .to_owned()
270}
271
272fn optional_number_string(value: &Value, field: &str) -> Option<NumberString> {
273    value.get(field).and_then(Value::as_str).map(Into::into)
274}
rust_okx/ws/event.rs

rust_okx/ws/
event.rs
