Only allow setting Bus sync handler and AppSrc/Sink callbacks once

Re-setting them is not thread-safe and can cause segfaults or worse.

See https://gitlab.freedesktop.org/gstreamer/gstreamer/issues/506
and https://gitlab.freedesktop.org/gstreamer/gst-plugins-base/issues/729

gstreamer-app/Cargo.toml

@@ -24,6 +24,7 @@ gstreamer-app-sys = { version = "0.8", features = ["v1_8"] }
 glib = { version = "0.9" }
 gstreamer = { version = "0.15", path = "../gstreamer" }
 gstreamer-base = { version = "0.15", path = "../gstreamer-base" }
+lazy_static = "1.0"
 
 [build-dependencies]
 rustdoc-stripper = { version = "0.1", optional = true }

gstreamer-app/src/app_sink.rs

@@ -30,6 +30,11 @@ use {
     },
 };
 
+lazy_static! {
+    static ref SET_ONCE_QUARK: glib::Quark =
+        glib::Quark::from_string("gstreamer-rs-app-sink-callbacks");
+}
+
 #[allow(clippy::type_complexity)]
 pub struct AppSinkCallbacks {
     eos: Option<RefCell<Box<dyn FnMut(&AppSink) + Send + 'static>>>,
@@ -185,8 +190,16 @@ unsafe extern "C" fn destroy_callbacks(ptr: gpointer) {
 impl AppSink {
     pub fn set_callbacks(&self, callbacks: AppSinkCallbacks) {
         unsafe {
+            let sink = self.to_glib_none().0;
+            if !gobject_sys::g_object_get_qdata(sink as *mut _, SET_ONCE_QUARK.to_glib()).is_null()
+            {
+                panic!("AppSink callbacks can only be set once");
+            }
+
+            gobject_sys::g_object_set_qdata(sink as *mut _, SET_ONCE_QUARK.to_glib(), 1 as *mut _);
+
             gst_app_sys::gst_app_sink_set_callbacks(
-                self.to_glib_none().0,
+                sink,
                 mut_override(&callbacks.callbacks),
                 Box::into_raw(Box::new(callbacks)) as *mut _,
                 Some(destroy_callbacks),
@@ -303,13 +316,6 @@ impl AppSinkStream {
     }
 }
 
-#[cfg(any(feature = "v1_10"))]
-impl Drop for AppSinkStream {
-    fn drop(&mut self) {
-        self.app_sink.set_callbacks(AppSinkCallbacks::new().build());
-    }
-}
-
 #[cfg(any(feature = "v1_10"))]
 impl Stream for AppSinkStream {
     type Item = gst::Sample;

gstreamer-app/src/app_src.rs

@@ -19,6 +19,11 @@ use std::sync::{Arc, Mutex};
 use std::task::{Context, Poll, Waker};
 use AppSrc;
 
+lazy_static! {
+    static ref SET_ONCE_QUARK: glib::Quark =
+        glib::Quark::from_string("gstreamer-rs-app-src-callbacks");
+}
+
 #[allow(clippy::type_complexity)]
 pub struct AppSrcCallbacks {
     need_data: Option<RefCell<Box<dyn FnMut(&AppSrc, u32) + Send + 'static>>>,
@@ -200,8 +205,15 @@ impl AppSrc {
 
     pub fn set_callbacks(&self, callbacks: AppSrcCallbacks) {
         unsafe {
+            let src = self.to_glib_none().0;
+            if !gobject_sys::g_object_get_qdata(src as *mut _, SET_ONCE_QUARK.to_glib()).is_null() {
+                panic!("AppSrc callbacks can only be set once");
+            }
+
+            gobject_sys::g_object_set_qdata(src as *mut _, SET_ONCE_QUARK.to_glib(), 1 as *mut _);
+
             gst_app_sys::gst_app_src_set_callbacks(
-                self.to_glib_none().0,
+                src,
                 mut_override(&callbacks.callbacks),
                 Box::into_raw(Box::new(callbacks)) as *mut _,
                 Some(destroy_callbacks),
@@ -271,12 +283,6 @@ impl AppSrcSink {
     }
 }
 
-impl Drop for AppSrcSink {
-    fn drop(&mut self) {
-        self.app_src.set_callbacks(AppSrcCallbacks::new().build());
-    }
-}
-
 impl Sink<gst::Sample> for AppSrcSink {
     type Error = gst::FlowError;
 

gstreamer-app/src/lib.rs

@@ -17,6 +17,9 @@ extern crate gstreamer_app_sys as gst_app_sys;
 extern crate gstreamer_base as gst_base;
 extern crate gstreamer_sys as gst_sys;
 
+#[macro_use]
+extern crate lazy_static;
+
 #[macro_use]
 extern crate glib;
 

gstreamer/src/bus.rs

@@ -18,7 +18,6 @@ use gst_sys;
 use std::cell::RefCell;
 use std::mem::transmute;
 use std::pin::Pin;
-use std::ptr;
 use std::task::{Context, Poll};
 
 use Bus;
@@ -26,6 +25,10 @@ use BusSyncReply;
 use Message;
 use MessageType;
 
+lazy_static! {
+    static ref SET_ONCE_QUARK: glib::Quark = glib::Quark::from_string("gstreamer-rs-sync-handler");
+}
+
 unsafe extern "C" fn trampoline_watch<F: FnMut(&Bus, &Message) -> Continue + 'static>(
     bus: *mut gst_sys::GstBus,
     msg: *mut gst_sys::GstMessage,
@@ -158,8 +161,15 @@ impl Bus {
         F: Fn(&Bus, &Message) -> BusSyncReply + Send + Sync + 'static,
     {
         unsafe {
+            let bus = self.to_glib_none().0;
+            if !gobject_sys::g_object_get_qdata(bus as *mut _, SET_ONCE_QUARK.to_glib()).is_null() {
+                panic!("Bus sync handler can only be set once");
+            }
+
+            gobject_sys::g_object_set_qdata(bus as *mut _, SET_ONCE_QUARK.to_glib(), 1 as *mut _);
+
             gst_sys::gst_bus_set_sync_handler(
-                self.to_glib_none().0,
+                bus,
                 Some(trampoline_sync::<F>),
                 into_raw_sync(func),
                 Some(destroy_closure_sync::<F>),
@@ -167,12 +177,6 @@ impl Bus {
         }
     }
 
-    pub fn unset_sync_handler(&self) {
-        unsafe {
-            gst_sys::gst_bus_set_sync_handler(self.to_glib_none().0, None, ptr::null_mut(), None)
-        }
-    }
-
     pub fn iter(&self) -> Iter {
         self.iter_timed(0.into())
     }
@@ -272,12 +276,6 @@ impl BusStream {
     }
 }
 
-impl Drop for BusStream {
-    fn drop(&mut self) {
-        self.bus.unset_sync_handler();
-    }
-}
-
 impl Stream for BusStream {
     type Item = Message;