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
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
|
//! Implementation of anything directly subscriber related
use core::future::Future;
use core::marker::PhantomData;
use core::ops::{Deref, DerefMut};
use core::pin::Pin;
use core::task::{Context, Poll};
use super::{PubSubBehavior, PubSubChannel, WaitResult};
use crate::blocking_mutex::raw::RawMutex;
/// A subscriber to a channel
pub struct Sub<'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> {
/// The message id of the next message we are yet to receive
next_message_id: u64,
/// The channel we are a subscriber to
channel: &'a PSB,
_phantom: PhantomData<T>,
}
impl<'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> Sub<'a, PSB, T> {
pub(super) fn new(next_message_id: u64, channel: &'a PSB) -> Self {
Self {
next_message_id,
channel,
_phantom: Default::default(),
}
}
/// Wait for a published message
pub fn next_message<'s>(&'s mut self) -> SubscriberWaitFuture<'s, 'a, PSB, T> {
SubscriberWaitFuture { subscriber: self }
}
/// Wait for a published message (ignoring lag results)
pub async fn next_message_pure(&mut self) -> T {
loop {
match self.next_message().await {
WaitResult::Lagged(_) => continue,
WaitResult::Message(message) => break message,
}
}
}
/// Try to see if there's a published message we haven't received yet.
///
/// This function does not peek. The message is received if there is one.
pub fn try_next_message(&mut self) -> Option<WaitResult<T>> {
match self.channel.get_message_with_context(&mut self.next_message_id, None) {
Poll::Ready(result) => Some(result),
Poll::Pending => None,
}
}
/// Try to see if there's a published message we haven't received yet (ignoring lag results).
///
/// This function does not peek. The message is received if there is one.
pub fn try_next_message_pure(&mut self) -> Option<T> {
loop {
match self.try_next_message() {
Some(WaitResult::Lagged(_)) => continue,
Some(WaitResult::Message(message)) => break Some(message),
None => break None,
}
}
}
}
impl<'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> Drop for Sub<'a, PSB, T> {
fn drop(&mut self) {
self.channel.unregister_subscriber(self.next_message_id)
}
}
impl<'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> Unpin for Sub<'a, PSB, T> {}
/// Warning: The stream implementation ignores lag results and returns all messages.
/// This might miss some messages without you knowing it.
impl<'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> futures_util::Stream for Sub<'a, PSB, T> {
type Item = T;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
match self
.channel
.get_message_with_context(&mut self.next_message_id, Some(cx))
{
Poll::Ready(WaitResult::Message(message)) => Poll::Ready(Some(message)),
Poll::Ready(WaitResult::Lagged(_)) => {
cx.waker().wake_by_ref();
Poll::Pending
}
Poll::Pending => Poll::Pending,
}
}
}
/// A subscriber that holds a dynamic reference to the channel
pub struct DynSubscriber<'a, T: Clone>(pub(super) Sub<'a, dyn PubSubBehavior<T> + 'a, T>);
impl<'a, T: Clone> Deref for DynSubscriber<'a, T> {
type Target = Sub<'a, dyn PubSubBehavior<T> + 'a, T>;
fn deref(&self) -> &Self::Target {
&self.0
}
}
impl<'a, T: Clone> DerefMut for DynSubscriber<'a, T> {
fn deref_mut(&mut self) -> &mut Self::Target {
&mut self.0
}
}
/// A subscriber that holds a generic reference to the channel
pub struct Subscriber<'a, M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usize>(
pub(super) Sub<'a, PubSubChannel<M, T, CAP, SUBS, PUBS>, T>,
);
impl<'a, M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usize> Deref
for Subscriber<'a, M, T, CAP, SUBS, PUBS>
{
type Target = Sub<'a, PubSubChannel<M, T, CAP, SUBS, PUBS>, T>;
fn deref(&self) -> &Self::Target {
&self.0
}
}
impl<'a, M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usize> DerefMut
for Subscriber<'a, M, T, CAP, SUBS, PUBS>
{
fn deref_mut(&mut self) -> &mut Self::Target {
&mut self.0
}
}
/// Future for the subscriber wait action
pub struct SubscriberWaitFuture<'s, 'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> {
subscriber: &'s mut Sub<'a, PSB, T>,
}
impl<'s, 'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> Future for SubscriberWaitFuture<'s, 'a, PSB, T> {
type Output = WaitResult<T>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
self.subscriber
.channel
.get_message_with_context(&mut self.subscriber.next_message_id, Some(cx))
}
}
impl<'s, 'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> Unpin for SubscriberWaitFuture<'s, 'a, PSB, T> {}
|
