bus.rs 6.83 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13
// Copyright (C) 2017 Sebastian Dröge <sebastian@centricular.com>
//
// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
// option. This file may not be copied, modified, or distributed
// except according to those terms.

use std::cell::RefCell;
use std::mem::transmute;
use ffi;
use glib;
use glib::translate::*;
14
use glib::source::{CallbackGuard, Continue, Priority, SourceId};
15 16
use glib_ffi;
use glib_ffi::{gboolean, gpointer};
17
use std::ptr;
18

Sebastian Dröge's avatar
Sebastian Dröge committed
19 20 21 22 23 24
#[cfg(any(all(unix, feature = "v1_14"), feature = "dox"))]
use std::os::unix;

#[cfg(any(all(not(unix), feature = "v1_14"), feature = "dox"))]
use std::os::windows;

25 26 27 28 29 30 31 32 33
use Bus;
use BusSyncReply;
use Message;

unsafe extern "C" fn trampoline_watch(
    bus: *mut ffi::GstBus,
    msg: *mut ffi::GstMessage,
    func: gpointer,
) -> gboolean {
34
    let _guard = CallbackGuard::new();
35
    #[cfg_attr(feature = "cargo-clippy", allow(transmute_ptr_to_ref))]
36
    let func: &RefCell<Box<FnMut(&Bus, &Message) -> Continue + Send + 'static>> = transmute(func);
37
    (&mut *func.borrow_mut())(&from_glib_borrow(bus), &Message::from_glib_borrow(msg)).to_glib()
38 39 40
}

unsafe extern "C" fn destroy_closure_watch(ptr: gpointer) {
41
    let _guard = CallbackGuard::new();
42 43 44
    Box::<RefCell<Box<FnMut(&Bus, &Message) -> Continue + Send + 'static>>>::from_raw(
        ptr as *mut _,
    );
45 46 47
}

fn into_raw_watch<F: FnMut(&Bus, &Message) -> Continue + Send + 'static>(func: F) -> gpointer {
48
    #[cfg_attr(feature = "cargo-clippy", allow(type_complexity))]
49 50 51 52 53 54 55 56 57 58
    let func: Box<RefCell<Box<FnMut(&Bus, &Message) -> Continue + Send + 'static>>> =
        Box::new(RefCell::new(Box::new(func)));
    Box::into_raw(func) as gpointer
}

unsafe extern "C" fn trampoline_sync(
    bus: *mut ffi::GstBus,
    msg: *mut ffi::GstMessage,
    func: gpointer,
) -> ffi::GstBusSyncReply {
59
    let _guard = CallbackGuard::new();
60
    #[cfg_attr(feature = "cargo-clippy", allow(transmute_ptr_to_ref))]
61
    let f: &&(Fn(&Bus, &Message) -> BusSyncReply + Send + Sync + 'static) = transmute(func);
62
    f(&from_glib_borrow(bus), &Message::from_glib_borrow(msg)).to_glib()
63 64 65
}

unsafe extern "C" fn destroy_closure_sync(ptr: gpointer) {
66
    let _guard = CallbackGuard::new();
67
    Box::<Box<Fn(&Bus, &Message) -> BusSyncReply + Send + Sync + 'static>>::from_raw(ptr as *mut _);
68 69
}

70 71 72 73
fn into_raw_sync<F: Fn(&Bus, &Message) -> BusSyncReply + Send + Sync + 'static>(
    func: F,
) -> gpointer {
    let func: Box<Box<Fn(&Bus, &Message) -> BusSyncReply + Send + Sync + 'static>> =
74
        Box::new(Box::new(func));
75 76 77 78
    Box::into_raw(func) as gpointer
}

impl Bus {
79 80 81 82 83 84
    pub fn add_signal_watch_full(&self, priority: Priority) {
        unsafe {
            ffi::gst_bus_add_signal_watch_full(self.to_glib_none().0, priority.to_glib());
        }
    }

85 86 87 88 89 90 91 92 93
    pub fn create_watch<'a, N: Into<Option<&'a str>>, F>(
        &self,
        name: N,
        priority: Priority,
        func: F,
    ) -> Option<glib::Source>
    where
        F: FnMut(&Bus, &Message) -> Continue + Send + 'static,
    {
94
        skip_assert_initialized!();
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
        unsafe {
            let source = ffi::gst_bus_create_watch(self.to_glib_none().0);
            let trampoline = trampoline_watch as gpointer;
            glib_ffi::g_source_set_callback(
                source,
                Some(transmute(trampoline)),
                into_raw_watch(func),
                Some(destroy_closure_watch),
            );
            glib_ffi::g_source_set_priority(source, priority.to_glib());

            let name = name.into();
            if let Some(name) = name {
                glib_ffi::g_source_set_name(source, name.to_glib_none().0);
            }

            from_glib_full(source)
        }
    }

