wireshift-core 0.1.1

//! Completion types and typed request handles.

use std::any::Any;
use std::marker::PhantomData;
use std::sync::mpsc::{Receiver, TryRecvError};
use std::sync::Arc;
use std::time::{Duration, Instant};

use crate::error::{Error, Result};
use crate::ring::RingInner;

/// High-level completion kind.
///
/// ```rust
/// use wireshift::CompletionKind;
///
/// assert_eq!(CompletionKind::Completed.as_str(), "completed");
/// ```
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[non_exhaustive]
pub enum CompletionKind {
    /// The operation completed successfully.
    Completed,
    /// The operation failed.
    Failed,
    /// The operation was canceled.
    Canceled,
}

impl CompletionKind {
    /// Returns a stable string representation.
    #[must_use]
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Completed => "completed",
            Self::Failed => "failed",
            Self::Canceled => "canceled",
        }
    }
}

/// Metadata for one completed submission.
///
/// ```no_run
/// use wireshift::{CompletionKind, Ring, RingConfig, ops::Nop};
///
/// let ring = Ring::new(RingConfig::default())?;
/// let request = ring.submit(Nop)?;
/// let event = ring.complete(None)?;
/// assert_eq!(event.kind(), CompletionKind::Completed);
/// # drop(request);
/// # Ok::<(), Box<dyn std::error::Error>>(())
/// ```
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct CompletionEvent {
    id: u64,
    op_name: &'static str,
    kind: CompletionKind,
}

impl CompletionEvent {
    pub(crate) fn new(id: u64, op_name: &'static str, kind: CompletionKind) -> Self {
        Self { id, op_name, kind }
    }

    /// Returns the request identifier.
    #[must_use]
    pub fn id(&self) -> u64 {
        self.id
    }

    /// Returns the operation name.
    #[must_use]
    pub fn op_name(&self) -> &'static str {
        self.op_name
    }

    /// Returns the completion kind.
    #[must_use]
    pub fn kind(&self) -> CompletionKind {
        self.kind
    }
}

/// Typed handle for a submitted operation.
///
/// ```no_run
/// use wireshift::{Ring, RingConfig, ops::Nop};
///
/// let ring = Ring::new(RingConfig::default())?;
/// let request = ring.submit(Nop)?;
/// request.wait(None)?;
/// # Ok::<(), Box<dyn std::error::Error>>(())
/// ```
pub struct Request<T> {
    pub(crate) id: u64,
    pub(crate) ring: Arc<RingInner>,
    pub(crate) receiver: Receiver<Result<Box<dyn Any + Send>>>,
    pub(crate) marker: PhantomData<T>,
}

impl<T: Send + 'static> Request<T> {
    /// Returns the stable request identifier.
    #[must_use]
    pub fn id(&self) -> u64 {
        self.id
    }

    /// Waits until the operation completes or the optional timeout expires.
    pub fn wait(self, timeout: Option<Duration>) -> Result<T> {
        let deadline = timeout
            .or(self.ring.config.default_timeout)
            .map(|duration| Instant::now() + duration);
        loop {
            match self.receiver.try_recv() {
                Ok(result) => {
                    let boxed = result?;
                    return boxed.downcast::<T>().map(|value| *value).map_err(|_| {
                        Error::completion(
                            "typed completion downcast failed",
                            "ensure the request is awaited with the same output type it was submitted with",
                        )
                    });
                }
                Err(TryRecvError::Disconnected) => {
                    return Err(Error::completion(
                        "request completion channel disconnected",
                        "keep the ring alive until all in-flight requests are completed",
                    ));
                }
                Err(TryRecvError::Empty) => {}
            }
            let remaining = deadline.map(|target| target.saturating_duration_since(Instant::now()));
            if matches!(remaining, Some(duration) if duration.is_zero()) {
                return Err(Error::timeout(
                    "request timed out while waiting for completion",
                    "increase the timeout or reduce backend load",
                ));
            }
            let step = remaining
                .unwrap_or_else(|| Duration::from_millis(50))
                .min(Duration::from_millis(50));
            match self.ring.pump_completion(Some(step)) {
                Ok(_) | Err(Error::Timeout { .. }) => {}
                Err(error) => return Err(error),
            }
        }
    }
}