tradingview/live/handler/
handler.rs

1use serde_json::Value;
2use tokio::sync::mpsc;
3
4use crate::{
5    Error,
6    live::{handler::command::Command, models::TradingViewDataEvent},
7};
8
9/// Default capacity for the bounded command channel.
10/// 256 commands of buffering before senders experience backpressure.
11pub const DEFAULT_COMMAND_CHANNEL_CAPACITY: usize = 256;
12
13/// Bounded sender for the command channel.
14pub type CommandTx = mpsc::Sender<Command>;
15
16/// Bounded receiver for the command channel.
17pub type CommandRx = mpsc::Receiver<Command>;
18
19// =============================================================================
20// Handler trait
21// =============================================================================
22
23/// Core event handler trait — object-safe, supports `Arc<dyn Handler>`.
24///
25/// Unlike v1, this trait does **not** require `Clone` or a `new()`
26/// constructor.  Use [`HandlerFactory`] for construction and `Arc<dyn
27/// Handler>` when shared ownership is needed.
28pub trait Handler: Send + Sync + 'static {
29    /// Called when a TradingView data event is received.
30    fn handle_events(&self, event: TradingViewDataEvent, message: &[Value]);
31
32    /// Called when quote data is received (e.g., price updates).
33    fn handle_quote_data(&self, message: &[Value]);
34
35    /// Called when series/historical data is received.
36    fn handle_series_data(&self, event: TradingViewDataEvent, messages: &[Value]);
37
38    /// Called when an error occurs in the WebSocket or command pipeline.
39    fn notify_error(&self, error: Error, message: &[Value]);
40}
41
42/// Factory trait for constructing [`Handler`] implementations.
43///
44/// Separating construction from the handler trait enables dependency
45/// injection and cleaner initialization patterns.
46pub trait HandlerFactory: Send + Sync + 'static {
47    /// The concrete handler type produced by this factory.
48    type Handler: Handler;
49
50    /// Create a new handler instance, passing the command sender so the
51    /// handler can issue commands back to the WebSocket client.
52    fn create(&self, command_tx: CommandTx) -> Self::Handler;
53}
54
55// =============================================================================
56// Tests
57// =============================================================================
58#[cfg(test)]
59mod tests {
60    use super::*;
61    use tokio::sync::mpsc;
62
63    // ------------------------------------------------------------------
64    // Channel tests (existing)
65    // ------------------------------------------------------------------
66
67    #[test]
68    fn test_command_tx_is_bounded() {
69        let (tx, _rx) = mpsc::channel::<Command>(DEFAULT_COMMAND_CHANNEL_CAPACITY);
70        let _command_tx: CommandTx = tx;
71    }
72
73    #[test]
74    fn test_command_rx_is_bounded() {
75        let (_tx, rx) = mpsc::channel::<Command>(DEFAULT_COMMAND_CHANNEL_CAPACITY);
76        let _command_rx: CommandRx = rx;
77    }
78
79    #[test]
80    fn test_default_capacity_is_reasonable() {
81        const { assert!(DEFAULT_COMMAND_CHANNEL_CAPACITY >= 64) };
82        const { assert!(DEFAULT_COMMAND_CHANNEL_CAPACITY <= 4096) };
83    }
84
85    #[tokio::test]
86    async fn test_bounded_channel_backpressure() {
87        let (tx, mut rx) = mpsc::channel::<u32>(4);
88        for i in 0..4 {
89            tx.send(i).await.expect("send should succeed");
90        }
91        let consumer = tokio::spawn(async move {
92            let mut drained = Vec::new();
93            while let Some(val) = rx.recv().await {
94                drained.push(val);
95                if drained.len() == 8 {
96                    break;
97                }
98            }
99            drained
100        });
101        for i in 4..8 {
102            tx.send(i).await.expect("send after drain");
103        }
104        drop(tx);
105        let drained = consumer.await.unwrap();
106        assert_eq!(drained, vec![0, 1, 2, 3, 4, 5, 6, 7]);
107    }
108
109    #[tokio::test]
110    async fn test_try_send_backpressure() {
111        let (tx, mut _rx) = mpsc::channel::<u32>(2);
112        assert!(tx.try_send(1).is_ok());
113        assert!(tx.try_send(2).is_ok());
114        assert!(tx.try_send(3).is_err());
115    }
116
117    // ------------------------------------------------------------------
118    // Handler v2 tests
119    // ------------------------------------------------------------------
120
121    /// A minimal handler implementation using the new v2 traits.
122    struct TestHandler {
123        events: std::sync::Mutex<Vec<String>>,
124    }
125
126    impl Handler for TestHandler {
127        fn handle_events(&self, _event: TradingViewDataEvent, message: &[Value]) {
128            self.events
129                .lock()
130                .unwrap()
131                .push(format!("event: {:?}", message));
132        }
133        fn handle_quote_data(&self, message: &[Value]) {
134            self.events
135                .lock()
136                .unwrap()
137                .push(format!("quote: {:?}", message));
138        }
139        fn handle_series_data(&self, _event: TradingViewDataEvent, messages: &[Value]) {
140            self.events
141                .lock()
142                .unwrap()
143                .push(format!("series: {:?}", messages));
144        }
145        fn notify_error(&self, _error: Error, message: &[Value]) {
146            self.events
147                .lock()
148                .unwrap()
149                .push(format!("error: {:?}", message));
150        }
151    }
152
153    struct TestHandlerFactory;
154    impl HandlerFactory for TestHandlerFactory {
155        type Handler = TestHandler;
156        fn create(&self, _command_tx: CommandTx) -> Self::Handler {
157            TestHandler {
158                events: std::sync::Mutex::new(Vec::new()),
159            }
160        }
161    }
162
163    #[test]
164    fn test_new_handler_compiles_and_works() {
165        let (_tx, _rx) = mpsc::channel::<Command>(4);
166        let factory = TestHandlerFactory;
167        let handler = factory.create(_tx);
168        handler.handle_events(
169            TradingViewDataEvent::OnChartData,
170            &[serde_json::json!({"test": true})],
171        );
172        let events = handler.events.lock().unwrap();
173        assert_eq!(events.len(), 1);
174    }
175
176    #[test]
177    fn test_handler_is_object_safe() {
178        let (_tx, _rx) = mpsc::channel::<Command>(4);
179        let factory = TestHandlerFactory;
180        let handler = factory.create(_tx);
181        // Verify we can use Arc<dyn Handler>
182        let _arc: std::sync::Arc<dyn Handler> = std::sync::Arc::new(handler);
183    }
184}
tradingview/live/handler/handler.rs

tradingview/live/handler/
handler.rs
