binance_sdk/spot/websocket_api/apis/
user_data_stream_api.rs

1/*
2 * Spot WebSocket API
3 *
4 * Access market data, manage accounts, and trade on Binance Spot.
5 *
6 * The version of the OpenAPI document: 1.0.0
7 *
8 *
9 * NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
10 * https://openapi-generator.tech
11 * Do not edit the class manually.
12 */
13
14#![allow(unused_imports)]
15use anyhow::Context;
16use async_trait::async_trait;
17use derive_builder::Builder;
18use rust_decimal::prelude::*;
19use serde::{Deserialize, Serialize};
20use serde_json::Value;
21use std::{collections::BTreeMap, sync::Arc};
22
23use crate::common::{
24    errors::WebsocketError,
25    models::{ParamBuildError, WebsocketApiResponse},
26    utils::remove_empty_value,
27    websocket::{WebsocketApi, WebsocketMessageSendOptions},
28};
29use crate::spot::websocket_api::models;
30
31#[async_trait]
32pub trait UserDataStreamApi: Send + Sync {
33    async fn session_subscriptions(
34        &self,
35        params: SessionSubscriptionsParams,
36    ) -> anyhow::Result<WebsocketApiResponse<Vec<models::SessionSubscriptionsResponseResultInner>>>;
37    async fn user_data_stream_subscribe(
38        &self,
39        params: UserDataStreamSubscribeParams,
40    ) -> anyhow::Result<WebsocketApiResponse<Box<models::UserDataStreamSubscribeResponseResult>>>;
41    async fn user_data_stream_subscribe_signature(
42        &self,
43        params: UserDataStreamSubscribeSignatureParams,
44    ) -> anyhow::Result<WebsocketApiResponse<Box<models::UserDataStreamSubscribeResponseResult>>>;
45    async fn user_data_stream_unsubscribe(
46        &self,
47        params: UserDataStreamUnsubscribeParams,
48    ) -> anyhow::Result<WebsocketApiResponse<serde_json::Value>>;
49}
50
51#[derive(Clone)]
52pub struct UserDataStreamApiClient {
53    websocket_api_base: Arc<WebsocketApi>,
54}
55
56impl UserDataStreamApiClient {
57    pub fn new(websocket_api_base: Arc<WebsocketApi>) -> Self {
58        Self { websocket_api_base }
59    }
60}
61
62/// Request parameters for the [`session_subscriptions`] operation.
63///
64/// This struct holds all of the inputs you can pass when calling
65/// [`session_subscriptions`](#method.session_subscriptions).
66#[derive(Clone, Debug, Builder, Deserialize, Default)]
67#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
68pub struct SessionSubscriptionsParams {
69    /// Client-generated request identifier.
70    ///
71    /// This field is **optional.
72    #[builder(setter(into), default)]
73    #[serde(rename = "id", default)]
74    pub id: Option<String>,
75}
76
77impl SessionSubscriptionsParams {
78    /// Create a builder for [`session_subscriptions`].
79    ///
80    #[must_use]
81    pub fn builder() -> SessionSubscriptionsParamsBuilder {
82        SessionSubscriptionsParamsBuilder::default()
83    }
84}
85/// Request parameters for the [`user_data_stream_subscribe`] operation.
86///
87/// This struct holds all of the inputs you can pass when calling
88/// [`user_data_stream_subscribe`](#method.user_data_stream_subscribe).
89#[derive(Clone, Debug, Builder, Deserialize, Default)]
90#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
91pub struct UserDataStreamSubscribeParams {
92    /// Client-generated request identifier.
93    ///
94    /// This field is **optional.
95    #[builder(setter(into), default)]
96    #[serde(rename = "id", default)]
97    pub id: Option<String>,
98}
99
100impl UserDataStreamSubscribeParams {
101    /// Create a builder for [`user_data_stream_subscribe`].
102    ///
103    #[must_use]
104    pub fn builder() -> UserDataStreamSubscribeParamsBuilder {
105        UserDataStreamSubscribeParamsBuilder::default()
106    }
107}
108/// Request parameters for the [`user_data_stream_subscribe_signature`] operation.
109///
110/// This struct holds all of the inputs you can pass when calling
111/// [`user_data_stream_subscribe_signature`](#method.user_data_stream_subscribe_signature).
112#[derive(Clone, Debug, Builder, Deserialize, Default)]
113#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
114pub struct UserDataStreamSubscribeSignatureParams {
115    /// Client-generated request identifier.
116    ///
117    /// This field is **optional.
118    #[builder(setter(into), default)]
119    #[serde(rename = "id", default)]
120    pub id: Option<String>,
121    /// Supports up to three decimal places of precision (e.g., 6000.346) so that microseconds may be specified.
122    ///
123    /// This field is **optional.
124    #[builder(setter(into), default)]
125    #[serde(rename = "recvWindow", default)]
126    pub recv_window: Option<rust_decimal::Decimal>,
127}
128
129impl UserDataStreamSubscribeSignatureParams {
130    /// Create a builder for [`user_data_stream_subscribe_signature`].
131    ///
132    #[must_use]
133    pub fn builder() -> UserDataStreamSubscribeSignatureParamsBuilder {
134        UserDataStreamSubscribeSignatureParamsBuilder::default()
135    }
136}
137/// Request parameters for the [`user_data_stream_unsubscribe`] operation.
138///
139/// This struct holds all of the inputs you can pass when calling
140/// [`user_data_stream_unsubscribe`](#method.user_data_stream_unsubscribe).
141#[derive(Clone, Debug, Builder, Deserialize, Default)]
142#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
143pub struct UserDataStreamUnsubscribeParams {
144    /// Client-generated request identifier.
145    ///
146    /// This field is **optional.
147    #[builder(setter(into), default)]
148    #[serde(rename = "id", default)]
149    pub id: Option<String>,
150    /// When called with no parameter, this will close all subscriptions. When called with the subscriptionId parameter, this will attempt to close the subscription with that subscription id, if it exists.
151    ///
152    /// This field is **optional.
153    #[builder(setter(into), default)]
154    #[serde(rename = "subscriptionId", default)]
155    pub subscription_id: Option<i32>,
156}
157
158impl UserDataStreamUnsubscribeParams {
159    /// Create a builder for [`user_data_stream_unsubscribe`].
160    ///
161    #[must_use]
162    pub fn builder() -> UserDataStreamUnsubscribeParamsBuilder {
163        UserDataStreamUnsubscribeParamsBuilder::default()
164    }
165}
166
167#[async_trait]
168impl UserDataStreamApi for UserDataStreamApiClient {
169    async fn session_subscriptions(
170        &self,
171        params: SessionSubscriptionsParams,
172    ) -> anyhow::Result<WebsocketApiResponse<Vec<models::SessionSubscriptionsResponseResultInner>>>
173    {
174        let SessionSubscriptionsParams { id } = params;
175
176        let mut payload: BTreeMap<String, Value> = BTreeMap::new();
177        if let Some(value) = id {
178            payload.insert("id".to_string(), serde_json::json!(value));
179        }
180        let payload = remove_empty_value(payload);
181
182        self.websocket_api_base
183            .send_message::<Vec<models::SessionSubscriptionsResponseResultInner>>(
184                "/session.subscriptions".trim_start_matches('/'),
185                payload,
186                WebsocketMessageSendOptions::new(),
187            )
188            .await
189            .map_err(anyhow::Error::from)?
190            .into_iter()
191            .next()
192            .ok_or(WebsocketError::NoResponse)
193            .map_err(anyhow::Error::from)
194    }
195
196    async fn user_data_stream_subscribe(
197        &self,
198        params: UserDataStreamSubscribeParams,
199    ) -> anyhow::Result<WebsocketApiResponse<Box<models::UserDataStreamSubscribeResponseResult>>>
200    {
201        let UserDataStreamSubscribeParams { id } = params;
202
203        let mut payload: BTreeMap<String, Value> = BTreeMap::new();
204        if let Some(value) = id {
205            payload.insert("id".to_string(), serde_json::json!(value));
206        }
207        let payload = remove_empty_value(payload);
208
209        self.websocket_api_base
210            .send_message::<Box<models::UserDataStreamSubscribeResponseResult>>(
211                "/userDataStream.subscribe".trim_start_matches('/'),
212                payload,
213                WebsocketMessageSendOptions::new(),
214            )
215            .await
216            .map_err(anyhow::Error::from)?
217            .into_iter()
218            .next()
219            .ok_or(WebsocketError::NoResponse)
220            .map_err(anyhow::Error::from)
221    }
222
223    async fn user_data_stream_subscribe_signature(
224        &self,
225        params: UserDataStreamSubscribeSignatureParams,
226    ) -> anyhow::Result<WebsocketApiResponse<Box<models::UserDataStreamSubscribeResponseResult>>>
227    {
228        let UserDataStreamSubscribeSignatureParams { id, recv_window } = params;
229
230        let mut payload: BTreeMap<String, Value> = BTreeMap::new();
231        if let Some(value) = id {
232            payload.insert("id".to_string(), serde_json::json!(value));
233        }
234        if let Some(value) = recv_window {
235            payload.insert("recvWindow".to_string(), serde_json::json!(value));
236        }
237        let payload = remove_empty_value(payload);
238
239        self.websocket_api_base
240            .send_message::<Box<models::UserDataStreamSubscribeResponseResult>>(
241                "/userDataStream.subscribe.signature".trim_start_matches('/'),
242                payload,
243                WebsocketMessageSendOptions::new().signed(),
244            )
245            .await
246            .map_err(anyhow::Error::from)?
247            .into_iter()
248            .next()
249            .ok_or(WebsocketError::NoResponse)
250            .map_err(anyhow::Error::from)
251    }
252
253    async fn user_data_stream_unsubscribe(
254        &self,
255        params: UserDataStreamUnsubscribeParams,
256    ) -> anyhow::Result<WebsocketApiResponse<serde_json::Value>> {
257        let UserDataStreamUnsubscribeParams {
258            id,
259            subscription_id,
260        } = params;
261
262        let mut payload: BTreeMap<String, Value> = BTreeMap::new();
263        if let Some(value) = id {
264            payload.insert("id".to_string(), serde_json::json!(value));
265        }
266        if let Some(value) = subscription_id {
267            payload.insert("subscriptionId".to_string(), serde_json::json!(value));
268        }
269        let payload = remove_empty_value(payload);
270
271        self.websocket_api_base
272            .send_message::<serde_json::Value>(
273                "/userDataStream.unsubscribe".trim_start_matches('/'),
274                payload,
275                WebsocketMessageSendOptions::new(),
276            )
277            .await
278            .map_err(anyhow::Error::from)?
279            .into_iter()
280            .next()
281            .ok_or(WebsocketError::NoResponse)
282            .map_err(anyhow::Error::from)
283    }
284}
285
286#[cfg(all(test, feature = "spot"))]
287mod tests {
288    use super::*;
289    use crate::TOKIO_SHARED_RT;
290    use crate::common::websocket::{WebsocketApi, WebsocketConnection, WebsocketHandler};
291    use crate::config::ConfigurationWebsocketApi;
292    use crate::errors::WebsocketError;
293    use crate::models::WebsocketApiRateLimit;
294    use serde_json::{Value, json};
295    use tokio::spawn;
296    use tokio::sync::mpsc::{UnboundedReceiver, unbounded_channel};
297    use tokio::time::{Duration, timeout};
298    use tokio_tungstenite::tungstenite::Message;
299
300    async fn setup() -> (
301        Arc<WebsocketApi>,
302        Arc<WebsocketConnection>,
303        UnboundedReceiver<Message>,
304    ) {
305        let conn = WebsocketConnection::new("test-conn");
306        let (tx, rx) = unbounded_channel::<Message>();
307        {
308            let mut conn_state = conn.state.lock().await;
309            conn_state.ws_write_tx = Some(tx);
310        }
311
312        let config = ConfigurationWebsocketApi::builder()
313            .api_key("key")
314            .api_secret("secret")
315            .build()
316            .expect("Failed to build configuration");
317        let ws_api = WebsocketApi::new(config, vec![conn.clone()]);
318        conn.set_handler(ws_api.clone() as Arc<dyn WebsocketHandler>)
319            .await;
320        ws_api.clone().connect().await.unwrap();
321
322        (ws_api, conn, rx)
323    }
324
325    #[test]
326    fn session_subscriptions_success() {
327        TOKIO_SHARED_RT.block_on(async {
328            let (ws_api, conn, mut rx) = setup().await;
329            let client = UserDataStreamApiClient::new(ws_api.clone());
330
331            let handle = spawn(async move {
332                let params = SessionSubscriptionsParams::builder().build().unwrap();
333                client.session_subscriptions(params).await
334            });
335
336            let sent = timeout(Duration::from_secs(1), rx.recv()).await.expect("send should occur").expect("channel closed");
337            let Message::Text(text) = sent else { panic!() };
338            let v: Value = serde_json::from_str(&text).unwrap();
339            let id = v["id"].as_str().unwrap();
340            assert_eq!(v["method"], "/session.subscriptions".trim_start_matches('/'));
341            let mut resp_json: Value = serde_json::from_str(r#"{"id":"d3df5a22-88ea-4fe0-9f4e-0fcea5d418b7","status":200,"result":[{"subscriptionId":0}]}"#).unwrap_or_else(|_| serde_json::json!({}));
342            resp_json["id"] = id.into();
343
344            let raw_data = resp_json.get("result").or_else(|| resp_json.get("response")).expect("no response in JSON");
345            let expected_data: Vec<models::SessionSubscriptionsResponseResultInner> = serde_json::from_value(raw_data.clone()).expect("should parse raw response");
346            let empty_array = Value::Array(vec![]);
347            let raw_rate_limits = resp_json.get("rateLimits").unwrap_or(&empty_array);
348            let expected_rate_limits: Option<Vec<WebsocketApiRateLimit>> =
349                match raw_rate_limits.as_array() {
350                    Some(arr) if arr.is_empty() => None,
351                    Some(_) => Some(serde_json::from_value(raw_rate_limits.clone()).expect("should parse rateLimits array")),
352                    None => None,
353                };
354
355            WebsocketHandler::on_message(&*ws_api, resp_json.to_string(), conn.clone()).await;
356
357            let response = timeout(Duration::from_secs(1), handle).await.expect("task done").expect("no panic").expect("no error");
358
359
360            let response_rate_limits = response.rate_limits.clone();
361            let response_data = response.data().expect("deserialize data");
362
363            assert_eq!(response_rate_limits, expected_rate_limits);
364            assert_eq!(response_data, expected_data);
365        });
366    }
367
368    #[test]
369    fn session_subscriptions_error_response() {
370        TOKIO_SHARED_RT.block_on(async {
371            let (ws_api, conn, mut rx) = setup().await;
372            let client = UserDataStreamApiClient::new(ws_api.clone());
373
374            let handle = tokio::spawn(async move {
375                let params = SessionSubscriptionsParams::builder().build().unwrap();
376                client.session_subscriptions(params).await
377            });
378
379            let sent = timeout(Duration::from_secs(1), rx.recv()).await.unwrap().unwrap();
380            let Message::Text(text) = sent else { panic!() };
381            let v: Value = serde_json::from_str(&text).unwrap();
382            let id = v["id"].as_str().unwrap().to_string();
383
384            let resp_json = json!({
385                "id": id,
386                "status": 400,
387                    "error": {
388                        "code": -2010,
389                        "msg": "Account has insufficient balance for requested action.",
390                    },
391                    "rateLimits": [
392                        {
393                            "rateLimitType": "ORDERS",
394                            "interval": "SECOND",
395                            "intervalNum": 10,
396                            "limit": 50,
397                            "count": 13
398                        },
399                    ],
400            });
401            WebsocketHandler::on_message(&*ws_api, resp_json.to_string(), conn.clone()).await;
402
403            let join = timeout(Duration::from_secs(1), handle).await.unwrap();
404            match join {
405                Ok(Err(e)) => {
406                    let msg = e.to_string();
407                    assert!(
408                        msg.contains("Server‐side response error (code -2010): Account has insufficient balance for requested action."),
409                        "Expected error msg to contain server error, got: {msg}"
410                    );
411                }
412                Ok(Ok(_)) => panic!("Expected error"),
413                Err(_) => panic!("Task panicked"),
414            }
415        });
416    }
417
418    #[test]
419    fn session_subscriptions_request_timeout() {
420        TOKIO_SHARED_RT.block_on(async {
421            let (ws_api, _conn, mut rx) = setup().await;
422            let client = UserDataStreamApiClient::new(ws_api.clone());
423
424            let handle = spawn(async move {
425                let params = SessionSubscriptionsParams::builder().build().unwrap();
426                client.session_subscriptions(params).await
427            });
428
429            let sent = timeout(Duration::from_secs(1), rx.recv())
430                .await
431                .expect("send should occur")
432                .expect("channel closed");
433            let Message::Text(text) = sent else {
434                panic!("expected Message Text")
435            };
436
437            let _: Value = serde_json::from_str(&text).unwrap();
438
439            let result = handle.await.expect("task completed");
440            match result {
441                Err(e) => {
442                    if let Some(inner) = e.downcast_ref::<WebsocketError>() {
443                        assert!(matches!(inner, WebsocketError::Timeout));
444                    } else {
445                        panic!("Unexpected error type: {:?}", e);
446                    }
447                }
448                Ok(_) => panic!("Expected timeout error"),
449            }
450        });
451    }
452
453    #[test]
454    fn user_data_stream_subscribe_success() {
455        TOKIO_SHARED_RT.block_on(async {
456            let (ws_api, conn, mut rx) = setup().await;
457            let client = UserDataStreamApiClient::new(ws_api.clone());
458
459            let handle = spawn(async move {
460                let params = UserDataStreamSubscribeParams::builder().build().unwrap();
461                client.user_data_stream_subscribe(params).await
462            });
463
464            let sent = timeout(Duration::from_secs(1), rx.recv()).await.expect("send should occur").expect("channel closed");
465            let Message::Text(text) = sent else { panic!() };
466            let v: Value = serde_json::from_str(&text).unwrap();
467            let id = v["id"].as_str().unwrap();
468            assert_eq!(v["method"], "/userDataStream.subscribe".trim_start_matches('/'));
469            let mut resp_json: Value = serde_json::from_str(r#"{"id":"d3df8a21-98ea-4fe0-8f4e-0fcea5d418b7","status":200,"result":{"subscriptionId":0}}"#).unwrap_or_else(|_| serde_json::json!({}));
470            resp_json["id"] = id.into();
471
472            let raw_data = resp_json.get("result").or_else(|| resp_json.get("response")).expect("no response in JSON");
473            let expected_data: Box<models::UserDataStreamSubscribeResponseResult> = serde_json::from_value(raw_data.clone()).expect("should parse raw response");
474            let empty_array = Value::Array(vec![]);
475            let raw_rate_limits = resp_json.get("rateLimits").unwrap_or(&empty_array);
476            let expected_rate_limits: Option<Vec<WebsocketApiRateLimit>> =
477                match raw_rate_limits.as_array() {
478                    Some(arr) if arr.is_empty() => None,
479                    Some(_) => Some(serde_json::from_value(raw_rate_limits.clone()).expect("should parse rateLimits array")),
480                    None => None,
481                };
482
483            WebsocketHandler::on_message(&*ws_api, resp_json.to_string(), conn.clone()).await;
484
485            let response = timeout(Duration::from_secs(1), handle).await.expect("task done").expect("no panic").expect("no error");
486
487
488            let response_rate_limits = response.rate_limits.clone();
489            let response_data = response.data().expect("deserialize data");
490
491            assert_eq!(response_rate_limits, expected_rate_limits);
492            assert_eq!(response_data, expected_data);
493        });
494    }
495
496    #[test]
497    fn user_data_stream_subscribe_error_response() {
498        TOKIO_SHARED_RT.block_on(async {
499            let (ws_api, conn, mut rx) = setup().await;
500            let client = UserDataStreamApiClient::new(ws_api.clone());
501
502            let handle = tokio::spawn(async move {
503                let params = UserDataStreamSubscribeParams::builder().build().unwrap();
504                client.user_data_stream_subscribe(params).await
505            });
506
507            let sent = timeout(Duration::from_secs(1), rx.recv()).await.unwrap().unwrap();
508            let Message::Text(text) = sent else { panic!() };
509            let v: Value = serde_json::from_str(&text).unwrap();
510            let id = v["id"].as_str().unwrap().to_string();
511
512            let resp_json = json!({
513                "id": id,
514                "status": 400,
515                    "error": {
516                        "code": -2010,
517                        "msg": "Account has insufficient balance for requested action.",
518                    },
519                    "rateLimits": [
520                        {
521                            "rateLimitType": "ORDERS",
522                            "interval": "SECOND",
523                            "intervalNum": 10,
524                            "limit": 50,
525                            "count": 13
526                        },
527                    ],
528            });
529            WebsocketHandler::on_message(&*ws_api, resp_json.to_string(), conn.clone()).await;
530
531            let join = timeout(Duration::from_secs(1), handle).await.unwrap();
532            match join {
533                Ok(Err(e)) => {
534                    let msg = e.to_string();
535                    assert!(
536                        msg.contains("Server‐side response error (code -2010): Account has insufficient balance for requested action."),
537                        "Expected error msg to contain server error, got: {msg}"
538                    );
539                }
540                Ok(Ok(_)) => panic!("Expected error"),
541                Err(_) => panic!("Task panicked"),
542            }
543        });
544    }
545
546    #[test]
547    fn user_data_stream_subscribe_request_timeout() {
548        TOKIO_SHARED_RT.block_on(async {
549            let (ws_api, _conn, mut rx) = setup().await;
550            let client = UserDataStreamApiClient::new(ws_api.clone());
551
552            let handle = spawn(async move {
553                let params = UserDataStreamSubscribeParams::builder().build().unwrap();
554                client.user_data_stream_subscribe(params).await
555            });
556
557            let sent = timeout(Duration::from_secs(1), rx.recv())
558                .await
559                .expect("send should occur")
560                .expect("channel closed");
561            let Message::Text(text) = sent else {
562                panic!("expected Message Text")
563            };
564
565            let _: Value = serde_json::from_str(&text).unwrap();
566
567            let result = handle.await.expect("task completed");
568            match result {
569                Err(e) => {
570                    if let Some(inner) = e.downcast_ref::<WebsocketError>() {
571                        assert!(matches!(inner, WebsocketError::Timeout));
572                    } else {
573                        panic!("Unexpected error type: {:?}", e);
574                    }
575                }
576                Ok(_) => panic!("Expected timeout error"),
577            }
578        });
579    }
580
581    #[test]
582    fn user_data_stream_subscribe_signature_success() {
583        TOKIO_SHARED_RT.block_on(async {
584            let (ws_api, conn, mut rx) = setup().await;
585            let client = UserDataStreamApiClient::new(ws_api.clone());
586
587            let handle = spawn(async move {
588                let params = UserDataStreamSubscribeSignatureParams::builder().build().unwrap();
589                client.user_data_stream_subscribe_signature(params).await
590            });
591
592            let sent = timeout(Duration::from_secs(1), rx.recv()).await.expect("send should occur").expect("channel closed");
593            let Message::Text(text) = sent else { panic!() };
594            let v: Value = serde_json::from_str(&text).unwrap();
595            let id = v["id"].as_str().unwrap();
596            assert_eq!(v["method"], "/userDataStream.subscribe.signature".trim_start_matches('/'));
597            let mut resp_json: Value = serde_json::from_str(r#"{"id":"d3df8a22-98ea-4fe0-9f4e-0fcea5d418b7","status":200,"result":{"subscriptionId":0}}"#).unwrap_or_else(|_| serde_json::json!({}));
598            resp_json["id"] = id.into();
599
600            let raw_data = resp_json.get("result").or_else(|| resp_json.get("response")).expect("no response in JSON");
601            let expected_data: Box<models::UserDataStreamSubscribeResponseResult> = serde_json::from_value(raw_data.clone()).expect("should parse raw response");
602            let empty_array = Value::Array(vec![]);
603            let raw_rate_limits = resp_json.get("rateLimits").unwrap_or(&empty_array);
604            let expected_rate_limits: Option<Vec<WebsocketApiRateLimit>> =
605                match raw_rate_limits.as_array() {
606                    Some(arr) if arr.is_empty() => None,
607                    Some(_) => Some(serde_json::from_value(raw_rate_limits.clone()).expect("should parse rateLimits array")),
608                    None => None,
609                };
610
611            WebsocketHandler::on_message(&*ws_api, resp_json.to_string(), conn.clone()).await;
612
613            let response = timeout(Duration::from_secs(1), handle).await.expect("task done").expect("no panic").expect("no error");
614
615
616            let response_rate_limits = response.rate_limits.clone();
617            let response_data = response.data().expect("deserialize data");
618
619            assert_eq!(response_rate_limits, expected_rate_limits);
620            assert_eq!(response_data, expected_data);
621        });
622    }
623
624    #[test]
625    fn user_data_stream_subscribe_signature_error_response() {
626        TOKIO_SHARED_RT.block_on(async {
627            let (ws_api, conn, mut rx) = setup().await;
628            let client = UserDataStreamApiClient::new(ws_api.clone());
629
630            let handle = tokio::spawn(async move {
631                let params = UserDataStreamSubscribeSignatureParams::builder().build().unwrap();
632                client.user_data_stream_subscribe_signature(params).await
633            });
634
635            let sent = timeout(Duration::from_secs(1), rx.recv()).await.unwrap().unwrap();
636            let Message::Text(text) = sent else { panic!() };
637            let v: Value = serde_json::from_str(&text).unwrap();
638            let id = v["id"].as_str().unwrap().to_string();
639
640            let resp_json = json!({
641                "id": id,
642                "status": 400,
643                    "error": {
644                        "code": -2010,
645                        "msg": "Account has insufficient balance for requested action.",
646                    },
647                    "rateLimits": [
648                        {
649                            "rateLimitType": "ORDERS",
650                            "interval": "SECOND",
651                            "intervalNum": 10,
652                            "limit": 50,
653                            "count": 13
654                        },
655                    ],
656            });
657            WebsocketHandler::on_message(&*ws_api, resp_json.to_string(), conn.clone()).await;
658
659            let join = timeout(Duration::from_secs(1), handle).await.unwrap();
660            match join {
661                Ok(Err(e)) => {
662                    let msg = e.to_string();
663                    assert!(
664                        msg.contains("Server‐side response error (code -2010): Account has insufficient balance for requested action."),
665                        "Expected error msg to contain server error, got: {msg}"
666                    );
667                }
668                Ok(Ok(_)) => panic!("Expected error"),
669                Err(_) => panic!("Task panicked"),
670            }
671        });
672    }
673
674    #[test]
675    fn user_data_stream_subscribe_signature_request_timeout() {
676        TOKIO_SHARED_RT.block_on(async {
677            let (ws_api, _conn, mut rx) = setup().await;
678            let client = UserDataStreamApiClient::new(ws_api.clone());
679
680            let handle = spawn(async move {
681                let params = UserDataStreamSubscribeSignatureParams::builder()
682                    .build()
683                    .unwrap();
684                client.user_data_stream_subscribe_signature(params).await
685            });
686
687            let sent = timeout(Duration::from_secs(1), rx.recv())
688                .await
689                .expect("send should occur")
690                .expect("channel closed");
691            let Message::Text(text) = sent else {
692                panic!("expected Message Text")
693            };
694
695            let _: Value = serde_json::from_str(&text).unwrap();
696
697            let result = handle.await.expect("task completed");
698            match result {
699                Err(e) => {
700                    if let Some(inner) = e.downcast_ref::<WebsocketError>() {
701                        assert!(matches!(inner, WebsocketError::Timeout));
702                    } else {
703                        panic!("Unexpected error type: {:?}", e);
704                    }
705                }
706                Ok(_) => panic!("Expected timeout error"),
707            }
708        });
709    }
710
711    #[test]
712    fn user_data_stream_unsubscribe_success() {
713        TOKIO_SHARED_RT.block_on(async {
714            let (ws_api, conn, mut rx) = setup().await;
715            let client = UserDataStreamApiClient::new(ws_api.clone());
716
717            let handle = spawn(async move {
718                let params = UserDataStreamUnsubscribeParams::builder().build().unwrap();
719                client.user_data_stream_unsubscribe(params).await
720            });
721
722            let sent = timeout(Duration::from_secs(1), rx.recv())
723                .await
724                .expect("send should occur")
725                .expect("channel closed");
726            let Message::Text(text) = sent else { panic!() };
727            let v: Value = serde_json::from_str(&text).unwrap();
728            let id = v["id"].as_str().unwrap();
729            assert_eq!(
730                v["method"],
731                "/userDataStream.unsubscribe".trim_start_matches('/')
732            );
733            let mut resp_json: Value = serde_json::from_str(
734                r#"{"id":"d3df8a21-98ea-4fe0-8f4e-0fcea5d418b7","status":200,"result":{}}"#,
735            )
736            .unwrap_or_else(|_| serde_json::json!({}));
737            resp_json["id"] = id.into();
738
739            let raw_data = resp_json
740                .get("result")
741                .or_else(|| resp_json.get("response"))
742                .expect("no response in JSON");
743            let expected_data: serde_json::Value =
744                serde_json::from_value(raw_data.clone()).expect("should parse raw response");
745            let empty_array = Value::Array(vec![]);
746            let raw_rate_limits = resp_json.get("rateLimits").unwrap_or(&empty_array);
747            let expected_rate_limits: Option<Vec<WebsocketApiRateLimit>> =
748                match raw_rate_limits.as_array() {
749                    Some(arr) if arr.is_empty() => None,
750                    Some(_) => Some(
751                        serde_json::from_value(raw_rate_limits.clone())
752                            .expect("should parse rateLimits array"),
753                    ),
754                    None => None,
755                };
756
757            WebsocketHandler::on_message(&*ws_api, resp_json.to_string(), conn.clone()).await;
758
759            let response = timeout(Duration::from_secs(1), handle)
760                .await
761                .expect("task done")
762                .expect("no panic")
763                .expect("no error");
764
765            let response_rate_limits = response.rate_limits.clone();
766            let response_data = response.data().expect("deserialize data");
767
768            assert_eq!(response_rate_limits, expected_rate_limits);
769            assert_eq!(response_data, expected_data);
770        });
771    }
772
773    #[test]
774    fn user_data_stream_unsubscribe_error_response() {
775        TOKIO_SHARED_RT.block_on(async {
776            let (ws_api, conn, mut rx) = setup().await;
777            let client = UserDataStreamApiClient::new(ws_api.clone());
778
779            let handle = tokio::spawn(async move {
780                let params = UserDataStreamUnsubscribeParams::builder().build().unwrap();
781                client.user_data_stream_unsubscribe(params).await
782            });
783
784            let sent = timeout(Duration::from_secs(1), rx.recv()).await.unwrap().unwrap();
785            let Message::Text(text) = sent else { panic!() };
786            let v: Value = serde_json::from_str(&text).unwrap();
787            let id = v["id"].as_str().unwrap().to_string();
788
789            let resp_json = json!({
790                "id": id,
791                "status": 400,
792                    "error": {
793                        "code": -2010,
794                        "msg": "Account has insufficient balance for requested action.",
795                    },
796                    "rateLimits": [
797                        {
798                            "rateLimitType": "ORDERS",
799                            "interval": "SECOND",
800                            "intervalNum": 10,
801                            "limit": 50,
802                            "count": 13
803                        },
804                    ],
805            });
806            WebsocketHandler::on_message(&*ws_api, resp_json.to_string(), conn.clone()).await;
807
808            let join = timeout(Duration::from_secs(1), handle).await.unwrap();
809            match join {
810                Ok(Err(e)) => {
811                    let msg = e.to_string();
812                    assert!(
813                        msg.contains("Server‐side response error (code -2010): Account has insufficient balance for requested action."),
814                        "Expected error msg to contain server error, got: {msg}"
815                    );
816                }
817                Ok(Ok(_)) => panic!("Expected error"),
818                Err(_) => panic!("Task panicked"),
819            }
820        });
821    }
822
823    #[test]
824    fn user_data_stream_unsubscribe_request_timeout() {
825        TOKIO_SHARED_RT.block_on(async {
826            let (ws_api, _conn, mut rx) = setup().await;
827            let client = UserDataStreamApiClient::new(ws_api.clone());
828
829            let handle = spawn(async move {
830                let params = UserDataStreamUnsubscribeParams::builder().build().unwrap();
831                client.user_data_stream_unsubscribe(params).await
832            });
833
834            let sent = timeout(Duration::from_secs(1), rx.recv())
835                .await
836                .expect("send should occur")
837                .expect("channel closed");
838            let Message::Text(text) = sent else {
839                panic!("expected Message Text")
840            };
841
842            let _: Value = serde_json::from_str(&text).unwrap();
843
844            let result = handle.await.expect("task completed");
845            match result {
846                Err(e) => {
847                    if let Some(inner) = e.downcast_ref::<WebsocketError>() {
848                        assert!(matches!(inner, WebsocketError::Timeout));
849                    } else {
850                        panic!("Unexpected error type: {:?}", e);
851                    }
852                }
853                Ok(_) => panic!("Expected timeout error"),
854            }
855        });
856    }
857}
binance_sdk/spot/websocket_api/apis/user_data_stream_api.rs

binance_sdk/spot/websocket_api/apis/
user_data_stream_api.rs
