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

acta_maker_sdk/ws/maker/
data.rs
