threadshare: Set element flags correctly for sources and sinks

And let proxysink post EOS events

gst-plugin-threadshare/Cargo.toml

@@ -5,6 +5,8 @@ authors = ["Sebastian Dröge <sebastian@centricular.com>"]
 license = "LGPL-2.1+"
 
 [dependencies]
+glib-sys = { git = "https://github.com/gtk-rs/sys" }
+gstreamer-sys = { git = "https://github.com/sdroege/gstreamer-sys" }
 glib = { git = "https://github.com/gtk-rs/glib" }
 gstreamer = { git = "https://github.com/sdroege/gstreamer-rs" }
 gst-plugin = { git = "https://github.com/sdroege/gst-plugin-rs" }

gst-plugin-threadshare/src/appsrc.rs

@@ -213,6 +213,8 @@ impl AppSrc {
         });
         element.add_pad(&src_pad).unwrap();
 
+        ::set_element_flags(element, gst::ElementFlags::SOURCE);
+
         Box::new(Self {
             cat: gst::DebugCategory::new(
                 "ts-appsrc",

gst-plugin-threadshare/src/lib.rs

@@ -17,6 +17,9 @@
 
 #![crate_type = "cdylib"]
 
+extern crate glib_sys as glib_ffi;
+extern crate gstreamer_sys as gst_ffi;
+
 extern crate glib;
 #[macro_use]
 extern crate gst_plugin;
@@ -66,3 +69,38 @@ plugin_define!(
     b"https://github.com/sdroege/gst-plugin-threadshare\0",
     b"2018-03-01\0"
 );
+
+pub fn set_element_flags<T: glib::IsA<gst::Object> + glib::IsA<gst::Element>>(
+    element: &T,
+    flags: gst::ElementFlags,
+) {
+    unsafe {
+        use glib::translate::ToGlib;
+        use gst_ffi;
+
+        let ptr: *mut gst_ffi::GstObject = element.to_glib_none().0;
+        let _guard = MutexGuard::lock(&(*ptr).lock);
+        (*ptr).flags |= flags.to_glib();
+    }
+}
+
+struct MutexGuard<'a>(&'a glib_ffi::GMutex);
+
+impl<'a> MutexGuard<'a> {
+    pub fn lock(mutex: &'a glib_ffi::GMutex) -> Self {
+        use glib::translate::mut_override;
+        unsafe {
+            glib_ffi::g_mutex_lock(mut_override(mutex));
+        }
+        MutexGuard(mutex)
+    }
+}
+
+impl<'a> Drop for MutexGuard<'a> {
+    fn drop(&mut self) {
+        use glib::translate::mut_override;
+        unsafe {
+            glib_ffi::g_mutex_unlock(mut_override(self.0));
+        }
+    }
+}

gst-plugin-threadshare/src/proxy.rs

@@ -343,6 +343,8 @@ impl ProxySink {
 
         element.add_pad(&sink_pad).unwrap();
 
+        ::set_element_flags(element, gst::ElementFlags::SINK);
+
         Box::new(Self {
             cat: gst::DebugCategory::new(
                 "ts-proxysink",
@@ -541,6 +543,9 @@ impl ProxySink {
         gst_log!(self.cat, obj: pad, "Handling event {:?}", event);
 
         match event.view() {
+            EventView::Eos(..) => {
+                let _ = element.post_message(&gst::Message::new_eos().src(Some(element)).build());
+            }
             EventView::FlushStart(..) => {
                 let _ = self.stop(element);
             }
@@ -773,6 +778,8 @@ impl ProxySrc {
         });
         element.add_pad(&src_pad).unwrap();
 
+        ::set_element_flags(element, gst::ElementFlags::SOURCE);
+
         Box::new(Self {
             cat: gst::DebugCategory::new(
                 "ts-proxysrc",

gst-plugin-threadshare/src/udpsrc.rs

@@ -198,6 +198,8 @@ impl UdpSrc {
         });
         element.add_pad(&src_pad).unwrap();
 
+        ::set_element_flags(element, gst::ElementFlags::SOURCE);
+
         Box::new(Self {
             cat: gst::DebugCategory::new(
                 "ts-udpsrc",