[package]
name = "sigq"
version = "0.13.2"
authors = ["Jan Danielsson <jan.danielsson@qrnch.com>"]
edition = "2021"
license = "0BSD"
categories = [ "asynchronous", "concurrency", "data-structures" ]
keywords = [ "threads", "sync" ]
repository = "https://repos.qrnch.tech/pub/sigq"
description = "Queue that signals waiting consumers about node availability"

//! _sigq_ is a FIFO queue that supports pushing and poping nodes from
//! threads/tasks, crossing sync/async boundaries.  The interface to interact
//! with the queue is a pair of end-points.  The [`Pusher`] is used to add data
//! to the queue, and the [`Puller`] is used to pull data off the queue.
//!
//! The `Pusher` has a [`push()`](Pusher::push) method that is used to push new
//! nodes onto the queue.
//!
//! The `Puller` has a blocking [`pop()`](Puller::pop) and a
//! [`apop()`](Puller::apop) that returns a `Future` for getting the next node
//! off the queue.  These will return immediately with the next node if
//! available, or block and wait for a new node to be pushed onto the queue.
//! [`try_pop()`](Puller::try_pop) can be used as a non-blocking way to get the
//! next node, if available.
//!
//! ```
//! let (pusher, puller) = sigq::new();
//! pusher.push(42).unwrap();
//! assert_eq!(puller.pop(), Ok(42));
//! assert_eq!(puller.try_pop(), Ok(None));
//! ```
//!
//! # Semantics
//! - Dropping the last `Pusher` end-point will cause waiting `Puller`'s to
//!   wake up and return `Err(StaleErr)` if there are no more nodes on the
//!   queue.
//! - Dropping the last `Puller` end-point will:
//!   - Immediately drop all the nodes in the queue.
//!   - Cause the `Puller`'s to return `Err(StaleErr)` if new nodes are
//!     attempted to be added to the queue.

pub(crate) mod pull;
pub(crate) mod push;

use std::{
  collections::VecDeque, sync::atomic::AtomicUsize, sync::Arc, task::Waker
};

use parking_lot::{Condvar, Mutex};

use indexmap::IndexMap;

pub use pull::Puller;
pub use push::Pusher;

/// Error value used to indicate that there are no remote end-points available.
///
/// If a `Puller` method returns this it means the queue has no more associated
/// `Pusher`'s, which implies that no new nodes can become available.
///
/// If a `Pusher` method returns this it means that the queue has no more
/// associated `Puller`'s, which implies that there's nothing to take nodes off
/// the queue any longer.
#[derive(Debug, PartialEq)]
pub struct StaleErr;

/// Inner shared data.
///
/// This is read/write data, and hence protected by a mutex.
struct Inner<I> {

impl<I> Puller<I> {
  /// Pull the oldest node off the queue and return it.
  ///
  /// If no nodes are available on the queue, then block and wait for one to
  /// become available.
  ///
  /// Returns `Err(StaleErr)` if there are no more items in queue and there are
  /// no more [`Pusher`](super::Pusher) objects associated with this
  /// `Puller`.
  #[cfg_attr(feature = "inline-more", inline)]
  pub fn pop(&self) -> Result<I, StaleErr> {
    let mut inner = self.0.inner.lock();
    loop {
      if inner.npushers == 0 {
        break Err(StaleErr);
      } else {

  /// Pull the oldest node off the queue and return it.
  ///
  /// If a node is available on the queue then take it off and return it.
  ///
  /// If no nodes are available and there are no more [`Pusher`](super::Pusher)
  /// objects associated with this `Puller`, then return `Err(StaleErr)`.
  ///
  /// If no nodes are available and there's at least one associated `Pusher`
  /// exists then return `Ok(None)`.
  #[cfg_attr(feature = "inline-more", inline)]
  pub fn try_pop(&self) -> Result<Option<I>, StaleErr> {
    let mut inner = self.0.inner.lock();
    if let Some(n) = inner.q.pop_front() {
      Ok(Some(n))
    } else if inner.npushers == 0 {

  pub fn was_empty(&self) -> bool {
    let inner = self.0.inner.lock();
    inner.q.is_empty()
  }
}

impl<I> Drop for Puller<I> {
  /// Drop a `Puller` instance.
  ///
  /// If this is the last `Puller` end-point of a sigq instance, then the inner
  /// queue will be cleared (i.e. all its elements will be immediately
  /// dropped).
  fn drop(&mut self) {
    let mut inner = self.0.inner.lock();
    inner.npullers -= 1;

    // If this is the last puller then remove all thr nodes.
    // The nodes may contain some kind of context that must be notified that
    // the node will never reach its destination.

    let mut inner = self.0.inner.lock();
    inner.npushers += 1;
    Self(Arc::clone(&self.0))
  }
}

impl<I> Drop for Pusher<I> {
  /// Drop a `Pusher` instance.
  ///
  /// When the final instance of a sigq's instance's `Pusher` is dropped, wake
  /// up any `Puller`'s waiting for new nodes to arrive.
  fn drop(&mut self) {
    let mut inner = self.0.inner.lock();
    inner.npushers -= 1;

    // If this was the last pusher then wake any pullers that are waiting to
    // receive new items.  (When they discover that no pushers remain they will
    // return None).

# Change Log

## [Unreleased]

### Added

### Changed

### Removed


## [0.13.2] - 2023-07-26

### Added

### Changed

- Documentation updates.

### Removed


## [0.13.1] - 2023-07-25

### Added