Struct EventBatcher 

pub struct EventBatcher<T> { /* private fields */ }

Event batcher with time-based debouncing Accumulates events and flushes them periodically

Implementations

impl<T> EventBatcher<T>

pub fn new(window_ms: u64) -> Self

Create a new batcher with specified time window (milliseconds)

Examples found in repository

examples/04_reasoning_streaming.rs (line 19)

async fn main() -> Result<()> {
    let api_key = std::env::var("OPENAI_API_KEY")?;
    let client = OpenAIClient::new(api_key)?;

    let request = ResponseRequest::new(
        "gpt-5",
        vec![Message::human("Solve this problem: If a train travels 120 km in 2 hours, what is its average speed? And after that make a dynamic programming problem that can be solved in O(n) time and space complexity.")]
    ).with_reasoning(ReasoningConfig::medium());

    println!("Streaming response with reasoning (batched for network efficiency):\n");
    
    let mut stream = client.reason_stream(request).await?;
    let mut batcher = EventBatcher::new(50); // 50ms batching window
    let mut reasoning_displayed = false;
    let start_time = Instant::now();
    let mut total_events = 0;
    let mut total_batches = 0;

    loop {
        tokio::select! {
            // Receive events from stream
            event_result = stream.next() => {
                match event_result {
                    Some(Ok(event)) => {
                        total_events += 1;
                        batcher.push(event);
                    }
                    Some(Err(e)) => {
                        eprintln!("Stream error: {}", e);
                        break;
                    }
                    None => {
                        // Stream ended, flush remaining events
                        if !batcher.is_empty() {
                            let batch = batcher.take();
                            total_batches += 1;
                            display_batch(&batch, &mut reasoning_displayed)?;
                        }
                        break;
                    }
                }
            }
            
            // Flush batch when timer expires
            _ = batcher.ticker().tick() => {
                if !batcher.is_empty() {
                    let batch = batcher.take();
                    total_batches += 1;
                    display_batch(&batch, &mut reasoning_displayed)?;
                }
            }
        }
    }

    let elapsed = start_time.elapsed();
    println!("\n\nDone.");
    println!("Stats: {} events in {} batches ({}% reduction in network calls)", 
        total_events, 
        total_batches,
        if total_events > 0 { 100 - (total_batches * 100 / total_events) } else { 0 }
    );
    println!("Time elapsed: {:.2}s", elapsed.as_secs_f64());

    Ok(())
}

pub fn push(&mut self, event: T)

Add an event to the current batch

Examples found in repository

examples/04_reasoning_streaming.rs (line 32)

async fn main() -> Result<()> {
    let api_key = std::env::var("OPENAI_API_KEY")?;
    let client = OpenAIClient::new(api_key)?;

    let request = ResponseRequest::new(
        "gpt-5",
        vec![Message::human("Solve this problem: If a train travels 120 km in 2 hours, what is its average speed? And after that make a dynamic programming problem that can be solved in O(n) time and space complexity.")]
    ).with_reasoning(ReasoningConfig::medium());

    println!("Streaming response with reasoning (batched for network efficiency):\n");
    
    let mut stream = client.reason_stream(request).await?;
    let mut batcher = EventBatcher::new(50); // 50ms batching window
    let mut reasoning_displayed = false;
    let start_time = Instant::now();
    let mut total_events = 0;
    let mut total_batches = 0;

    loop {
        tokio::select! {
            // Receive events from stream
            event_result = stream.next() => {
                match event_result {
                    Some(Ok(event)) => {
                        total_events += 1;
                        batcher.push(event);
                    }
                    Some(Err(e)) => {
                        eprintln!("Stream error: {}", e);
                        break;
                    }
                    None => {
                        // Stream ended, flush remaining events
                        if !batcher.is_empty() {
                            let batch = batcher.take();
                            total_batches += 1;
                            display_batch(&batch, &mut reasoning_displayed)?;
                        }
                        break;
                    }
                }
            }
            
            // Flush batch when timer expires
            _ = batcher.ticker().tick() => {
                if !batcher.is_empty() {
                    let batch = batcher.take();
                    total_batches += 1;
                    display_batch(&batch, &mut reasoning_displayed)?;
                }
            }
        }
    }

    let elapsed = start_time.elapsed();
    println!("\n\nDone.");
    println!("Stats: {} events in {} batches ({}% reduction in network calls)", 
        total_events, 
        total_batches,
        if total_events > 0 { 100 - (total_batches * 100 / total_events) } else { 0 }
    );
    println!("Time elapsed: {:.2}s", elapsed.as_secs_f64());

    Ok(())
}

pub fn should_flush_now(&self) -> bool

Check if it’s time to flush (non-blocking) Note: This is a simplified check. For production use, integrate with tokio::select!

pub fn take(&mut self) -> Vec<T>

Take the current batch, leaving an empty one

Examples found in repository

examples/04_reasoning_streaming.rs (line 41)

