wireshift_core/strategy/
callback.rs

1//! An async callback trait mapping to be executed on success conditions.
2use std::time::Duration;
3
4use crate::completion::CompletionEvent;
5use crate::error::{Error, Result};
6use crate::{BackendFactory, Ring};
7
8/// Drains completions and invokes a caller-provided callback for each one.
9///
10/// ```no_run
11/// use wireshift::{Ring, RingConfig, strategy::CallbackDrain};
12///
13/// let ring = Ring::new(RingConfig::default())?;
14/// let drain = CallbackDrain::new(ring, 4);
15/// let _processed = drain.drain(None, |event| {
16///     let _ = event.id();
17/// })?;
18/// # Ok::<(), Box<dyn std::error::Error>>(())
19/// ```
20#[derive(Clone)]
21pub struct CallbackDrain<F: BackendFactory> {
22    ring: Ring<F>,
23    limit: usize,
24}
25
26impl<F: BackendFactory> CallbackDrain<F> {
27    /// Creates a callback-driven drain adapter.
28    #[must_use]
29    pub fn new(ring: Ring<F>, limit: usize) -> Self {
30        Self {
31            ring,
32            limit: limit.max(1),
33        }
34    }
35
36    /// Drains up to `limit` completions, invoking `callback` for each one.
37    pub fn drain<C>(&self, timeout: Option<Duration>, mut callback: C) -> Result<usize>
38    where
39        C: FnMut(CompletionEvent),
40    {
41        let mut processed = 0;
42        for index in 0..self.limit {
43            let wait = if index == 0 {
44                timeout
45            } else {
46                Some(Duration::from_millis(1))
47            };
48            match self.ring.complete(wait) {
49                Ok(event) => {
50                    callback(event);
51                    processed += 1;
52                }
53                Err(Error::Timeout { .. }) if processed > 0 => break,
54                Err(error) => return Err(error),
55            }
56        }
57        Ok(processed)
58    }
59}
wireshift_core/strategy/callback.rs

wireshift_core/strategy/
callback.rs
