generic: Update for subclassing API changes

generic/file/src/filesink/imp.rs

@@ -23,7 +23,7 @@ use std::sync::Mutex;
 
 use url::Url;
 
-use file_location::FileLocation;
+use crate::file_location::FileLocation;
 
 const DEFAULT_LOCATION: Option<FileLocation> = None;
 
@@ -77,7 +77,7 @@ lazy_static! {
 impl FileSink {
     fn set_location(
         &self,
-        element: &gst_base::BaseSink,
+        element: &super::FileSink,
         location: Option<FileLocation>,
     ) -> Result<(), glib::Error> {
         let state = self.state.lock().unwrap();
@@ -119,6 +119,7 @@ impl FileSink {
 
 impl ObjectSubclass for FileSink {
     const NAME: &'static str = "RsFileSink";
+    type Type = super::FileSink;
     type ParentType = gst_base::BaseSink;
     type Instance = gst::subclass::ElementInstanceStruct<Self>;
     type Class = subclass::simple::ClassStruct<Self>;
@@ -136,7 +137,7 @@ impl ObjectSubclass for FileSink {
         type_.add_interface::<gst::URIHandler>();
     }
 
-    fn class_init(klass: &mut subclass::simple::ClassStruct<Self>) {
+    fn class_init(klass: &mut Self::Class) {
         klass.set_metadata(
             "File Sink",
             "Sink/File",
@@ -159,33 +160,26 @@ impl ObjectSubclass for FileSink {
 }
 
 impl ObjectImpl for FileSink {
-    fn set_property(&self, obj: &glib::Object, id: usize, value: &glib::Value) {
+    fn set_property(&self, obj: &Self::Type, id: usize, value: &glib::Value) {
         let prop = &PROPERTIES[id];
         match *prop {
             subclass::Property("location", ..) => {
-                let element = obj.downcast_ref::<gst_base::BaseSink>().unwrap();
-
                 let res = match value.get::<String>() {
                     Ok(Some(location)) => FileLocation::try_from_path_str(location)
-                        .and_then(|file_location| self.set_location(&element, Some(file_location))),
+                        .and_then(|file_location| self.set_location(obj, Some(file_location))),
-                    Ok(None) => self.set_location(&element, None),
+                    Ok(None) => self.set_location(obj, None),
                     Err(_) => unreachable!("type checked upstream"),
                 };
 
                 if let Err(err) = res {
-                    gst_error!(
+                    gst_error!(CAT, obj: obj, "Failed to set property `location`: {}", err);
-                        CAT,
-                        obj: element,
-                        "Failed to set property `location`: {}",
-                        err
-                    );
                 }
             }
             _ => unimplemented!(),
         };
     }
 
-    fn get_property(&self, _obj: &glib::Object, id: usize) -> Result<glib::Value, ()> {
+    fn get_property(&self, _obj: &Self::Type, id: usize) -> Result<glib::Value, ()> {
         let prop = &PROPERTIES[id];
         match *prop {
             subclass::Property("location", ..) => {
@@ -205,7 +199,7 @@ impl ObjectImpl for FileSink {
 impl ElementImpl for FileSink {}
 
 impl BaseSinkImpl for FileSink {
-    fn start(&self, element: &gst_base::BaseSink) -> Result<(), gst::ErrorMessage> {
+    fn start(&self, element: &Self::Type) -> Result<(), gst::ErrorMessage> {
         let mut state = self.state.lock().unwrap();
         if let State::Started { .. } = *state {
             unreachable!("FileSink already started");
@@ -237,7 +231,7 @@ impl BaseSinkImpl for FileSink {
         Ok(())
     }
 
-    fn stop(&self, element: &gst_base::BaseSink) -> Result<(), gst::ErrorMessage> {
+    fn stop(&self, element: &Self::Type) -> Result<(), gst::ErrorMessage> {
         let mut state = self.state.lock().unwrap();
         if let State::Stopped = *state {
             return Err(gst_error_msg!(
@@ -256,7 +250,7 @@ impl BaseSinkImpl for FileSink {
 
     fn render(
         &self,
-        element: &gst_base::BaseSink,
+        element: &Self::Type,
         buffer: &gst::Buffer,
     ) -> Result<gst::FlowSuccess, gst::FlowError> {
         let mut state = self.state.lock().unwrap();
@@ -293,7 +287,7 @@ impl BaseSinkImpl for FileSink {
 }
 
 impl URIHandlerImpl for FileSink {
-    fn get_uri(&self, _element: &gst::URIHandler) -> Option<String> {
+    fn get_uri(&self, _element: &Self::Type) -> Option<String> {
         let settings = self.settings.lock().unwrap();
 
         // Conversion to Url already checked while building the `FileLocation`
@@ -304,9 +298,7 @@ impl URIHandlerImpl for FileSink {
         })
     }
 
-    fn set_uri(&self, element: &gst::URIHandler, uri: &str) -> Result<(), glib::Error> {
+    fn set_uri(&self, element: &Self::Type, uri: &str) -> Result<(), glib::Error> {
-        let element = element.dynamic_cast_ref::<gst_base::BaseSink>().unwrap();
-
         // Special case for "file://" as this is used by some applications to test
         // with `gst_element_make_from_uri` if there's an element that supports the URI protocol
 
@@ -326,12 +318,3 @@ impl URIHandlerImpl for FileSink {
         vec!["file".to_string()]
     }
 }
-
-pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
-    gst::Element::register(
-        Some(plugin),
-        "rsfilesink",
-        gst::Rank::None,
-        FileSink::get_type(),
-    )
-}

generic/file/src/filesink/mod.rs

@@ -0,0 +1,29 @@
+// Copyright (C) 2016-2017 Sebastian Dröge <sebastian@centricular.com>
+//               2016 Luis de Bethencourt <luisbg@osg.samsung.com>
+//               2018 François Laignel <fengalin@free.fr>
+//
+// 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 glib::prelude::*;
+
+mod imp;
+
+glib_wrapper! {
+    pub struct FileSink(ObjectSubclass<imp::FileSink>) @extends gst_base::BaseSink, gst::Element, gst::Object, @implements gst::URIHandler;
+}
+
+unsafe impl Send for FileSink {}
+unsafe impl Sync for FileSink {}
+
+pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
+    gst::Element::register(
+        Some(plugin),
+        "rsfilesink",
+        gst::Rank::None,
+        FileSink::static_type(),
+    )
+}

generic/file/src/filesrc/imp.rs

@@ -23,7 +23,7 @@ use std::sync::Mutex;
 
 use url::Url;
 
-use file_location::FileLocation;
+use crate::file_location::FileLocation;
 
 const DEFAULT_LOCATION: Option<FileLocation> = None;
 
@@ -77,7 +77,7 @@ lazy_static! {
 impl FileSrc {
     fn set_location(
         &self,
-        element: &gst_base::BaseSrc,
+        element: &super::FileSrc,
         location: Option<FileLocation>,
     ) -> Result<(), glib::Error> {
         let state = self.state.lock().unwrap();
@@ -133,6 +133,7 @@ impl FileSrc {
 
 impl ObjectSubclass for FileSrc {
     const NAME: &'static str = "RsFileSrc";
+    type Type = super::FileSrc;
     type ParentType = gst_base::BaseSrc;
     type Instance = gst::subclass::ElementInstanceStruct<Self>;
     type Class = subclass::simple::ClassStruct<Self>;
@@ -150,7 +151,7 @@ impl ObjectSubclass for FileSrc {
         type_.add_interface::<gst::URIHandler>();
     }
 
-    fn class_init(klass: &mut subclass::simple::ClassStruct<Self>) {
+    fn class_init(klass: &mut Self::Class) {
         klass.set_metadata(
             "File Source",
             "Source/File",
@@ -173,33 +174,26 @@ impl ObjectSubclass for FileSrc {
 }
 
 impl ObjectImpl for FileSrc {
-    fn set_property(&self, obj: &glib::Object, id: usize, value: &glib::Value) {
+    fn set_property(&self, obj: &Self::Type, id: usize, value: &glib::Value) {
         let prop = &PROPERTIES[id];
         match *prop {
             subclass::Property("location", ..) => {
-                let element = obj.downcast_ref::<gst_base::BaseSrc>().unwrap();
-
                 let res = match value.get::<String>() {
                     Ok(Some(location)) => FileLocation::try_from_path_str(location)
-                        .and_then(|file_location| self.set_location(&element, Some(file_location))),
+                        .and_then(|file_location| self.set_location(obj, Some(file_location))),
-                    Ok(None) => self.set_location(&element, None),
+                    Ok(None) => self.set_location(obj, None),
                     Err(_) => unreachable!("type checked upstream"),
                 };
 
                 if let Err(err) = res {
-                    gst_error!(
+                    gst_error!(CAT, obj: obj, "Failed to set property `location`: {}", err);
-                        CAT,
-                        obj: element,
-                        "Failed to set property `location`: {}",
-                        err
-                    );
                 }
             }
             _ => unimplemented!(),
         };
     }
 
-    fn get_property(&self, _obj: &glib::Object, id: usize) -> Result<glib::Value, ()> {
+    fn get_property(&self, _obj: &Self::Type, id: usize) -> Result<glib::Value, ()> {
         let prop = &PROPERTIES[id];
         match *prop {
             subclass::Property("location", ..) => {
@@ -215,22 +209,21 @@ impl ObjectImpl for FileSrc {
         }
     }
 
-    fn constructed(&self, obj: &glib::Object) {
+    fn constructed(&self, obj: &Self::Type) {
         self.parent_constructed(obj);
 
-        let element = obj.downcast_ref::<gst_base::BaseSrc>().unwrap();
+        obj.set_format(gst::Format::Bytes);
-        element.set_format(gst::Format::Bytes);
     }
 }
 
 impl ElementImpl for FileSrc {}
 
 impl BaseSrcImpl for FileSrc {
-    fn is_seekable(&self, _src: &gst_base::BaseSrc) -> bool {
+    fn is_seekable(&self, _src: &Self::Type) -> bool {
         true
     }
 
-    fn get_size(&self, _src: &gst_base::BaseSrc) -> Option<u64> {
+    fn get_size(&self, _src: &Self::Type) -> Option<u64> {
         let state = self.state.lock().unwrap();
         if let State::Started { ref file, .. } = *state {
             file.metadata().ok().map(|m| m.len())
@@ -239,7 +232,7 @@ impl BaseSrcImpl for FileSrc {
         }
     }
 
-    fn start(&self, element: &gst_base::BaseSrc) -> Result<(), gst::ErrorMessage> {
+    fn start(&self, element: &Self::Type) -> Result<(), gst::ErrorMessage> {
         let mut state = self.state.lock().unwrap();
         if let State::Started { .. } = *state {
             unreachable!("FileSrc already started");
@@ -273,7 +266,7 @@ impl BaseSrcImpl for FileSrc {
         Ok(())
     }
 
-    fn stop(&self, element: &gst_base::BaseSrc) -> Result<(), gst::ErrorMessage> {
+    fn stop(&self, element: &Self::Type) -> Result<(), gst::ErrorMessage> {
         let mut state = self.state.lock().unwrap();
         if let State::Stopped = *state {
             return Err(gst_error_msg!(
@@ -291,7 +284,7 @@ impl BaseSrcImpl for FileSrc {
 
     fn fill(
         &self,
-        element: &gst_base::BaseSrc,
+        element: &Self::Type,
         offset: u64,
         _length: u32,
         buffer: &mut gst::BufferRef,
@@ -347,7 +340,7 @@ impl BaseSrcImpl for FileSrc {
 }
 
 impl URIHandlerImpl for FileSrc {
-    fn get_uri(&self, _element: &gst::URIHandler) -> Option<String> {
+    fn get_uri(&self, _element: &Self::Type) -> Option<String> {
         let settings = self.settings.lock().unwrap();
 
         // Conversion to Url already checked while building the `FileLocation`
@@ -358,9 +351,7 @@ impl URIHandlerImpl for FileSrc {
         })
     }
 
-    fn set_uri(&self, element: &gst::URIHandler, uri: &str) -> Result<(), glib::Error> {
+    fn set_uri(&self, element: &Self::Type, uri: &str) -> Result<(), glib::Error> {
-        let element = element.dynamic_cast_ref::<gst_base::BaseSrc>().unwrap();
-
         // Special case for "file://" as this is used by some applications to test
         // with `gst_element_make_from_uri` if there's an element that supports the URI protocol
 
@@ -380,12 +371,3 @@ impl URIHandlerImpl for FileSrc {
         vec!["file".to_string()]
     }
 }
-
-pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
-    gst::Element::register(
-        Some(plugin),
-        "rsfilesrc",
-        gst::Rank::None,
-        FileSrc::get_type(),
-    )
-}

generic/file/src/filesrc/mod.rs

@@ -0,0 +1,28 @@
+// Copyright (C) 2016-2017 Sebastian Dröge <sebastian@centricular.com>
+//               2018 François Laignel <fengalin@free.fr>
+//
+// 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 glib::prelude::*;
+
+mod imp;
+
+glib_wrapper! {
+    pub struct FileSrc(ObjectSubclass<imp::FileSrc>) @extends gst_base::BaseSrc, gst::Element, gst::Object, @implements gst::URIHandler;
+}
+
+unsafe impl Send for FileSrc {}
+unsafe impl Sync for FileSrc {}
+
+pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
+    gst::Element::register(
+        Some(plugin),
+        "rsfilesrc",
+        gst::Rank::None,
+        FileSrc::static_type(),
+    )
+}

generic/sodium/src/decrypter/imp.rs

@@ -123,7 +123,7 @@ impl State {
     // retrieval
     fn decrypt_into_adapter(
         &mut self,
-        element: &gst::Element,
+        element: &super::Decrypter,
         pad: &gst::Pad,
         buffer: &gst::Buffer,
         chunk_index: u64,
@@ -261,7 +261,7 @@ fn add_nonce(initial_nonce: box_::Nonce, chunk_index: u64) -> box_::Nonce {
     box_::Nonce::from_slice(&nonce).expect("Failed to convert slice back to Nonce")
 }
 
-struct Decrypter {
+pub struct Decrypter {
     srcpad: gst::Pad,
     sinkpad: gst::Pad,
     props: Mutex<Props>,
@@ -272,7 +272,7 @@ impl Decrypter {
     fn src_activatemode_function(
         &self,
         _pad: &gst::Pad,
-        element: &gst::Element,
+        element: &super::Decrypter,
         mode: gst::PadMode,
         active: bool,
     ) -> Result<(), gst::LoggableError> {
@@ -295,7 +295,12 @@ impl Decrypter {
         }
     }
 
-    fn src_query(&self, pad: &gst::Pad, element: &gst::Element, query: &mut gst::QueryRef) -> bool {
+    fn src_query(
+        &self,
+        pad: &gst::Pad,
+        element: &super::Decrypter,
+        query: &mut gst::QueryRef,
+    ) -> bool {
         use gst::QueryView;
 
         gst_log!(CAT, obj: pad, "Handling query {:?}", query);
@@ -349,7 +354,7 @@ impl Decrypter {
                 };
 
                 // subtract static offsets
-                let size = size - super::HEADERS_SIZE as u64;
+                let size = size - crate::HEADERS_SIZE as u64;
 
                 // calculate the number of chunks that exist in the stream
                 let total_chunks =
@@ -366,7 +371,7 @@ impl Decrypter {
         }
     }
 
-    fn check_headers(&self, element: &gst::Element) -> Result<(), gst::LoggableError> {
+    fn check_headers(&self, element: &super::Decrypter) -> Result<(), gst::LoggableError> {
         let is_none = {
             let mutex_state = self.state.lock().unwrap();
             let state = mutex_state.as_ref().unwrap();
@@ -440,12 +445,12 @@ impl Decrypter {
     fn pull_requested_buffer(
         &self,
         pad: &gst::Pad,
-        element: &gst::Element,
+        element: &super::Decrypter,
         requested_size: u32,
         block_size: u32,
         chunk_index: u64,
     ) -> Result<gst::Buffer, gst::FlowError> {
-        let pull_offset = super::HEADERS_SIZE as u64
+        let pull_offset = crate::HEADERS_SIZE as u64
             + (chunk_index * block_size as u64)
             + (chunk_index * box_::MACBYTES as u64);
 
@@ -508,7 +513,7 @@ impl Decrypter {
     fn get_range(
         &self,
         pad: &gst::Pad,
-        element: &gst::Element,
+        element: &super::Decrypter,
         offset: u64,
         buffer: Option<&mut gst::BufferRef>,
         requested_size: u32,
@@ -555,13 +560,14 @@ impl Decrypter {
 
 impl ObjectSubclass for Decrypter {
     const NAME: &'static str = "RsSodiumDecryptor";
+    type Type = super::Decrypter;
     type ParentType = gst::Element;
     type Instance = gst::subclass::ElementInstanceStruct<Self>;
     type Class = subclass::simple::ClassStruct<Self>;
 
     glib_object_subclass!();
 
-    fn with_class(klass: &subclass::simple::ClassStruct<Self>) -> Self {
+    fn with_class(klass: &Self::Class) -> Self {
         let templ = klass.get_pad_template("sink").unwrap();
         let sinkpad = gst::Pad::from_template(&templ, Some("sink"));
 
@@ -608,7 +614,7 @@ impl ObjectSubclass for Decrypter {
         }
     }
 
-    fn class_init(klass: &mut subclass::simple::ClassStruct<Self>) {
+    fn class_init(klass: &mut Self::Class) {
         klass.set_metadata(
             "Decrypter",
             "Generic",
@@ -639,15 +645,14 @@ impl ObjectSubclass for Decrypter {
 }
 
 impl ObjectImpl for Decrypter {
-    fn constructed(&self, obj: &glib::Object) {
+    fn constructed(&self, obj: &Self::Type) {
         self.parent_constructed(obj);
 
-        let element = obj.downcast_ref::<gst::Element>().unwrap();
+        obj.add_pad(&self.sinkpad).unwrap();
-        element.add_pad(&self.sinkpad).unwrap();
+        obj.add_pad(&self.srcpad).unwrap();
-        element.add_pad(&self.srcpad).unwrap();
     }
 
-    fn set_property(&self, _obj: &glib::Object, id: usize, value: &glib::Value) {
+    fn set_property(&self, _obj: &Self::Type, id: usize, value: &glib::Value) {
         let prop = &PROPERTIES[id];
 
         match *prop {
@@ -665,7 +670,7 @@ impl ObjectImpl for Decrypter {
         }
     }
 
-    fn get_property(&self, _obj: &glib::Object, id: usize) -> Result<glib::Value, ()> {
+    fn get_property(&self, _obj: &Self::Type, id: usize) -> Result<glib::Value, ()> {
         let prop = &PROPERTIES[id];
 
         match *prop {
@@ -682,7 +687,7 @@ impl ObjectImpl for Decrypter {
 impl ElementImpl for Decrypter {
     fn change_state(
         &self,
-        element: &gst::Element,
+        element: &Self::Type,
         transition: gst::StateChange,
     ) -> Result<gst::StateChangeSuccess, gst::StateChangeError> {
         gst_debug!(CAT, obj: element, "Changing state {:?}", transition);
@@ -716,12 +721,3 @@ impl ElementImpl for Decrypter {
         Ok(success)
     }
 }
-
-pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
-    gst::Element::register(
-        Some(plugin),
-        "sodiumdecrypter",
-        gst::Rank::None,
-        Decrypter::get_type(),
-    )
-}

generic/sodium/src/decrypter/mod.rs

@@ -0,0 +1,43 @@
+// decrypter.rs
+//
+// Copyright 2019 Jordan Petridis <jordan@centricular.com>
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to
+// deal in the Software without restriction, including without limitation the
+// rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
+// sell copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+// IN THE SOFTWARE.
+//
+// SPDX-License-Identifier: MIT
+
+use glib::prelude::*;
+
+mod imp;
+
+glib_wrapper! {
+    pub struct Decrypter(ObjectSubclass<imp::Decrypter>) @extends gst::Element, gst::Object;
+}
+
+unsafe impl Send for Decrypter {}
+unsafe impl Sync for Decrypter {}
+
+pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
+    gst::Element::register(
+        Some(plugin),
+        "sodiumdecrypter",
+        gst::Rank::None,
+        Decrypter::static_type(),
+    )
+}

generic/sodium/src/encrypter/imp.rs

@@ -190,7 +190,7 @@ impl State {
     }
 }
 
-struct Encrypter {
+pub struct Encrypter {
     srcpad: gst::Pad,
     sinkpad: gst::Pad,
     props: Mutex<Props>,
@@ -201,7 +201,7 @@ impl Encrypter {
     fn sink_chain(
         &self,
         pad: &gst::Pad,
-        element: &gst::Element,
+        element: &super::Encrypter,
         buffer: gst::Buffer,
     ) -> Result<gst::FlowSuccess, gst::FlowError> {
         gst_log!(CAT, obj: pad, "Handling buffer {:?}", buffer);
@@ -251,7 +251,7 @@ impl Encrypter {
         Ok(gst::FlowSuccess::Ok)
     }
 
-    fn sink_event(&self, pad: &gst::Pad, element: &gst::Element, event: gst::Event) -> bool {
+    fn sink_event(&self, pad: &gst::Pad, element: &super::Encrypter, event: gst::Event) -> bool {
         use gst::EventView;
 
         gst_log!(CAT, obj: pad, "Handling event {:?}", event);
@@ -306,7 +306,7 @@ impl Encrypter {
         }
     }
 
-    fn src_event(&self, pad: &gst::Pad, element: &gst::Element, event: gst::Event) -> bool {
+    fn src_event(&self, pad: &gst::Pad, element: &super::Encrypter, event: gst::Event) -> bool {
         use gst::EventView;
 
         gst_log!(CAT, obj: pad, "Handling event {:?}", event);
@@ -317,7 +317,12 @@ impl Encrypter {
         }
     }
 
-    fn src_query(&self, pad: &gst::Pad, element: &gst::Element, query: &mut gst::QueryRef) -> bool {
+    fn src_query(
+        &self,
+        pad: &gst::Pad,
+        element: &super::Encrypter,
+        query: &mut gst::QueryRef,
+    ) -> bool {
         use gst::QueryView;
 
         gst_log!(CAT, obj: pad, "Handling query {:?}", query);
@@ -371,7 +376,7 @@ impl Encrypter {
                 let size = size + total_chunks * box_::MACBYTES as u64;
 
                 // add static offsets
-                let size = size + super::HEADERS_SIZE as u64;
+                let size = size + crate::HEADERS_SIZE as u64;
 
                 gst_debug!(CAT, obj: pad, "Setting duration bytes: {}", size);
                 q.set(gst::format::Bytes::from(size));
@@ -385,13 +390,14 @@ impl Encrypter {
 
 impl ObjectSubclass for Encrypter {
     const NAME: &'static str = "RsSodiumEncrypter";
+    type Type = super::Encrypter;
     type ParentType = gst::Element;
     type Instance = gst::subclass::ElementInstanceStruct<Self>;
     type Class = subclass::simple::ClassStruct<Self>;
 
     glib_object_subclass!();
 
-    fn with_class(klass: &subclass::simple::ClassStruct<Self>) -> Self {
+    fn with_class(klass: &Self::Class) -> Self {
         let templ = klass.get_pad_template("sink").unwrap();
         let sinkpad = gst::Pad::builder_with_template(&templ, Some("sink"))
             .chain_function(|pad, parent, buffer| {
@@ -439,7 +445,7 @@ impl ObjectSubclass for Encrypter {
         }
     }
 
-    fn class_init(klass: &mut subclass::simple::ClassStruct<Self>) {
+    fn class_init(klass: &mut Self::Class) {
         klass.set_metadata(
             "Encrypter",
             "Generic",
@@ -470,15 +476,14 @@ impl ObjectSubclass for Encrypter {
 }
 
 impl ObjectImpl for Encrypter {
-    fn constructed(&self, obj: &glib::Object) {
+    fn constructed(&self, obj: &Self::Type) {
         self.parent_constructed(obj);
 
-        let element = obj.downcast_ref::<gst::Element>().unwrap();
+        obj.add_pad(&self.sinkpad).unwrap();
-        element.add_pad(&self.sinkpad).unwrap();
+        obj.add_pad(&self.srcpad).unwrap();
-        element.add_pad(&self.srcpad).unwrap();
     }
 
-    fn set_property(&self, _obj: &glib::Object, id: usize, value: &glib::Value) {
+    fn set_property(&self, _obj: &Self::Type, id: usize, value: &glib::Value) {
         let prop = &PROPERTIES[id];
 
         match *prop {
@@ -501,7 +506,7 @@ impl ObjectImpl for Encrypter {
         }
     }
 
-    fn get_property(&self, _obj: &glib::Object, id: usize) -> Result<glib::Value, ()> {
+    fn get_property(&self, _obj: &Self::Type, id: usize) -> Result<glib::Value, ()> {
         let prop = &PROPERTIES[id];
 
         match *prop {
@@ -523,7 +528,7 @@ impl ObjectImpl for Encrypter {
 impl ElementImpl for Encrypter {
     fn change_state(
         &self,
-        element: &gst::Element,
+        element: &Self::Type,
         transition: gst::StateChange,
     ) -> Result<gst::StateChangeSuccess, gst::StateChangeError> {
         gst_debug!(CAT, obj: element, "Changing state {:?}", transition);
@@ -557,12 +562,3 @@ impl ElementImpl for Encrypter {
         Ok(success)
     }
 }
-
-pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
-    gst::Element::register(
-        Some(plugin),
-        "sodiumencrypter",
-        gst::Rank::None,
-        Encrypter::get_type(),
-    )
-}

generic/sodium/src/encrypter/mod.rs

@@ -0,0 +1,43 @@
+// encrypter.rs
+//
+// Copyright 2019 Jordan Petridis <jordan@centricular.com>
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to
+// deal in the Software without restriction, including without limitation the
+// rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
+// sell copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+// IN THE SOFTWARE.
+//
+// SPDX-License-Identifier: MIT
+
+use glib::prelude::*;
+
+mod imp;
+
+glib_wrapper! {
+    pub struct Encrypter(ObjectSubclass<imp::Encrypter>) @extends gst::Element, gst::Object;
+}
+
+unsafe impl Send for Encrypter {}
+unsafe impl Sync for Encrypter {}
+
+pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
+    gst::Element::register(
+        Some(plugin),
+        "sodiumencrypter",
+        gst::Rank::None,
+        Encrypter::static_type(),
+    )
+}

generic/threadshare/src/appsrc/imp.rs

@@ -176,7 +176,7 @@ impl AppSrcPadHandler {
         self.0.state.lock().await.need_segment = true;
     }
 
-    async fn push_prelude(&self, pad: &PadSrcRef<'_>, _element: &gst::Element) {
+    async fn push_prelude(&self, pad: &PadSrcRef<'_>, _element: &super::AppSrc) {
         let mut state = self.0.state.lock().await;
         if state.need_initial_events {
             gst_debug!(CAT, obj: pad.gst_pad(), "Pushing initial events");
@@ -207,7 +207,7 @@ impl AppSrcPadHandler {
     async fn push_item(
         &self,
         pad: &PadSrcRef<'_>,
-        element: &gst::Element,
+        element: &super::AppSrc,
         item: StreamItem,
     ) -> Result<gst::FlowSuccess, gst::FlowError> {
         gst_log!(CAT, obj: pad.gst_pad(), "Handling {:?}", item);
@@ -316,7 +316,7 @@ impl PadSrcHandler for AppSrcPadHandler {
 
 #[derive(Debug)]
 struct AppSrcTask {
-    element: gst::Element,
+    element: super::AppSrc,
     src_pad: PadSrcWeak,
     src_pad_handler: AppSrcPadHandler,
     receiver: mpsc::Receiver<StreamItem>,
@@ -324,7 +324,7 @@ struct AppSrcTask {
 
 impl AppSrcTask {
     fn new(
-        element: &gst::Element,
+        element: &super::AppSrc,
         src_pad: &PadSrc,
         src_pad_handler: &AppSrcPadHandler,
         receiver: mpsc::Receiver<StreamItem>,
@@ -422,7 +422,7 @@ impl TaskImpl for AppSrcTask {
 }
 
 #[derive(Debug)]
-struct AppSrc {
+pub struct AppSrc {
     src_pad: PadSrc,
     src_pad_handler: AppSrcPadHandler,
     task: Task,
@@ -431,7 +431,7 @@ struct AppSrc {
 }
 
 impl AppSrc {
-    fn push_buffer(&self, element: &gst::Element, mut buffer: gst::Buffer) -> bool {
+    fn push_buffer(&self, element: &super::AppSrc, mut buffer: gst::Buffer) -> bool {
         let state = self.task.lock_state();
         if *state != TaskState::Started && *state != TaskState::Paused {
             gst_debug!(CAT, obj: element, "Rejecting buffer due to element state");
@@ -469,7 +469,7 @@ impl AppSrc {
         }
     }
 
-    fn end_of_stream(&self, element: &gst::Element) -> bool {
+    fn end_of_stream(&self, element: &super::AppSrc) -> bool {
         let mut sender = self.sender.lock().unwrap();
         let sender = match sender.as_mut() {
             Some(sender) => sender,
@@ -485,7 +485,7 @@ impl AppSrc {
         }
     }
 
-    fn prepare(&self, element: &gst::Element) -> Result<(), gst::ErrorMessage> {
+    fn prepare(&self, element: &super::AppSrc) -> Result<(), gst::ErrorMessage> {
         let settings = self.settings.lock().unwrap();
         gst_debug!(CAT, obj: element, "Preparing");
 
@@ -526,7 +526,7 @@ impl AppSrc {
         Ok(())
     }
 
-    fn unprepare(&self, element: &gst::Element) {
+    fn unprepare(&self, element: &super::AppSrc) {
         gst_debug!(CAT, obj: element, "Unpreparing");
 
         *self.sender.lock().unwrap() = None;
@@ -535,21 +535,21 @@ impl AppSrc {
         gst_debug!(CAT, obj: element, "Unprepared");
     }
 
-    fn stop(&self, element: &gst::Element) -> Result<(), gst::ErrorMessage> {
+    fn stop(&self, element: &super::AppSrc) -> Result<(), gst::ErrorMessage> {
         gst_debug!(CAT, obj: element, "Stopping");
         self.task.stop()?;
         gst_debug!(CAT, obj: element, "Stopped");
         Ok(())
     }
 
-    fn start(&self, element: &gst::Element) -> Result<(), gst::ErrorMessage> {
+    fn start(&self, element: &super::AppSrc) -> Result<(), gst::ErrorMessage> {
         gst_debug!(CAT, obj: element, "Starting");
         self.task.start()?;
         gst_debug!(CAT, obj: element, "Started");
         Ok(())
     }
 
-    fn pause(&self, element: &gst::Element) -> Result<(), gst::ErrorMessage> {
+    fn pause(&self, element: &super::AppSrc) -> Result<(), gst::ErrorMessage> {
         gst_debug!(CAT, obj: element, "Pausing");
         self.task.pause()?;
         gst_debug!(CAT, obj: element, "Paused");
@@ -559,13 +559,14 @@ impl AppSrc {
 
 impl ObjectSubclass for AppSrc {
     const NAME: &'static str = "RsTsAppSrc";
+    type Type = super::AppSrc;
     type ParentType = gst::Element;
     type Instance = gst::subclass::ElementInstanceStruct<Self>;
     type Class = subclass::simple::ClassStruct<Self>;
 
     glib_object_subclass!();
 
-    fn class_init(klass: &mut subclass::simple::ClassStruct<Self>) {
+    fn class_init(klass: &mut Self::Class) {
         klass.set_metadata(
             "Thread-sharing app source",
             "Source/Generic",
@@ -593,7 +594,7 @@ impl ObjectSubclass for AppSrc {
             bool::static_type(),
             |_, args| {
                 let element = args[0]
-                    .get::<gst::Element>()
+                    .get::<super::AppSrc>()
                     .expect("signal arg")
                     .expect("missing signal arg");
                 let buffer = args[1]
@@ -613,7 +614,7 @@ impl ObjectSubclass for AppSrc {
             bool::static_type(),
             |_, args| {
                 let element = args[0]
-                    .get::<gst::Element>()
+                    .get::<super::AppSrc>()
                     .expect("signal arg")
                     .expect("missing signal arg");
                 let appsrc = Self::from_instance(&element);
@@ -622,7 +623,7 @@ impl ObjectSubclass for AppSrc {
         );
     }
 
-    fn with_class(klass: &subclass::simple::ClassStruct<Self>) -> Self {
+    fn with_class(klass: &Self::Class) -> Self {
         let src_pad_handler = AppSrcPadHandler::default();
 
         Self {
@@ -639,7 +640,7 @@ impl ObjectSubclass for AppSrc {
 }
 
 impl ObjectImpl for AppSrc {
-    fn set_property(&self, _obj: &glib::Object, id: usize, value: &glib::Value) {
+    fn set_property(&self, _obj: &Self::Type, id: usize, value: &glib::Value) {
         let prop = &PROPERTIES[id];
 
         let mut settings = self.settings.lock().unwrap();
@@ -666,7 +667,7 @@ impl ObjectImpl for AppSrc {
         }
     }
 
-    fn get_property(&self, _obj: &glib::Object, id: usize) -> Result<glib::Value, ()> {
+    fn get_property(&self, _obj: &Self::Type, id: usize) -> Result<glib::Value, ()> {
         let prop = &PROPERTIES[id];
 
         let settings = self.settings.lock().unwrap();
@@ -680,20 +681,19 @@ impl ObjectImpl for AppSrc {
         }
     }
 
-    fn constructed(&self, obj: &glib::Object) {
+    fn constructed(&self, obj: &Self::Type) {
         self.parent_constructed(obj);
 
-        let element = obj.downcast_ref::<gst::Element>().unwrap();
+        obj.add_pad(self.src_pad.gst_pad()).unwrap();
-        element.add_pad(self.src_pad.gst_pad()).unwrap();
 
-        super::set_element_flags(element, gst::ElementFlags::SOURCE);
+        crate::set_element_flags(obj, gst::ElementFlags::SOURCE);
     }
 }
 
 impl ElementImpl for AppSrc {
     fn change_state(
         &self,
-        element: &gst::Element,
+        element: &Self::Type,
         transition: gst::StateChange,
     ) -> Result<gst::StateChangeSuccess, gst::StateChangeError> {
         gst_trace!(CAT, obj: element, "Changing state {:?}", transition);
@@ -735,12 +735,3 @@ impl ElementImpl for AppSrc {
         Ok(success)
     }
 }
-
-pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
-    gst::Element::register(
-        Some(plugin),
-        "ts-appsrc",
-        gst::Rank::None,
-        AppSrc::get_type(),
-    )
-}

generic/threadshare/src/appsrc/mod.rs

@@ -0,0 +1,40 @@
+// Copyright (C) 2018 Sebastian Dröge <sebastian@centricular.com>
+// Copyright (C) 2019-2020 François Laignel <fengalin@free.fr>
+//
+// This library is free software; you can redistribute it and/or
+// modify it under the terms of the GNU Library General Public
+// License as published by the Free Software Foundation; either
+// version 2 of the License, or (at your option) any later version.
+//
+// This library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+// Library General Public License for more details.
+//
+// You should have received a copy of the GNU Library General Public
+// License along with this library; if not, write to the
+// Free Software Foundation, Inc., 51 Franklin Street, Suite 500,
+// Boston, MA 02110-1335, USA.
+
+use glib::glib_wrapper;
+use glib::prelude::*;
+
+mod imp;
+
+glib_wrapper! {
+    pub struct AppSrc(ObjectSubclass<imp::AppSrc>) @extends gst::Element, gst::Object;
+}
+
+// GStreamer elements need to be thread-safe. For the private implementation this is automatically
+// enforced but for the public wrapper type we need to specify this manually.
+unsafe impl Send for AppSrc {}
+unsafe impl Sync for AppSrc {}
+
+pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
+    gst::Element::register(
+        Some(plugin),
+        "ts-appsrc",
+        gst::Rank::None,
+        AppSrc::static_type(),
+    )
+}

generic/threadshare/src/inputselector/imp.rs

@@ -110,7 +110,7 @@ struct InputSelectorPadSinkHandler(Arc<Mutex<InputSelectorPadSinkHandlerInner>>)
 
 impl InputSelectorPadSinkHandler {
     /* Wait until specified time */
-    async fn sync(&self, element: &gst::Element, running_time: gst::ClockTime) {
+    async fn sync(&self, element: &super::InputSelector, running_time: gst::ClockTime) {
         let now = element.get_current_running_time();
 
         if let Some(delay) = running_time
@@ -124,7 +124,7 @@ impl InputSelectorPadSinkHandler {
     async fn handle_item(
         &self,
         pad: &PadSinkRef<'_>,
-        element: &gst::Element,
+        element: &super::InputSelector,
         mut buffer: gst::Buffer,
     ) -> Result<gst::FlowSuccess, gst::FlowError> {
         let inputselector = InputSelector::from_instance(element);
@@ -199,7 +199,7 @@ impl PadSinkHandler for InputSelectorPadSinkHandler {
         buffer: gst::Buffer,
     ) -> BoxFuture<'static, Result<gst::FlowSuccess, gst::FlowError>> {
         let this = self.clone();
-        let element = element.clone();
+        let element = element.clone().downcast::<super::InputSelector>().unwrap();
         let pad_weak = pad.downgrade();
         async move {
             let pad = pad_weak.upgrade().expect("PadSink no longer exists");
@@ -216,7 +216,7 @@ impl PadSinkHandler for InputSelectorPadSinkHandler {
         list: gst::BufferList,
     ) -> BoxFuture<'static, Result<gst::FlowSuccess, gst::FlowError>> {
         let this = self.clone();
-        let element = element.clone();
+        let element = element.clone().downcast::<super::InputSelector>().unwrap();
         let pad_weak = pad.downgrade();
         async move {
             let pad = pad_weak.upgrade().expect("PadSink no longer exists");
@@ -400,7 +400,7 @@ impl Default for Pads {
 }
 
 #[derive(Debug)]
-struct InputSelector {
+pub struct InputSelector {
     src_pad: PadSrc,
     state: Mutex<State>,
     settings: Mutex<Settings>,
@@ -416,7 +416,7 @@ lazy_static! {
 }
 
 impl InputSelector {
-    fn unprepare(&self, element: &gst::Element) -> Result<(), ()> {
+    fn unprepare(&self, element: &super::InputSelector) -> Result<(), ()> {
         let mut state = self.state.lock().unwrap();
         gst_debug!(CAT, obj: element, "Unpreparing");
         *state = State::default();
@@ -428,13 +428,14 @@ impl InputSelector {
 
 impl ObjectSubclass for InputSelector {
     const NAME: &'static str = "RsTsInputSelector";
+    type Type = super::InputSelector;
     type ParentType = gst::Element;
     type Instance = gst::subclass::ElementInstanceStruct<Self>;
     type Class = subclass::simple::ClassStruct<Self>;
 
     glib_object_subclass!();
 
-    fn class_init(klass: &mut subclass::simple::ClassStruct<Self>) {
+    fn class_init(klass: &mut Self::Class) {
         klass.set_metadata(
             "Thread-sharing input selector",
             "Generic",
@@ -465,7 +466,7 @@ impl ObjectSubclass for InputSelector {
         klass.install_properties(&PROPERTIES);
     }
 
-    fn with_class(klass: &subclass::simple::ClassStruct<Self>) -> Self {
+    fn with_class(klass: &Self::Class) -> Self {
         Self {
             src_pad: PadSrc::new(
                 gst::Pad::from_template(&klass.get_pad_template("src").unwrap(), Some("src")),
@@ -479,7 +480,7 @@ impl ObjectSubclass for InputSelector {
 }
 
 impl ObjectImpl for InputSelector {
-    fn set_property(&self, _obj: &glib::Object, id: usize, value: &glib::Value) {
+    fn set_property(&self, _obj: &Self::Type, id: usize, value: &glib::Value) {
         let prop = &PROPERTIES[id];
 
         match *prop {
@@ -526,7 +527,7 @@ impl ObjectImpl for InputSelector {
         }
     }
 
-    fn get_property(&self, _obj: &glib::Object, id: usize) -> Result<glib::Value, ()> {
+    fn get_property(&self, _obj: &Self::Type, id: usize) -> Result<glib::Value, ()> {
         let prop = &PROPERTIES[id];
 
         match *prop {
@@ -547,20 +548,18 @@ impl ObjectImpl for InputSelector {
         }
     }
 
-    fn constructed(&self, obj: &glib::Object) {
+    fn constructed(&self, obj: &Self::Type) {
         self.parent_constructed(obj);
 
-        let element = obj.downcast_ref::<gst::Element>().unwrap();
+        obj.add_pad(self.src_pad.gst_pad()).unwrap();
-        element.add_pad(self.src_pad.gst_pad()).unwrap();
+        obj.set_element_flags(gst::ElementFlags::PROVIDE_CLOCK | gst::ElementFlags::REQUIRE_CLOCK);
-        element
-            .set_element_flags(gst::ElementFlags::PROVIDE_CLOCK | gst::ElementFlags::REQUIRE_CLOCK);
     }
 }
 
 impl ElementImpl for InputSelector {
     fn change_state(
         &self,
-        element: &gst::Element,
+        element: &Self::Type,
         transition: gst::StateChange,
     ) -> Result<gst::StateChangeSuccess, gst::StateChangeError> {
         gst_trace!(CAT, obj: element, "Changing state {:?}", transition);
@@ -586,7 +585,7 @@ impl ElementImpl for InputSelector {
 
     fn request_new_pad(
         &self,
-        element: &gst::Element,
+        element: &Self::Type,
         templ: &gst::PadTemplate,
         _name: Option<String>,
         _caps: Option<&gst::Caps>,
@@ -615,7 +614,7 @@ impl ElementImpl for InputSelector {
         Some(ret)
     }
 
-    fn release_pad(&self, element: &gst::Element, pad: &gst::Pad) {
+    fn release_pad(&self, element: &Self::Type, pad: &gst::Pad) {
         let mut pads = self.pads.lock().unwrap();
         let sink_pad = pads.sink_pads.remove(pad).unwrap();
         drop(sink_pad);
@@ -625,16 +624,7 @@ impl ElementImpl for InputSelector {
         let _ = element.post_message(gst::message::Latency::builder().src(element).build());
     }
 
-    fn provide_clock(&self, _element: &gst::Element) -> Option<gst::Clock> {
+    fn provide_clock(&self, _element: &Self::Type) -> Option<gst::Clock> {
         Some(gst::SystemClock::obtain())
     }
 }
-
-pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
-    gst::Element::register(
-        Some(plugin),
-        "ts-input-selector",
-        gst::Rank::None,
-        InputSelector::get_type(),
-    )
-}

generic/threadshare/src/inputselector/mod.rs

@@ -0,0 +1,39 @@
+// Copyright (C) 2020 Mathieu Duponchelle <mathieu@centricular.com>
+//
+// This library is free software; you can redistribute it and/or
+// modify it under the terms of the GNU Library General Public
+// License as published by the Free Software Foundation; either
+// version 2 of the License, or (at your option) any later version.
+//
+// This library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+// Library General Public License for more details.
+//
+// You should have received a copy of the GNU Library General Public
+// License along with this library; if not, write to the
+// Free Software Foundation, Inc., 51 Franklin Street, Suite 500,
+// Boston, MA 02110-1335, USA.
+
+use glib::glib_wrapper;
+use glib::prelude::*;
+
+mod imp;
+
+glib_wrapper! {
+    pub struct InputSelector(ObjectSubclass<imp::InputSelector>) @extends gst::Element, gst::Object;
+}
+
+// GStreamer elements need to be thread-safe. For the private implementation this is automatically
+// enforced but for the public wrapper type we need to specify this manually.
+unsafe impl Send for InputSelector {}
+unsafe impl Sync for InputSelector {}
+
+pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
+    gst::Element::register(
+        Some(plugin),
+        "ts-input-selector",
+        gst::Rank::None,
+        InputSelector::static_type(),
+    )
+}

generic/threadshare/src/jitterbuffer/ffi.rs

@@ -0,0 +1,115 @@
+// Copyright (C) 2019 Sebastian Dröge <sebastian@centricular.com>
+//
+// This library is free software; you can redistribute it and/or
+// modify it under the terms of the GNU Library General Public
+// License as published by the Free Software Foundation; either
+// version 2 of the License, or (at your option) any later version.
+//
+// This library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+// Library General Public License for more details.
+//
+// You should have received a copy of the GNU Library General Public
+// License along with this library; if not, write to the
+// Free Software Foundation, Inc., 51 Franklin Street, Suite 500,
+// Boston, MA 02110-1335, USA.
+
+use glib_ffi::{gboolean, gpointer, GList, GType};
+use glib_sys as glib_ffi;
+
+use gst_ffi::GstClockTime;
+use gstreamer_sys as gst_ffi;
+use libc::{c_int, c_uint, c_ulonglong, c_ushort, c_void};
+
+#[repr(C)]
+#[derive(Copy, Clone)]
+pub struct RTPJitterBufferItem {
+    pub data: gpointer,
+    pub next: *mut GList,
+    pub prev: *mut GList,
+    pub r#type: c_uint,
+    pub dts: GstClockTime,
+    pub pts: GstClockTime,
+    pub seqnum: c_uint,
+    pub count: c_uint,
+    pub rtptime: c_uint,
+}
+
+#[repr(C)]
+pub struct RTPJitterBuffer(c_void);
+
+#[repr(C)]
+#[derive(Copy, Clone)]
+pub struct RTPPacketRateCtx {
+    probed: gboolean,
+    clock_rate: c_int,
+    last_seqnum: c_ushort,
+    last_ts: c_ulonglong,
+    avg_packet_rate: c_uint,
+}
+
+pub type RTPJitterBufferMode = c_int;
+pub const RTP_JITTER_BUFFER_MODE_NONE: RTPJitterBufferMode = 0;
+pub const RTP_JITTER_BUFFER_MODE_SLAVE: RTPJitterBufferMode = 1;
+pub const RTP_JITTER_BUFFER_MODE_BUFFER: RTPJitterBufferMode = 2;
+pub const RTP_JITTER_BUFFER_MODE_SYNCED: RTPJitterBufferMode = 4;
+
+extern "C" {
+    pub fn rtp_jitter_buffer_new() -> *mut RTPJitterBuffer;
+    pub fn rtp_jitter_buffer_get_type() -> GType;
+    #[allow(dead_code)]
+    pub fn rtp_jitter_buffer_get_mode(jbuf: *mut RTPJitterBuffer) -> RTPJitterBufferMode;
+    #[allow(dead_code)]
+    pub fn rtp_jitter_buffer_set_mode(jbuf: *mut RTPJitterBuffer, mode: RTPJitterBufferMode);
+    #[allow(dead_code)]
+    pub fn rtp_jitter_buffer_get_delay(jbuf: *mut RTPJitterBuffer) -> GstClockTime;
+    pub fn rtp_jitter_buffer_set_delay(jbuf: *mut RTPJitterBuffer, delay: GstClockTime);
+    pub fn rtp_jitter_buffer_set_clock_rate(jbuf: *mut RTPJitterBuffer, clock_rate: c_uint);
+    #[allow(dead_code)]
+    pub fn rtp_jitter_buffer_get_clock_rate(jbuf: *mut RTPJitterBuffer) -> c_uint;
+    pub fn rtp_jitter_buffer_reset_skew(jbuf: *mut RTPJitterBuffer);
+
+    pub fn rtp_jitter_buffer_flush(jbuf: *mut RTPJitterBuffer, free_func: glib_ffi::GFunc);
+    pub fn rtp_jitter_buffer_find_earliest(
+        jbuf: *mut RTPJitterBuffer,
+        pts: *mut GstClockTime,
+        seqnum: *mut c_uint,
+    );
+    pub fn rtp_jitter_buffer_calculate_pts(
+        jbuf: *mut RTPJitterBuffer,
+        dts: GstClockTime,
+        estimated_dts: gboolean,
+        rtptime: c_uint,
+        base_time: GstClockTime,
+        gap: c_int,
+        is_rtx: gboolean,
+    ) -> GstClockTime;
+    pub fn rtp_jitter_buffer_insert(
+        jbuf: *mut RTPJitterBuffer,
+        item: *mut RTPJitterBufferItem,
+        head: *mut gboolean,
+        percent: *mut c_int,
+    ) -> gboolean;
+    pub fn rtp_jitter_buffer_pop(
+        jbuf: *mut RTPJitterBuffer,
+        percent: *mut c_int,
+    ) -> *mut RTPJitterBufferItem;
+    pub fn rtp_jitter_buffer_peek(jbuf: *mut RTPJitterBuffer) -> *mut RTPJitterBufferItem;
+
+    pub fn gst_rtp_packet_rate_ctx_reset(ctx: *mut RTPPacketRateCtx, clock_rate: c_int);
+    pub fn gst_rtp_packet_rate_ctx_update(
+        ctx: *mut RTPPacketRateCtx,
+        seqnum: c_ushort,
+        ts: c_uint,
+    ) -> c_uint;
+    pub fn gst_rtp_packet_rate_ctx_get_max_dropout(
+        ctx: *mut RTPPacketRateCtx,
+        time_ms: c_int,
+    ) -> c_uint;
+    #[allow(dead_code)]
+    pub fn gst_rtp_packet_rate_ctx_get_max_disorder(
+        ctx: *mut RTPPacketRateCtx,
+        time_ms: c_int,
+    ) -> c_uint;
+}

generic/threadshare/src/jitterbuffer/mod.rs

@@ -15,472 +15,29 @@
 // Free Software Foundation, Inc., 51 Franklin Street, Suite 500,
 // Boston, MA 02110-1335, USA.
 
-use glib_sys as glib_ffi;
+use glib::prelude::*;
-use gstreamer_sys as gst_ffi;
-
-use std::ptr;
-use std::u32;
 
+mod ffi;
+mod imp;
 #[allow(clippy::module_inception)]
 pub mod jitterbuffer;
 
-pub mod ffi {
-    use glib_ffi::{gboolean, gpointer, GList, GType};
-    use glib_sys as glib_ffi;
-
-    use gst_ffi::GstClockTime;
-    use gstreamer_sys as gst_ffi;
-    use libc::{c_int, c_uint, c_ulonglong, c_ushort, c_void};
-
-    #[repr(C)]
-    #[derive(Copy, Clone)]
-    pub struct RTPJitterBufferItem {
-        pub data: gpointer,
-        pub next: *mut GList,
-        pub prev: *mut GList,
-        pub r#type: c_uint,
-        pub dts: GstClockTime,
-        pub pts: GstClockTime,
-        pub seqnum: c_uint,
-        pub count: c_uint,
-        pub rtptime: c_uint,
-    }
-
-    #[repr(C)]
-    pub struct RTPJitterBuffer(c_void);
-
-    #[repr(C)]
-    #[derive(Copy, Clone)]
-    pub struct RTPPacketRateCtx {
-        probed: gboolean,
-        clock_rate: c_int,
-        last_seqnum: c_ushort,
-        last_ts: c_ulonglong,
-        avg_packet_rate: c_uint,
-    }
-
-    pub type RTPJitterBufferMode = c_int;
-    pub const RTP_JITTER_BUFFER_MODE_NONE: RTPJitterBufferMode = 0;
-    pub const RTP_JITTER_BUFFER_MODE_SLAVE: RTPJitterBufferMode = 1;
-    pub const RTP_JITTER_BUFFER_MODE_BUFFER: RTPJitterBufferMode = 2;
-    pub const RTP_JITTER_BUFFER_MODE_SYNCED: RTPJitterBufferMode = 4;
-
-    extern "C" {
-        pub fn rtp_jitter_buffer_new() -> *mut RTPJitterBuffer;
-        pub fn rtp_jitter_buffer_get_type() -> GType;
-        #[allow(dead_code)]
-        pub fn rtp_jitter_buffer_get_mode(jbuf: *mut RTPJitterBuffer) -> RTPJitterBufferMode;
-        #[allow(dead_code)]
-        pub fn rtp_jitter_buffer_set_mode(jbuf: *mut RTPJitterBuffer, mode: RTPJitterBufferMode);
-        #[allow(dead_code)]
-        pub fn rtp_jitter_buffer_get_delay(jbuf: *mut RTPJitterBuffer) -> GstClockTime;
-        pub fn rtp_jitter_buffer_set_delay(jbuf: *mut RTPJitterBuffer, delay: GstClockTime);
-        pub fn rtp_jitter_buffer_set_clock_rate(jbuf: *mut RTPJitterBuffer, clock_rate: c_uint);
-        #[allow(dead_code)]
-        pub fn rtp_jitter_buffer_get_clock_rate(jbuf: *mut RTPJitterBuffer) -> c_uint;
-        pub fn rtp_jitter_buffer_reset_skew(jbuf: *mut RTPJitterBuffer);
-
-        pub fn rtp_jitter_buffer_flush(jbuf: *mut RTPJitterBuffer, free_func: glib_ffi::GFunc);
-        pub fn rtp_jitter_buffer_find_earliest(
-            jbuf: *mut RTPJitterBuffer,
-            pts: *mut GstClockTime,
-            seqnum: *mut c_uint,
-        );
-        pub fn rtp_jitter_buffer_calculate_pts(
-            jbuf: *mut RTPJitterBuffer,
-            dts: GstClockTime,
-            estimated_dts: gboolean,
-            rtptime: c_uint,
-            base_time: GstClockTime,
-            gap: c_int,
-            is_rtx: gboolean,
-        ) -> GstClockTime;
-        pub fn rtp_jitter_buffer_insert(
-            jbuf: *mut RTPJitterBuffer,
-            item: *mut RTPJitterBufferItem,
-            head: *mut gboolean,
-            percent: *mut c_int,
-        ) -> gboolean;
-        pub fn rtp_jitter_buffer_pop(
-            jbuf: *mut RTPJitterBuffer,
-            percent: *mut c_int,
-        ) -> *mut RTPJitterBufferItem;
-        pub fn rtp_jitter_buffer_peek(jbuf: *mut RTPJitterBuffer) -> *mut RTPJitterBufferItem;
-
-        pub fn gst_rtp_packet_rate_ctx_reset(ctx: *mut RTPPacketRateCtx, clock_rate: c_int);
-        pub fn gst_rtp_packet_rate_ctx_update(
-            ctx: *mut RTPPacketRateCtx,
-            seqnum: c_ushort,
-            ts: c_uint,
-        ) -> c_uint;
-        pub fn gst_rtp_packet_rate_ctx_get_max_dropout(
-            ctx: *mut RTPPacketRateCtx,
-            time_ms: c_int,
-        ) -> c_uint;
-        #[allow(dead_code)]
-        pub fn gst_rtp_packet_rate_ctx_get_max_disorder(
-            ctx: *mut RTPPacketRateCtx,
-            time_ms: c_int,
-        ) -> c_uint;
-    }
-}
-
 use glib::glib_wrapper;
-use glib::prelude::*;
-use glib::translate::*;
-
-use std::mem;
 
 glib_wrapper! {
-    pub struct RTPJitterBuffer(Object<ffi::RTPJitterBuffer>);
+    pub struct JitterBuffer(ObjectSubclass<imp::JitterBuffer>) @extends gst::Element, gst::Object;
-
-    match fn {
-        get_type => || ffi::rtp_jitter_buffer_get_type(),
-    }
 }
 
-unsafe impl glib::SendUnique for RTPJitterBuffer {
+// GStreamer elements need to be thread-safe. For the private implementation this is automatically
-    fn is_unique(&self) -> bool {
+// enforced but for the public wrapper type we need to specify this manually.
-        self.ref_count() == 1
+unsafe impl Send for JitterBuffer {}
-    }
+unsafe impl Sync for JitterBuffer {}
-}
+
-
+pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
-impl ToGlib for RTPJitterBufferMode {
+    gst::Element::register(
-    type GlibType = ffi::RTPJitterBufferMode;
+        Some(plugin),
-
+        "ts-jitterbuffer",
-    fn to_glib(&self) -> ffi::RTPJitterBufferMode {
+        gst::Rank::None,
-        match *self {
+        JitterBuffer::static_type(),
-            RTPJitterBufferMode::None => ffi::RTP_JITTER_BUFFER_MODE_NONE,
+    )
-            RTPJitterBufferMode::Slave => ffi::RTP_JITTER_BUFFER_MODE_SLAVE,
-            RTPJitterBufferMode::Buffer => ffi::RTP_JITTER_BUFFER_MODE_BUFFER,
-            RTPJitterBufferMode::Synced => ffi::RTP_JITTER_BUFFER_MODE_SYNCED,
-            RTPJitterBufferMode::__Unknown(value) => value,
-        }
-    }
-}
-
-impl FromGlib<ffi::RTPJitterBufferMode> for RTPJitterBufferMode {
-    fn from_glib(value: ffi::RTPJitterBufferMode) -> Self {
-        match value {
-            0 => RTPJitterBufferMode::None,
-            1 => RTPJitterBufferMode::Slave,
-            2 => RTPJitterBufferMode::Buffer,
-            4 => RTPJitterBufferMode::Synced,
-            value => RTPJitterBufferMode::__Unknown(value),
-        }
-    }
-}
-
-pub struct RTPJitterBufferItem(Option<ptr::NonNull<ffi::RTPJitterBufferItem>>);
-
-unsafe impl Send for RTPJitterBufferItem {}
-
-impl RTPJitterBufferItem {
-    pub fn new(
-        buffer: gst::Buffer,
-        dts: gst::ClockTime,
-        pts: gst::ClockTime,
-        seqnum: Option<u16>,
-        rtptime: u32,
-    ) -> RTPJitterBufferItem {
-        unsafe {
-            let ptr = ptr::NonNull::new(glib_sys::g_slice_alloc0(mem::size_of::<
-                ffi::RTPJitterBufferItem,
-            >()) as *mut ffi::RTPJitterBufferItem)
-            .expect("Allocation failed");
-            ptr::write(
-                ptr.as_ptr(),
-                ffi::RTPJitterBufferItem {
-                    data: buffer.into_ptr() as *mut _,
-                    next: ptr::null_mut(),
-                    prev: ptr::null_mut(),
-                    r#type: 0,
-                    dts: dts.to_glib(),
-                    pts: pts.to_glib(),
-                    seqnum: seqnum.map(|s| s as u32).unwrap_or(u32::MAX),
-                    count: 1,
-                    rtptime,
-                },
-            );
-
-            RTPJitterBufferItem(Some(ptr))
-        }
-    }
-
-    pub fn into_buffer(mut self) -> gst::Buffer {
-        unsafe {
-            let item = self.0.take().expect("Invalid wrapper");
-            let buf = item.as_ref().data as *mut gst_ffi::GstBuffer;
-            glib_sys::g_slice_free1(
-                mem::size_of::<ffi::RTPJitterBufferItem>(),
-                item.as_ptr() as *mut _,
-            );
-            from_glib_full(buf)
-        }
-    }
-
-    pub fn get_dts(&self) -> gst::ClockTime {
-        unsafe {
-            let item = self.0.as_ref().expect("Invalid wrapper");
-            if item.as_ref().dts == gst_ffi::GST_CLOCK_TIME_NONE {
-                gst::CLOCK_TIME_NONE
-            } else {
-                gst::ClockTime(Some(item.as_ref().dts))
-            }
-        }
-    }
-
-    pub fn get_pts(&self) -> gst::ClockTime {
-        unsafe {
-            let item = self.0.as_ref().expect("Invalid wrapper");
-            if item.as_ref().pts == gst_ffi::GST_CLOCK_TIME_NONE {
-                gst::CLOCK_TIME_NONE
-            } else {
-                gst::ClockTime(Some(item.as_ref().pts))
-            }
-        }
-    }
-
-    pub fn get_seqnum(&self) -> Option<u16> {
-        unsafe {
-            let item = self.0.as_ref().expect("Invalid wrapper");
-            if item.as_ref().seqnum == u32::MAX {
-                None
-            } else {
-                Some(item.as_ref().seqnum as u16)
-            }
-        }
-    }
-
-    #[allow(dead_code)]
-    pub fn get_rtptime(&self) -> u32 {
-        unsafe {
-            let item = self.0.as_ref().expect("Invalid wrapper");
-            item.as_ref().rtptime
-        }
-    }
-}
-
-impl Drop for RTPJitterBufferItem {
-    fn drop(&mut self) {
-        unsafe {
-            if let Some(ref item) = self.0 {
-                if !item.as_ref().data.is_null() {
-                    gst_ffi::gst_mini_object_unref(item.as_ref().data as *mut _);
-                }
-
-                glib_sys::g_slice_free1(
-                    mem::size_of::<ffi::RTPJitterBufferItem>(),
-                    item.as_ptr() as *mut _,
-                );
-            }
-        }
-    }
-}
-
-pub struct RTPPacketRateCtx(Box<ffi::RTPPacketRateCtx>);
-
-unsafe impl Send for RTPPacketRateCtx {}
-
-impl RTPPacketRateCtx {
-    pub fn new() -> RTPPacketRateCtx {
-        unsafe {
-            let mut ptr = std::mem::MaybeUninit::uninit();
-            ffi::gst_rtp_packet_rate_ctx_reset(ptr.as_mut_ptr(), -1);
-            RTPPacketRateCtx(Box::new(ptr.assume_init()))
-        }
-    }
-
-    pub fn reset(&mut self, clock_rate: i32) {
-        unsafe { ffi::gst_rtp_packet_rate_ctx_reset(&mut *self.0, clock_rate) }
-    }
-
-    pub fn update(&mut self, seqnum: u16, ts: u32) -> u32 {
-        unsafe { ffi::gst_rtp_packet_rate_ctx_update(&mut *self.0, seqnum, ts) }
-    }
-
-    pub fn get_max_dropout(&mut self, time_ms: i32) -> u32 {
-        unsafe { ffi::gst_rtp_packet_rate_ctx_get_max_dropout(&mut *self.0, time_ms) }
-    }
-
-    #[allow(dead_code)]
-    pub fn get_max_disorder(&mut self, time_ms: i32) -> u32 {
-        unsafe { ffi::gst_rtp_packet_rate_ctx_get_max_disorder(&mut *self.0, time_ms) }
-    }
-}
-
-impl Default for RTPPacketRateCtx {
-    fn default() -> Self {
-        RTPPacketRateCtx::new()
-    }
-}
-
-#[derive(Debug, Eq, PartialEq, Ord, PartialOrd, Hash, Clone, Copy)]
-pub enum RTPJitterBufferMode {
-    r#None,
-    Slave,
-    Buffer,
-    Synced,
-    __Unknown(i32),
-}
-
-impl RTPJitterBuffer {
-    pub fn new() -> RTPJitterBuffer {
-        unsafe { from_glib_full(ffi::rtp_jitter_buffer_new()) }
-    }
-
-    #[allow(dead_code)]
-    pub fn get_mode(&self) -> RTPJitterBufferMode {
-        unsafe { from_glib(ffi::rtp_jitter_buffer_get_mode(self.to_glib_none().0)) }
-    }
-
-    #[allow(dead_code)]
-    pub fn set_mode(&self, mode: RTPJitterBufferMode) {
-        unsafe { ffi::rtp_jitter_buffer_set_mode(self.to_glib_none().0, mode.to_glib()) }
-    }
-
-    #[allow(dead_code)]
-    pub fn get_delay(&self) -> gst::ClockTime {
-        unsafe { from_glib(ffi::rtp_jitter_buffer_get_delay(self.to_glib_none().0)) }
-    }
-
-    pub fn set_delay(&self, delay: gst::ClockTime) {
-        unsafe { ffi::rtp_jitter_buffer_set_delay(self.to_glib_none().0, delay.to_glib()) }
-    }
-
-    pub fn set_clock_rate(&self, clock_rate: u32) {
-        unsafe { ffi::rtp_jitter_buffer_set_clock_rate(self.to_glib_none().0, clock_rate) }
-    }
-
-    #[allow(dead_code)]
-    pub fn get_clock_rate(&self) -> u32 {
-        unsafe { ffi::rtp_jitter_buffer_get_clock_rate(self.to_glib_none().0) }
-    }
-
-    pub fn calculate_pts(
-        &self,
-        dts: gst::ClockTime,
-        estimated_dts: bool,
-        rtptime: u32,
-        base_time: gst::ClockTime,
-        gap: i32,
-        is_rtx: bool,
-    ) -> gst::ClockTime {
-        unsafe {
-            let pts = ffi::rtp_jitter_buffer_calculate_pts(
-                self.to_glib_none().0,
-                dts.to_glib(),
-                estimated_dts.to_glib(),
-                rtptime,
-                base_time.to_glib(),
-                gap,
-                is_rtx.to_glib(),
-            );
-
-            if pts == gst_ffi::GST_CLOCK_TIME_NONE {
-                gst::CLOCK_TIME_NONE
-            } else {
-                pts.into()
-            }
-        }
-    }
-
-    pub fn insert(&self, mut item: RTPJitterBufferItem) -> (bool, bool, i32) {
-        unsafe {
-            let mut head = mem::MaybeUninit::uninit();
-            let mut percent = mem::MaybeUninit::uninit();
-            let ptr = item.0.take().expect("Invalid wrapper");
-            let ret: bool = from_glib(ffi::rtp_jitter_buffer_insert(
-                self.to_glib_none().0,
-                ptr.as_ptr(),
-                head.as_mut_ptr(),
-                percent.as_mut_ptr(),
-            ));
-            if !ret {
-                item.0 = Some(ptr);
-            }
-            (ret, from_glib(head.assume_init()), percent.assume_init())
-        }
-    }
-
-    pub fn find_earliest(&self) -> (gst::ClockTime, Option<u16>) {
-        unsafe {
-            let mut pts = mem::MaybeUninit::uninit();
-            let mut seqnum = mem::MaybeUninit::uninit();
-
-            ffi::rtp_jitter_buffer_find_earliest(
-                self.to_glib_none().0,
-                pts.as_mut_ptr(),
-                seqnum.as_mut_ptr(),
-            );
-            let pts = pts.assume_init();
-            let seqnum = seqnum.assume_init();
-
-            let seqnum = if seqnum == u32::MAX {
-                None
-            } else {
-                Some(seqnum as u16)
-            };
-
-            if pts == gst_ffi::GST_CLOCK_TIME_NONE {
-                (gst::CLOCK_TIME_NONE, seqnum)
-            } else {
-                (pts.into(), seqnum)
-            }
-        }
-    }
-
-    pub fn pop(&self) -> (Option<RTPJitterBufferItem>, i32) {
-        unsafe {
-            let mut percent = mem::MaybeUninit::uninit();
-            let item = ffi::rtp_jitter_buffer_pop(self.to_glib_none().0, percent.as_mut_ptr());
-
-            (
-                if item.is_null() {
-                    None
-                } else {
-                    Some(RTPJitterBufferItem(Some(ptr::NonNull::new_unchecked(item))))
-                },
-                percent.assume_init(),
-            )
-        }
-    }
-
-    pub fn peek(&self) -> (gst::ClockTime, Option<u16>) {
-        unsafe {
-            let item = ffi::rtp_jitter_buffer_peek(self.to_glib_none().0);
-            if item.is_null() {
-                (gst::CLOCK_TIME_NONE, None)
-            } else {
-                let seqnum = (*item).seqnum;
-                let seqnum = if seqnum == u32::MAX {
-                    None
-                } else {
-                    Some(seqnum as u16)
-                };
-                ((*item).pts.into(), seqnum)
-            }
-        }
-    }
-
-    pub fn flush(&self) {
-        unsafe extern "C" fn free_item(item: glib_ffi::gpointer, _: glib_ffi::gpointer) {
-            let _ =
-                RTPJitterBufferItem(Some(ptr::NonNull::new(item as *mut _).expect("NULL item")));
-        }
-
-        unsafe {
-            ffi::rtp_jitter_buffer_flush(self.to_glib_none().0, Some(free_item));
-        }
-    }
-
-    pub fn reset_skew(&self) {
-        unsafe { ffi::rtp_jitter_buffer_reset_skew(self.to_glib_none().0) }
-    }
-}
-
-impl Default for RTPJitterBuffer {
-    fn default() -> Self {
-        RTPJitterBuffer::new()
-    }
 }

generic/threadshare/src/lib.rs

@@ -53,7 +53,7 @@ fn plugin_init(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
     queue::register(plugin)?;
     proxy::register(plugin)?;
     appsrc::register(plugin)?;
-    jitterbuffer::jitterbuffer::register(plugin)?;
+    jitterbuffer::register(plugin)?;
     inputselector::register(plugin)?;
 
     Ok(())

generic/threadshare/src/proxy/imp.rs

@@ -41,7 +41,7 @@ use crate::runtime::{
     Context, PadSink, PadSinkRef, PadSinkWeak, PadSrc, PadSrcRef, PadSrcWeak, Task,
 };
 
-use super::dataqueue::{DataQueue, DataQueueItem};
+use crate::dataqueue::{DataQueue, DataQueueItem};
 
 lazy_static! {
     static ref PROXY_CONTEXTS: StdMutex<HashMap<String, Weak<StdMutex<ProxyContextInner>>>> =
@@ -301,7 +301,7 @@ impl PadSinkHandler for ProxySinkPadHandler {
         buffer: gst::Buffer,
     ) -> BoxFuture<'static, Result<gst::FlowSuccess, gst::FlowError>> {
         let pad_weak = pad.downgrade();
-        let element = element.clone();
+        let element = element.clone().downcast::<super::ProxySink>().unwrap();
 
         async move {
             let pad = pad_weak.upgrade().expect("PadSink no longer exists");
@@ -322,7 +322,7 @@ impl PadSinkHandler for ProxySinkPadHandler {
         list: gst::BufferList,
     ) -> BoxFuture<'static, Result<gst::FlowSuccess, gst::FlowError>> {
         let pad_weak = pad.downgrade();
-        let element = element.clone();
+        let element = element.clone().downcast::<super::ProxySink>().unwrap();
         async move {
             let pad = pad_weak.upgrade().expect("PadSink no longer exists");
             gst_log!(SINK_CAT, obj: pad.gst_pad(), "Handling {:?}", list);
@@ -357,7 +357,7 @@ impl PadSinkHandler for ProxySinkPadHandler {
         };
 
         if let EventView::FlushStart(..) = event.view() {
-            proxysink.stop(&element);
+            proxysink.stop(element.downcast_ref::<super::ProxySink>().unwrap());
         }
 
         if let Some(src_pad) = src_pad {
@@ -381,7 +381,7 @@ impl PadSinkHandler for ProxySinkPadHandler {
         gst_log!(SINK_CAT, obj: pad.gst_pad(), "Handling serialized {:?}", event);
 
         let pad_weak = pad.downgrade();
-        let element = element.clone();
+        let element = element.clone().downcast::<super::ProxySink>().unwrap();
         async move {
             let pad = pad_weak.upgrade().expect("PadSink no longer exists");
             let proxysink = ProxySink::from_instance(&element);
@@ -406,7 +406,7 @@ impl PadSinkHandler for ProxySinkPadHandler {
 }
 
 #[derive(Debug)]
-struct ProxySink {
+pub struct ProxySink {
     sink_pad: PadSink,
     proxy_ctx: StdMutex<Option<ProxyContext>>,
     settings: StdMutex<SettingsSink>,
@@ -421,7 +421,7 @@ lazy_static! {
 }
 
 impl ProxySink {
-    async fn schedule_pending_queue(&self, element: &gst::Element) {
+    async fn schedule_pending_queue(&self, element: &super::ProxySink) {
         loop {
             let more_queue_space_receiver = {
                 let proxy_ctx = self.proxy_ctx.lock().unwrap();
@@ -476,7 +476,7 @@ impl ProxySink {
 
     async fn enqueue_item(
         &self,
-        element: &gst::Element,
+        element: &super::ProxySink,
         item: DataQueueItem,
     ) -> Result<gst::FlowSuccess, gst::FlowError> {
         let wait_fut = {
@@ -589,7 +589,7 @@ impl ProxySink {
         shared_ctx.last_res
     }
 
-    fn prepare(&self, element: &gst::Element) -> Result<(), gst::ErrorMessage> {
+    fn prepare(&self, element: &super::ProxySink) -> Result<(), gst::ErrorMessage> {
         gst_debug!(SINK_CAT, obj: element, "Preparing");
 
         let proxy_context = self.settings.lock().unwrap().proxy_context.to_string();
@@ -614,13 +614,13 @@ impl ProxySink {
         Ok(())
     }
 
-    fn unprepare(&self, element: &gst::Element) {
+    fn unprepare(&self, element: &super::ProxySink) {
         gst_debug!(SINK_CAT, obj: element, "Unpreparing");
         *self.proxy_ctx.lock().unwrap() = None;
         gst_debug!(SINK_CAT, obj: element, "Unprepared");
     }
 
-    fn start(&self, element: &gst::Element) {
+    fn start(&self, element: &super::ProxySink) {
         let proxy_ctx = self.proxy_ctx.lock().unwrap();
         let mut shared_ctx = proxy_ctx.as_ref().unwrap().lock_shared();
 
@@ -637,7 +637,7 @@ impl ProxySink {
         gst_debug!(SINK_CAT, obj: element, "Started");
     }
 
-    fn stop(&self, element: &gst::Element) {
+    fn stop(&self, element: &super::ProxySink) {
         let proxy_ctx = self.proxy_ctx.lock().unwrap();
         let mut shared_ctx = proxy_ctx.as_ref().unwrap().lock_shared();
 
@@ -652,13 +652,14 @@ impl ProxySink {
 
 impl ObjectSubclass for ProxySink {
     const NAME: &'static str = "RsTsProxySink";
+    type Type = super::ProxySink;
     type ParentType = gst::Element;
     type Instance = gst::subclass::ElementInstanceStruct<Self>;
     type Class = subclass::simple::ClassStruct<Self>;
 
     glib_object_subclass!();
 
-    fn class_init(klass: &mut subclass::simple::ClassStruct<Self>) {
+    fn class_init(klass: &mut Self::Class) {
         klass.set_metadata(
             "Thread-sharing proxy sink",
             "Sink/Generic",
@@ -680,7 +681,7 @@ impl ObjectSubclass for ProxySink {
         klass.install_properties(&PROPERTIES_SINK);
     }
 
-    fn with_class(klass: &subclass::simple::ClassStruct<Self>) -> Self {
+    fn with_class(klass: &Self::Class) -> Self {
         Self {
             sink_pad: PadSink::new(
                 gst::Pad::from_template(&klass.get_pad_template("sink").unwrap(), Some("sink")),
@@ -693,7 +694,7 @@ impl ObjectSubclass for ProxySink {
 }
 
 impl ObjectImpl for ProxySink {
-    fn set_property(&self, _obj: &glib::Object, id: usize, value: &glib::Value) {
+    fn set_property(&self, _obj: &Self::Type, id: usize, value: &glib::Value) {
         let prop = &PROPERTIES_SINK[id];
 
         let mut settings = self.settings.lock().unwrap();
@@ -708,7 +709,7 @@ impl ObjectImpl for ProxySink {
         }
     }
 
-    fn get_property(&self, _obj: &glib::Object, id: usize) -> Result<glib::Value, ()> {
+    fn get_property(&self, _obj: &Self::Type, id: usize) -> Result<glib::Value, ()> {
         let prop = &PROPERTIES_SINK[id];
 
         let settings = self.settings.lock().unwrap();
@@ -718,20 +719,19 @@ impl ObjectImpl for ProxySink {
         }
     }
 
-    fn constructed(&self, obj: &glib::Object) {
+    fn constructed(&self, obj: &Self::Type) {
         self.parent_constructed(obj);
 
-        let element = obj.downcast_ref::<gst::Element>().unwrap();
+        obj.add_pad(self.sink_pad.gst_pad()).unwrap();
-        element.add_pad(self.sink_pad.gst_pad()).unwrap();
 
-        super::set_element_flags(element, gst::ElementFlags::SINK);
+        crate::set_element_flags(obj, gst::ElementFlags::SINK);
     }
 }
 
 impl ElementImpl for ProxySink {
     fn change_state(
         &self,
-        element: &gst::Element,
+        element: &Self::Type,
         transition: gst::StateChange,
     ) -> Result<gst::StateChangeSuccess, gst::StateChangeError> {
         gst_trace!(SINK_CAT, obj: element, "Changing state {:?}", transition);
@@ -909,13 +909,13 @@ impl PadSrcHandler for ProxySrcPadHandler {
 
 #[derive(Debug)]
 struct ProxySrcTask {
-    element: gst::Element,
+    element: super::ProxySrc,
     src_pad: PadSrcWeak,
     dataqueue: DataQueue,
 }
 
 impl ProxySrcTask {
-    fn new(element: &gst::Element, src_pad: &PadSrc, dataqueue: DataQueue) -> Self {
+    fn new(element: &super::ProxySrc, src_pad: &PadSrc, dataqueue: DataQueue) -> Self {
         ProxySrcTask {
             element: element.clone(),
             src_pad: src_pad.downgrade(),
@@ -1043,7 +1043,7 @@ impl TaskImpl for ProxySrcTask {
 }
 
 #[derive(Debug)]
-struct ProxySrc {
+pub struct ProxySrc {
     src_pad: PadSrc,
     task: Task,
     proxy_ctx: StdMutex<Option<ProxyContext>>,
@@ -1060,7 +1060,7 @@ lazy_static! {
 }
 
 impl ProxySrc {
-    fn prepare(&self, element: &gst::Element) -> Result<(), gst::ErrorMessage> {
+    fn prepare(&self, element: &super::ProxySrc) -> Result<(), gst::ErrorMessage> {
         gst_debug!(SRC_CAT, obj: element, "Preparing");
 
         let settings = self.settings.lock().unwrap().clone();
@@ -1126,7 +1126,7 @@ impl ProxySrc {
         Ok(())
     }
 
-    fn unprepare(&self, element: &gst::Element) {
+    fn unprepare(&self, element: &super::ProxySrc) {
         gst_debug!(SRC_CAT, obj: element, "Unpreparing");
 
         {
@@ -1143,21 +1143,21 @@ impl ProxySrc {
         gst_debug!(SRC_CAT, obj: element, "Unprepared");
     }
 
-    fn stop(&self, element: &gst::Element) -> Result<(), gst::ErrorMessage> {
+    fn stop(&self, element: &super::ProxySrc) -> Result<(), gst::ErrorMessage> {
         gst_debug!(SRC_CAT, obj: element, "Stopping");
         self.task.stop()?;
         gst_debug!(SRC_CAT, obj: element, "Stopped");
         Ok(())
     }
 
-    fn start(&self, element: &gst::Element) -> Result<(), gst::ErrorMessage> {
+    fn start(&self, element: &super::ProxySrc) -> Result<(), gst::ErrorMessage> {
         gst_debug!(SRC_CAT, obj: element, "Starting");
         self.task.start()?;
         gst_debug!(SRC_CAT, obj: element, "Started");
         Ok(())
     }
 
-    fn pause(&self, element: &gst::Element) -> Result<(), gst::ErrorMessage> {
+    fn pause(&self, element: &super::ProxySrc) -> Result<(), gst::ErrorMessage> {
         gst_debug!(SRC_CAT, obj: element, "Pausing");
         self.task.pause()?;
         gst_debug!(SRC_CAT, obj: element, "Paused");
@@ -1167,13 +1167,14 @@ impl ProxySrc {
 
 impl ObjectSubclass for ProxySrc {
     const NAME: &'static str = "RsTsProxySrc";
+    type Type = super::ProxySrc;
     type ParentType = gst::Element;
     type Instance = gst::subclass::ElementInstanceStruct<Self>;
     type Class = subclass::simple::ClassStruct<Self>;
 
     glib_object_subclass!();
 
-    fn class_init(klass: &mut subclass::simple::ClassStruct<Self>) {
+    fn class_init(klass: &mut Self::Class) {
         klass.set_metadata(
             "Thread-sharing proxy source",
             "Source/Generic",
@@ -1199,7 +1200,7 @@ impl ObjectSubclass for ProxySrc {
         unreachable!()
     }
 
-    fn with_class(klass: &subclass::simple::ClassStruct<Self>) -> Self {
+    fn with_class(klass: &Self::Class) -> Self {
         Self {
             src_pad: PadSrc::new(
                 gst::Pad::from_template(&klass.get_pad_template("src").unwrap(), Some("src")),
@@ -1214,7 +1215,7 @@ impl ObjectSubclass for ProxySrc {
 }
 
 impl ObjectImpl for ProxySrc {
-    fn set_property(&self, _obj: &glib::Object, id: usize, value: &glib::Value) {
+    fn set_property(&self, _obj: &Self::Type, id: usize, value: &glib::Value) {
         let prop = &PROPERTIES_SRC[id];
 
         let mut settings = self.settings.lock().unwrap();
@@ -1247,7 +1248,7 @@ impl ObjectImpl for ProxySrc {
         }
     }
 
-    fn get_property(&self, _obj: &glib::Object, id: usize) -> Result<glib::Value, ()> {
+    fn get_property(&self, _obj: &Self::Type, id: usize) -> Result<glib::Value, ()> {
         let prop = &PROPERTIES_SRC[id];
 
         let settings = self.settings.lock().unwrap();
@@ -1262,20 +1263,19 @@ impl ObjectImpl for ProxySrc {
         }
     }
 
-    fn constructed(&self, obj: &glib::Object) {
+    fn constructed(&self, obj: &Self::Type) {
         self.parent_constructed(obj);
 
-        let element = obj.downcast_ref::<gst::Element>().unwrap();
+        obj.add_pad(self.src_pad.gst_pad()).unwrap();
-        element.add_pad(self.src_pad.gst_pad()).unwrap();
 
-        super::set_element_flags(element, gst::ElementFlags::SOURCE);
+        crate::set_element_flags(obj, gst::ElementFlags::SOURCE);
     }
 }
 
 impl ElementImpl for ProxySrc {
     fn change_state(
         &self,
-        element: &gst::Element,
+        element: &Self::Type,
         transition: gst::StateChange,
     ) -> Result<gst::StateChangeSuccess, gst::StateChangeError> {
         gst_trace!(SRC_CAT, obj: element, "Changing state {:?}", transition);
@@ -1317,18 +1317,3 @@ impl ElementImpl for ProxySrc {
         Ok(success)
     }
 }
-
-pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
-    gst::Element::register(
-        Some(plugin),
-        "ts-proxysink",
-        gst::Rank::None,
-        ProxySink::get_type(),
-    )?;
-    gst::Element::register(
-        Some(plugin),
-        "ts-proxysrc",
-        gst::Rank::None,
-        ProxySrc::get_type(),
-    )
-}

generic/threadshare/src/proxy/mod.rs

@@ -0,0 +1,54 @@
+// Copyright (C) 2018 Sebastian Dröge <sebastian@centricular.com>
+//
+// This library is free software; you can redistribute it and/or
+// modify it under the terms of the GNU Library General Public
+// License as published by the Free Software Foundation; either
+// version 2 of the License, or (at your option) any later version.
+//
+// This library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+// Library General Public License for more details.
+//
+// You should have received a copy of the GNU Library General Public
+// License along with this library; if not, write to the
+// Free Software Foundation, Inc., 51 Franklin Street, Suite 500,
+// Boston, MA 02110-1335, USA.
+
+use glib::glib_wrapper;
+use glib::prelude::*;
+
+mod imp;
+
+glib_wrapper! {
+    pub struct ProxySink(ObjectSubclass<imp::ProxySink>) @extends gst::Element, gst::Object;
+}
+
+// GStreamer elements need to be thread-safe. For the private implementation this is automatically
+// enforced but for the public wrapper type we need to specify this manually.
+unsafe impl Send for ProxySink {}
+unsafe impl Sync for ProxySink {}
+
+glib_wrapper! {
+    pub struct ProxySrc(ObjectSubclass<imp::ProxySrc>) @extends gst::Element, gst::Object;
+}
+
+// GStreamer elements need to be thread-safe. For the private implementation this is automatically
+// enforced but for the public wrapper type we need to specify this manually.
+unsafe impl Send for ProxySrc {}
+unsafe impl Sync for ProxySrc {}
+
+pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
+    gst::Element::register(
+        Some(plugin),
+        "ts-proxysink",
+        gst::Rank::None,
+        ProxySink::static_type(),
+    )?;
+    gst::Element::register(
+        Some(plugin),
+        "ts-proxysrc",
+        gst::Rank::None,
+        ProxySrc::static_type(),
+    )
+}

generic/threadshare/src/queue/imp.rs

@@ -37,7 +37,7 @@ use std::{u32, u64};
 use crate::runtime::prelude::*;
 use crate::runtime::{Context, PadSink, PadSinkRef, PadSrc, PadSrcRef, PadSrcWeak, Task};
 
-use super::dataqueue::{DataQueue, DataQueueItem};
+use crate::dataqueue::{DataQueue, DataQueueItem};
 
 const DEFAULT_MAX_SIZE_BUFFERS: u32 = 200;
 const DEFAULT_MAX_SIZE_BYTES: u32 = 1024 * 1024;
@@ -149,7 +149,7 @@ impl PadSinkHandler for QueuePadSinkHandler {
         buffer: gst::Buffer,
     ) -> BoxFuture<'static, Result<gst::FlowSuccess, gst::FlowError>> {
         let pad_weak = pad.downgrade();
-        let element = element.clone();
+        let element = element.clone().downcast::<super::Queue>().unwrap();
         async move {
             let pad = pad_weak.upgrade().expect("PadSink no longer exists");
             gst_log!(CAT, obj: pad.gst_pad(), "Handling {:?}", buffer);
@@ -169,7 +169,7 @@ impl PadSinkHandler for QueuePadSinkHandler {
         list: gst::BufferList,
     ) -> BoxFuture<'static, Result<gst::FlowSuccess, gst::FlowError>> {
         let pad_weak = pad.downgrade();
-        let element = element.clone();
+        let element = element.clone().downcast::<super::Queue>().unwrap();
         async move {
             let pad = pad_weak.upgrade().expect("PadSink no longer exists");
             gst_log!(CAT, obj: pad.gst_pad(), "Handling {:?}", list);
@@ -221,7 +221,7 @@ impl PadSinkHandler for QueuePadSinkHandler {
         gst_log!(CAT, obj: pad.gst_pad(), "Handling serialized {:?}", event);
 
         let pad_weak = pad.downgrade();
-        let element = element.clone();
+        let element = element.clone().downcast::<super::Queue>().unwrap();
         async move {
             let pad = pad_weak.upgrade().expect("PadSink no longer exists");
             let queue = Queue::from_instance(&element);
@@ -379,13 +379,13 @@ impl PadSrcHandler for QueuePadSrcHandler {
 
 #[derive(Debug)]
 struct QueueTask {
-    element: gst::Element,
+    element: super::Queue,
     src_pad: PadSrcWeak,
     dataqueue: DataQueue,
 }
 
 impl QueueTask {
-    fn new(element: &gst::Element, src_pad: &PadSrc, dataqueue: DataQueue) -> Self {
+    fn new(element: &super::Queue, src_pad: &PadSrc, dataqueue: DataQueue) -> Self {
         QueueTask {
             element: element.clone(),
             src_pad: src_pad.downgrade(),
@@ -503,7 +503,7 @@ impl TaskImpl for QueueTask {
 }
 
 #[derive(Debug)]
-struct Queue {
+pub struct Queue {
     sink_pad: PadSink,
     src_pad: PadSrc,
     task: Task,
@@ -561,7 +561,7 @@ impl Queue {
     /* Schedules emptying of the pending queue. If there is an upstream
      * TaskContext, the new task is spawned, it is otherwise
      * returned, for the caller to block on */
-    async fn schedule_pending_queue(&self, element: &gst::Element) {
+    async fn schedule_pending_queue(&self, element: &super::Queue) {
         loop {
             let more_queue_space_receiver = {
                 let dataqueue = self.dataqueue.lock().unwrap();
@@ -604,7 +604,7 @@ impl Queue {
 
     async fn enqueue_item(
         &self,
-        element: &gst::Element,
+        element: &super::Queue,
         item: DataQueueItem,
     ) -> Result<gst::FlowSuccess, gst::FlowError> {
         let wait_fut = {
@@ -672,7 +672,7 @@ impl Queue {
         *self.last_res.lock().unwrap()
     }
 
-    fn prepare(&self, element: &gst::Element) -> Result<(), gst::ErrorMessage> {
+    fn prepare(&self, element: &super::Queue) -> Result<(), gst::ErrorMessage> {
         gst_debug!(CAT, obj: element, "Preparing");
 
         let settings = self.settings.lock().unwrap().clone();
@@ -721,7 +721,7 @@ impl Queue {
         Ok(())
     }
 
-    fn unprepare(&self, element: &gst::Element) {
+    fn unprepare(&self, element: &super::Queue) {
         gst_debug!(CAT, obj: element, "Unpreparing");
 
         self.task.unprepare().unwrap();
@@ -734,14 +734,14 @@ impl Queue {
         gst_debug!(CAT, obj: element, "Unprepared");
     }
 
-    fn stop(&self, element: &gst::Element) -> Result<(), gst::ErrorMessage> {
+    fn stop(&self, element: &super::Queue) -> Result<(), gst::ErrorMessage> {
         gst_debug!(CAT, obj: element, "Stopping");
         self.task.stop()?;
         gst_debug!(CAT, obj: element, "Stopped");
         Ok(())
     }
 
-    fn start(&self, element: &gst::Element) -> Result<(), gst::ErrorMessage> {
+    fn start(&self, element: &super::Queue) -> Result<(), gst::ErrorMessage> {
         gst_debug!(CAT, obj: element, "Starting");
         self.task.start()?;
         gst_debug!(CAT, obj: element, "Started");
@@ -751,13 +751,14 @@ impl Queue {
 
 impl ObjectSubclass for Queue {
     const NAME: &'static str = "RsTsQueue";
+    type Type = super::Queue;
     type ParentType = gst::Element;
     type Instance = gst::subclass::ElementInstanceStruct<Self>;
     type Class = subclass::simple::ClassStruct<Self>;
 
     glib_object_subclass!();
 
-    fn class_init(klass: &mut subclass::simple::ClassStruct<Self>) {
+    fn class_init(klass: &mut Self::Class) {
         klass.set_metadata(
             "Thread-sharing queue",
             "Generic",
@@ -788,7 +789,7 @@ impl ObjectSubclass for Queue {
         klass.install_properties(&PROPERTIES);
     }
 
-    fn with_class(klass: &subclass::simple::ClassStruct<Self>) -> Self {
+    fn with_class(klass: &Self::Class) -> Self {
         Self {
             sink_pad: PadSink::new(
                 gst::Pad::from_template(&klass.get_pad_template("sink").unwrap(), Some("sink")),
@@ -808,7 +809,7 @@ impl ObjectSubclass for Queue {
 }
 
 impl ObjectImpl for Queue {
-    fn set_property(&self, _obj: &glib::Object, id: usize, value: &glib::Value) {
+    fn set_property(&self, _obj: &Self::Type, id: usize, value: &glib::Value) {
         let prop = &PROPERTIES[id];
 
         let mut settings = self.settings.lock().unwrap();
@@ -835,7 +836,7 @@ impl ObjectImpl for Queue {
         }
     }
 
-    fn get_property(&self, _obj: &glib::Object, id: usize) -> Result<glib::Value, ()> {
+    fn get_property(&self, _obj: &Self::Type, id: usize) -> Result<glib::Value, ()> {
         let prop = &PROPERTIES[id];
 
         let settings = self.settings.lock().unwrap();
@@ -849,19 +850,18 @@ impl ObjectImpl for Queue {
         }
     }
 
-    fn constructed(&self, obj: &glib::Object) {
+    fn constructed(&self, obj: &Self::Type) {
         self.parent_constructed(obj);
 
-        let element = obj.downcast_ref::<gst::Element>().unwrap();
+        obj.add_pad(self.sink_pad.gst_pad()).unwrap();
-        element.add_pad(self.sink_pad.gst_pad()).unwrap();
+        obj.add_pad(self.src_pad.gst_pad()).unwrap();
-        element.add_pad(self.src_pad.gst_pad()).unwrap();
     }
 }
 
 impl ElementImpl for Queue {
     fn change_state(
         &self,
-        element: &gst::Element,
+        element: &Self::Type,
         transition: gst::StateChange,
     ) -> Result<gst::StateChangeSuccess, gst::StateChangeError> {
         gst_trace!(CAT, obj: element, "Changing state {:?}", transition);
@@ -891,7 +891,3 @@ impl ElementImpl for Queue {
         Ok(success)
     }
 }
-
-pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
-    gst::Element::register(Some(plugin), "ts-queue", gst::Rank::None, Queue::get_type())
-}

generic/threadshare/src/queue/mod.rs

@@ -0,0 +1,39 @@
+// Copyright (C) 2018 Sebastian Dröge <sebastian@centricular.com>
+//
+// This library is free software; you can redistribute it and/or
+// modify it under the terms of the GNU Library General Public
+// License as published by the Free Software Foundation; either
+// version 2 of the License, or (at your option) any later version.
+//
+// This library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+// Library General Public License for more details.
+//
+// You should have received a copy of the GNU Library General Public
+// License along with this library; if not, write to the
+// Free Software Foundation, Inc., 51 Franklin Street, Suite 500,
+// Boston, MA 02110-1335, USA.
+
+use glib::glib_wrapper;
+use glib::prelude::*;
+
+mod imp;
+
+glib_wrapper! {
+    pub struct Queue(ObjectSubclass<imp::Queue>) @extends gst::Element, gst::Object;
+}
+
+// GStreamer elements need to be thread-safe. For the private implementation this is automatically
+// enforced but for the public wrapper type we need to specify this manually.
+unsafe impl Send for Queue {}
+unsafe impl Sync for Queue {}
+
+pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
+    gst::Element::register(
+        Some(plugin),
+        "ts-queue",
+        gst::Rank::None,
+        Queue::static_type(),
+    )
+}

generic/threadshare/src/runtime/pad.rs

@@ -402,7 +402,11 @@ impl PadSrc {
                         },
                         move |imp, element| {
                             let this_ref = PadSrcRef::new(inner_arc);
-                            handler.src_activate(&this_ref, imp, element)
+                            handler.src_activate(
+                                &this_ref,
+                                imp,
+                                element.dynamic_cast_ref::<gst::Element>().unwrap(),
+                            )
                         },
                     )
                 });
@@ -425,7 +429,13 @@ impl PadSrc {
                         move |imp, element| {
                             let this_ref = PadSrcRef::new(inner_arc);
                             this_ref.activate_mode_hook(mode, active)?;
-                            handler.src_activatemode(&this_ref, imp, element, mode, active)
+                            handler.src_activatemode(
+                                &this_ref,
+                                imp,
+                                element.dynamic_cast_ref::<gst::Element>().unwrap(),
+                                mode,
+                                active,
+                            )
                         },
                     )
                 });
@@ -443,7 +453,12 @@ impl PadSrc {
                         || Err(FlowError::Error),
                         move |imp, element| {
                             let this_ref = PadSrcRef::new(inner_arc);
-                            handler.src_event_full(&this_ref, imp, &element, event)
+                            handler.src_event_full(
+                                &this_ref,
+                                imp,
+                                element.dynamic_cast_ref::<gst::Element>().unwrap(),
+                                event,
+                            )
                         },
                     )
                 });
@@ -459,7 +474,7 @@ impl PadSrc {
                     move |imp, element| {
                         let this_ref = PadSrcRef::new(inner_arc);
                         if !query.is_serialized() {
-                            handler.src_query(&this_ref, imp, &element, query)
+                            handler.src_query(&this_ref, imp, element.dynamic_cast_ref::<gst::Element>().unwrap(), query)
                         } else {
                             gst_fixme!(RUNTIME_CAT, obj: this_ref.gst_pad(), "Serialized Query not supported");
                             false
@@ -509,6 +524,8 @@ impl Deref for PadSrc {
 /// [`pad` module]: index.html
 pub trait PadSinkHandler: Clone + Send + Sync + 'static {
     type ElementImpl: ElementImpl + ObjectSubclass;
+    // FIXME: Once associated type bounds are stable we should use ObjectSubclass::Type below
+    // instead of &gst::Element
 
     fn sink_activate(
         &self,
@@ -806,7 +823,11 @@ impl PadSink {
                         },
                         move |imp, element| {
                             let this_ref = PadSinkRef::new(inner_arc);
-                            handler.sink_activate(&this_ref, imp, element)
+                            handler.sink_activate(
+                                &this_ref,
+                                imp,
+                                element.dynamic_cast_ref::<gst::Element>().unwrap(),
+                            )
                         },
                     )
                 });
@@ -830,7 +851,13 @@ impl PadSink {
                             let this_ref = PadSinkRef::new(inner_arc);
                             this_ref.activate_mode_hook(mode, active)?;
 
-                            handler.sink_activatemode(&this_ref, imp, element, mode, active)
+                            handler.sink_activatemode(
+                                &this_ref,
+                                imp,
+                                element.dynamic_cast_ref::<gst::Element>().unwrap(),
+                                mode,
+                                active,
+                            )
                         },
                     )
                 });
@@ -848,10 +875,12 @@ impl PadSink {
                             if Context::current_has_sub_tasks() {
                                 let this_weak = PadSinkWeak(Arc::downgrade(&inner_arc));
                                 let handler = handler.clone();
-                                let element = element.clone();
+                                let element =
+                                    element.clone().dynamic_cast::<gst::Element>().unwrap();
                                 let delayed_fut = async move {
-                                    let imp =
+                                    let imp = <H::ElementImpl as ObjectSubclass>::from_instance(
-                                        <H::ElementImpl as ObjectSubclass>::from_instance(&element);
+                                        element.unsafe_cast_ref(),
+                                    );
                                     let this_ref =
                                         this_weak.upgrade().ok_or(gst::FlowError::Flushing)?;
                                     handler.sink_chain(&this_ref, imp, &element, buffer).await
@@ -861,8 +890,12 @@ impl PadSink {
                                 Ok(gst::FlowSuccess::Ok)
                             } else {
                                 let this_ref = PadSinkRef::new(inner_arc);
-                                let chain_fut =
+                                let chain_fut = handler.sink_chain(
-                                    handler.sink_chain(&this_ref, imp, &element, buffer);
+                                    &this_ref,
+                                    imp,
+                                    element.dynamic_cast_ref::<gst::Element>().unwrap(),
+                                    buffer,
+                                );
                                 this_ref.handle_future(chain_fut)
                             }
                         },
@@ -882,10 +915,12 @@ impl PadSink {
                             if Context::current_has_sub_tasks() {
                                 let this_weak = PadSinkWeak(Arc::downgrade(&inner_arc));
                                 let handler = handler.clone();
-                                let element = element.clone();
+                                let element =
+                                    element.clone().dynamic_cast::<gst::Element>().unwrap();
                                 let delayed_fut = async move {
-                                    let imp =
+                                    let imp = <H::ElementImpl as ObjectSubclass>::from_instance(
-                                        <H::ElementImpl as ObjectSubclass>::from_instance(&element);
+                                        element.unsafe_cast_ref(),
+                                    );
                                     let this_ref =
                                         this_weak.upgrade().ok_or(gst::FlowError::Flushing)?;
                                     handler
@@ -897,8 +932,12 @@ impl PadSink {
                                 Ok(gst::FlowSuccess::Ok)
                             } else {
                                 let this_ref = PadSinkRef::new(inner_arc);
-                                let chain_list_fut =
+                                let chain_list_fut = handler.sink_chain_list(
-                                    handler.sink_chain_list(&this_ref, imp, &element, list);
+                                    &this_ref,
+                                    imp,
+                                    element.dynamic_cast_ref::<gst::Element>().unwrap(),
+                                    list,
+                                );
                                 this_ref.handle_future(chain_list_fut)
                             }
                         },
@@ -921,10 +960,11 @@ impl PadSink {
                                 if Context::current_has_sub_tasks() {
                                     let this_weak = PadSinkWeak(Arc::downgrade(&inner_arc));
                                     let handler = handler.clone();
-                                    let element = element.clone();
+                                    let element =
+                                        element.clone().dynamic_cast::<gst::Element>().unwrap();
                                     let delayed_fut = async move {
                                         let imp = <H::ElementImpl as ObjectSubclass>::from_instance(
-                                            &element,
+                                            element.unsafe_cast_ref(),
                                         );
                                         let this_ref =
                                             this_weak.upgrade().ok_or(gst::FlowError::Flushing)?;
@@ -942,13 +982,21 @@ impl PadSink {
                                 } else {
                                     let this_ref = PadSinkRef::new(inner_arc);
                                     let event_fut = handler.sink_event_full_serialized(
-                                        &this_ref, imp, &element, event,
+                                        &this_ref,
+                                        imp,
+                                        element.dynamic_cast_ref::<gst::Element>().unwrap(),
+                                        event,
                                     );
                                     this_ref.handle_future(event_fut)
                                 }
                             } else {
                                 let this_ref = PadSinkRef::new(inner_arc);
-                                handler.sink_event_full(&this_ref, imp, &element, event)
+                                handler.sink_event_full(
+                                    &this_ref,
+                                    imp,
+                                    element.dynamic_cast_ref::<gst::Element>().unwrap(),
+                                    event,
+                                )
                             }
                         },
                     )
@@ -965,7 +1013,7 @@ impl PadSink {
                     move |imp, element| {
                         let this_ref = PadSinkRef::new(inner_arc);
                         if !query.is_serialized() {
-                            handler.sink_query(&this_ref, imp, &element, query)
+                            handler.sink_query(&this_ref, imp, element.dynamic_cast_ref::<gst::Element>().unwrap(), query)
                         } else {
                             gst_fixme!(RUNTIME_CAT, obj: this_ref.gst_pad(), "Serialized Query not supported");
                             false

generic/threadshare/src/tcpclientsrc/imp.rs

@@ -44,7 +44,7 @@ use crate::runtime::prelude::*;
 use crate::runtime::task;
 use crate::runtime::{Context, PadSrc, PadSrcRef, PadSrcWeak, Task, TaskState};
 
-use super::socket::{Socket, SocketError, SocketRead};
+use crate::socket::{Socket, SocketError, SocketRead};
 
 const DEFAULT_HOST: Option<&str> = Some("127.0.0.1");
 const DEFAULT_PORT: i32 = 4953;
@@ -139,7 +139,7 @@ static PROPERTIES: [subclass::Property; 6] = [
     }),
 ];
 
-pub struct TcpClientReader(tokio::net::TcpStream);
+struct TcpClientReader(tokio::net::TcpStream);
 
 impl TcpClientReader {
     pub fn new(socket: tokio::net::TcpStream) -> Self {
@@ -201,7 +201,7 @@ impl TcpClientSrcPadHandler {
         self.0.state.lock().await.need_segment = true;
     }
 
-    async fn push_prelude(&self, pad: &PadSrcRef<'_>, _element: &gst::Element) {
+    async fn push_prelude(&self, pad: &PadSrcRef<'_>, _element: &super::TcpClientSrc) {
         let mut state = self.0.state.lock().await;
         if state.need_initial_events {
             gst_debug!(CAT, obj: pad.gst_pad(), "Pushing initial events");
@@ -232,7 +232,7 @@ impl TcpClientSrcPadHandler {
     async fn push_buffer(
         &self,
         pad: &PadSrcRef<'_>,
-        element: &gst::Element,
+        element: &super::TcpClientSrc,
         buffer: gst::Buffer,
     ) -> Result<gst::FlowSuccess, gst::FlowError> {
         gst_log!(CAT, obj: pad.gst_pad(), "Handling {:?}", buffer);
@@ -328,7 +328,7 @@ impl PadSrcHandler for TcpClientSrcPadHandler {
 }
 
 struct TcpClientSrcTask {
-    element: gst::Element,
+    element: super::TcpClientSrc,
     src_pad: PadSrcWeak,
     src_pad_handler: TcpClientSrcPadHandler,
     saddr: SocketAddr,
@@ -338,7 +338,7 @@ struct TcpClientSrcTask {
 
 impl TcpClientSrcTask {
     fn new(
-        element: &gst::Element,
+        element: &super::TcpClientSrc,
         src_pad: &PadSrc,
         src_pad_handler: &TcpClientSrcPadHandler,
         saddr: SocketAddr,
@@ -371,7 +371,7 @@ impl TaskImpl for TcpClientSrcTask {
 
             self.socket = Some(
                 Socket::try_new(
-                    self.element.clone(),
+                    self.element.clone().upcast(),
                     self.buffer_pool.take().unwrap(),
                     TcpClientReader::new(socket),
                 )
@@ -496,7 +496,7 @@ impl TaskImpl for TcpClientSrcTask {
     }
 }
 
-struct TcpClientSrc {
+pub struct TcpClientSrc {
     src_pad: PadSrc,
     src_pad_handler: TcpClientSrcPadHandler,
     task: Task,
@@ -512,7 +512,7 @@ lazy_static! {
 }
 
 impl TcpClientSrc {
-    fn prepare(&self, element: &gst::Element) -> Result<(), gst::ErrorMessage> {
+    fn prepare(&self, element: &super::TcpClientSrc) -> Result<(), gst::ErrorMessage> {
         let settings = self.settings.lock().unwrap().clone();
 
         gst_debug!(CAT, obj: element, "Preparing");
@@ -581,27 +581,27 @@ impl TcpClientSrc {
         Ok(())
     }
 
-    fn unprepare(&self, element: &gst::Element) {
+    fn unprepare(&self, element: &super::TcpClientSrc) {
         gst_debug!(CAT, obj: element, "Unpreparing");
         self.task.unprepare().unwrap();
         gst_debug!(CAT, obj: element, "Unprepared");
     }
 
-    fn stop(&self, element: &gst::Element) -> Result<(), gst::ErrorMessage> {
+    fn stop(&self, element: &super::TcpClientSrc) -> Result<(), gst::ErrorMessage> {
         gst_debug!(CAT, obj: element, "Stopping");
         self.task.stop()?;
         gst_debug!(CAT, obj: element, "Stopped");
         Ok(())
     }
 
-    fn start(&self, element: &gst::Element) -> Result<(), gst::ErrorMessage> {
+    fn start(&self, element: &super::TcpClientSrc) -> Result<(), gst::ErrorMessage> {
         gst_debug!(CAT, obj: element, "Starting");
         self.task.start()?;
         gst_debug!(CAT, obj: element, "Started");
         Ok(())
     }
 
-    fn pause(&self, element: &gst::Element) -> Result<(), gst::ErrorMessage> {
+    fn pause(&self, element: &super::TcpClientSrc) -> Result<(), gst::ErrorMessage> {
         gst_debug!(CAT, obj: element, "Pausing");
         self.task.pause()?;
         gst_debug!(CAT, obj: element, "Paused");
@@ -611,13 +611,14 @@ impl TcpClientSrc {
 
 impl ObjectSubclass for TcpClientSrc {
     const NAME: &'static str = "RsTsTcpClientSrc";
+    type Type = super::TcpClientSrc;
     type ParentType = gst::Element;
     type Instance = gst::subclass::ElementInstanceStruct<Self>;
     type Class = subclass::simple::ClassStruct<Self>;
 
     glib_object_subclass!();
 
-    fn class_init(klass: &mut subclass::simple::ClassStruct<Self>) {
+    fn class_init(klass: &mut Self::Class) {
         klass.set_metadata(
             "Thread-sharing TCP client source",
             "Source/Network",
@@ -638,7 +639,7 @@ impl ObjectSubclass for TcpClientSrc {
         klass.install_properties(&PROPERTIES);
     }
 
-    fn with_class(klass: &subclass::simple::ClassStruct<Self>) -> Self {
+    fn with_class(klass: &Self::Class) -> Self {
         let src_pad_handler = TcpClientSrcPadHandler::default();
 
         Self {
@@ -654,7 +655,7 @@ impl ObjectSubclass for TcpClientSrc {
 }
 
 impl ObjectImpl for TcpClientSrc {
-    fn set_property(&self, _obj: &glib::Object, id: usize, value: &glib::Value) {
+    fn set_property(&self, _obj: &Self::Type, id: usize, value: &glib::Value) {
         let prop = &PROPERTIES[id];
 
         let mut settings = self.settings.lock().unwrap();
@@ -684,7 +685,7 @@ impl ObjectImpl for TcpClientSrc {
         }
     }
 
-    fn get_property(&self, _obj: &glib::Object, id: usize) -> Result<glib::Value, ()> {
+    fn get_property(&self, _obj: &Self::Type, id: usize) -> Result<glib::Value, ()> {
         let prop = &PROPERTIES[id];
 
         let settings = self.settings.lock().unwrap();
@@ -699,20 +700,19 @@ impl ObjectImpl for TcpClientSrc {
         }
     }
 
-    fn constructed(&self, obj: &glib::Object) {
+    fn constructed(&self, obj: &Self::Type) {
         self.parent_constructed(obj);
 
-        let element = obj.downcast_ref::<gst::Element>().unwrap();
+        obj.add_pad(self.src_pad.gst_pad()).unwrap();
-        element.add_pad(self.src_pad.gst_pad()).unwrap();
 
-        super::set_element_flags(element, gst::ElementFlags::SOURCE);
+        crate::set_element_flags(obj, gst::ElementFlags::SOURCE);
     }
 }
 
 impl ElementImpl for TcpClientSrc {
     fn change_state(
         &self,
-        element: &gst::Element,
+        element: &Self::Type,
         transition: gst::StateChange,
     ) -> Result<gst::StateChangeSuccess, gst::StateChangeError> {
         gst_trace!(CAT, obj: element, "Changing state {:?}", transition);
@@ -754,12 +754,3 @@ impl ElementImpl for TcpClientSrc {
         Ok(success)
     }
 }
-
-pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
-    gst::Element::register(
-        Some(plugin),
-        "ts-tcpclientsrc",
-        gst::Rank::None,
-        TcpClientSrc::get_type(),
-    )
-}

generic/threadshare/src/tcpclientsrc/mod.rs

@@ -0,0 +1,40 @@
+// Copyright (C) 2018 Sebastian Dröge <sebastian@centricular.com>
+// Copyright (C) 2018 LEE Dongjun <redongjun@gmail.com>
+//
+// This library is free software; you can redistribute it and/or
+// modify it under the terms of the GNU Library General Public
+// License as published by the Free Software Foundation; either
+// version 2 of the License, or (at your option) any later version.
+//
+// This library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+// Library General Public License for more details.
+//
+// You should have received a copy of the GNU Library General Public
+// License along with this library; if not, write to the
+// Free Software Foundation, Inc., 51 Franklin Street, Suite 500,
+// Boston, MA 02110-1335, USA.
+
+use glib::glib_wrapper;
+use glib::prelude::*;
+
+mod imp;
+
+glib_wrapper! {
+    pub struct TcpClientSrc(ObjectSubclass<imp::TcpClientSrc>) @extends gst::Element, gst::Object;
+}
+
+// GStreamer elements need to be thread-safe. For the private implementation this is automatically
+// enforced but for the public wrapper type we need to specify this manually.
+unsafe impl Send for TcpClientSrc {}
+unsafe impl Sync for TcpClientSrc {}
+
+pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
+    gst::Element::register(
+        Some(plugin),
+        "ts-tcpclientsrc",
+        gst::Rank::None,
+        TcpClientSrc::static_type(),
+    )
+}

generic/threadshare/src/udpsink/imp.rs

@@ -562,7 +562,7 @@ impl UdpSinkPadHandler {
 
     async fn render(
         &self,
-        element: &gst::Element,
+        element: &super::UdpSink,
         buffer: gst::Buffer,
     ) -> Result<gst::FlowSuccess, gst::FlowError> {
         let (
@@ -691,7 +691,7 @@ impl UdpSinkPadHandler {
     }
 
     /* Wait until specified time */
-    async fn sync(&self, element: &gst::Element, running_time: gst::ClockTime) {
+    async fn sync(&self, element: &super::UdpSink, running_time: gst::ClockTime) {
         let now = element.get_current_running_time();
 
         if let Some(delay) = running_time
@@ -702,7 +702,7 @@ impl UdpSinkPadHandler {
         }
     }
 
-    async fn handle_event(&self, element: &gst::Element, event: gst::Event) {
+    async fn handle_event(&self, element: &super::UdpSink, event: gst::Event) {
         match event.view() {
             EventView::Eos(_) => {
                 let _ = element.post_message(gst::message::Eos::builder().src(element).build());
@@ -726,7 +726,7 @@ impl PadSinkHandler for UdpSinkPadHandler {
         buffer: gst::Buffer,
     ) -> BoxFuture<'static, Result<gst::FlowSuccess, gst::FlowError>> {
         let sender = Arc::clone(&self.0.read().unwrap().sender);
-        let element = element.clone();
+        let element = element.clone().downcast::<super::UdpSink>().unwrap();
 
         async move {
             if let Some(sender) = sender.lock().await.as_mut() {
@@ -748,7 +748,7 @@ impl PadSinkHandler for UdpSinkPadHandler {
         list: gst::BufferList,
     ) -> BoxFuture<'static, Result<gst::FlowSuccess, gst::FlowError>> {
         let sender = Arc::clone(&self.0.read().unwrap().sender);
-        let element = element.clone();
+        let element = element.clone().downcast::<super::UdpSink>().unwrap();
 
         async move {
             if let Some(sender) = sender.lock().await.as_mut() {
@@ -773,7 +773,7 @@ impl PadSinkHandler for UdpSinkPadHandler {
         event: gst::Event,
     ) -> BoxFuture<'static, bool> {
         let sender = Arc::clone(&self.0.read().unwrap().sender);
-        let element = element.clone();
+        let element = element.clone().downcast::<super::UdpSink>().unwrap();
 
         async move {
             if let EventView::FlushStop(_) = event.view() {
@@ -807,13 +807,13 @@ impl PadSinkHandler for UdpSinkPadHandler {
 
 #[derive(Debug)]
 struct UdpSinkTask {
-    element: gst::Element,
+    element: super::UdpSink,
     sink_pad_handler: UdpSinkPadHandler,
     receiver: Option<mpsc::Receiver<TaskItem>>,
 }
 
 impl UdpSinkTask {
-    fn new(element: &gst::Element, sink_pad_handler: &UdpSinkPadHandler) -> Self {
+    fn new(element: &super::UdpSink, sink_pad_handler: &UdpSinkPadHandler) -> Self {
         UdpSinkTask {
             element: element.clone(),
             sink_pad_handler: sink_pad_handler.clone(),
@@ -877,7 +877,7 @@ enum SocketFamily {
 }
 
 #[derive(Debug)]
-struct UdpSink {
+pub struct UdpSink {
     sink_pad: PadSink,
     sink_pad_handler: UdpSinkPadHandler,
     task: Task,
@@ -889,7 +889,7 @@ impl UdpSink {
         &self,
         family: SocketFamily,
         context: &Context,
-        element: &gst::Element,
+        element: &super::UdpSink,
     ) -> Result<(), gst::ErrorMessage> {
         let mut settings = self.settings.lock().unwrap();
 
@@ -1028,7 +1028,7 @@ impl UdpSink {
         Ok(())
     }
 
-    fn prepare(&self, element: &gst::Element) -> Result<(), gst::ErrorMessage> {
+    fn prepare(&self, element: &super::UdpSink) -> Result<(), gst::ErrorMessage> {
         gst_debug!(CAT, obj: element, "Preparing");
 
         let context = {
@@ -1060,7 +1060,7 @@ impl UdpSink {
         Ok(())
     }
 
-    fn unprepare(&self, element: &gst::Element) {
+    fn unprepare(&self, element: &super::UdpSink) {
         gst_debug!(CAT, obj: element, "Unpreparing");
 
         self.task.unprepare().unwrap();
@@ -1069,14 +1069,14 @@ impl UdpSink {
         gst_debug!(CAT, obj: element, "Unprepared");
     }
 
-    fn stop(&self, element: &gst::Element) -> Result<(), gst::ErrorMessage> {
+    fn stop(&self, element: &super::UdpSink) -> Result<(), gst::ErrorMessage> {
         gst_debug!(CAT, obj: element, "Stopping");
         self.task.stop()?;
         gst_debug!(CAT, obj: element, "Stopped");
         Ok(())
     }
 
-    fn start(&self, element: &gst::Element) -> Result<(), gst::ErrorMessage> {
+    fn start(&self, element: &super::UdpSink) -> Result<(), gst::ErrorMessage> {
         gst_debug!(CAT, obj: element, "Starting");
         self.task.start()?;
         gst_debug!(CAT, obj: element, "Started");
@@ -1101,7 +1101,7 @@ impl UdpSink {
     }
 }
 
-fn try_into_socket_addr(element: &gst::Element, host: &str, port: i32) -> Result<SocketAddr, ()> {
+fn try_into_socket_addr(element: &super::UdpSink, host: &str, port: i32) -> Result<SocketAddr, ()> {
     let addr: IpAddr = match host.parse() {
         Err(err) => {
             gst_error!(CAT, obj: element, "Failed to parse host {}: {}", host, err);
@@ -1123,13 +1123,14 @@ fn try_into_socket_addr(element: &gst::Element, host: &str, port: i32) -> Result
 
 impl ObjectSubclass for UdpSink {
     const NAME: &'static str = "RsTsUdpSink";
+    type Type = super::UdpSink;
     type ParentType = gst::Element;
     type Instance = gst::subclass::ElementInstanceStruct<Self>;
     type Class = subclass::simple::ClassStruct<Self>;
 
     glib_object_subclass!();
 
-    fn class_init(klass: &mut subclass::simple::ClassStruct<Self>) {
+    fn class_init(klass: &mut Self::Class) {
         klass.set_metadata(
             "Thread-sharing UDP sink",
             "Sink/Network",
@@ -1154,7 +1155,7 @@ impl ObjectSubclass for UdpSink {
             glib::types::Type::Unit,
             |_, args| {
                 let element = args[0]
-                    .get::<gst::Element>()
+                    .get::<super::UdpSink>()
                     .expect("signal arg")
                     .expect("missing signal arg");
                 let host = args[1]
@@ -1182,7 +1183,7 @@ impl ObjectSubclass for UdpSink {
             glib::types::Type::Unit,
             |_, args| {
                 let element = args[0]
-                    .get::<gst::Element>()
+                    .get::<super::UdpSink>()
                     .expect("signal arg")
                     .expect("missing signal arg");
                 let host = args[1]
@@ -1211,7 +1212,7 @@ impl ObjectSubclass for UdpSink {
             glib::types::Type::Unit,
             |_, args| {
                 let element = args[0]
-                    .get::<gst::Element>()
+                    .get::<super::UdpSink>()
                     .expect("signal arg")
                     .expect("missing signal arg");
 
@@ -1225,7 +1226,7 @@ impl ObjectSubclass for UdpSink {
         klass.install_properties(&PROPERTIES);
     }
 
-    fn with_class(klass: &subclass::simple::ClassStruct<Self>) -> Self {
+    fn with_class(klass: &Self::Class) -> Self {
         let settings = Arc::new(StdMutex::new(Settings::default()));
         let sink_pad_handler = UdpSinkPadHandler::new(Arc::clone(&settings));
 
@@ -1242,9 +1243,8 @@ impl ObjectSubclass for UdpSink {
 }
 
 impl ObjectImpl for UdpSink {
-    fn set_property(&self, obj: &glib::Object, id: usize, value: &glib::Value) {
+    fn set_property(&self, obj: &Self::Type, id: usize, value: &glib::Value) {
         let prop = &PROPERTIES[id];
-        let element = obj.downcast_ref::<gst::Element>().unwrap();
 
         let mut settings = self.settings.lock().unwrap();
         match *prop {
@@ -1315,15 +1315,9 @@ impl ObjectImpl for UdpSink {
                         rsplit[0]
                             .parse::<i32>()
                             .map_err(|err| {
-                                gst_error!(
+                                gst_error!(CAT, obj: obj, "Invalid port {}: {}", rsplit[0], err);
-                                    CAT,
-                                    obj: element,
-                                    "Invalid port {}: {}",
-                                    rsplit[0],
-                                    err
-                                );
                             })
-                            .and_then(|port| try_into_socket_addr(&element, rsplit[1], port))
+                            .and_then(|port| try_into_socket_addr(&obj, rsplit[1], port))
                             .ok()
                     } else {
                         None
@@ -1346,7 +1340,7 @@ impl ObjectImpl for UdpSink {
         }
     }
 
-    fn get_property(&self, _obj: &glib::Object, id: usize) -> Result<glib::Value, ()> {
+    fn get_property(&self, _obj: &Self::Type, id: usize) -> Result<glib::Value, ()> {
         let prop = &PROPERTIES[id];
 
         let settings = self.settings.lock().unwrap();
@@ -1399,20 +1393,19 @@ impl ObjectImpl for UdpSink {
         }
     }
 
-    fn constructed(&self, obj: &glib::Object) {
+    fn constructed(&self, obj: &Self::Type) {
         self.parent_constructed(obj);
 
-        let element = obj.downcast_ref::<gst::Element>().unwrap();
+        obj.add_pad(self.sink_pad.gst_pad()).unwrap();
-        element.add_pad(self.sink_pad.gst_pad()).unwrap();
 
-        super::set_element_flags(element, gst::ElementFlags::SINK);
+        crate::set_element_flags(obj, gst::ElementFlags::SINK);
     }
 }
 
 impl ElementImpl for UdpSink {
     fn change_state(
         &self,
-        element: &gst::Element,
+        element: &Self::Type,
         transition: gst::StateChange,
     ) -> Result<gst::StateChangeSuccess, gst::StateChangeError> {
         gst_trace!(CAT, obj: element, "Changing state {:?}", transition);
@@ -1439,7 +1432,7 @@ impl ElementImpl for UdpSink {
         self.parent_change_state(element, transition)
     }
 
-    fn send_event(&self, _element: &gst::Element, event: gst::Event) -> bool {
+    fn send_event(&self, _element: &Self::Type, event: gst::Event) -> bool {
         match event.view() {
             EventView::Latency(ev) => {
                 self.sink_pad_handler.set_latency(ev.get_latency());
@@ -1450,12 +1443,3 @@ impl ElementImpl for UdpSink {
         }
     }
 }
-
-pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
-    gst::Element::register(
-        Some(plugin),
-        "ts-udpsink",
-        gst::Rank::None,
-        UdpSink::get_type(),
-    )
-}

generic/threadshare/src/udpsink/mod.rs

@@ -0,0 +1,39 @@
+// Copyright (C) 2019 Mathieu Duponchelle <mathieu@centricular.com>
+//
+// This library is free software; you can redistribute it and/or
+// modify it under the terms of the GNU Library General Public
+// License as published by the Free Software Foundation; either
+// version 2 of the License, or (at your option) any later version.
+//
+// This library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+// Library General Public License for more details.
+//
+// You should have received a copy of the GNU Library General Public
+// License along with this library; if not, write to the
+// Free Software Foundation, Inc., 51 Franklin Street, Suite 500,
+// Boston, MA 02110-1335, USA.
+
+use glib::glib_wrapper;
+use glib::prelude::*;
+
+mod imp;
+
+glib_wrapper! {
+    pub struct UdpSink(ObjectSubclass<imp::UdpSink>) @extends gst::Element, gst::Object;
+}
+
+// GStreamer elements need to be thread-safe. For the private implementation this is automatically
+// enforced but for the public wrapper type we need to specify this manually.
+unsafe impl Send for UdpSink {}
+unsafe impl Sync for UdpSink {}
+
+pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
+    gst::Element::register(
+        Some(plugin),
+        "ts-udpsink",
+        gst::Rank::None,
+        UdpSink::static_type(),
+    )
+}

generic/threadshare/src/udpsrc/imp.rs

@@ -41,7 +41,7 @@ use std::u16;
 use crate::runtime::prelude::*;
 use crate::runtime::{Context, PadSrc, PadSrcRef, PadSrcWeak, Task};
 
-use super::socket::{wrap_socket, GioSocketWrapper, Socket, SocketError, SocketRead};
+use crate::socket::{wrap_socket, GioSocketWrapper, Socket, SocketError, SocketRead};
 
 const DEFAULT_ADDRESS: Option<&str> = Some("0.0.0.0");
 const DEFAULT_PORT: i32 = 5000;
@@ -185,7 +185,7 @@ static PROPERTIES: [subclass::Property; 10] = [
 ];
 
 #[derive(Debug)]
-pub struct UdpReader(tokio::net::UdpSocket);
+struct UdpReader(tokio::net::UdpSocket);
 
 impl UdpReader {
     fn new(socket: tokio::net::UdpSocket) -> Self {
@@ -254,7 +254,7 @@ impl UdpSrcPadHandler {
         self.0.state.lock().await.need_segment = true;
     }
 
-    async fn push_prelude(&self, pad: &PadSrcRef<'_>, _element: &gst::Element) {
+    async fn push_prelude(&self, pad: &PadSrcRef<'_>, _element: &super::UdpSrc) {
         let mut state = self.0.state.lock().await;
         if state.need_initial_events {
             gst_debug!(CAT, obj: pad.gst_pad(), "Pushing initial events");
@@ -285,7 +285,7 @@ impl UdpSrcPadHandler {
     async fn push_buffer(
         &self,
         pad: &PadSrcRef<'_>,
-        element: &gst::Element,
+        element: &super::UdpSrc,
         buffer: gst::Buffer,
     ) -> Result<gst::FlowSuccess, gst::FlowError> {
         gst_log!(CAT, obj: pad.gst_pad(), "Handling {:?}", buffer);
@@ -377,7 +377,7 @@ impl PadSrcHandler for UdpSrcPadHandler {
 }
 
 struct UdpSrcTask {
-    element: gst::Element,
+    element: super::UdpSrc,
     src_pad: PadSrcWeak,
     src_pad_handler: UdpSrcPadHandler,
     socket: Socket<UdpReader>,
@@ -385,7 +385,7 @@ struct UdpSrcTask {
 
 impl UdpSrcTask {
     fn new(
-        element: &gst::Element,
+        element: &super::UdpSrc,
         src_pad: &PadSrc,
         src_pad_handler: &UdpSrcPadHandler,
         socket: Socket<UdpReader>,
@@ -510,7 +510,7 @@ impl TaskImpl for UdpSrcTask {
     }
 }
 
-struct UdpSrc {
+pub struct UdpSrc {
     src_pad: PadSrc,
     src_pad_handler: UdpSrcPadHandler,
     task: Task,
@@ -526,7 +526,7 @@ lazy_static! {
 }
 
 impl UdpSrc {
-    fn prepare(&self, element: &gst::Element) -> Result<(), gst::ErrorMessage> {
+    fn prepare(&self, element: &super::UdpSrc) -> Result<(), gst::ErrorMessage> {
         let mut settings_guard = self.settings.lock().unwrap();
 
         gst_debug!(CAT, obj: element, "Preparing");
@@ -719,13 +719,17 @@ impl UdpSrc {
             )
         })?;
 
-        let socket = Socket::try_new(element.clone(), buffer_pool, UdpReader::new(socket))
+        let socket = Socket::try_new(
-            .map_err(|err| {
+            element.clone().upcast(),
-                gst_error_msg!(
+            buffer_pool,
-                    gst::ResourceError::OpenRead,
+            UdpReader::new(socket),
-                    ["Failed to prepare socket {:?}", err]
+        )
-                )
+        .map_err(|err| {
-            })?;
+            gst_error_msg!(
+                gst::ResourceError::OpenRead,
+                ["Failed to prepare socket {:?}", err]
+            )
+        })?;
 
         element.notify("used-socket");
 
@@ -749,7 +753,7 @@ impl UdpSrc {
         Ok(())
     }
 
-    fn unprepare(&self, element: &gst::Element) {
+    fn unprepare(&self, element: &super::UdpSrc) {
         gst_debug!(CAT, obj: element, "Unpreparing");
 
         self.settings.lock().unwrap().used_socket = None;
@@ -760,21 +764,21 @@ impl UdpSrc {
         gst_debug!(CAT, obj: element, "Unprepared");
     }
 
-    fn stop(&self, element: &gst::Element) -> Result<(), gst::ErrorMessage> {
+    fn stop(&self, element: &super::UdpSrc) -> Result<(), gst::ErrorMessage> {
         gst_debug!(CAT, obj: element, "Stopping");
         self.task.stop()?;
         gst_debug!(CAT, obj: element, "Stopped");
         Ok(())
     }
 
-    fn start(&self, element: &gst::Element) -> Result<(), gst::ErrorMessage> {
+    fn start(&self, element: &super::UdpSrc) -> Result<(), gst::ErrorMessage> {
         gst_debug!(CAT, obj: element, "Starting");
         self.task.start()?;
         gst_debug!(CAT, obj: element, "Started");
         Ok(())
     }
 
-    fn pause(&self, element: &gst::Element) -> Result<(), gst::ErrorMessage> {
+    fn pause(&self, element: &super::UdpSrc) -> Result<(), gst::ErrorMessage> {
         gst_debug!(CAT, obj: element, "Pausing");
         self.task.pause()?;
         gst_debug!(CAT, obj: element, "Paused");
@@ -784,13 +788,14 @@ impl UdpSrc {
 
 impl ObjectSubclass for UdpSrc {
     const NAME: &'static str = "RsTsUdpSrc";
+    type Type = super::UdpSrc;
     type ParentType = gst::Element;
     type Instance = gst::subclass::ElementInstanceStruct<Self>;
     type Class = subclass::simple::ClassStruct<Self>;
 
     glib_object_subclass!();
 
-    fn class_init(klass: &mut subclass::simple::ClassStruct<Self>) {
+    fn class_init(klass: &mut Self::Class) {
         klass.set_metadata(
             "Thread-sharing UDP source",
             "Source/Network",
@@ -827,7 +832,7 @@ impl ObjectSubclass for UdpSrc {
         }
     }
 
-    fn with_class(klass: &subclass::simple::ClassStruct<Self>) -> Self {
+    fn with_class(klass: &Self::Class) -> Self {
         let src_pad_handler = UdpSrcPadHandler::default();
 
         Self {
@@ -843,7 +848,7 @@ impl ObjectSubclass for UdpSrc {
 }
 
 impl ObjectImpl for UdpSrc {
-    fn set_property(&self, _obj: &glib::Object, id: usize, value: &glib::Value) {
+    fn set_property(&self, _obj: &Self::Type, id: usize, value: &glib::Value) {
         let prop = &PROPERTIES[id];
 
         let mut settings = self.settings.lock().unwrap();
@@ -888,7 +893,7 @@ impl ObjectImpl for UdpSrc {
         }
     }
 
-    fn get_property(&self, _obj: &glib::Object, id: usize) -> Result<glib::Value, ()> {
+    fn get_property(&self, _obj: &Self::Type, id: usize) -> Result<glib::Value, ()> {
         let prop = &PROPERTIES[id];
 
         let settings = self.settings.lock().unwrap();
@@ -917,20 +922,19 @@ impl ObjectImpl for UdpSrc {
         }
     }
 
-    fn constructed(&self, obj: &glib::Object) {
+    fn constructed(&self, obj: &Self::Type) {
         self.parent_constructed(obj);
 
-        let element = obj.downcast_ref::<gst::Element>().unwrap();
+        obj.add_pad(self.src_pad.gst_pad()).unwrap();
-        element.add_pad(self.src_pad.gst_pad()).unwrap();
 
-        super::set_element_flags(element, gst::ElementFlags::SOURCE);
+        crate::set_element_flags(obj, gst::ElementFlags::SOURCE);
     }
 }
 
 impl ElementImpl for UdpSrc {
     fn change_state(
         &self,
-        element: &gst::Element,
+        element: &Self::Type,
         transition: gst::StateChange,
     ) -> Result<gst::StateChangeSuccess, gst::StateChangeError> {
         gst_trace!(CAT, obj: element, "Changing state {:?}", transition);
@@ -972,12 +976,3 @@ impl ElementImpl for UdpSrc {
         Ok(success)
     }
 }
-
-pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
-    gst::Element::register(
-        Some(plugin),
-        "ts-udpsrc",
-        gst::Rank::None,
-        UdpSrc::get_type(),
-    )
-}

generic/threadshare/src/udpsrc/mod.rs

@@ -0,0 +1,39 @@
+// Copyright (C) 2018 Sebastian Dröge <sebastian@centricular.com>
+//
+// This library is free software; you can redistribute it and/or
+// modify it under the terms of the GNU Library General Public
+// License as published by the Free Software Foundation; either
+// version 2 of the License, or (at your option) any later version.
+//
+// This library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+// Library General Public License for more details.
+//
+// You should have received a copy of the GNU Library General Public
+// License along with this library; if not, write to the
+// Free Software Foundation, Inc., 51 Franklin Street, Suite 500,
+// Boston, MA 02110-1335, USA.
+
+use glib::glib_wrapper;
+use glib::prelude::*;
+
+mod imp;
+
+glib_wrapper! {
+    pub struct UdpSrc(ObjectSubclass<imp::UdpSrc>) @extends gst::Element, gst::Object;
+}
+
+// GStreamer elements need to be thread-safe. For the private implementation this is automatically
+// enforced but for the public wrapper type we need to specify this manually.
+unsafe impl Send for UdpSrc {}
+unsafe impl Sync for UdpSrc {}
+
+pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
+    gst::Element::register(
+        Some(plugin),
+        "ts-udpsrc",
+        gst::Rank::None,
+        UdpSrc::static_type(),
+    )
+}