gstreamer/
bus_futures.rs

1// Take a look at the license at the top of the repository in the LICENSE file.
2
3use std::{
4    pin::Pin,
5    sync::{Arc, Mutex},
6    task::{Context, Poll},
7};
8
9use futures_channel::mpsc::{self, UnboundedReceiver};
10use futures_core::Stream;
11use futures_util::{StreamExt, stream::FusedStream};
12use glib::object::ObjectExt as _;
13
14use crate::{Bus, BusSyncReply, Message};
15
16#[derive(Debug)]
17pub struct BusStream {
18    bus: glib::WeakRef<Bus>,
19    receiver: UnboundedReceiver<Message>,
20}
21
22impl BusStream {
23    pub(crate) fn new(bus: &Bus) -> Self {
24        skip_assert_initialized!();
25
26        let mutex = Arc::new(Mutex::new(()));
27        let (sender, receiver) = mpsc::unbounded();
28
29        // Use a mutex to ensure that the sync handler is not putting any messages into the sender
30        // until we have removed all previously queued messages from the bus.
31        // This makes sure that the messages are staying in order.
32        //
33        // We could use the bus' object lock here but a separate mutex seems safer.
34        let _mutex_guard = mutex.lock().unwrap();
35        bus.set_sync_handler({
36            let sender = sender.clone();
37            let mutex = mutex.clone();
38
39            move |_bus, message| {
40                let _mutex_guard = mutex.lock().unwrap();
41
42                let _ = sender.unbounded_send(message.to_owned());
43
44                BusSyncReply::Drop
45            }
46        });
47
48        // First pop all messages that might've been previously queued before creating the bus stream.
49        while let Some(message) = bus.pop() {
50            let _ = sender.unbounded_send(message);
51        }
52
53        Self {
54            bus: bus.downgrade(),
55            receiver,
56        }
57    }
58}
59
60impl Drop for BusStream {
61    fn drop(&mut self) {
62        if let Some(bus) = self.bus.upgrade() {
63            bus.unset_sync_handler();
64        }
65    }
66}
67
68impl Stream for BusStream {
69    type Item = Message;
70
71    fn poll_next(mut self: Pin<&mut Self>, context: &mut Context) -> Poll<Option<Self::Item>> {
72        self.receiver.poll_next_unpin(context)
73    }
74}
75
76impl FusedStream for BusStream {
77    fn is_terminated(&self) -> bool {
78        self.receiver.is_terminated()
79    }
80}
81
82#[cfg(test)]
83mod tests {
84    use super::*;
85
86    #[test]
87    fn test_bus_stream() {
88        crate::init().unwrap();
89
90        let bus = Bus::new();
91        let bus_stream = bus.stream();
92
93        let eos_message = crate::message::Eos::new();
94        bus.post(eos_message).unwrap();
95
96        let bus_future = StreamExt::into_future(bus_stream);
97        let (message, _) = futures_executor::block_on(bus_future);
98
99        match message.unwrap().view() {
100            crate::MessageView::Eos(_) => (),
101            _ => unreachable!(),
102        }
103    }
104}
gstreamer/bus_futures.rs

gstreamer/
bus_futures.rs
