workflow_egui/runtime/
channel.rs

1use crate::runtime::runtime;
2use workflow_core::channel::{
3    Receiver, RecvError, SendError, Sender as ChannelSender, TryRecvError, TrySendError, unbounded,
4};
5
6#[derive(Debug, Clone)]
7/// A channel sender that requests a UI repaint after each successful send so
8/// that messages produced by async tasks are promptly reflected in the egui UI.
9pub struct Sender<T> {
10    sender: ChannelSender<T>,
11}
12
13impl<T> Sender<T> {
14    /// Wraps an underlying channel sender in a repaint-aware sender.
15    pub fn new(sender: ChannelSender<T>) -> Self {
16        Self { sender }
17    }
18
19    /// Sends a message, awaiting capacity if needed, and requests a UI repaint.
20    pub async fn send(&self, msg: T) -> Result<(), SendError<T>> {
21        self.sender.send(msg).await?;
22        runtime().request_repaint();
23        Ok(())
24    }
25
26    /// Sends a message without blocking and requests a UI repaint.
27    pub fn try_send(&self, msg: T) -> Result<(), TrySendError<T>> {
28        self.sender.try_send(msg)?;
29        runtime().request_repaint();
30
31        Ok(())
32    }
33
34    /// Returns the number of receivers currently associated with the channel.
35    pub fn receiver_count(&self) -> usize {
36        self.sender.receiver_count()
37    }
38
39    /// Returns the number of senders currently associated with the channel.
40    pub fn sender_count(&self) -> usize {
41        self.sender.sender_count()
42    }
43}
44
45#[derive(Debug, Clone)]
46/// A bundled sender/receiver pair whose sends request a UI repaint, used for
47/// communication between async tasks and the egui event loop.
48pub struct Channel<T = ()> {
49    /// The repaint-aware sending half of the channel.
50    pub sender: Sender<T>,
51    /// The receiving half of the channel.
52    pub receiver: Receiver<T>,
53}
54
55impl<T> Channel<T> {
56    /// Creates a new channel with an unbounded message capacity.
57    pub fn unbounded() -> Self {
58        let (sender, receiver) = unbounded();
59        Self {
60            sender: Sender::new(sender),
61            receiver,
62        }
63    }
64
65    /// Discards all currently-queued messages from the channel.
66    pub fn drain(&self) -> std::result::Result<(), TryRecvError> {
67        while !self.receiver.is_empty() {
68            self.receiver.try_recv()?;
69        }
70        Ok(())
71    }
72
73    /// Awaits and returns the next message from the channel.
74    pub async fn recv(&self) -> Result<T, RecvError> {
75        self.receiver.recv().await
76    }
77
78    /// Receives a message without blocking, failing if the channel is empty.
79    pub fn try_recv(&self) -> Result<T, TryRecvError> {
80        self.receiver.try_recv()
81    }
82
83    /// Sends a message, awaiting capacity if the channel is bounded and full.
84    pub async fn send(&self, msg: T) -> Result<(), SendError<T>> {
85        self.sender.send(msg).await
86    }
87
88    /// Sends a message without blocking, failing if the channel is full or closed.
89    pub fn try_send(&self, msg: T) -> Result<(), TrySendError<T>> {
90        self.sender.try_send(msg)
91    }
92
93    /// Returns the number of messages currently queued in the channel.
94    pub fn len(&self) -> usize {
95        self.receiver.len()
96    }
97
98    /// Returns `true` if there are no messages currently queued.
99    pub fn is_empty(&self) -> bool {
100        self.receiver.is_empty()
101    }
102
103    /// Returns the number of receivers currently associated with the channel.
104    pub fn receiver_count(&self) -> usize {
105        self.sender.receiver_count()
106    }
107
108    /// Returns the number of senders currently associated with the channel.
109    pub fn sender_count(&self) -> usize {
110        self.sender.sender_count()
111    }
112
113    /// Returns a non-blocking iterator over the currently-queued messages.
114    pub fn iter(&self) -> ChannelIterator<T> {
115        ChannelIterator::new(self.receiver.clone())
116    }
117}
118
119/// Non-blocking iterator that yields currently-queued messages from a channel
120/// receiver, stopping once the channel is momentarily empty.
121pub struct ChannelIterator<T> {
122    receiver: Receiver<T>,
123}
124
125impl<T> ChannelIterator<T> {
126    /// Creates an iterator that drains messages from the given receiver.
127    pub fn new(receiver: Receiver<T>) -> Self {
128        ChannelIterator { receiver }
129    }
130}
131
132impl<T> Iterator for ChannelIterator<T> {
133    type Item = T;
134    fn next(&mut self) -> Option<T> {
135        if self.receiver.is_empty() {
136            None
137        } else {
138            self.receiver.try_recv().ok()
139        }
140    }
141}
workflow_egui/runtime/channel.rs

workflow_egui/runtime/
channel.rs
