1use std::{
4 future,
5 mem::transmute,
6 pin::Pin,
7 sync::{Arc, Mutex},
8 task::{Context, Poll},
9};
10
11use futures_channel::mpsc::{self, UnboundedReceiver};
12use futures_core::Stream;
13use futures_util::{stream::FusedStream, StreamExt};
14use glib::{
15 ffi::{gboolean, gpointer},
16 prelude::*,
17 source::Priority,
18 translate::*,
19 ControlFlow,
20};
21
22use crate::{ffi, Bus, BusSyncReply, Message, MessageType};
23
24unsafe extern "C" fn trampoline_watch<F: FnMut(&Bus, &Message) -> ControlFlow + Send + 'static>(
25 bus: *mut ffi::GstBus,
26 msg: *mut ffi::GstMessage,
27 func: gpointer,
28) -> gboolean {
29 let func: &mut F = &mut *(func as *mut F);
30 func(&from_glib_borrow(bus), &Message::from_glib_borrow(msg)).into_glib()
31}
32
33unsafe extern "C" fn destroy_closure_watch<
34 F: FnMut(&Bus, &Message) -> ControlFlow + Send + 'static,
35>(
36 ptr: gpointer,
37) {
38 let _ = Box::<F>::from_raw(ptr as *mut _);
39}
40
41fn into_raw_watch<F: FnMut(&Bus, &Message) -> ControlFlow + Send + 'static>(func: F) -> gpointer {
42 #[allow(clippy::type_complexity)]
43 let func: Box<F> = Box::new(func);
44 Box::into_raw(func) as gpointer
45}
46
47unsafe extern "C" fn trampoline_watch_local<F: FnMut(&Bus, &Message) -> ControlFlow + 'static>(
48 bus: *mut ffi::GstBus,
49 msg: *mut ffi::GstMessage,
50 func: gpointer,
51) -> gboolean {
52 let func: &mut glib::thread_guard::ThreadGuard<F> =
53 &mut *(func as *mut glib::thread_guard::ThreadGuard<F>);
54 (func.get_mut())(&from_glib_borrow(bus), &Message::from_glib_borrow(msg)).into_glib()
55}
56
57unsafe extern "C" fn destroy_closure_watch_local<
58 F: FnMut(&Bus, &Message) -> ControlFlow + 'static,
59>(
60 ptr: gpointer,
61) {
62 let _ = Box::<glib::thread_guard::ThreadGuard<F>>::from_raw(ptr as *mut _);
63}
64
65fn into_raw_watch_local<F: FnMut(&Bus, &Message) -> ControlFlow + 'static>(func: F) -> gpointer {
66 #[allow(clippy::type_complexity)]
67 let func: Box<glib::thread_guard::ThreadGuard<F>> =
68 Box::new(glib::thread_guard::ThreadGuard::new(func));
69 Box::into_raw(func) as gpointer
70}
71
72unsafe extern "C" fn trampoline_sync<
73 F: Fn(&Bus, &Message) -> BusSyncReply + Send + Sync + 'static,
74>(
75 bus: *mut ffi::GstBus,
76 msg: *mut ffi::GstMessage,
77 func: gpointer,
78) -> ffi::GstBusSyncReply {
79 let f: &F = &*(func as *const F);
80 let res = f(&from_glib_borrow(bus), &Message::from_glib_borrow(msg)).into_glib();
81
82 if res == ffi::GST_BUS_DROP {
83 ffi::gst_mini_object_unref(msg as *mut _);
84 }
85
86 res
87}
88
89unsafe extern "C" fn destroy_closure_sync<
90 F: Fn(&Bus, &Message) -> BusSyncReply + Send + Sync + 'static,
91>(
92 ptr: gpointer,
93) {
94 let _ = Box::<F>::from_raw(ptr as *mut _);
95}
96
97fn into_raw_sync<F: Fn(&Bus, &Message) -> BusSyncReply + Send + Sync + 'static>(
98 func: F,
99) -> gpointer {
100 let func: Box<F> = Box::new(func);
101 Box::into_raw(func) as gpointer
102}
103
104impl Bus {
105 #[doc(alias = "gst_bus_add_signal_watch")]
123 #[doc(alias = "gst_bus_add_signal_watch_full")]
124 pub fn add_signal_watch_full(&self, priority: Priority) {
125 unsafe {
126 ffi::gst_bus_add_signal_watch_full(self.to_glib_none().0, priority.into_glib());
127 }
128 }
129
130 #[doc(alias = "gst_bus_create_watch")]
141 pub fn create_watch<F>(&self, name: Option<&str>, priority: Priority, func: F) -> glib::Source
142 where
143 F: FnMut(&Bus, &Message) -> ControlFlow + Send + 'static,
144 {
145 skip_assert_initialized!();
146 unsafe {
147 let source = ffi::gst_bus_create_watch(self.to_glib_none().0);
148 glib::ffi::g_source_set_callback(
149 source,
150 Some(transmute::<
151 *mut (),
152 unsafe extern "C" fn(glib::ffi::gpointer) -> i32,
153 >(trampoline_watch::<F> as *mut ())),
154 into_raw_watch(func),
155 Some(destroy_closure_watch::<F>),
156 );
157 glib::ffi::g_source_set_priority(source, priority.into_glib());
158
159 if let Some(name) = name {
160 glib::ffi::g_source_set_name(source, name.to_glib_none().0);
161 }
162
163 from_glib_full(source)
164 }
165 }
166
167 #[doc(alias = "gst_bus_add_watch")]
192 #[doc(alias = "gst_bus_add_watch_full")]
193 pub fn add_watch<F>(&self, func: F) -> Result<BusWatchGuard, glib::BoolError>
194 where
195 F: FnMut(&Bus, &Message) -> ControlFlow + Send + 'static,
196 {
197 unsafe {
198 let res = ffi::gst_bus_add_watch_full(
199 self.to_glib_none().0,
200 glib::ffi::G_PRIORITY_DEFAULT,
201 Some(trampoline_watch::<F>),
202 into_raw_watch(func),
203 Some(destroy_closure_watch::<F>),
204 );
205
206 if res == 0 {
207 Err(glib::bool_error!("Bus already has a watch"))
208 } else {
209 Ok(BusWatchGuard { bus: self.clone() })
210 }
211 }
212 }
213
214 #[doc(alias = "gst_bus_add_watch")]
215 #[doc(alias = "gst_bus_add_watch_full")]
216 pub fn add_watch_local<F>(&self, func: F) -> Result<BusWatchGuard, glib::BoolError>
217 where
218 F: FnMut(&Bus, &Message) -> ControlFlow + 'static,
219 {
220 unsafe {
221 let ctx = glib::MainContext::ref_thread_default();
222 let _acquire = ctx
223 .acquire()
224 .expect("thread default main context already acquired by another thread");
225
226 let res = ffi::gst_bus_add_watch_full(
227 self.to_glib_none().0,
228 glib::ffi::G_PRIORITY_DEFAULT,
229 Some(trampoline_watch_local::<F>),
230 into_raw_watch_local(func),
231 Some(destroy_closure_watch_local::<F>),
232 );
233
234 if res == 0 {
235 Err(glib::bool_error!("Bus already has a watch"))
236 } else {
237 Ok(BusWatchGuard { bus: self.clone() })
238 }
239 }
240 }
241
242 #[doc(alias = "gst_bus_set_sync_handler")]
256 pub fn set_sync_handler<F>(&self, func: F)
257 where
258 F: Fn(&Bus, &Message) -> BusSyncReply + Send + Sync + 'static,
259 {
260 unsafe {
261 let bus = self.to_glib_none().0;
262
263 #[cfg(not(feature = "v1_18"))]
264 {
265 static SET_ONCE_QUARK: std::sync::OnceLock<glib::Quark> =
266 std::sync::OnceLock::new();
267
268 let set_once_quark = SET_ONCE_QUARK
269 .get_or_init(|| glib::Quark::from_str("gstreamer-rs-sync-handler"));
270
271 if crate::version() < (1, 16, 3, 0) {
274 if !glib::gobject_ffi::g_object_get_qdata(
275 bus as *mut _,
276 set_once_quark.into_glib(),
277 )
278 .is_null()
279 {
280 panic!("Bus sync handler can only be set once");
281 }
282
283 glib::gobject_ffi::g_object_set_qdata(
284 bus as *mut _,
285 set_once_quark.into_glib(),
286 1 as *mut _,
287 );
288 }
289 }
290
291 ffi::gst_bus_set_sync_handler(
292 bus,
293 Some(trampoline_sync::<F>),
294 into_raw_sync(func),
295 Some(destroy_closure_sync::<F>),
296 )
297 }
298 }
299
300 pub fn unset_sync_handler(&self) {
301 #[cfg(not(feature = "v1_18"))]
302 {
303 if crate::version() < (1, 16, 3, 0) {
306 return;
307 }
308 }
309
310 unsafe {
311 use std::ptr;
312
313 ffi::gst_bus_set_sync_handler(self.to_glib_none().0, None, ptr::null_mut(), None)
314 }
315 }
316
317 #[doc(alias = "gst_bus_pop")]
318 pub fn iter(&self) -> Iter {
319 self.iter_timed(Some(crate::ClockTime::ZERO))
320 }
321
322 #[doc(alias = "gst_bus_timed_pop")]
323 pub fn iter_timed(&self, timeout: impl Into<Option<crate::ClockTime>>) -> Iter {
324 Iter {
325 bus: self,
326 timeout: timeout.into(),
327 }
328 }
329
330 #[doc(alias = "gst_bus_pop_filtered")]
331 pub fn iter_filtered<'a>(
332 &'a self,
333 msg_types: &'a [MessageType],
334 ) -> impl Iterator<Item = Message> + 'a {
335 self.iter_timed_filtered(Some(crate::ClockTime::ZERO), msg_types)
336 }
337
338 #[doc(alias = "gst_bus_timed_pop_filtered")]
339 pub fn iter_timed_filtered<'a>(
340 &'a self,
341 timeout: impl Into<Option<crate::ClockTime>>,
342 msg_types: &'a [MessageType],
343 ) -> impl Iterator<Item = Message> + 'a {
344 self.iter_timed(timeout)
345 .filter(move |msg| msg_types.contains(&msg.type_()))
346 }
347
348 #[doc(alias = "gst_bus_timed_pop_filtered")]
366 pub fn timed_pop_filtered(
367 &self,
368 timeout: impl Into<Option<crate::ClockTime>> + Clone,
369 msg_types: &[MessageType],
370 ) -> Option<Message> {
371 loop {
372 let msg = self.timed_pop(timeout.clone())?;
373 if msg_types.contains(&msg.type_()) {
374 return Some(msg);
375 }
376 }
377 }
378
379 #[doc(alias = "gst_bus_pop_filtered")]
393 pub fn pop_filtered(&self, msg_types: &[MessageType]) -> Option<Message> {
394 loop {
395 let msg = self.pop()?;
396 if msg_types.contains(&msg.type_()) {
397 return Some(msg);
398 }
399 }
400 }
401
402 pub fn stream(&self) -> BusStream {
403 BusStream::new(self)
404 }
405
406 pub fn stream_filtered<'a>(
407 &self,
408 message_types: &'a [MessageType],
409 ) -> impl FusedStream<Item = Message> + Unpin + Send + 'a {
410 self.stream().filter(move |message| {
411 let message_type = message.type_();
412
413 future::ready(message_types.contains(&message_type))
414 })
415 }
416}
417
418#[derive(Debug)]
419pub struct Iter<'a> {
420 bus: &'a Bus,
421 timeout: Option<crate::ClockTime>,
422}
423
424impl Iterator for Iter<'_> {
425 type Item = Message;
426
427 fn next(&mut self) -> Option<Message> {
428 self.bus.timed_pop(self.timeout)
429 }
430}
431
432#[derive(Debug)]
433pub struct BusStream {
434 bus: glib::WeakRef<Bus>,
435 receiver: UnboundedReceiver<Message>,
436}
437
438impl BusStream {
439 fn new(bus: &Bus) -> Self {
440 skip_assert_initialized!();
441
442 let mutex = Arc::new(Mutex::new(()));
443 let (sender, receiver) = mpsc::unbounded();
444
445 let _mutex_guard = mutex.lock().unwrap();
451 bus.set_sync_handler({
452 let sender = sender.clone();
453 let mutex = mutex.clone();
454
455 move |_bus, message| {
456 let _mutex_guard = mutex.lock().unwrap();
457
458 let _ = sender.unbounded_send(message.to_owned());
459
460 BusSyncReply::Drop
461 }
462 });
463
464 while let Some(message) = bus.pop() {
466 let _ = sender.unbounded_send(message);
467 }
468
469 Self {
470 bus: bus.downgrade(),
471 receiver,
472 }
473 }
474}
475
476impl Drop for BusStream {
477 fn drop(&mut self) {
478 if let Some(bus) = self.bus.upgrade() {
479 bus.unset_sync_handler();
480 }
481 }
482}
483
484impl Stream for BusStream {
485 type Item = Message;
486
487 fn poll_next(mut self: Pin<&mut Self>, context: &mut Context) -> Poll<Option<Self::Item>> {
488 self.receiver.poll_next_unpin(context)
489 }
490}
491
492impl FusedStream for BusStream {
493 fn is_terminated(&self) -> bool {
494 self.receiver.is_terminated()
495 }
496}
497
498#[derive(Debug)]
503#[must_use = "if unused the bus watch will immediately be removed"]
504pub struct BusWatchGuard {
505 bus: Bus,
506}
507
508impl Drop for BusWatchGuard {
509 fn drop(&mut self) {
510 let _ = self.bus.remove_watch();
511 }
512}
513
514#[cfg(test)]
515mod tests {
516 use std::sync::{Arc, Mutex};
517
518 use super::*;
519
520 #[test]
521 fn test_sync_handler() {
522 crate::init().unwrap();
523
524 let bus = Bus::new();
525 let msgs = Arc::new(Mutex::new(Vec::new()));
526 let msgs_clone = msgs.clone();
527 bus.set_sync_handler(move |_, msg| {
528 msgs_clone.lock().unwrap().push(msg.clone());
529 BusSyncReply::Pass
530 });
531
532 bus.post(crate::message::Eos::new()).unwrap();
533
534 let msgs = msgs.lock().unwrap();
535 assert_eq!(msgs.len(), 1);
536 match msgs[0].view() {
537 crate::MessageView::Eos(_) => (),
538 _ => unreachable!(),
539 }
540 }
541
542 #[test]
543 fn test_bus_stream() {
544 crate::init().unwrap();
545
546 let bus = Bus::new();
547 let bus_stream = bus.stream();
548
549 let eos_message = crate::message::Eos::new();
550 bus.post(eos_message).unwrap();
551
552 let bus_future = bus_stream.into_future();
553 let (message, _) = futures_executor::block_on(bus_future);
554
555 match message.unwrap().view() {
556 crate::MessageView::Eos(_) => (),
557 _ => unreachable!(),
558 }
559 }
560}