diff --git a/services/api_gateway/src/grpc/broker_gateway_proxy.rs b/services/api_gateway/src/grpc/broker_gateway_proxy.rs index 82f5564d0..258fabe6e 100644 --- a/services/api_gateway/src/grpc/broker_gateway_proxy.rs +++ b/services/api_gateway/src/grpc/broker_gateway_proxy.rs @@ -4,6 +4,7 @@ use futures::Stream; use std::pin::Pin; +use std::time::Duration; use tonic::{Request, Response, Status}; use tracing::{error, instrument}; @@ -13,7 +14,8 @@ use crate::broker_gateway::{ CancelOrderRequest, CancelOrderResponse, ExecutionEvent, GetAccountStateRequest, GetAccountStateResponse, GetPositionsRequest, GetPositionsResponse, GetSessionStatusRequest, GetSessionStatusResponse, HealthCheckRequest, HealthCheckResponse, - RouteOrderRequest, RouteOrderResponse, StreamExecutionsRequest, + RouteOrderRequest, RouteOrderResponse, StreamAccountStateRequest, StreamExecutionsRequest, + StreamSessionStatusRequest, }; #[derive(Debug, Clone)] @@ -122,4 +124,74 @@ impl BrokerGatewayService for BrokerGatewayProxy { e }) } + + type StreamAccountStateStream = + Pin> + Send>>; + + #[instrument(skip(self, request), err)] + async fn stream_account_state( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + let interval_secs = if req.interval_seconds == 0 { + 3 + } else { + req.interval_seconds.clamp(1, 60) + }; + let mut client = self.client.clone(); + + let stream = async_stream::stream! { + let mut interval = tokio::time::interval(Duration::from_secs(u64::from(interval_secs))); + loop { + interval.tick().await; + match client.get_account_state(Request::new(GetAccountStateRequest { + account_id: String::new(), + })).await { + Ok(resp) => yield Ok(resp.into_inner()), + Err(e) => { + error!("Backend GetAccountState failed: {}", e); + yield Err(e); + break; + } + } + } + }; + + Ok(Response::new(Box::pin(stream))) + } + + type StreamSessionStatusStream = + Pin> + Send>>; + + #[instrument(skip(self, request), err)] + async fn stream_session_status( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + let interval_secs = if req.interval_seconds == 0 { + 5 + } else { + req.interval_seconds.clamp(1, 60) + }; + let mut client = self.client.clone(); + + let stream = async_stream::stream! { + let mut interval = tokio::time::interval(Duration::from_secs(u64::from(interval_secs))); + loop { + interval.tick().await; + match client.get_session_status(Request::new(GetSessionStatusRequest::default())).await { + Ok(resp) => yield Ok(resp.into_inner()), + Err(e) => { + error!("Backend GetSessionStatus failed: {}", e); + yield Err(e); + break; + } + } + } + }; + + Ok(Response::new(Box::pin(stream))) + } } diff --git a/services/api_gateway/src/grpc/data_acquisition_proxy.rs b/services/api_gateway/src/grpc/data_acquisition_proxy.rs index 04bf844c9..f892b8a4a 100644 --- a/services/api_gateway/src/grpc/data_acquisition_proxy.rs +++ b/services/api_gateway/src/grpc/data_acquisition_proxy.rs @@ -3,6 +3,9 @@ //! Forwards to data-acquisition-service:50057 (separate service). //! All RPCs are unary (no streaming) — simplest proxy. +use futures::Stream; +use std::pin::Pin; +use std::time::Duration; use tonic::{Request, Response, Status}; use tracing::{error, instrument}; @@ -12,7 +15,7 @@ use crate::data_acquisition::{ CancelDownloadRequest, CancelDownloadResponse, GetDownloadStatusRequest, GetDownloadStatusResponse, HealthCheckRequest, HealthCheckResponse, ListDownloadJobsRequest, ListDownloadJobsResponse, ScheduleDownloadRequest, - ScheduleDownloadResponse, + ScheduleDownloadResponse, StreamDownloadStatusRequest, }; #[derive(Debug, Clone)] @@ -98,4 +101,38 @@ impl DataAcquisitionService for DataAcquisitionProxy { e }) } + + type StreamDownloadStatusStream = + Pin> + Send>>; + + #[instrument(skip(self, request), err)] + async fn stream_download_status( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + let interval_secs = if req.interval_seconds == 0 { + 5 + } else { + req.interval_seconds.clamp(1, 60) + }; + let mut client = self.client.clone(); + + let stream = async_stream::stream! { + let mut interval = tokio::time::interval(Duration::from_secs(u64::from(interval_secs))); + loop { + interval.tick().await; + match client.list_download_jobs(Request::new(ListDownloadJobsRequest::default())).await { + Ok(resp) => yield Ok(resp.into_inner()), + Err(e) => { + error!("Backend ListDownloadJobs failed: {}", e); + yield Err(e); + break; + } + } + } + }; + + Ok(Response::new(Box::pin(stream))) + } } diff --git a/services/api_gateway/src/grpc/ml_proxy.rs b/services/api_gateway/src/grpc/ml_proxy.rs index b2d5d4cee..224eef24c 100644 --- a/services/api_gateway/src/grpc/ml_proxy.rs +++ b/services/api_gateway/src/grpc/ml_proxy.rs @@ -4,6 +4,7 @@ use futures::Stream; use std::pin::Pin; +use std::time::Duration; use tonic::{Request, Response, Status}; use tracing::{error, instrument}; @@ -15,7 +16,8 @@ use crate::ml_inference::{ GetModelPerformanceRequest, GetModelPerformanceResponse, GetModelStatusRequest, GetModelStatusResponse, GetPredictionRequest, GetPredictionResponse, ModelMetricsEvent, PredictionEvent, RetrainModelRequest, RetrainModelResponse, SignalStrengthEvent, - StreamModelMetricsRequest, StreamPredictionsRequest, StreamSignalStrengthRequest, + StreamModelMetricsRequest, StreamModelStatusRequest, StreamPredictionsRequest, + StreamSignalStrengthRequest, }; #[derive(Debug, Clone)] @@ -37,6 +39,8 @@ impl MlService for MlServiceProxy { Pin> + Send>>; type StreamSignalStrengthStream = Pin> + Send>>; + type StreamModelStatusStream = + Pin> + Send>>; #[instrument(skip(self, request), err)] async fn get_prediction( @@ -189,4 +193,38 @@ impl MlService for MlServiceProxy { })?; Ok(Response::new(Box::pin(stream.into_inner()))) } + + #[instrument(skip(self, request), err)] + async fn stream_model_status( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + let interval_secs = if req.interval_seconds == 0 { + 5 + } else { + req.interval_seconds.clamp(1, 60) + }; + let model_name = req.model_name; + let mut client = self.client.clone(); + + let stream = async_stream::stream! { + let mut interval = tokio::time::interval(Duration::from_secs(u64::from(interval_secs))); + loop { + interval.tick().await; + match client.get_model_status(Request::new(GetModelStatusRequest { + model_name: if model_name.is_empty() { None } else { Some(model_name.clone()) }, + })).await { + Ok(resp) => yield Ok(resp.into_inner()), + Err(e) => { + error!("Backend GetModelStatus failed: {}", e); + yield Err(e); + break; + } + } + } + }; + + Ok(Response::new(Box::pin(stream))) + } } diff --git a/services/api_gateway/src/grpc/monitoring_handler.rs b/services/api_gateway/src/grpc/monitoring_handler.rs index 9a405b087..98380c36f 100644 --- a/services/api_gateway/src/grpc/monitoring_handler.rs +++ b/services/api_gateway/src/grpc/monitoring_handler.rs @@ -21,15 +21,16 @@ use tracing::{error, info, instrument}; use crate::monitoring::monitoring_service_server::{MonitoringService, MonitoringServiceServer}; use crate::monitoring::{ - AcknowledgeAlertRequest, AcknowledgeAlertResponse, AlertEvent, EpochFinancialSnapshot, - GetActiveAlertsRequest, GetActiveAlertsResponse, GetEpochHistoryRequest, - GetEpochHistoryResponse, GetHealthCheckRequest, GetHealthCheckResponse, - GetLatencyMetricsRequest, GetLatencyMetricsResponse, GetLiveTrainingMetricsRequest, - GetLiveTrainingMetricsResponse, GetMetricsRequest, GetMetricsResponse, - GetSystemStatusRequest, GetSystemStatusResponse, GetThroughputMetricsRequest, - GetThroughputMetricsResponse, GpuSnapshot, HealthCheck, HealthStatus, Metric, MetricType, - MetricsEvent, ServiceHealth, ServiceState, ServiceStatus, StreamTrainingMetricsRequest, - SystemHealth, SystemMetrics, SystemStatus, SystemStatusEvent, TrainingSession, + AcknowledgeAlertRequest, AcknowledgeAlertResponse, AlertEvent, AlertEventType, + EpochFinancialSnapshot, GetActiveAlertsRequest, GetActiveAlertsResponse, + GetEpochHistoryRequest, GetEpochHistoryResponse, GetHealthCheckRequest, + GetHealthCheckResponse, GetLatencyMetricsRequest, GetLatencyMetricsResponse, + GetLiveTrainingMetricsRequest, GetLiveTrainingMetricsResponse, GetMetricsRequest, + GetMetricsResponse, GetSystemStatusRequest, GetSystemStatusResponse, + GetThroughputMetricsRequest, GetThroughputMetricsResponse, GpuSnapshot, HealthCheck, + HealthStatus, Metric, MetricType, MetricsEvent, ServiceHealth, ServiceState, ServiceStatus, + StreamTrainingMetricsRequest, SystemHealth, SystemMetrics, SystemStatus, + SystemStatusChangeType, SystemStatusEvent, TrainingSession, }; // ============================================================================ @@ -494,13 +495,101 @@ impl MonitoringService for MonitoringServiceHandler { Ok(Response::new(response)) } + #[instrument(skip(self, request), fields(request_id = %uuid::Uuid::new_v4()), err)] async fn stream_system_status( &self, - _request: Request, + request: Request, ) -> Result, Status> { - Err(Status::unimplemented( - "StreamSystemStatus not yet implemented", - )) + let req = request.into_inner(); + let interval_secs = req.update_frequency_seconds.unwrap_or(0); + let interval_secs = if interval_secs == 0 { + 5u32 + } else { + (interval_secs as u32).clamp(1, 60) + }; + let backends = self.backends.clone(); + let start_time = self.start_time; + let service_filter = req.service_names; + + info!( + "StreamSystemStatus: interval={}s, filter={:?}", + interval_secs, + if service_filter.is_empty() { + "" + } else { + "filtered" + } + ); + + let stream = async_stream::stream! { + let mut interval = tokio::time::interval(Duration::from_secs(u64::from(interval_secs))); + loop { + interval.tick().await; + + let checks: Vec<_> = backends + .iter() + .filter(|(name, _)| { + service_filter.is_empty() || service_filter.contains(name) + }) + .map(|(name, url)| { + let name = name.clone(); + let url = url.clone(); + async move { Self::check_backend_health(&name, &url).await } + }) + .collect(); + + let statuses = futures::future::join_all(checks).await; + + let healthy_count = statuses + .iter() + .filter(|s| s.health == i32::from(ServiceHealth::Healthy)) + .count(); + let total = statuses.len(); + + #[allow(clippy::cast_possible_truncation)] + let overall_health = if healthy_count == total { + SystemHealth::Healthy + } else if healthy_count == 0 { + SystemHealth::Critical + } else if healthy_count * 2 >= total { + SystemHealth::Degraded + } else { + SystemHealth::Unhealthy + }; + + let critical_issues: Vec = statuses + .iter() + .filter(|s| s.health != i32::from(ServiceHealth::Healthy)) + .map(|s| { + format!( + "{}: {}", + s.service_name, + s.error_message.as_deref().unwrap_or("unhealthy") + ) + }) + .collect(); + + #[allow(clippy::cast_possible_truncation)] + let system_status = SystemStatus { + overall_health: overall_health.into(), + healthy_services: healthy_count as i32, + total_services: total as i32, + critical_issues, + system_uptime_seconds: start_time.elapsed().as_secs() as i64, + system_metrics: Some(SystemMetrics::default()), + }; + + let event = SystemStatusEvent { + system_status: Some(system_status), + change_type: SystemStatusChangeType::Unspecified.into(), + timestamp: chrono::Utc::now().timestamp(), + }; + + yield Ok(event); + } + }; + + Ok(Response::new(Box::pin(stream))) } #[instrument(skip(self, request), err)] @@ -606,13 +695,77 @@ impl MonitoringService for MonitoringServiceHandler { })) } + #[instrument(skip(self, request), fields(request_id = %uuid::Uuid::new_v4()), err)] async fn stream_metrics( &self, - _request: Request, + request: Request, ) -> Result, Status> { - Err(Status::unimplemented( - "StreamMetrics not yet implemented", - )) + let req = request.into_inner(); + let interval_secs = req.update_frequency_seconds.unwrap_or(0); + let interval_secs = if interval_secs == 0 { + 5u32 + } else { + (interval_secs as u32).clamp(1, 60) + }; + let metric_names: Vec = if req.metric_names.is_empty() { + vec![ + "cpu_usage".to_owned(), + "memory_usage".to_owned(), + "gpu_utilization".to_owned(), + ] + } else { + req.metric_names + }; + let prom = self.prom.clone(); + + info!( + "StreamMetrics: interval={}s, metrics={:?}", + interval_secs, metric_names + ); + + let stream = async_stream::stream! { + let mut interval = tokio::time::interval(Duration::from_secs(u64::from(interval_secs))); + loop { + interval.tick().await; + + let mut metrics = Vec::new(); + for name in &metric_names { + match prom.query(name).await { + Ok(results) => { + for r in results { + if let Ok(value) = r.value.1.parse::() { + metrics.push(Metric { + name: r + .metric + .get("__name__") + .cloned() + .unwrap_or_else(|| name.clone()), + metric_type: MetricType::Gauge.into(), + value, + unit: String::new(), + labels: r.metric, + timestamp: chrono::Utc::now().timestamp(), + statistics: None, + }); + } + } + } + Err(e) => { + error!("StreamMetrics: Prometheus query for '{}' failed: {}", name, e); + } + } + } + + let event = MetricsEvent { + metrics, + timestamp: chrono::Utc::now().timestamp(), + }; + + yield Ok(event); + } + }; + + Ok(Response::new(Box::pin(stream))) } async fn get_latency_metrics( @@ -636,13 +789,33 @@ impl MonitoringService for MonitoringServiceHandler { type StreamAlertsStream = Pin> + Send>>; + #[instrument(skip(self, request), fields(request_id = %uuid::Uuid::new_v4()), err)] async fn stream_alerts( &self, - _request: Request, + request: Request, ) -> Result, Status> { - Err(Status::unimplemented( - "StreamAlerts not yet implemented", - )) + let _req = request.into_inner(); + + info!("StreamAlerts: interval=10s (placeholder, no alertmanager integration yet)"); + + let stream = async_stream::stream! { + let mut interval = tokio::time::interval(Duration::from_secs(10)); + loop { + interval.tick().await; + + // Placeholder: yields an empty AlertEvent each tick. + // Will be wired to Prometheus Alertmanager when available. + let event = AlertEvent { + alert: None, + event_type: AlertEventType::Unspecified.into(), + timestamp: chrono::Utc::now().timestamp(), + }; + + yield Ok(event); + } + }; + + Ok(Response::new(Box::pin(stream))) } async fn acknowledge_alert( diff --git a/services/api_gateway/src/grpc/risk_proxy.rs b/services/api_gateway/src/grpc/risk_proxy.rs index 96ed95324..25a15ddda 100644 --- a/services/api_gateway/src/grpc/risk_proxy.rs +++ b/services/api_gateway/src/grpc/risk_proxy.rs @@ -4,6 +4,7 @@ use futures::Stream; use std::pin::Pin; +use std::time::Duration; use tonic::{Request, Response, Status}; use tracing::{error, instrument}; @@ -13,7 +14,8 @@ use crate::risk::{ EmergencyStopRequest, EmergencyStopResponse, GetCircuitBreakerStatusRequest, GetCircuitBreakerStatusResponse, GetPositionRiskRequest, GetPositionRiskResponse, GetRiskMetricsRequest, GetRiskMetricsResponse, GetVaRRequest, GetVaRResponse, - RiskAlertEvent, StreamRiskAlertsRequest, StreamVaRRequest, VaREvent, ValidateOrderRequest, + RiskAlertEvent, StreamCircuitBreakerStatusRequest, StreamRiskAlertsRequest, + StreamRiskMetricsRequest, StreamVaRRequest, VaREvent, ValidateOrderRequest, ValidateOrderResponse, }; @@ -33,6 +35,10 @@ impl RiskService for RiskServiceProxy { type StreamVaRUpdatesStream = Pin> + Send>>; type StreamRiskAlertsStream = Pin> + Send>>; + type StreamCircuitBreakerStatusStream = + Pin> + Send>>; + type StreamRiskMetricsStream = + Pin> + Send>>; #[instrument(skip(self, request), err)] async fn get_va_r( @@ -153,4 +159,72 @@ impl RiskService for RiskServiceProxy { e }) } + + #[instrument(skip(self, request), err)] + async fn stream_circuit_breaker_status( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + let interval_secs = if req.interval_seconds == 0 { + 2 + } else { + req.interval_seconds.clamp(1, 60) + }; + let symbol = req.symbol; + let mut client = self.client.clone(); + + let stream = async_stream::stream! { + let mut interval = tokio::time::interval(Duration::from_secs(u64::from(interval_secs))); + loop { + interval.tick().await; + match client.get_circuit_breaker_status(Request::new(GetCircuitBreakerStatusRequest { + symbol: symbol.clone(), + })).await { + Ok(resp) => yield Ok(resp.into_inner()), + Err(e) => { + error!("Backend GetCircuitBreakerStatus failed: {}", e); + yield Err(e); + break; + } + } + } + }; + + Ok(Response::new(Box::pin(stream))) + } + + #[instrument(skip(self, request), err)] + async fn stream_risk_metrics( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + let interval_secs = if req.interval_seconds == 0 { + 3 + } else { + req.interval_seconds.clamp(1, 60) + }; + let portfolio_id = req.portfolio_id; + let mut client = self.client.clone(); + + let stream = async_stream::stream! { + let mut interval = tokio::time::interval(Duration::from_secs(u64::from(interval_secs))); + loop { + interval.tick().await; + match client.get_risk_metrics(Request::new(GetRiskMetricsRequest { + portfolio_id: portfolio_id.clone(), + })).await { + Ok(resp) => yield Ok(resp.into_inner()), + Err(e) => { + error!("Backend GetRiskMetrics failed: {}", e); + yield Err(e); + break; + } + } + } + }; + + Ok(Response::new(Box::pin(stream))) + } } diff --git a/services/api_gateway/src/grpc/trading_agent_proxy.rs b/services/api_gateway/src/grpc/trading_agent_proxy.rs index 28431faa6..3cff1a7e8 100644 --- a/services/api_gateway/src/grpc/trading_agent_proxy.rs +++ b/services/api_gateway/src/grpc/trading_agent_proxy.rs @@ -7,8 +7,10 @@ //! - Efficient streaming support for agent activity events //! - Health checking integration -use futures::Stream; use std::pin::Pin; +use std::time::Duration; + +use futures::Stream; use tonic::{Request, Response, Status}; use tracing::{error, info, instrument, warn}; @@ -56,6 +58,7 @@ use crate::trading_agent::{ SelectUniverseRequest, SelectUniverseResponse, StreamAgentActivityRequest, + StreamAgentStatusRequest, SubmitAgentOrdersRequest, SubmitAgentOrdersResponse, @@ -101,6 +104,9 @@ impl TradingAgentService for TradingAgentProxy { /// Server streaming type for agent activity events type StreamAgentActivityStream = Pin> + Send>>; + /// Server streaming type for poll-based agent status + type StreamAgentStatusStream = + Pin> + Send>>; // ===== Universe Management Methods ===== @@ -499,6 +505,40 @@ impl TradingAgentService for TradingAgentProxy { info!("HealthCheck request forwarded successfully"); Ok(response) } + + /// Poll-to-stream adapter for agent status + /// + /// Polls GetAgentStatus at a configurable interval and yields results as a server stream. + #[instrument(skip(self, request), fields(request_id = %uuid::Uuid::new_v4()), err)] + async fn stream_agent_status( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + let interval_secs = if req.interval_seconds == 0 { + 3 + } else { + req.interval_seconds.clamp(1, 60) + }; + let mut client = self.client.clone(); + + let stream = async_stream::stream! { + let mut interval = tokio::time::interval(Duration::from_secs(u64::from(interval_secs))); + loop { + interval.tick().await; + match client.get_agent_status(Request::new(GetAgentStatusRequest::default())).await { + Ok(resp) => yield Ok(resp.into_inner()), + Err(e) => { + error!("Backend GetAgentStatus failed: {}", e); + yield Err(e); + break; + } + } + } + }; + + Ok(Response::new(Box::pin(stream))) + } } #[cfg(test)] diff --git a/services/api_gateway/src/grpc/trading_direct_proxy.rs b/services/api_gateway/src/grpc/trading_direct_proxy.rs index 3b476cdd9..12b5b4a81 100644 --- a/services/api_gateway/src/grpc/trading_direct_proxy.rs +++ b/services/api_gateway/src/grpc/trading_direct_proxy.rs @@ -3,8 +3,10 @@ //! Forwards requests from fxt CLI (package: trading) directly to trading-service backend. //! Distinct from TradingServiceProxy which serves the monolithic foxhunt.tli.TradingService. -use futures::Stream; use std::pin::Pin; +use std::time::Duration; + +use futures::Stream; use tonic::{Request, Response, Status}; use tracing::{error, instrument}; @@ -18,8 +20,9 @@ use crate::trading_backend::{ GetRegimeStateResponse, GetRegimeTransitionsRequest, GetRegimeTransitionsResponse, MarketDataEvent, MlOrderRequest, MlOrderResponse, MlPerformanceRequest, MlPerformanceResponse, MlPredictionsRequest, MlPredictionsResponse, OrderEvent, - PositionEvent, StreamExecutionsRequest, StreamMarketDataRequest, StreamOrdersRequest, - StreamPositionsRequest, SubmitOrderRequest, SubmitOrderResponse, + PositionEvent, StreamExecutionsRequest, StreamMarketDataRequest, StreamOrderBookRequest, + StreamOrdersRequest, StreamPortfolioSummaryRequest, StreamPositionsRequest, + SubmitOrderRequest, SubmitOrderResponse, }; #[derive(Debug, Clone)] @@ -42,6 +45,10 @@ impl TradingService for TradingDirectProxy { Pin> + Send>>; type StreamExecutionsStream = Pin> + Send>>; + type StreamPortfolioSummaryStream = + Pin> + Send>>; + type StreamOrderBookStream = + Pin> + Send>>; #[instrument(skip(self, request), err)] async fn submit_order( @@ -256,4 +263,74 @@ impl TradingService for TradingDirectProxy { e }) } + + #[instrument(skip(self, request), err)] + async fn stream_portfolio_summary( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + let interval_secs = if req.interval_seconds == 0 { + 3 + } else { + req.interval_seconds.clamp(1, 60) + }; + let account_id = req.account_id; + let mut client = self.client.clone(); + + let stream = async_stream::stream! { + let mut interval = tokio::time::interval(Duration::from_secs(u64::from(interval_secs))); + loop { + interval.tick().await; + match client.get_portfolio_summary(Request::new(GetPortfolioSummaryRequest { + account_id: account_id.clone(), + })).await { + Ok(resp) => yield Ok(resp.into_inner()), + Err(e) => { + error!("Backend GetPortfolioSummary failed: {}", e); + yield Err(e); + break; + } + } + } + }; + + Ok(Response::new(Box::pin(stream))) + } + + #[instrument(skip(self, request), err)] + async fn stream_order_book( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + let interval_secs = if req.interval_seconds == 0 { + 1 + } else { + req.interval_seconds.clamp(1, 60) + }; + let symbol = req.symbol; + let depth = req.depth; + let mut client = self.client.clone(); + + let stream = async_stream::stream! { + let mut interval = tokio::time::interval(Duration::from_secs(u64::from(interval_secs))); + loop { + interval.tick().await; + match client.get_order_book(Request::new(GetOrderBookRequest { + symbol: symbol.clone(), + depth: if depth == 0 { None } else { Some(depth) }, + })).await { + Ok(resp) => yield Ok(resp.into_inner()), + Err(e) => { + error!("Backend GetOrderBook failed: {}", e); + yield Err(e); + break; + } + } + } + }; + + Ok(Response::new(Box::pin(stream))) + } }