tradestation-api 0.1.1

//! Brokerage streaming methods (orders, positions) on `Client`, split out
//! of the single-file `streaming` module.

use super::*;
use crate::{Client, Error};

impl Client {
    /// Stream order status updates for the specified accounts.
    pub async fn stream_orders(
        &mut self,
        account_ids: &[&str],
    ) -> Result<BoxStream<StreamOrder>, Error> {
        let ids = account_ids.join(",");
        let path = format!("/v3/brokerage/stream/accounts/{}/orders", ids);
        let headers = self.auth_headers().await?;
        let url = format!("{}{}", self.base_url(), &path);

        let resp = self.http.get(&url).headers(headers).send().await?;

        if !resp.status().is_success() {
            let status = resp.status().as_u16();
            let body = resp.text().await.unwrap_or_default();
            return Err(Error::Api {
                status,
                message: body,
            });
        }

        let stream = async_stream::try_stream! {
            let mut bytes_stream = resp.bytes_stream();
            let mut buffer = String::new();

            use futures::StreamExt;
            while let Some(chunk) = bytes_stream.next().await {
                let chunk = chunk.map_err(Error::Http)?;
                buffer.push_str(&String::from_utf8_lossy(&chunk));

                while let Some(newline_pos) = buffer.find('\n') {
                    let line = buffer[..newline_pos].trim().to_string();
                    buffer = buffer[newline_pos + 1..].to_string();

                    if line.is_empty() {
                        continue;
                    }

                    match serde_json::from_str::<StreamOrder>(&line) {
                        Ok(item) => yield item,
                        Err(e) => {
                            tracing::warn!("Failed to parse stream order: {e}, line: {line}");
                        }
                    }
                }
            }
        };

        Ok(Box::pin(stream))
    }

    /// Stream order updates for specific order IDs.
    pub async fn stream_orders_by_id(
        &mut self,
        account_ids: &[&str],
        order_ids: &[&str],
    ) -> Result<BoxStream<StreamOrder>, Error> {
        let ids = account_ids.join(",");
        let oids = order_ids.join(",");
        let path = format!("/v3/brokerage/stream/accounts/{}/orders/{}", ids, oids);
        let headers = self.auth_headers().await?;
        let url = format!("{}{}", self.base_url(), &path);

        let resp = self.http.get(&url).headers(headers).send().await?;

        if !resp.status().is_success() {
            let status = resp.status().as_u16();
            let body = resp.text().await.unwrap_or_default();
            return Err(Error::Api {
                status,
                message: body,
            });
        }

        let stream = async_stream::try_stream! {
            let mut bytes_stream = resp.bytes_stream();
            let mut buffer = String::new();

            use futures::StreamExt;
            while let Some(chunk) = bytes_stream.next().await {
                let chunk = chunk.map_err(Error::Http)?;
                buffer.push_str(&String::from_utf8_lossy(&chunk));

                while let Some(newline_pos) = buffer.find('\n') {
                    let line = buffer[..newline_pos].trim().to_string();
                    buffer = buffer[newline_pos + 1..].to_string();

                    if line.is_empty() {
                        continue;
                    }

                    match serde_json::from_str::<StreamOrder>(&line) {
                        Ok(item) => yield item,
                        Err(e) => {
                            tracing::warn!("Failed to parse stream order: {e}, line: {line}");
                        }
                    }
                }
            }
        };

        Ok(Box::pin(stream))
    }

    /// Stream position updates for the specified accounts.
    pub async fn stream_positions(
        &mut self,
        account_ids: &[&str],
    ) -> Result<BoxStream<StreamPosition>, Error> {
        let ids = account_ids.join(",");
        let path = format!("/v3/brokerage/stream/accounts/{}/positions", ids);
        let headers = self.auth_headers().await?;
        let url = format!("{}{}", self.base_url(), &path);

        let resp = self.http.get(&url).headers(headers).send().await?;

        if !resp.status().is_success() {
            let status = resp.status().as_u16();
            let body = resp.text().await.unwrap_or_default();
            return Err(Error::Api {
                status,
                message: body,
            });
        }

        let stream = async_stream::try_stream! {
            let mut bytes_stream = resp.bytes_stream();
            let mut buffer = String::new();

            use futures::StreamExt;
            while let Some(chunk) = bytes_stream.next().await {
                let chunk = chunk.map_err(Error::Http)?;
                buffer.push_str(&String::from_utf8_lossy(&chunk));

                while let Some(newline_pos) = buffer.find('\n') {
                    let line = buffer[..newline_pos].trim().to_string();
                    buffer = buffer[newline_pos + 1..].to_string();

                    if line.is_empty() {
                        continue;
                    }

                    match serde_json::from_str::<StreamPosition>(&line) {
                        Ok(item) => yield item,
                        Err(e) => {
                            tracing::warn!("Failed to parse stream position: {e}, line: {line}");
                        }
                    }
                }
            }
        };

        Ok(Box::pin(stream))
    }
}