use std::fmt;
use std::sync::{Arc, Mutex, MutexGuard};
pub trait UiAction: Clone + PartialEq + fmt::Debug + Send + Sync + 'static {}
impl<T> UiAction for T where T: Clone + PartialEq + fmt::Debug + Send + Sync + 'static {}
type Shared<A> = Arc<Mutex<Vec<A>>>;
fn lock<A>(shared: &Shared<A>) -> MutexGuard<'_, Vec<A>> {
shared.lock().unwrap_or_else(|poisoned| poisoned.into_inner())
}
#[derive(Clone)]
pub struct Sink<A: UiAction> {
shared: Shared<A>,
}
impl<A: UiAction> Sink<A> {
pub fn push(&self, action: A) {
lock(&self.shared).push(action);
}
pub fn extend(&self, actions: impl IntoIterator<Item = A>) {
lock(&self.shared).extend(actions);
}
pub fn pending(&self) -> usize {
lock(&self.shared).len()
}
pub fn drain(&self) -> Vec<A> {
drain_buffer(&self.shared)
}
}
impl<A: UiAction> fmt::Debug for Sink<A> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Sink").field("pending", &self.pending()).finish()
}
}
#[derive(Clone)]
pub struct ActionQueue<A: UiAction> {
shared: Shared<A>,
}
impl<A: UiAction> ActionQueue<A> {
pub fn new() -> Self {
Self {
shared: Arc::new(Mutex::new(Vec::new())),
}
}
pub fn pair() -> (Self, Sink<A>) {
let queue = Self::new();
let sink = queue.sink();
(queue, sink)
}
pub fn sink(&self) -> Sink<A> {
Sink {
shared: Arc::clone(&self.shared),
}
}
pub fn push(&self, action: A) {
lock(&self.shared).push(action);
}
pub fn pending(&self) -> usize {
lock(&self.shared).len()
}
pub fn is_empty(&self) -> bool {
self.pending() == 0
}
pub fn drain(&self) -> Vec<A> {
drain_buffer(&self.shared)
}
pub fn drain_into(&self, out: &mut Vec<A>) {
let mut buffer = lock(&self.shared);
out.append(&mut buffer);
}
pub fn absorb(&self, intents: &mut Intents<A>) -> usize {
intents.extend(self.drain())
}
}
impl<A: UiAction> Default for ActionQueue<A> {
fn default() -> Self {
Self::new()
}
}
impl<A: UiAction> fmt::Debug for ActionQueue<A> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("ActionQueue")
.field("pending", &self.pending())
.finish()
}
}
fn drain_buffer<A>(shared: &Shared<A>) -> Vec<A> {
std::mem::take(&mut *lock(shared))
}
#[derive(Clone, Debug, Default, PartialEq)]
pub struct Intents<A: UiAction> {
pending: Vec<A>,
}
impl<A: UiAction> Intents<A> {
pub fn new() -> Self {
Self { pending: Vec::new() }
}
pub fn requests(&self) -> &[A] {
&self.pending
}
pub fn is_empty(&self) -> bool {
self.pending.is_empty()
}
pub fn len(&self) -> usize {
self.pending.len()
}
pub fn clear(&mut self) {
self.pending.clear();
}
pub fn request(&mut self, action: A) {
self.pending.push(action);
}
pub fn extend(&mut self, actions: impl IntoIterator<Item = A>) -> usize {
let before = self.pending.len();
self.pending.extend(actions);
self.pending.len() - before
}
pub fn requested(&self, action: &A) -> bool {
self.pending.contains(action)
}
pub fn count(&self, action: &A) -> usize {
self.pending.iter().filter(|pending| *pending == action).count()
}
pub fn take(&mut self, action: &A) -> bool {
match self.pending.iter().position(|pending| pending == action) {
Some(index) => {
self.pending.remove(index);
true
}
None => false,
}
}
pub fn take_matching(&mut self, mut predicate: impl FnMut(&A) -> bool) -> Option<A> {
let index = self.pending.iter().position(|action| predicate(action))?;
Some(self.pending.remove(index))
}
pub fn take_all(&mut self) -> Vec<A> {
std::mem::take(&mut self.pending)
}
}
impl<A: UiAction> FromIterator<A> for Intents<A> {
fn from_iter<T: IntoIterator<Item = A>>(iter: T) -> Self {
Self {
pending: iter.into_iter().collect(),
}
}
}
pub fn absorb<A: UiAction>(queue: &ActionQueue<A>, intents: &mut Intents<A>) -> usize {
queue.absorb(intents)
}
#[cfg(test)]
mod tests {
use super::*;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum Act {
Play,
Pause,
Pick(u8),
}
#[test]
fn drains_in_push_order() {
let (queue, sink) = ActionQueue::pair();
sink.push(Act::Play);
sink.push(Act::Pause);
sink.push(Act::Play);
assert_eq!(queue.drain(), vec![Act::Play, Act::Pause, Act::Play]);
}
#[test]
fn drain_empties_the_buffer_once() {
let (queue, sink) = ActionQueue::pair();
sink.push(Act::Play);
assert_eq!(queue.drain(), vec![Act::Play]);
assert!(queue.drain().is_empty());
assert!(queue.is_empty());
assert_eq!(queue.pending(), 0);
}
#[test]
fn a_batch_never_contains_what_arrives_after_the_drain_started() {
let (queue, sink) = ActionQueue::pair();
sink.push(Act::Play);
let batch = queue.drain();
sink.push(Act::Pause);
assert_eq!(batch, vec![Act::Play]);
assert_eq!(queue.drain(), vec![Act::Pause]);
}
#[test]
fn clones_push_into_the_same_buffer() {
let (queue, sink) = ActionQueue::pair();
let clone = sink.clone();
clone.push(Act::Play);
assert_eq!(sink.pending(), 1);
assert_eq!(queue.drain(), vec![Act::Play]);
}
#[test]
fn drain_into_appends_to_the_caller_buffer() {
let (queue, sink) = ActionQueue::pair();
let mut out = vec![Act::Play];
sink.push(Act::Pause);
queue.drain_into(&mut out);
assert_eq!(out, vec![Act::Play, Act::Pause]);
}
#[test]
fn a_panic_while_pushing_keeps_the_queued_actions() {
let (queue, sink) = ActionQueue::pair();
sink.push(Act::Play);
let shared = Arc::clone(&queue.shared);
let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(move || {
let _guard = shared.lock().unwrap();
panic!("pusher panicked while holding the queue");
}));
sink.push(Act::Pause);
assert_eq!(queue.drain(), vec![Act::Play, Act::Pause]);
}
#[test]
fn intents_consume_one_occurrence_at_a_time() {
let mut intents = Intents::new();
intents.extend([Act::Play, Act::Pause, Act::Play]);
assert_eq!(intents.count(&Act::Play), 2);
assert!(intents.requested(&Act::Play));
assert!(intents.take(&Act::Play));
assert!(intents.take(&Act::Play));
assert!(!intents.take(&Act::Play));
assert!(!intents.requested(&Act::Play));
assert!(intents.take(&Act::Pause));
assert!(intents.is_empty());
}
#[test]
fn take_matching_keeps_the_other_requests() {
let mut intents = Intents::new();
intents.extend([Act::Play, Act::Pick(3), Act::Pick(7)]);
assert_eq!(
intents.take_matching(|action| matches!(action, Act::Pick(_))),
Some(Act::Pick(3))
);
assert_eq!(intents.requests(), [Act::Play, Act::Pick(7)]);
assert_eq!(intents.take_matching(|action| matches!(action, Act::Pick(_))), Some(Act::Pick(7)));
assert!(intents.take_matching(|action| matches!(action, Act::Pick(_))).is_none());
}
#[test]
fn absorb_reports_how_many_actions_arrived() {
let (queue, sink) = ActionQueue::pair();
let mut intents = Intents::new();
assert_eq!(queue.absorb(&mut intents), 0);
sink.extend([Act::Play, Act::Pause]);
assert_eq!(queue.absorb(&mut intents), 2);
assert_eq!(intents.requests(), [Act::Play, Act::Pause]);
assert_eq!(intents.take_all(), vec![Act::Play, Act::Pause]);
assert!(intents.is_empty());
}
}