use crate::utils::ptr_ext::MutPtrExt;
use crate::utils::ptr_ext::PtrExt;
use derivative::Derivative;
use std::cell::Cell;
use std::cell::UnsafeCell;
use std::collections::VecDeque;
use std::mem;
use std::pin::Pin;
use std::task::Context;
use std::task::Poll;
use std::task::Waker;
#[derive(Derivative)]
#[derivative(Default(bound = ""))]
pub struct AsyncQueue<T> {
data: UnsafeCell<VecDeque<T>>,
waiter: Cell<Option<Waker>>,
}
impl<T> AsyncQueue<T> {
pub fn push(&self, t: T) {
unsafe {
self.data.get().deref_mut().push_back(t);
}
if let Some(waiter) = self.waiter.take() {
waiter.wake();
}
}
pub fn push_front(&self, t: T) {
unsafe {
self.data.get().deref_mut().push_front(t);
}
if let Some(waiter) = self.waiter.take() {
waiter.wake();
}
}
pub fn try_pop(&self) -> Option<T> {
unsafe { self.data.get().deref_mut().pop_front() }
}
pub fn pop(&self) -> AsyncQueuePop<'_, T> {
AsyncQueuePop { queue: self }
}
pub fn non_empty(&self) -> AsyncQueueNonEmpty<'_, T> {
AsyncQueueNonEmpty { queue: self }
}
pub fn is_empty(&self) -> bool {
unsafe { self.data.get().deref().is_empty() }
}
pub fn is_not_empty(&self) -> bool {
!self.is_empty()
}
pub fn clear(&self) {
unsafe {
mem::take(self.data.get().deref_mut());
}
self.waiter.take();
}
pub fn move_to(&self, other: &mut VecDeque<T>) {
unsafe {
other.append(self.data.get().deref_mut());
}
}
pub fn swap(&self, other: &mut VecDeque<T>) {
unsafe {
mem::swap(other, self.data.get().deref_mut());
}
}
}
pub struct AsyncQueuePop<'a, T> {
queue: &'a AsyncQueue<T>,
}
impl<T> Future for AsyncQueuePop<'_, T> {
type Output = T;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
if let Some(t) = self.queue.try_pop() {
Poll::Ready(t)
} else {
self.queue.waiter.set(Some(cx.waker().clone()));
Poll::Pending
}
}
}
pub struct AsyncQueueNonEmpty<'a, T> {
queue: &'a AsyncQueue<T>,
}
impl<T> Future for AsyncQueueNonEmpty<'_, T> {
type Output = ();
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
if unsafe { self.queue.data.get().deref().len() } > 0 {
Poll::Ready(())
} else {
self.queue.waiter.set(Some(cx.waker().clone()));
Poll::Pending
}
}
}