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

acta_maker_sdk/ws/maker/
data.rs
