Packet leak when reading to a "not empty" packet. #260

use anyhow::{bail, Result};
use std::sync::{
    atomic::{AtomicBool, Ordering},
    Arc,
};
use std::time::Duration;
use tokio::sync::mpsc;

fn main() -> Result<()> {
    ffmpeg::init()?;

    let shutdown = Arc::new(AtomicBool::new(false));
    let shutdown_clone = shutdown.clone();

    ctrlc::set_handler(move || {
        println!("Ctrl+C received");
        shutdown_clone.store(true, Ordering::SeqCst);
    })
    .expect("Error setting Ctrl-C handler");

    let rt = tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()?;

    rt.block_on(async {
        let (packet_tx, mut packet_rx) = mpsc::channel::<ffmpeg::Packet>(1);

        // Dummy receiver - consumes packets to keep the channel moving
        let rx_handle = tokio::spawn(async move {
            let mut count = 0u64;
            let mut last_log = std::time::Instant::now();
            while let Some(_) = packet_rx.recv().await {
                count += 1;
                if last_log.elapsed() > Duration::from_secs(5) {
                    println!("Received {} packets so far...", count);
                    last_log = std::time::Instant::now();
                }
            }
            println!("Receiver finished. Total packets: {}", count);
        });

        // Producer - Running in blocking thread as per original architecture
        let shutdown_loop = shutdown.clone();
        let producer_handle = tokio::task::spawn_blocking(move || {
            while !shutdown_loop.load(Ordering::SeqCst) {
                // Using "testsrc" to generate dummy video frames locally
                if let Err(e) = run_dummy_leak_loop("lavfi", "testsrc=duration=100:size=1280x720:rate=30", &shutdown_loop, packet_tx.clone()) {
                    eprintln!("Stream error: {}", e);
                    if !shutdown_loop.load(Ordering::SeqCst) {
                        std::thread::sleep(Duration::from_secs(1));
                    }
                }
            }
        });

        let _ = producer_handle.await;
        let _ = rx_handle.await;
        Ok(())
    })
}

fn run_dummy_leak_loop(
    format: &str,
    url: &str,
    shutdown_sig: &Arc<AtomicBool>,
    packet_tx: mpsc::Sender<ffmpeg::Packet>,
) -> Result<()> {
    let mut opts = ffmpeg::Dictionary::new();
    
    println!("Opening dummy source: {}", url);
    let mut ictx = ffmpeg::format::input_with_dictionary(&url, opts)?;

    let video_stream_index = ictx
        .streams()
        .best(ffmpeg::media::Type::Video)
        .ok_or_else(|| anyhow::anyhow!("No video stream found"))?
        .index();

    println!("Source opened. Starting leak loop...");

    let mut packet_count = 0u64;
    
    // --- THE LEAK SOURCE ---
    // Creating the packet outside the loop. When packet.read() is called inside 
    // the loop, it may not correctly free the previous buffer allocated by the 
    // C API, causing memory to climb.
    let mut packet = ffmpeg::Packet::empty(); 

    loop {
        if shutdown_sig.load(Ordering::SeqCst) {
            break;
        }

        match packet.read(&mut ictx) {
            Ok(_) => { /* Successfully read into the existing packet buffer */ }
            Err(ffmpeg::Error::Eof) => {
                println!("Source finished (EOF)");
                break;
            }
            Err(e) => {
                eprintln!("Failed to read packet: {}", e);
                bail!("Read error");
            }
        }

        if packet.stream() != video_stream_index {
            continue;
        }

        packet_count += 1;
        
        // Cloning the packet sends a reference-counted handle or a copy 
        // to the receiver, but the 'mut packet' above persists.
        if let Err(_) = packet_tx.blocking_send(packet.clone()) {
            break;
        }
    }

    println!("Loop exited. Packets processed: {}", packet_count);
    Ok(())
}