    pub fn add_watch<F>(&self, func: F) -> SourceId
    where
        F: FnMut(&Bus, &Message) -> Continue + Send + 'static,
    {
        unsafe {
            from_glib(ffi::gst_bus_add_watch_full(
                self.to_glib_none().0,
                glib_ffi::G_PRIORITY_DEFAULT,
                Some(trampoline_watch),
                into_raw_watch(func),
                Some(destroy_closure_watch),
            ))
        }
    }

    pub fn set_sync_handler<F>(&self, func: F)
    where
132
        F: Fn(&Bus, &Message) -> BusSyncReply + Send + Sync + 'static,
133 134 135 136 137 138 139 140 141 142
    {
        unsafe {
            ffi::gst_bus_set_sync_handler(
                self.to_glib_none().0,
                Some(trampoline_sync),
                into_raw_sync(func),
                Some(destroy_closure_sync),
            )
        }
    }
143 144

    pub fn unset_sync_handler(&self) {
145
        unsafe { ffi::gst_bus_set_sync_handler(self.to_glib_none().0, None, ptr::null_mut(), None) }
146
    }
Sebastian Dröge's avatar
Sebastian Dröge committed
147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166

    #[cfg(any(all(unix, feature = "v1_14"), feature = "dox"))]
    pub fn get_pollfd(&self) -> unix::io::RawFd {
        unsafe {
            let mut pollfd: glib_ffi::GPollFD = mem::zeroed();
            ffi::gst_bus_get_pollfd(self.to_glib_none().0, &mut pollfd);

            pollfd.fd
        }
    }

    #[cfg(any(all(not(unix), feature = "v1_14"), feature = "dox"))]
    pub fn get_pollfd(&self) -> windows::io::RawHandle {
        unsafe {
            let mut pollfd: glib_ffi::GPollFD = mem::zeroed();
            ffi::gst_bus_get_pollfd(self.to_glib_none().0, &mut pollfd);

            pollfd.fd as *mut _
        }
    }
167
}
168

169 170 171 172 173 174 175 176 177 178 179 180 181 182 183
#[cfg(any(feature = "futures", feature = "dox"))]
mod futures {
    use std::sync::{Arc, Mutex};
    use futures;
    use futures::{Async, Poll};
    use futures::task::Task;
    use futures::stream::Stream;
    use super::*;

    pub struct BusStream(Bus, Arc<Mutex<Option<Task>>>);

    impl BusStream {
        pub fn new(bus: &Bus) -> Self {
            skip_assert_initialized!();
            let task = Arc::new(Mutex::new(None));
184
            let task_clone = Arc::clone(&task);
185 186 187 188 189 190 191 192 193 194 195 196 197 198

            bus.set_sync_handler(move |_, _| {
                let mut task = task_clone.lock().unwrap();
                if let Some(task) = task.take() {
                    // FIXME: Force type...
                    let task: Task = task;
                    task.notify();
                }

                BusSyncReply::Pass
            });

            BusStream(bus.clone(), task)
        }
199 200
    }

201 202 203 204
    impl Drop for BusStream {
        fn drop(&mut self) {
            self.0.unset_sync_handler();
        }
205 206
    }

207 208 209
    impl Stream for BusStream {
        type Item = Message;
        type Error = ();
210

211 212
        fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
            let mut task = self.1.lock().unwrap();
213

214 215 216 217 218 219 220
            let msg = self.0.pop();
            if let Some(msg) = msg {
                Ok(Async::Ready(Some(msg)))
            } else {
                *task = Some(futures::task::current());
                Ok(Async::NotReady)
            }
221 222 223
        }
    }
}
224 225 226

#[cfg(any(feature = "futures", feature = "dox"))]
pub use bus::futures::BusStream;