whisk 0.13.0

Simple and fast lockless async channels
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131

use alloc::sync::Arc;
use core::{
    future::Future,
    pin::Pin,
    task::{Context, Poll},
};

use crate::{wake_list::WakeHandle, Queue};

/// An MPMC channel with both send and receive capabilities
///
/// Enable the **`futures_core_3`** feature for `Channel` to implement
/// [`Stream`](futures_core_3::Stream) (generic `T` must be `Option<Item>`).
///
/// Enable the **`event_iterator`** feature for `Channel` to implement
/// [`EventIterator`](event_iterator::EventIterator).
pub struct Channel<T = (), U: ?Sized = ()>(Arc<Queue<T, U>>, WakeHandle);

impl<T, U: ?Sized> Drop for Channel<T, U> {
    fn drop(&mut self) {
        // Drop to avoid use after free
        self.1 = WakeHandle::new();
    }
}

impl<T> Channel<T> {
    /// Create a new channel.
    #[inline(always)]
    pub fn new() -> Self {
        Self(Arc::new(Queue::new()), WakeHandle::new())
    }
}

impl<T, U> Channel<T, U> {
    /// Create a new channel with associated data.
    #[inline(always)]
    pub fn with(user_data: U) -> Self {
        Self::from(Arc::new(Queue::with(user_data)))
    }
}

impl<T, U: ?Sized> Channel<T, U> {
    /// Send a message on this channel.
    #[inline(always)]
    pub async fn send(&self, message: T) {
        self.0.send(message).await
    }

    /// Receive a message from this channel.
    #[inline(always)]
    pub async fn recv(&self) -> T {
        self.0.recv().await
    }
}

impl<T, U: ?Sized> Clone for Channel<T, U> {
    fn clone(&self) -> Self {
        Self(Arc::clone(&self.0), WakeHandle::new())
    }
}

impl<T, U: ?Sized> core::fmt::Debug for Channel<T, U> {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        f.debug_struct("Channel").finish_non_exhaustive()
    }
}

impl<T, U: ?Sized + Default> Default for Channel<T, U> {
    fn default() -> Self {
        Self::with(U::default())
    }
}

impl<T, U: ?Sized> core::ops::Deref for Channel<T, U> {
    type Target = U;

    fn deref(&self) -> &Self::Target {
        &self.0.user
    }
}

impl<T, U: ?Sized> Future for Channel<T, U> {
    type Output = T;

    #[inline(always)]
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<T> {
        let this = self.get_mut();
        this.0.data.take(cx, &mut this.1)
    }
}

#[cfg(feature = "event_iterator")]
impl<T, U: ?Sized> event_iterator::EventIterator for Channel<T, U> {
    type Event<'me> = T where Self: 'me;

    #[inline(always)]
    fn poll_next(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<T>> {
        let this = self.get_mut();

        this.0.data.take(cx, &mut this.1).map(Some)
    }
}

#[cfg(feature = "futures_core_3")]
impl<T, U: ?Sized> futures_core_3::Stream for Channel<Option<T>, U> {
    type Item = T;

    #[inline(always)]
    fn poll_next(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<T>> {
        let this = self.get_mut();
        this.0.data.take(cx, &mut this.1)
    }
}

impl<T, U: ?Sized> From<Arc<Queue<T, U>>> for Channel<T, U> {
    fn from(inner: Arc<Queue<T, U>>) -> Self {
        Self(inner, WakeHandle::new())
    }
}

impl<T, U: ?Sized> From<Channel<T, U>> for Arc<Queue<T, U>> {
    fn from(channel: Channel<T, U>) -> Self {
        channel.0.clone()
    }
}