binance_sdk/stocks/websocket_streams/apis/
user_streams_api.rs

1/*
2 * Stocks Trading WebSocket Streams
3 *
4 * WebSocket stream definitions for Binance Stocks Trading. Base URL: wss://nbstream.binance.com/equity
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 async_trait::async_trait;
16use derive_builder::Builder;
17use serde::{Deserialize, Serialize};
18use serde_json::Value;
19use std::{collections::HashMap, sync::Arc};
20
21use crate::common::{
22    models::ParamBuildError,
23    utils::replace_websocket_streams_placeholders,
24    websocket::{WebsocketBase, WebsocketStream, WebsocketStreams, create_stream_handler},
25};
26use crate::models::StreamId;
27use crate::stocks::websocket_streams::models;
28
29#[async_trait]
30pub trait UserStreamsApi: Send + Sync {
31    async fn order_report_stream(
32        &self,
33        params: OrderReportStreamParams,
34    ) -> anyhow::Result<Arc<WebsocketStream<models::OrderReportStreamResponse>>>;
35}
36
37pub struct UserStreamsApiClient {
38    websocket_streams_base: Arc<WebsocketStreams>,
39}
40
41impl UserStreamsApiClient {
42    pub fn new(websocket_streams_base: Arc<WebsocketStreams>) -> Self {
43        Self {
44            websocket_streams_base,
45        }
46    }
47}
48
49/// Request parameters for the [`order_report_stream`] operation.
50///
51/// This struct holds all of the inputs you can pass when calling
52/// [`order_report_stream`](#method.order_report_stream).
53#[derive(Clone, Debug, Builder, Deserialize)]
54#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
55pub struct OrderReportStreamParams {
56    /// User data listen key obtained from the Listen Key endpoint.
57    ///
58    /// This field is **required.
59    #[builder(setter(into))]
60    #[serde(rename = "listenKey")]
61    pub listen_key: String,
62    /// Unique WebSocket request ID.
63    ///
64    /// This field is **optional.
65    #[builder(setter(into), default)]
66    #[serde(rename = "id", default)]
67    pub id: Option<String>,
68}
69
70impl OrderReportStreamParams {
71    /// Create a builder for [`order_report_stream`].
72    ///
73    /// Required parameters:
74    ///
75    /// * `listen_key` — User data listen key obtained from the Listen Key endpoint.
76    ///
77    #[must_use]
78    pub fn builder(listen_key: String) -> OrderReportStreamParamsBuilder {
79        OrderReportStreamParamsBuilder::default().listen_key(listen_key)
80    }
81}
82
83#[async_trait]
84impl UserStreamsApi for UserStreamsApiClient {
85    async fn order_report_stream(
86        &self,
87        params: OrderReportStreamParams,
88    ) -> anyhow::Result<Arc<WebsocketStream<models::OrderReportStreamResponse>>> {
89        let OrderReportStreamParams { listen_key, id } = params;
90
91        let pairs: &[(&str, Option<String>)] =
92            &[("listenKey", Some(listen_key.clone())), ("id", id.clone())];
93
94        let vars: HashMap<_, _> = pairs
95            .iter()
96            .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
97            .collect();
98
99        let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);
100
101        let stream = replace_websocket_streams_placeholders("/<listenKey>@orderReport", &vars);
102
103        Ok(create_stream_handler::<models::OrderReportStreamResponse>(
104            WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
105            stream,
106            id_opt.map(|s| {
107                if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
108                    if let Ok(n) = s.parse::<u32>() {
109                        return StreamId::Number(n);
110                    }
111                }
112                StreamId::Str(s)
113            }),
114            None,
115        )
116        .await)
117    }
118}
119
120#[cfg(all(test, feature = "stocks"))]
121mod tests {
122    use super::*;
123    use crate::TOKIO_SHARED_RT;
124    use crate::{
125        common::websocket::{WebsocketConnection, WebsocketHandler},
126        config::ConfigurationWebsocketStreams,
127    };
128    use serde_json::json;
129    use std::sync::atomic::{AtomicBool, Ordering};
130    use tokio::task::yield_now;
131
132    async fn make_streams_base() -> (Arc<WebsocketStreams>, Arc<WebsocketConnection>) {
133        let conn = WebsocketConnection::new("test");
134        let config = ConfigurationWebsocketStreams::builder()
135            .build()
136            .expect("Failed to build configuration");
137        let streams_base = WebsocketStreams::new(config, vec![conn.clone()], vec![]);
138        conn.set_handler(streams_base.clone() as Arc<dyn WebsocketHandler>)
139            .await;
140        (streams_base, conn)
141    }
142
143    #[test]
144    fn order_report_stream_should_execute_successfully() {
145        TOKIO_SHARED_RT.block_on(async {
146            let (streams_base, _) = make_streams_base().await;
147            let api = UserStreamsApiClient::new(streams_base.clone());
148
149            let id = "test-id-123".to_string();
150
151            let params = OrderReportStreamParams::builder(
152                "pqia91ma19a5s61cv6a81va65sdf19v8a65a1a5s6af0dkfj2a97b8a91d".to_string(),
153            )
154            .id(Some(id.clone()))
155            .build()
156            .unwrap();
157
158            let OrderReportStreamParams { listen_key, id } = params.clone();
159
160            let pairs: &[(&str, Option<String>)] =
161                &[("listenKey", Some(listen_key.clone())), ("id", id.clone())];
162
163            let vars: HashMap<_, _> = pairs
164                .iter()
165                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
166                .collect();
167            let stream = replace_websocket_streams_placeholders("/<listenKey>@orderReport", &vars);
168            let ws_stream = api
169                .order_report_stream(params)
170                .await
171                .expect("order_report_stream should return a WebsocketStream");
172
173            assert!(
174                streams_base.is_subscribed(&stream).await,
175                "expected stream '{stream}' to be subscribed"
176            );
177            assert_eq!(ws_stream.id, Some(StreamId::Str("test-id-123".to_string())));
178        });
179    }
180
181    #[test]
182    fn order_report_stream_should_handle_incoming_message() {
183        TOKIO_SHARED_RT.block_on(async {
184            let (streams_base, conn) = make_streams_base().await;
185            let api = UserStreamsApiClient::new(streams_base.clone());
186
187            let id = "test-id-123".to_string();
188
189            let params = OrderReportStreamParams::builder("pqia91ma19a5s61cv6a81va65sdf19v8a65a1a5s6af0dkfj2a97b8a91d".to_string(),).id(Some(id.clone())).build().unwrap();
190
191            let OrderReportStreamParams {
192                listen_key,id,
193            } = params.clone();
194
195            let pairs: &[(&str, Option<String>)] = &[
196                ("listenKey",
197                        Some(listen_key.clone())
198                ),
199                ("id",
200                        id.clone()
201                ),
202            ];
203
204            let vars: HashMap<_, _> = pairs
205                .iter()
206                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
207                .collect();
208            let stream = replace_websocket_streams_placeholders("/<listenKey>@orderReport", &vars);
209
210            let ws_stream = api.order_report_stream(params).await.unwrap();
211
212            let called = Arc::new(AtomicBool::new(false));
213            let called_with_message = called.clone();
214            ws_stream.on_message(move |_payload: models::OrderReportStreamResponse| {
215                called_with_message.store(true, Ordering::SeqCst);
216            });
217
218            let payload: Value = serde_json::from_str(r#"{"e":"orderReport","E":1710000000000,"x":"ORDER_UPDATE","i":"b0b6dd9d-8b9b-48a9-ba46-b9d54906e415","ai":"4ce9353c-66d1-46c2-898f-fce867ab0247","b":"EQ_AAPL","q":"USD","S":"buy","o":"limit","p":185.5,"Q":10,"N":1,"fq":5,"FN":927.5,"tc":1856.2,"Z":50,"n":"Regular","s":"partially_filled","T":1710000000000,"U":1710000060000}"#).unwrap_or_else(|_| serde_json::json!({}));
219            let msg = json!({
220                "stream": stream,
221                "data": payload,
222            });
223
224            streams_base.on_message(msg.to_string(), conn.clone()).await;
225            yield_now().await;
226
227            assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
228        });
229    }
230
231    #[test]
232    fn order_report_stream_should_not_fire_after_unsubscribe() {
233        TOKIO_SHARED_RT.block_on(async {
234            let (streams_base, conn) = make_streams_base().await;
235            let api = UserStreamsApiClient::new(streams_base.clone());
236
237            let id = "test-id-123".to_string();
238
239            let params = OrderReportStreamParams::builder("pqia91ma19a5s61cv6a81va65sdf19v8a65a1a5s6af0dkfj2a97b8a91d".to_string(),).id(Some(id.clone())).build().unwrap();
240
241            let OrderReportStreamParams {
242                listen_key,id,
243            } = params.clone();
244
245            let pairs: &[(&str, Option<String>)] = &[
246                ("listenKey",
247                        Some(listen_key.clone())
248                ),
249                ("id",
250                        id.clone()
251                ),
252            ];
253
254            let vars: HashMap<_, _> = pairs
255                .iter()
256                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
257                .collect();
258            let stream = replace_websocket_streams_placeholders("/<listenKey>@orderReport", &vars);
259
260            let ws_stream = api.order_report_stream(params).await.unwrap();
261
262            let called = Arc::new(AtomicBool::new(false));
263            let called_clone = called.clone();
264            ws_stream.on_message(move |_payload: models::OrderReportStreamResponse| {
265                called_clone.store(true, Ordering::SeqCst);
266            });
267
268            assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");
269
270            ws_stream.unsubscribe().await;
271
272            let payload: Value = serde_json::from_str(r#"{"e":"orderReport","E":1710000000000,"x":"ORDER_UPDATE","i":"b0b6dd9d-8b9b-48a9-ba46-b9d54906e415","ai":"4ce9353c-66d1-46c2-898f-fce867ab0247","b":"EQ_AAPL","q":"USD","S":"buy","o":"limit","p":185.5,"Q":10,"N":1,"fq":5,"FN":927.5,"tc":1856.2,"Z":50,"n":"Regular","s":"partially_filled","T":1710000000000,"U":1710000060000}"#).unwrap_or_else(|_| serde_json::json!({}));
273            let msg = json!({
274                "stream": stream,
275                "data": payload,
276            });
277
278            streams_base.on_message(msg.to_string(), conn.clone()).await;
279
280            yield_now().await;
281
282            assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
283        });
284    }
285}
binance_sdk/stocks/websocket_streams/apis/user_streams_api.rs

binance_sdk/stocks/websocket_streams/apis/
user_streams_api.rs
