wireshift-core 0.1.1

//! An async callback trait mapping to be executed on success conditions.
use std::time::Duration;

use crate::completion::CompletionEvent;
use crate::error::{Error, Result};
use crate::{BackendFactory, Ring};

/// Drains completions and invokes a caller-provided callback for each one.
///
/// ```no_run
/// use wireshift::{Ring, RingConfig, strategy::CallbackDrain};
///
/// let ring = Ring::new(RingConfig::default())?;
/// let drain = CallbackDrain::new(ring, 4);
/// let _processed = drain.drain(None, |event| {
///     let _ = event.id();
/// })?;
/// # Ok::<(), Box<dyn std::error::Error>>(())
/// ```
#[derive(Clone)]
pub struct CallbackDrain<F: BackendFactory> {
    ring: Ring<F>,
    limit: usize,
}

impl<F: BackendFactory> CallbackDrain<F> {
    /// Creates a callback-driven drain adapter.
    #[must_use]
    pub fn new(ring: Ring<F>, limit: usize) -> Self {
        Self {
            ring,
            limit: limit.max(1),
        }
    }

    /// Drains up to `limit` completions, invoking `callback` for each one.
    pub fn drain<C>(&self, timeout: Option<Duration>, mut callback: C) -> Result<usize>
    where
        C: FnMut(CompletionEvent),
    {
        let mut processed = 0;
        for index in 0..self.limit {
            let wait = if index == 0 {
                timeout
            } else {
                Some(Duration::from_millis(1))
            };
            match self.ring.complete(wait) {
                Ok(event) => {
                    callback(event);
                    processed += 1;
                }
                Err(Error::Timeout { .. }) if processed > 0 => break,
                Err(error) => return Err(error),
            }
        }
        Ok(processed)
    }
}