acta_maker_sdk/ws/maker/
data.rs

1use std::ops::Deref;
2use std::sync::Arc;
3use std::time::Duration;
4
5use crate::ws::managed::{
6    MakerWsEndpoint, ManagedMessageReceiver, ManagedWsConfig, ManagedWsError, ManagedWsEvent,
7    ManagedWsHandle, ManagedWsState, SendAwaitError, WaitUntilReadyError, spawn_managed_ws,
8};
9use crate::ws::types::*;
10use tokio::sync::broadcast;
11
12use super::MakerClientError;
13
14mod private {
15    pub trait Sealed {}
16}
17
18/// Request with a statically known data-plane response.
19pub trait MakerDataRequest: private::Sealed + Sized {
20    type Response;
21
22    const RESPONSE_TYPE: &'static str;
23
24    fn into_message(self) -> ClientMessage;
25    /// Extract the typed payload.
26    ///
27    /// # Errors
28    /// Returns the original message when its variant is not this request's
29    /// response.
30    fn into_response(message: ServerMessage) -> Result<Self::Response, Box<ServerMessage>>;
31}
32
33/// A typed response, owning its payload.
34pub struct MakerResponse<T> {
35    payload: T,
36}
37
38impl<T> MakerResponse<T> {
39    #[must_use]
40    pub const fn payload(&self) -> &T {
41        &self.payload
42    }
43
44    #[must_use]
45    pub fn into_payload(self) -> T {
46        self.payload
47    }
48}
49
50impl<T> Deref for MakerResponse<T> {
51    type Target = T;
52
53    fn deref(&self) -> &Self::Target {
54        &self.payload
55    }
56}
57
58#[derive(Debug, thiserror::Error)]
59#[non_exhaustive]
60pub enum MakerDataRequestError {
61    #[error(transparent)]
62    Session(#[from] ManagedWsError),
63    #[error(transparent)]
64    Transport(#[from] SendAwaitError),
65    #[error("server rejected request {request_id}")]
66    Rejected {
67        request_id: uuid::Uuid,
68        response: Arc<ServerMessage>,
69    },
70    #[error("expected {expected}, received a different correlated response")]
71    UnexpectedResponse {
72        expected: &'static str,
73        response: Arc<ServerMessage>,
74    },
75}
76
77impl MakerDataRequestError {
78    /// Borrow the server error without cloning its payload.
79    #[must_use]
80    pub fn server_error(&self) -> Option<&ServerError> {
81        match self {
82            Self::Rejected { response, .. } => match &**response {
83                ServerMessage::RequestError(error) => Some(&error.error),
84                _ => None,
85            },
86            _ => None,
87        }
88    }
89}
90
91/// Maker data-plane client.
92#[derive(Clone)]
93pub struct MakerDataClient {
94    inner: ManagedWsHandle,
95}
96
97impl MakerDataClient {
98    pub fn spawn(config: ManagedWsConfig) -> Result<Self, MakerClientError> {
99        let config = Self::prepare_config(config)?;
100        Ok(Self {
101            inner: spawn_managed_ws(config)?,
102        })
103    }
104
105    fn prepare_config(config: ManagedWsConfig) -> Result<ManagedWsConfig, MakerClientError> {
106        if config.endpoint != MakerWsEndpoint::Data {
107            return Err(MakerClientError::WrongEndpoint {
108                client: "MakerDataClient",
109                required: MakerWsEndpoint::Data,
110                actual: config.endpoint,
111            });
112        }
113        Ok(config)
114    }
115
116    pub async fn request<R: MakerDataRequest>(
117        &self,
118        request: R,
119        timeout: Duration,
120    ) -> Result<MakerResponse<R::Response>, MakerDataRequestError> {
121        let message = self
122            .inner
123            .send_await_ready(request.into_message(), timeout)
124            .await?;
125        if let ServerMessage::RequestError(error) = &*message {
126            return Err(MakerDataRequestError::Rejected {
127                request_id: error.request_id,
128                response: message,
129            });
130        }
131        let message = match Arc::try_unwrap(message) {
132            Ok(owned) => owned,
133            Err(shared) => (*shared).clone(),
134        };
135        match R::into_response(message) {
136            Ok(payload) => Ok(MakerResponse { payload }),
137            Err(other) => Err(MakerDataRequestError::UnexpectedResponse {
138                expected: R::RESPONSE_TYPE,
139                response: Arc::from(other),
140            }),
141        }
142    }
143
144    pub fn subscribe_messages(&self) -> ManagedMessageReceiver {
145        self.inner.subscribe_messages()
146    }
147
148    pub fn subscribe_events(&self) -> broadcast::Receiver<ManagedWsEvent> {
149        self.inner.subscribe_events()
150    }
151
152    #[must_use]
153    pub fn state(&self) -> ManagedWsState {
154        self.inner.state()
155    }
156
157    pub async fn wait_until_ready(&self) -> Result<u64, WaitUntilReadyError> {
158        self.inner.wait_until_ready().await
159    }
160
161    #[must_use]
162    pub const fn raw(&self) -> &ManagedWsHandle {
163        &self.inner
164    }
165
166    #[must_use]
167    pub fn into_raw(self) -> ManagedWsHandle {
168        self.inner
169    }
170
171    pub async fn close(&self) -> Result<(), ManagedWsError> {
172        self.inner.close().await
173    }
174}
175
176macro_rules! data_request {
177    ($request:ty, $client_variant:ident, $response:ty, $server_variant:ident) => {
178        impl private::Sealed for $request {}
179
180        impl MakerDataRequest for $request {
181            type Response = $response;
182
183            const RESPONSE_TYPE: &'static str = stringify!($server_variant);
184
185            fn into_message(self) -> ClientMessage {
186                ClientMessage::$client_variant(self)
187            }
188
189            fn into_response(message: ServerMessage) -> Result<Self::Response, Box<ServerMessage>> {
190                match message {
191                    ServerMessage::$server_variant(response) => Ok(response),
192                    other => Err(Box::new(other)),
193                }
194            }
195        }
196    };
197}
198
199data_request!(
200    GetActiveRfqsMessage,
201    GetActiveRfqs,
202    ActiveRfqsData,
203    ActiveRfqs
204);
205data_request!(GetMarketsMessage, GetMarkets, MarketsData, Markets);
206data_request!(
207    GetMarketDescriptorsMessage,
208    GetMarketDescriptors,
209    MarketDescriptorsData,
210    MarketDescriptors
211);
212data_request!(GetExpiriesMessage, GetExpiries, ExpiriesData, Expiries);
213data_request!(GetTokensMessage, GetTokens, TokensData, Tokens);
214data_request!(
215    GetIndicativePricesMessage,
216    GetIndicativePrices,
217    IndicativePricesMessage,
218    IndicativePrices
219);
220data_request!(
221    GetMakerPositionsMessage,
222    GetMakerPositions,
223    MakerPositionsMessage,
224    MakerPositions
225);
226data_request!(GetMyQuotesMessage, GetMyQuotes, MyQuotesMessage, MyQuotes);
227data_request!(
228    GetOrderStatusMessage,
229    GetOrderStatus,
230    OrderStatusMessage,
231    OrderStatus
232);
233data_request!(
234    GetMarketsForMakerMessage,
235    GetMarketsForMaker,
236    MakerMarketsMessage,
237    MakerMarkets
238);
239data_request!(GetTokenCapsMessage, GetTokenCaps, TokenCapsData, TokenCaps);
240data_request!(GetMyCapsMessage, GetMyCaps, MyCapsData, MyCaps);
241data_request!(CheckQuoteMessage, CheckQuote, QuoteCheckData, QuoteCheck);
242data_request!(GetMyTradesMessage, GetMyTrades, MyTradesMessage, MyTrades);
243data_request!(
244    GetEarnSummaryMessage,
245    GetEarnSummary,
246    EarnSummaryData,
247    EarnSummary
248);
249data_request!(GetMmSummaryMessage, GetMmSummary, MmSummaryData, MmSummary);
250data_request!(
251    GetTokenMarketsInfoMessage,
252    GetTokenMarketsInfo,
253    TokenMarketsInfoData,
254    TokenMarketsInfo
255);
256
257#[cfg(test)]
258mod tests {
259    use std::sync::Arc;
260
261    use super::*;
262    use crate::ws::managed::ManagedWsConfig;
263    use tokio::sync::mpsc::error::TryRecvError;
264
265    struct LeavesReadyWhileEncoding {
266        handle: ManagedWsHandle,
267        request_id: uuid::Uuid,
268    }
269
270    impl private::Sealed for LeavesReadyWhileEncoding {}
271
272    impl MakerDataRequest for LeavesReadyWhileEncoding {
273        type Response = MmSummaryData;
274
275        const RESPONSE_TYPE: &'static str = "MmSummary";
276
277        fn into_message(self) -> ClientMessage {
278            self.handle.inject_state(ManagedWsState::Authenticated {
279                connection_epoch: 7,
280            });
281            ClientMessage::GetMmSummary(GetMmSummaryMessage {
282                request_id: self.request_id,
283            })
284        }
285
286        fn into_response(message: ServerMessage) -> Result<Self::Response, Box<ServerMessage>> {
287            match message {
288                ServerMessage::MmSummary(response) => Ok(response),
289                other => Err(Box::new(other)),
290            }
291        }
292    }
293
294    fn config() -> ManagedWsConfig {
295        ManagedWsConfig::new(
296            "ws://localhost",
297            HelloData {
298                protocol_version: "1.0.0".to_string(),
299                features: Vec::new(),
300                client_name: None,
301                client_version: None,
302            },
303            Arc::new(crate::orders::BytesSigner::from_secret([1u8; 32])),
304        )
305    }
306
307    #[test]
308    fn data_client_rejects_quote_endpoint_before_spawning() {
309        assert!(matches!(
310            MakerDataClient::spawn(config()),
311            Err(MakerClientError::WrongEndpoint {
312                required: MakerWsEndpoint::Data,
313                actual: MakerWsEndpoint::Quote,
314                ..
315            })
316        ));
317    }
318
319    #[test]
320    fn data_request_maps_to_one_typed_response() {
321        let request_id = uuid::Uuid::new_v4();
322        let request = GetActiveRfqsMessage { request_id };
323        assert!(matches!(
324            request.into_message(),
325            ClientMessage::GetActiveRfqs(GetActiveRfqsMessage {
326                request_id: actual
327            }) if actual == request_id
328        ));
329
330        let response = ServerMessage::ActiveRfqs(ActiveRfqsData {
331            request_id,
332            rfqs: Vec::new(),
333        });
334        assert_eq!(
335            <GetActiveRfqsMessage as MakerDataRequest>::into_response(response)
336                .expect("typed response")
337                .request_id,
338            request_id
339        );
340        assert!(matches!(
341            <GetActiveRfqsMessage as MakerDataRequest>::into_response(ServerMessage::RequireInvite),
342            Err(boxed) if matches!(*boxed, ServerMessage::RequireInvite)
343        ));
344    }
345
346    #[tokio::test]
347    async fn typed_request_does_not_enqueue_an_unbound_query_after_leaving_ready() {
348        let (handle, mut commands) = ManagedWsHandle::test_handle(1, 1);
349        handle.inject_state(ManagedWsState::Ready {
350            connection_epoch: 7,
351        });
352        let client = MakerDataClient {
353            inner: handle.clone(),
354        };
355
356        let result = tokio::select! {
357            result = client.request(
358                LeavesReadyWhileEncoding {
359                    handle,
360                    request_id: uuid::Uuid::new_v4(),
361                },
362                Duration::from_secs(1),
363            ) => result,
364            _ = commands.recv() => panic!("typed data request must not enqueue after leaving Ready"),
365        };
366
367        assert!(matches!(
368            result,
369            Err(MakerDataRequestError::Transport(SendAwaitError::NotReady))
370        ));
371        assert!(matches!(commands.try_recv(), Err(TryRecvError::Empty)));
372    }
373}
acta_maker_sdk/ws/maker/data.rs

acta_maker_sdk/ws/maker/
data.rs
