df_hyperliquid_core/live/frame/
read.rs

1//! Reading one text frame ([`decode`]): its channel names the type its data reads into. A frame
2//! that does not read is a [`BadFrame`], naming the channel when it got that far.
3
4use std::{borrow::Cow, fmt};
5
6use crate::client::ExchangeResponse;
7use crate::live::{Ack, Answer, Frame, Method, Refusal, Reply};
8use crate::markets::context::{PerpCtx, SpotCtx};
9use crate::utils::Json;
10use serde::Deserialize;
11use serde_json::value::RawValue;
12
13/// A frame that did not read: the channel it named, when it got that far, and why.
14#[derive(Debug, Clone, PartialEq, Eq)]
15pub struct BadFrame {
16    pub channel: Option<String>,
17    pub detail: String,
18}
19
20impl fmt::Display for BadFrame {
21    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
22        match &self.channel {
23            Some(channel) => write!(f, "a `{channel}` frame did not read: {}", self.detail),
24            None => write!(f, "not a frame: {}", self.detail),
25        }
26    }
27}
28
29impl std::error::Error for BadFrame {}
30
31#[derive(Deserialize)]
32struct Head<'a> {
33    #[serde(borrow)]
34    channel: Cow<'a, str>,
35    #[serde(default, borrow)]
36    data: Option<&'a RawValue>,
37}
38
39#[derive(Deserialize)]
40struct RawAck<'a> {
41    method: Method,
42    #[serde(borrow)]
43    subscription: &'a RawValue,
44}
45
46#[derive(Deserialize)]
47struct RawReply<'a> {
48    id: u64,
49    #[serde(borrow)]
50    response: RawResponse<'a>,
51}
52
53#[derive(Deserialize)]
54struct RawResponse<'a> {
55    #[serde(rename = "type", borrow)]
56    kind: Cow<'a, str>,
57    #[serde(borrow)]
58    payload: &'a RawValue,
59}
60
61#[derive(Deserialize)]
62struct RawInfo<'a> {
63    #[serde(rename = "type")]
64    kind: String,
65    #[serde(default, borrow)]
66    data: Option<&'a RawValue>,
67}
68
69#[derive(Deserialize)]
70struct Active<C> {
71    coin: String,
72    ctx: C,
73}
74
75#[derive(Deserialize)]
76struct DexCtxs {
77    dex: String,
78    ctxs: Vec<PerpCtx>,
79}
80
81#[derive(Deserialize)]
82struct EveryDexCtxs {
83    ctxs: Vec<(String, Vec<PerpCtx>)>,
84}
85
86#[derive(Deserialize)]
87struct Note {
88    notification: String,
89}
90
91/// Reads one text frame.
92pub fn decode(text: &str) -> Result<Frame, BadFrame> {
93    let head: Head<'_> = serde_json::from_str(text)
94        .map_err(|error| BadFrame { channel: None, detail: format!("{error} in {}", shown(text)) })?;
95    let channel = head.channel.as_ref();
96    let data = head.data.map(RawValue::get);
97    let typed = Typed { channel, data };
98    Ok(match channel {
99        "pong" => Frame::Pong,
100        "subscriptionResponse" => {
101            let ack: RawAck<'_> = typed.read()?;
102            let echo = Json::parse(ack.subscription.get()).map_err(|error| typed.fail(error.to_string()))?;
103            Frame::Ack(Ack { method: ack.method, echo })
104        }
105        "error" => Frame::Error(Refusal { message: typed.read()? }),
106        "post" => Frame::Post(typed.reply()?),
107        "allMids" => Frame::AllMids(typed.read()?),
108        "l2Book" => Frame::Book(typed.read()?),
109        "trades" => Frame::Trades(typed.read()?),
110        "bbo" => Frame::Bbo(typed.read()?),
111        "candle" => Frame::Candle(typed.read()?),
112        "activeAssetCtx" => {
113            let active: Active<PerpCtx> = typed.read()?;
114            Frame::AssetCtx { coin: active.coin, ctx: active.ctx }
115        }
116        "activeSpotAssetCtx" => {
117            let active: Active<SpotCtx> = typed.read()?;
118            Frame::SpotAssetCtx { coin: active.coin, ctx: active.ctx }
119        }
120        "activeAssetData" => Frame::AssetData(typed.read()?),
121        "userFills" => Frame::UserFills(typed.read()?),
122        "orderUpdates" => Frame::OrderUpdates(typed.read()?),
123        "user" => Frame::UserEvent(typed.read()?),
124        "userFundings" => Frame::UserFundings(typed.read()?),
125        "userNonFundingLedgerUpdates" => Frame::Ledger(typed.read()?),
126        "clearinghouseState" => Frame::ClearinghouseState(typed.read()?),
127        "allDexsClearinghouseState" => Frame::AllDexsClearinghouseState(typed.read()?),
128        "openOrders" => Frame::OpenOrders(typed.read()?),
129        "spotState" => Frame::SpotState(typed.read()?),
130        "twapStates" => Frame::TwapStates(typed.read()?),
131        "userTwapSliceFills" => Frame::TwapSliceFills(typed.read()?),
132        "userTwapHistory" => Frame::TwapHistory(typed.read()?),
133        "notification" => Frame::Notification(typed.read::<Note>()?.notification),
134        "webData3" => Frame::WebData3(typed.read()?),
135        "allDexsAssetCtxs" => Frame::AllDexsAssetCtxs(typed.read::<EveryDexCtxs>()?.ctxs),
136        "assetCtxs" => {
137            let dex: DexCtxs = typed.read()?;
138            Frame::AssetCtxs { dex: dex.dex, ctxs: dex.ctxs }
139        }
140        "spotAssetCtxs" => Frame::SpotAssetCtxs(typed.read()?),
141        "outcomeMetaUpdates" => {
142            Frame::OutcomeMetaUpdates(Json::parse(typed.data()?).map_err(|error| typed.fail(error.to_string()))?)
143        }
144        "fastAssetCtxs" => Frame::FastAssetCtxs(typed.read()?),
145        "userHistoricalOrders" => Frame::HistoricalOrders(typed.read()?),
146        other => Frame::Other { channel: other.to_owned(), data: data.unwrap_or_default().to_owned() },
147    })
148}
149
150/// A frame's data, read for the channel that named it.
151struct Typed<'a> {
152    channel: &'a str,
153    data: Option<&'a str>,
154}
155
156impl<'a> Typed<'a> {
157    fn fail(&self, detail: String) -> BadFrame {
158        BadFrame { channel: Some(self.channel.to_owned()), detail }
159    }
160
161    fn data(&self) -> Result<&'a str, BadFrame> {
162        self.data.ok_or_else(|| self.fail("no data".to_owned()))
163    }
164
165    fn read<T: Deserialize<'a>>(&self) -> Result<T, BadFrame> {
166        let data = self.data()?;
167        serde_json::from_str(data).map_err(|error| self.fail(format!("{error} in {}", shown(data))))
168    }
169
170    fn reply(&self) -> Result<Reply, BadFrame> {
171        let data = self.data()?;
172        let reply: RawReply<'_> =
173            serde_json::from_str(data).map_err(|error| self.fail(format!("{error} in {}", shown(data))))?;
174        let payload = reply.response.payload.get();
175        let answer = match reply.response.kind.as_ref() {
176            "info" => {
177                let info: RawInfo<'_> = serde_json::from_str(payload).map_err(|error| self.fail(error.to_string()))?;
178                Answer::Info { kind: info.kind, data: info.data.map_or("null", RawValue::get).to_owned() }
179            }
180            "action" => Answer::Action(ExchangeResponse::parse(payload).map_err(|error| self.fail(error.to_string()))?),
181            "error" => Answer::Error(serde_json::from_str(payload).map_err(|error| self.fail(error.to_string()))?),
182            other => return Err(self.fail(format!("a post answered as `{other}`"))),
183        };
184        Ok(Reply { id: reply.id, answer })
185    }
186}
187
188/// The start of a text, for an error message.
189fn shown(text: &str) -> String {
190    let start: String = text.chars().take(160).collect();
191    match start.len() < text.len() {
192        true => format!("{start}..."),
193        false => start,
194    }
195}
196
197#[cfg(test)]
198#[path = "tests/read.rs"]
199mod tests;
df_hyperliquid_core/live/frame/read.rs

df_hyperliquid_core/live/frame/
read.rs
