tradestation_api/streaming/
brokerage_streams.rs

1//! Brokerage streaming methods (orders, positions) on `Client`, split out
2//! of the single-file `streaming` module.
3
4use super::*;
5use crate::{Client, Error};
6
7impl Client {
8    /// Stream order status updates for the specified accounts.
9    pub async fn stream_orders(
10        &mut self,
11        account_ids: &[&str],
12    ) -> Result<BoxStream<StreamOrder>, Error> {
13        let ids = account_ids.join(",");
14        let path = format!("/v3/brokerage/stream/accounts/{}/orders", ids);
15        let headers = self.auth_headers().await?;
16        let url = format!("{}{}", self.base_url(), &path);
17
18        let resp = self.http.get(&url).headers(headers).send().await?;
19
20        if !resp.status().is_success() {
21            let status = resp.status().as_u16();
22            let body = resp.text().await.unwrap_or_default();
23            return Err(Error::Api {
24                status,
25                message: body,
26            });
27        }
28
29        let stream = async_stream::try_stream! {
30            let mut bytes_stream = resp.bytes_stream();
31            let mut buffer = String::new();
32
33            use futures::StreamExt;
34            while let Some(chunk) = bytes_stream.next().await {
35                let chunk = chunk.map_err(Error::Http)?;
36                buffer.push_str(&String::from_utf8_lossy(&chunk));
37
38                while let Some(newline_pos) = buffer.find('\n') {
39                    let line = buffer[..newline_pos].trim().to_string();
40                    buffer = buffer[newline_pos + 1..].to_string();
41
42                    if line.is_empty() {
43                        continue;
44                    }
45
46                    match serde_json::from_str::<StreamOrder>(&line) {
47                        Ok(item) => yield item,
48                        Err(e) => {
49                            tracing::warn!("Failed to parse stream order: {e}, line: {line}");
50                        }
51                    }
52                }
53            }
54        };
55
56        Ok(Box::pin(stream))
57    }
58
59    /// Stream order updates for specific order IDs.
60    pub async fn stream_orders_by_id(
61        &mut self,
62        account_ids: &[&str],
63        order_ids: &[&str],
64    ) -> Result<BoxStream<StreamOrder>, Error> {
65        let ids = account_ids.join(",");
66        let oids = order_ids.join(",");
67        let path = format!("/v3/brokerage/stream/accounts/{}/orders/{}", ids, oids);
68        let headers = self.auth_headers().await?;
69        let url = format!("{}{}", self.base_url(), &path);
70
71        let resp = self.http.get(&url).headers(headers).send().await?;
72
73        if !resp.status().is_success() {
74            let status = resp.status().as_u16();
75            let body = resp.text().await.unwrap_or_default();
76            return Err(Error::Api {
77                status,
78                message: body,
79            });
80        }
81
82        let stream = async_stream::try_stream! {
83            let mut bytes_stream = resp.bytes_stream();
84            let mut buffer = String::new();
85
86            use futures::StreamExt;
87            while let Some(chunk) = bytes_stream.next().await {
88                let chunk = chunk.map_err(Error::Http)?;
89                buffer.push_str(&String::from_utf8_lossy(&chunk));
90
91                while let Some(newline_pos) = buffer.find('\n') {
92                    let line = buffer[..newline_pos].trim().to_string();
93                    buffer = buffer[newline_pos + 1..].to_string();
94
95                    if line.is_empty() {
96                        continue;
97                    }
98
99                    match serde_json::from_str::<StreamOrder>(&line) {
100                        Ok(item) => yield item,
101                        Err(e) => {
102                            tracing::warn!("Failed to parse stream order: {e}, line: {line}");
103                        }
104                    }
105                }
106            }
107        };
108
109        Ok(Box::pin(stream))
110    }
111
112    /// Stream position updates for the specified accounts.
113    pub async fn stream_positions(
114        &mut self,
115        account_ids: &[&str],
116    ) -> Result<BoxStream<StreamPosition>, Error> {
117        let ids = account_ids.join(",");
118        let path = format!("/v3/brokerage/stream/accounts/{}/positions", ids);
119        let headers = self.auth_headers().await?;
120        let url = format!("{}{}", self.base_url(), &path);
121
122        let resp = self.http.get(&url).headers(headers).send().await?;
123
124        if !resp.status().is_success() {
125            let status = resp.status().as_u16();
126            let body = resp.text().await.unwrap_or_default();
127            return Err(Error::Api {
128                status,
129                message: body,
130            });
131        }
132
133        let stream = async_stream::try_stream! {
134            let mut bytes_stream = resp.bytes_stream();
135            let mut buffer = String::new();
136
137            use futures::StreamExt;
138            while let Some(chunk) = bytes_stream.next().await {
139                let chunk = chunk.map_err(Error::Http)?;
140                buffer.push_str(&String::from_utf8_lossy(&chunk));
141
142                while let Some(newline_pos) = buffer.find('\n') {
143                    let line = buffer[..newline_pos].trim().to_string();
144                    buffer = buffer[newline_pos + 1..].to_string();
145
146                    if line.is_empty() {
147                        continue;
148                    }
149
150                    match serde_json::from_str::<StreamPosition>(&line) {
151                        Ok(item) => yield item,
152                        Err(e) => {
153                            tracing::warn!("Failed to parse stream position: {e}, line: {line}");
154                        }
155                    }
156                }
157            }
158        };
159
160        Ok(Box::pin(stream))
161    }
162}
tradestation_api/streaming/brokerage_streams.rs

tradestation_api/streaming/
brokerage_streams.rs