async fn main() -> Result<()> {
    let api_key = std::env::var("OPENAI_API_KEY")?;
    let client = OpenAIClient::new(api_key)?;

    let request = ResponseRequest::new(
        "gpt-5",
        vec![Message::human("Solve this problem: If a train travels 120 km in 2 hours, what is its average speed? And after that make a dynamic programming problem that can be solved in O(n) time and space complexity.")]
    ).with_reasoning(ReasoningConfig::medium());

    println!("Streaming response with reasoning (batched for network efficiency):\n");
    
    let mut stream = client.reason_stream(request).await?;
    let mut batcher = EventBatcher::new(50); // 50ms batching window
    let mut reasoning_displayed = false;
    let start_time = Instant::now();
    let mut total_events = 0;
    let mut total_batches = 0;

    loop {
        tokio::select! {
            // Receive events from stream
            event_result = stream.next() => {
                match event_result {
                    Some(Ok(event)) => {
                        total_events += 1;
                        batcher.push(event);
                    }
                    Some(Err(e)) => {
                        eprintln!("Stream error: {}", e);
                        break;
                    }
                    None => {
                        // Stream ended, flush remaining events
                        if !batcher.is_empty() {
                            let batch = batcher.take();
                            total_batches += 1;
                            display_batch(&batch, &mut reasoning_displayed)?;
                        }
                        break;
                    }
                }
            }
            
            // Flush batch when timer expires
            _ = batcher.ticker().tick() => {
                if !batcher.is_empty() {
                    let batch = batcher.take();
                    total_batches += 1;
                    display_batch(&batch, &mut reasoning_displayed)?;
                }
            }
        }
    }

    let elapsed = start_time.elapsed();
    println!("\n\nDone.");
    println!("Stats: {} events in {} batches ({}% reduction in network calls)", 
        total_events, 
        total_batches,
        if total_events > 0 { 100 - (total_batches * 100 / total_events) } else { 0 }
    );
    println!("Time elapsed: {:.2}s", elapsed.as_secs_f64());

    Ok(())
}

pub fn len(&self) -> usize

Current batch size

pub fn is_empty(&self) -> bool

Check if batch is empty

Examples found in repository

examples/04_reasoning_streaming.rs (line 40)

async fn main() -> Result<()> {
    let api_key = std::env::var("OPENAI_API_KEY")?;
    let client = OpenAIClient::new(api_key)?;

    let request = ResponseRequest::new(
        "gpt-5",
        vec![Message::human("Solve this problem: If a train travels 120 km in 2 hours, what is its average speed? And after that make a dynamic programming problem that can be solved in O(n) time and space complexity.")]
    ).with_reasoning(ReasoningConfig::medium());

    println!("Streaming response with reasoning (batched for network efficiency):\n");
    
    let mut stream = client.reason_stream(request).await?;
    let mut batcher = EventBatcher::new(50); // 50ms batching window
    let mut reasoning_displayed = false;
    let start_time = Instant::now();
    let mut total_events = 0;
    let mut total_batches = 0;

    loop {
        tokio::select! {
            // Receive events from stream
            event_result = stream.next() => {
                match event_result {
                    Some(Ok(event)) => {
                        total_events += 1;
                        batcher.push(event);
                    }
                    Some(Err(e)) => {
                        eprintln!("Stream error: {}", e);
                        break;
                    }
                    None => {
                        // Stream ended, flush remaining events
                        if !batcher.is_empty() {
                            let batch = batcher.take();
                            total_batches += 1;
                            display_batch(&batch, &mut reasoning_displayed)?;
                        }
                        break;
                    }
                }
            }
            
            // Flush batch when timer expires
            _ = batcher.ticker().tick() => {
                if !batcher.is_empty() {
                    let batch = batcher.take();
                    total_batches += 1;
                    display_batch(&batch, &mut reasoning_displayed)?;
                }
            }
        }
    }

    let elapsed = start_time.elapsed();
    println!("\n\nDone.");
    println!("Stats: {} events in {} batches ({}% reduction in network calls)", 
        total_events, 
        total_batches,
        if total_events > 0 { 100 - (total_batches * 100 / total_events) } else { 0 }
    );
    println!("Time elapsed: {:.2}s", elapsed.as_secs_f64());

    Ok(())
}

pub fn ticker(&mut self) -> &mut Interval

Get reference to ticker for use in tokio::select!

Examples found in repository

examples/04_reasoning_streaming.rs (line 51)

async fn main() -> Result<()> {
    let api_key = std::env::var("OPENAI_API_KEY")?;
    let client = OpenAIClient::new(api_key)?;

    let request = ResponseRequest::new(
        "gpt-5",
        vec![Message::human("Solve this problem: If a train travels 120 km in 2 hours, what is its average speed? And after that make a dynamic programming problem that can be solved in O(n) time and space complexity.")]
    ).with_reasoning(ReasoningConfig::medium());

    println!("Streaming response with reasoning (batched for network efficiency):\n");
    
    let mut stream = client.reason_stream(request).await?;
    let mut batcher = EventBatcher::new(50); // 50ms batching window
    let mut reasoning_displayed = false;
    let start_time = Instant::now();
    let mut total_events = 0;
    let mut total_batches = 0;

    loop {
        tokio::select! {
            // Receive events from stream
            event_result = stream.next() => {
                match event_result {
                    Some(Ok(event)) => {
                        total_events += 1;
                        batcher.push(event);
                    }
                    Some(Err(e)) => {
                        eprintln!("Stream error: {}", e);
                        break;
                    }
                    None => {
                        // Stream ended, flush remaining events
                        if !batcher.is_empty() {
                            let batch = batcher.take();
                            total_batches += 1;
                            display_batch(&batch, &mut reasoning_displayed)?;
                        }
                        break;
                    }
                }
            }
            
            // Flush batch when timer expires
            _ = batcher.ticker().tick() => {
                if !batcher.is_empty() {
                    let batch = batcher.take();
                    total_batches += 1;
                    display_batch(&batch, &mut reasoning_displayed)?;
                }
            }
        }
    }

    let elapsed = start_time.elapsed();
    println!("\n\nDone.");
    println!("Stats: {} events in {} batches ({}% reduction in network calls)", 
        total_events, 
        total_batches,
        if total_events > 0 { 100 - (total_batches * 100 / total_events) } else { 0 }
    );
    println!("Time elapsed: {:.2}s", elapsed.as_secs_f64());

    Ok(())
}

pub fn window_ms(&self) -> u64

Get window duration