tradingview/sink/
callback.rs

1//! Inline callback sink.
2//!
3//! The simplest possible sink: invokes a closure or function pointer for every
4//! batch of events. Ideal for debugging, logging, and lightweight consumers.
5
6use async_trait::async_trait;
7use std::sync::Arc;
8use tokio_util::sync::CancellationToken;
9
10use super::EventSink;
11use crate::Result;
12use crate::events::MarketEvent;
13
14/// A sink that invokes a user-supplied async function for each batch.
15///
16/// The callback receives a slice of [`MarketEvent`] and can process them
17/// however it likes. This is the simplest way to wire up custom logic without
18/// implementing [`EventSink`] directly.
19pub struct CallbackSink<F> {
20    callback: Arc<F>,
21    name: String,
22}
23
24impl<F> CallbackSink<F> {
25    /// Create a new callback sink.
26    ///
27    /// The name is used for logging and debugging.
28    pub fn new(name: impl Into<String>, callback: F) -> Self {
29        Self {
30            callback: Arc::new(callback),
31            name: name.into(),
32        }
33    }
34}
35
36// Fn variant — the async callback takes a slice of events.
37#[async_trait]
38impl<F, Fut> EventSink for CallbackSink<F>
39where
40    F: Fn(Vec<MarketEvent>) -> Fut + Send + Sync + 'static,
41    Fut: std::future::Future<Output = Result<()>> + Send + 'static,
42{
43    async fn accept(&self, events: &[MarketEvent]) -> Result<()> {
44        (self.callback)(events.to_vec()).await
45    }
46
47    fn name(&self) -> &str {
48        &self.name
49    }
50
51    async fn shutdown(&self, _token: CancellationToken) -> Result<()> {
52        Ok(())
53    }
54}
55
56/// A blocking/sync variant that accepts a `FnMut` closure.
57///
58/// This spawns the blocking work onto `tokio::task::spawn_blocking` so it
59/// doesn't starve the async runtime.
60pub struct BlockingCallbackSink<F> {
61    callback: Arc<std::sync::Mutex<F>>,
62    name: String,
63}
64
65impl<F> BlockingCallbackSink<F>
66where
67    F: FnMut(&[MarketEvent]) -> Result<()> + Send + 'static,
68{
69    /// Create a new blocking callback sink.
70    pub fn new(name: impl Into<String>, callback: F) -> Self {
71        Self {
72            callback: Arc::new(std::sync::Mutex::new(callback)),
73            name: name.into(),
74        }
75    }
76}
77
78#[async_trait]
79impl<F> EventSink for BlockingCallbackSink<F>
80where
81    F: FnMut(&[MarketEvent]) -> Result<()> + Send + 'static,
82{
83    async fn accept(&self, events: &[MarketEvent]) -> Result<()> {
84        let owned = events.to_vec();
85        let cb = Arc::clone(&self.callback);
86        tokio::task::spawn_blocking(move || {
87            let mut guard = cb.lock().map_err(|e| {
88                crate::Error::Internal(ustr::ustr(&format!("callback lock poisoned: {e}")))
89            })?;
90            guard(&owned)
91        })
92        .await
93        .map_err(|e| crate::Error::Internal(ustr::ustr(&format!("callback join: {e}"))))?
94    }
95
96    fn name(&self) -> &str {
97        &self.name
98    }
99
100    async fn shutdown(&self, _token: CancellationToken) -> Result<()> {
101        Ok(())
102    }
103}
tradingview/sink/callback.rs

tradingview/sink/
callback.rs
