opendeviationbar-streaming 13.79.0

//! ChannelSink: feeds bars to Python via tokio::sync::mpsc (Issue #178, Phase 1)

use super::traits::{BarSink, SinkError};
use crate::live_engine::CompletedBar;
use std::sync::atomic::{AtomicU64, Ordering};

/// Sends completed bars to a bounded mpsc channel.
/// The receiver end is consumed by Python via `PyOdbEngine.next_bar()`.
///
/// Uses `try_send()` (non-blocking) — if the channel is full, increments
/// `bars_dropped` and returns `SinkError::Recoverable`. The engine loop
/// is never blocked by a slow consumer. Charon dead-letter replay catches drops.
pub struct ChannelSink {
    tx: tokio::sync::mpsc::Sender<CompletedBar>,
    bars_sent: AtomicU64,
    bars_dropped: AtomicU64,
}

impl ChannelSink {
    pub fn new(tx: tokio::sync::mpsc::Sender<CompletedBar>) -> Self {
        Self {
            tx,
            bars_sent: AtomicU64::new(0),
            bars_dropped: AtomicU64::new(0),
        }
    }

    pub fn bars_sent(&self) -> u64 {
        self.bars_sent.load(Ordering::Relaxed)
    }

    pub fn bars_dropped(&self) -> u64 {
        self.bars_dropped.load(Ordering::Relaxed)
    }
}

impl BarSink for ChannelSink {
    fn on_bar(&mut self, bar: &CompletedBar) -> Result<(), SinkError> {
        match self.tx.try_send(bar.clone()) {
            Ok(()) => {
                self.bars_sent.fetch_add(1, Ordering::Relaxed);
                Ok(())
            }
            Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
                self.bars_dropped.fetch_add(1, Ordering::Relaxed);
                Err(SinkError::Recoverable("channel full".into()))
            }
            Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
                Err(SinkError::Unrecoverable("channel closed".into()))
            }
        }
    }

    fn flush(&mut self) -> Result<(), SinkError> {
        Ok(())
    }

    fn name(&self) -> &str {
        "channel"
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use opendeviationbar_core::{FixedPoint, OpenDeviationBar};
    use std::sync::Arc;

    fn make_bar() -> CompletedBar {
        let trade = opendeviationbar_core::Tick {
            ref_id: 1,
            price: FixedPoint::from_str("50000.0").unwrap(),
            volume: FixedPoint::from_str("1.0").unwrap(),
            first_sub_id: 1,
            last_sub_id: 1,
            timestamp: opendeviationbar_core::normalize_timestamp(1_700_000_000_000),
            is_buyer_maker: false,
            is_best_match: None,
            best_bid: None,
            best_ask: None,
        };
        CompletedBar {
            symbol: Arc::from("BTCUSDT"),
            threshold_decimal_bps: 250,
            bar: OpenDeviationBar::new(&trade),
        }
    }

    #[tokio::test]
    async fn test_channel_sink_send() {
        let (tx, mut rx) = tokio::sync::mpsc::channel(10);
        let mut sink = ChannelSink::new(tx);

        let bar = make_bar();
        assert!(sink.on_bar(&bar).is_ok());
        assert_eq!(sink.bars_sent(), 1);
        assert_eq!(sink.bars_dropped(), 0);

        let received = rx.recv().await.unwrap();
        assert_eq!(&*received.symbol, "BTCUSDT");
    }

    #[tokio::test]
    async fn test_channel_sink_full() {
        let (tx, _rx) = tokio::sync::mpsc::channel(1);
        let mut sink = ChannelSink::new(tx);

        let bar = make_bar();
        assert!(sink.on_bar(&bar).is_ok()); // fills the channel
        let result = sink.on_bar(&bar); // should be full
        assert!(result.is_err());
        assert_eq!(sink.bars_dropped(), 1);
    }

    #[test]
    fn test_channel_sink_closed() {
        let (tx, rx) = tokio::sync::mpsc::channel(10);
        drop(rx); // close the receiver
        let mut sink = ChannelSink::new(tx);

        let bar = make_bar();
        let result = sink.on_bar(&bar);
        assert!(matches!(result, Err(SinkError::Unrecoverable(_))));
    }
}