opendeviationbar-streaming 13.79.0

//! Command channel for on-demand gap-fill requests (Issue #257).
//!
//! Allows Python (via PyO3) to request gap-fills using `&self` (no borrow conflict
//! with `next_bar(&mut self)`). The engine's `tokio::select!` loop receives
//! commands and processes them inline.

use std::sync::Arc;
use tokio::sync::{mpsc, oneshot};

/// Request to fill a specific trade-ID gap.
#[derive(Debug)]
pub struct GapFillCommand {
    /// Symbol to fill (e.g., "BTCUSDT").
    pub symbol: Arc<str>,
    /// First missing trade ID (inclusive).
    pub from_tid: i64,
    /// First trade ID that IS present (exclusive).
    pub to_tid: i64,
    /// Response channel for the fill result.
    pub response_tx: oneshot::Sender<GapFillResult>,
}

/// Result of a gap-fill operation.
#[derive(Debug, Clone)]
pub struct GapFillResult {
    /// Number of trades fetched via REST.
    pub trades_fetched: usize,
    /// Number of bars produced from the fetched trades.
    pub bars_produced: usize,
    /// Wall-clock duration of the fill operation in milliseconds.
    pub duration_ms: u64,
    /// Whether the fill was partial (e.g., rate limited or too large).
    pub partial: bool,
}

/// Sender half for submitting gap-fill commands.
pub type GapFillSender = mpsc::Sender<GapFillCommand>;

/// Receiver half for the engine to consume gap-fill commands.
pub type GapFillReceiver = mpsc::Receiver<GapFillCommand>;

/// Create a paired (sender, receiver) for gap-fill commands.
///
/// `buffer` controls how many commands can be queued before backpressure.
/// Typically 16-32 is sufficient -- gap-fill commands are rare events.
pub fn gap_fill_channel(buffer: usize) -> (GapFillSender, GapFillReceiver) {
    mpsc::channel(buffer)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn test_gap_fill_channel_send_receive() {
        let (tx, mut rx) = gap_fill_channel(16);
        let (response_tx, response_rx) = oneshot::channel();

        let cmd = GapFillCommand {
            symbol: Arc::from("BTCUSDT"),
            from_tid: 100,
            to_tid: 200,
            response_tx,
        };

        tx.send(cmd).await.unwrap();

        let received = rx.recv().await.unwrap();
        assert_eq!(&*received.symbol, "BTCUSDT");
        assert_eq!(received.from_tid, 100);
        assert_eq!(received.to_tid, 200);

        // Send response back
        received
            .response_tx
            .send(GapFillResult {
                trades_fetched: 100,
                bars_produced: 5,
                duration_ms: 250,
                partial: false,
            })
            .unwrap();

        let result = response_rx.await.unwrap();
        assert_eq!(result.trades_fetched, 100);
        assert_eq!(result.bars_produced, 5);
        assert!(!result.partial);
    }

    #[test]
    fn test_gap_fill_channel_creation() {
        let (tx, _rx) = gap_fill_channel(32);
        // Verify we can clone the sender (needed for multiple callers)
        let _tx2 = tx.clone();
    }
}