Port Source over to the new infrastructure

gst-plugin-file/src/filesrc.rs

@@ -30,7 +30,7 @@ pub struct FileSrc {
 }
 
 impl FileSrc {
-    pub fn new(_src: &RsSrcWrapper) -> FileSrc {
+    pub fn new(_src: &RsBaseSrc) -> FileSrc {
         FileSrc {
             streaming_state: StreamingState::Stopped,
             cat: gst::DebugCategory::new(
@@ -41,7 +41,7 @@ impl FileSrc {
         }
     }
 
-    pub fn new_boxed(src: &RsSrcWrapper) -> Box<Source> {
+    pub fn new_boxed(src: &RsBaseSrc) -> Box<SourceImpl> {
         Box::new(FileSrc::new(src))
     }
 }
@@ -56,16 +56,16 @@ fn validate_uri(uri: &Url) -> Result<(), UriError> {
     Ok(())
 }
 
-impl Source for FileSrc {
+impl SourceImpl for FileSrc {
     fn uri_validator(&self) -> Box<UriValidator> {
         Box::new(validate_uri)
     }
 
-    fn is_seekable(&self, _src: &RsSrcWrapper) -> bool {
+    fn is_seekable(&self, _src: &RsBaseSrc) -> bool {
         true
     }
 
-    fn get_size(&self, _src: &RsSrcWrapper) -> Option<u64> {
+    fn get_size(&self, _src: &RsBaseSrc) -> Option<u64> {
         if let StreamingState::Started { ref file, .. } = self.streaming_state {
             file.metadata().ok().map(|m| m.len())
         } else {
@@ -73,7 +73,7 @@ impl Source for FileSrc {
         }
     }
 
-    fn start(&mut self, src: &RsSrcWrapper, uri: Url) -> Result<(), ErrorMessage> {
+    fn start(&mut self, src: &RsBaseSrc, uri: Url) -> Result<(), ErrorMessage> {
         if let StreamingState::Started { .. } = self.streaming_state {
             return Err(error_msg!(
                 gst::LibraryError::Failed,
@@ -121,7 +121,7 @@ impl Source for FileSrc {
         Ok(())
     }
 
-    fn stop(&mut self, _src: &RsSrcWrapper) -> Result<(), ErrorMessage> {
+    fn stop(&mut self, _src: &RsBaseSrc) -> Result<(), ErrorMessage> {
         self.streaming_state = StreamingState::Stopped;
 
         Ok(())
@@ -129,7 +129,7 @@ impl Source for FileSrc {
 
     fn fill(
         &mut self,
-        src: &RsSrcWrapper,
+        src: &RsBaseSrc,
         offset: u64,
         _: u32,
         buffer: &mut gst::BufferRef,
@@ -189,7 +189,7 @@ impl Source for FileSrc {
         Ok(())
     }
 
-    fn seek(&mut self, _src: &RsSrcWrapper, _: u64, _: Option<u64>) -> Result<(), ErrorMessage> {
+    fn seek(&mut self, _src: &RsBaseSrc, _: u64, _: Option<u64>) -> Result<(), ErrorMessage> {
         Ok(())
     }
 }

gst-plugin-http/src/httpsrc.rs

@@ -41,7 +41,7 @@ pub struct HttpSrc {
 }
 
 impl HttpSrc {
-    pub fn new(_src: &RsSrcWrapper) -> HttpSrc {
+    pub fn new(_src: &RsBaseSrc) -> HttpSrc {
         HttpSrc {
             streaming_state: StreamingState::Stopped,
             cat: gst::DebugCategory::new(
@@ -53,13 +53,13 @@ impl HttpSrc {
         }
     }
 
-    pub fn new_boxed(src: &RsSrcWrapper) -> Box<Source> {
+    pub fn new_boxed(src: &RsBaseSrc) -> Box<SourceImpl> {
         Box::new(HttpSrc::new(src))
     }
 
     fn do_request(
         &self,
-        src: &RsSrcWrapper,
+        src: &RsBaseSrc,
         uri: Url,
         start: u64,
         stop: Option<u64>,
@@ -152,44 +152,39 @@ fn validate_uri(uri: &Url) -> Result<(), UriError> {
     Ok(())
 }
 
-impl Source for HttpSrc {
+impl SourceImpl for HttpSrc {
     fn uri_validator(&self) -> Box<UriValidator> {
         Box::new(validate_uri)
     }
 
-    fn is_seekable(&self, _src: &RsSrcWrapper) -> bool {
+    fn is_seekable(&self, _src: &RsBaseSrc) -> bool {
         match self.streaming_state {
             StreamingState::Started { seekable, .. } => seekable,
             _ => false,
         }
     }
 
-    fn get_size(&self, _src: &RsSrcWrapper) -> Option<u64> {
+    fn get_size(&self, _src: &RsBaseSrc) -> Option<u64> {
         match self.streaming_state {
             StreamingState::Started { size, .. } => size,
             _ => None,
         }
     }
 
-    fn start(&mut self, src: &RsSrcWrapper, uri: Url) -> Result<(), ErrorMessage> {
+    fn start(&mut self, src: &RsBaseSrc, uri: Url) -> Result<(), ErrorMessage> {
         self.streaming_state = StreamingState::Stopped;
         self.streaming_state = try!(self.do_request(src, uri, 0, None));
 
         Ok(())
     }
 
-    fn stop(&mut self, _src: &RsSrcWrapper) -> Result<(), ErrorMessage> {
+    fn stop(&mut self, _src: &RsBaseSrc) -> Result<(), ErrorMessage> {
         self.streaming_state = StreamingState::Stopped;
 
         Ok(())
     }
 
-    fn seek(
+    fn seek(&mut self, src: &RsBaseSrc, start: u64, stop: Option<u64>) -> Result<(), ErrorMessage> {
-        &mut self,
-        src: &RsSrcWrapper,
-        start: u64,
-        stop: Option<u64>,
-    ) -> Result<(), ErrorMessage> {
         let (position, old_stop, uri) = match self.streaming_state {
             StreamingState::Started {
                 position,
@@ -214,7 +209,7 @@ impl Source for HttpSrc {
 
     fn fill(
         &mut self,
-        src: &RsSrcWrapper,
+        src: &RsBaseSrc,
         offset: u64,
         _: u32,
         buffer: &mut gst::BufferRef,

gst-plugin/src/base_src.rs

@@ -27,8 +27,12 @@ use element::*;
 
 pub trait BaseSrcImpl
     : mopa::Any + ObjectImpl + ElementImpl + Send + Sync + 'static {
-    fn start(&self, element: &gst_base::BaseSrc) -> bool;
+    fn start(&self, _element: &gst_base::BaseSrc) -> bool {
-    fn stop(&self, element: &gst_base::BaseSrc) -> bool;
+        true
+    }
+    fn stop(&self, _element: &gst_base::BaseSrc) -> bool {
+        true
+    }
     fn is_seekable(&self, _element: &gst_base::BaseSrc) -> bool {
         false
     }
@@ -216,9 +220,9 @@ impl ObjectImpl for Box<BaseSrcImpl> {
         imp.set_property(obj, id, value);
     }
 
-    fn get_property(&self, obj: &glib::Object, id: u32, value: &mut glib::Value) {
+    fn get_property(&self, obj: &glib::Object, id: u32) -> Result<glib::Value, ()> {
         let imp: &BaseSrcImpl = self.as_ref();
-        imp.get_property(obj, id, value);
+        imp.get_property(obj, id)
     }
 }
 

gst-plugin/src/element.rs

@@ -127,9 +127,9 @@ impl ObjectImpl for Box<ElementImpl> {
         imp.set_property(obj, id, value);
     }
 
-    fn get_property(&self, obj: &glib::Object, id: u32, value: &mut glib::Value) {
+    fn get_property(&self, obj: &glib::Object, id: u32) -> Result<glib::Value, ()> {
         let imp: &ElementImpl = self.as_ref();
-        imp.get_property(obj, id, value);
+        imp.get_property(obj, id)
     }
 }
 

gst-plugin/src/error.rs

@@ -110,8 +110,8 @@ impl ErrorMessage {
                 gst_ffi::GST_MESSAGE_ERROR,
                 error_domain,
                 error_code,
-                message.to_glib_none().0,
+                message.to_glib_full(),
-                debug.to_glib_none().0,
+                debug.to_glib_full(),
                 filename.to_glib_none().0,
                 function.to_glib_none().0,
                 line as i32,
@@ -215,7 +215,7 @@ impl Error for UriError {
     }
 }
 
-pub type UriValidator = Fn(&Url) -> Result<(), UriError>;
+pub type UriValidator = Fn(&Url) -> Result<(), UriError> + Send + Sync + 'static;
 
 #[macro_export]
 macro_rules! panic_to_error(

gst-plugin/src/object.rs

@@ -22,21 +22,13 @@ pub trait ObjectImpl: Send + Sync + 'static {
         unimplemented!()
     }
 
-    fn get_property(&self, _obj: &glib::Object, _id: u32, _value: &mut glib::Value) {
+    fn get_property(&self, _obj: &glib::Object, _id: u32) -> Result<glib::Value, ()> {
         unimplemented!()
     }
 
-    fn notify(&self, obj: &glib::Object, id: u32) {
+    fn notify(&self, obj: &glib::Object, name: &str) {
         unsafe {
-            let stash = obj.to_glib_none();
+            gobject_ffi::g_object_notify(obj.to_glib_none().0, name.to_glib_none().0);
-            let ptr: *mut gobject_ffi::GObject = stash.0;
-            let klass = *(ptr as *const *const gobject_ffi::GObjectClass);
-            let mut n_pspecs = 0;
-            let pspecs =
-                gobject_ffi::g_object_class_list_properties(klass as *mut _, &mut n_pspecs);
-
-            assert!(n_pspecs > id);
-            gobject_ffi::g_object_notify_by_pspec(ptr, *pspecs.offset(id as isize));
         }
     }
 }
@@ -68,7 +60,7 @@ where
         unimplemented!()
     }
 
-    fn get_property(_obj: &Self::RsType, _id: u32, _value: &mut glib::Value) {
+    fn get_property(_obj: &Self::RsType, _id: u32) -> Result<glib::Value, ()> {
         unimplemented!()
     }
 }
@@ -357,11 +349,14 @@ unsafe extern "C" fn get_property<T: ObjectType>(
     _pspec: *mut gobject_ffi::GParamSpec,
 ) {
     callback_guard!();
-    T::get_property(
+    match T::get_property(&from_glib_borrow(obj as *mut InstanceStruct<T>), id - 1) {
-        &from_glib_borrow(obj as *mut InstanceStruct<T>),
+        Ok(v) => {
-        id,
+            gobject_ffi::g_value_unset(value);
-        &mut *(value as *mut glib::Value),
+            ptr::write(value, ptr::read(v.to_glib_none().0));
-    );
+            mem::forget(v);
+        }
+        Err(_) => unimplemented!(),
+    }
 }
 
 unsafe extern "C" fn set_property<T: ObjectType>(
@@ -373,7 +368,7 @@ unsafe extern "C" fn set_property<T: ObjectType>(
     callback_guard!();
     T::set_property(
         &from_glib_borrow(obj as *mut InstanceStruct<T>),
-        id,
+        id - 1,
         &*(value as *mut glib::Value),
     );
 }
@@ -452,11 +447,15 @@ unsafe extern "C" fn sub_get_property<T: ObjectType>(
     callback_guard!();
     let instance = &*(obj as *mut InstanceStruct<T>);
     let imp = instance.get_impl();
-    imp.get_property(
+
-        &from_glib_borrow(obj),
+    match imp.get_property(&from_glib_borrow(obj), id - 1) {
-        id,
+        Ok(v) => {
-        &mut *(value as *mut glib::Value),
+            gobject_ffi::g_value_unset(value);
-    );
+            ptr::write(value, ptr::read(v.to_glib_none().0));
+            mem::forget(v);
+        }
+        Err(_) => unimplemented!(),
+    }
 }
 
 unsafe extern "C" fn sub_set_property<T: ObjectType>(
@@ -468,7 +467,11 @@ unsafe extern "C" fn sub_set_property<T: ObjectType>(
     callback_guard!();
     let instance = &*(obj as *mut InstanceStruct<T>);
     let imp = instance.get_impl();
-    imp.set_property(&from_glib_borrow(obj), id, &*(value as *mut glib::Value));
+    imp.set_property(
+        &from_glib_borrow(obj),
+        id - 1,
+        &*(value as *mut glib::Value),
+    );
 }
 
 unsafe extern "C" fn sub_init<T: ObjectType>(
@@ -476,6 +479,8 @@ unsafe extern "C" fn sub_init<T: ObjectType>(
     _klass: glib_ffi::gpointer,
 ) {
     callback_guard!();
+    // Get rid of floating reference, if any
+    gobject_ffi::g_object_ref_sink(obj as *mut gobject_ffi::GObject);
     let instance = &mut *(obj as *mut InstanceStruct<T>);
     let klass = &**(obj as *const *const ClassStruct<T>);
     let rs_instance: T::RsType = from_glib_borrow(obj as *mut InstanceStruct<T>);

gst-plugin/src/source.rs

@@ -6,126 +6,176 @@
 // option. This file may not be copied, modified, or distributed
 // except according to those terms.
 
-use libc::c_char;
-use std::ffi::{CStr, CString};
-use std::ptr;
-use std::mem;
 use std::u64;
 
-use std::panic::{self, AssertUnwindSafe};
-
 use std::sync::Mutex;
-use std::sync::atomic::{AtomicBool, Ordering};
 
 use url::Url;
 
 use error::*;
 
-use glib_ffi;
+use glib;
-use gobject_ffi;
-use gst_ffi;
-use gst_base_ffi;
-
-use glib::translate::*;
 use gst;
 use gst::prelude::*;
 use gst_base;
 use gst_base::prelude::*;
 
-pub struct SourceWrapper {
+use object::*;
-    cat: gst::DebugCategory,
+use element::*;
-    uri: Mutex<(Option<Url>, bool)>,
+use base_src::*;
-    uri_validator: Box<UriValidator>,
+use uri_handler::*;
-    source: Mutex<Box<Source>>,
-    panicked: AtomicBool,
-}
 
-pub trait Source: Send + 'static {
+pub use base_src::RsBaseSrc;
+
+pub trait SourceImpl: Send + 'static {
     fn uri_validator(&self) -> Box<UriValidator>;
 
-    fn is_seekable(&self, src: &RsSrcWrapper) -> bool;
+    fn is_seekable(&self, src: &RsBaseSrc) -> bool;
-    fn get_size(&self, src: &RsSrcWrapper) -> Option<u64>;
+    fn get_size(&self, src: &RsBaseSrc) -> Option<u64>;
 
-    fn start(&mut self, src: &RsSrcWrapper, uri: Url) -> Result<(), ErrorMessage>;
+    fn start(&mut self, src: &RsBaseSrc, uri: Url) -> Result<(), ErrorMessage>;
-    fn stop(&mut self, src: &RsSrcWrapper) -> Result<(), ErrorMessage>;
+    fn stop(&mut self, src: &RsBaseSrc) -> Result<(), ErrorMessage>;
     fn fill(
         &mut self,
-        src: &RsSrcWrapper,
+        src: &RsBaseSrc,
         offset: u64,
         length: u32,
         buffer: &mut gst::BufferRef,
     ) -> Result<(), FlowError>;
-    fn seek(
+    fn seek(&mut self, src: &RsBaseSrc, start: u64, stop: Option<u64>) -> Result<(), ErrorMessage>;
-        &mut self,
-        src: &RsSrcWrapper,
-        start: u64,
-        stop: Option<u64>,
-    ) -> Result<(), ErrorMessage>;
 }
 
-impl SourceWrapper {
+struct Source {
-    fn new(source: Box<Source>) -> SourceWrapper {
+    cat: gst::DebugCategory,
-        SourceWrapper {
+    uri: Mutex<(Option<Url>, bool)>,
+    uri_validator: Box<UriValidator>,
+    imp: Mutex<Box<SourceImpl>>,
+    push_only: bool,
+}
+
+static PROPERTIES: [Property; 1] = [
+    Property::String(
+        "uri",
+        "URI",
+        "URI to read from",
+        None,
+        PropertyMutability::ReadWrite,
+    ),
+];
+
+impl Source {
+    fn new(source: &RsBaseSrc, source_info: &SourceInfo) -> Self {
+        let source_impl = (source_info.create_instance)(source);
+
+        Self {
             cat: gst::DebugCategory::new(
-                "rssrc",
+                "rssource",
                 gst::DebugColorFlags::empty(),
                 "Rust source base class",
             ),
             uri: Mutex::new((None, false)),
-            uri_validator: source.uri_validator(),
+            uri_validator: source_impl.uri_validator(),
-            source: Mutex::new(source),
+            imp: Mutex::new(source_impl),
-            panicked: AtomicBool::new(false),
+            push_only: source_info.push_only,
         }
     }
 
-    fn set_uri(&self, src: &RsSrcWrapper, uri_str: Option<&str>) -> Result<(), UriError> {
+    fn class_init(klass: &mut RsBaseSrcClass, source_info: &SourceInfo) {
+        klass.set_metadata(
+            &source_info.long_name,
+            &source_info.classification,
+            &source_info.description,
+            &source_info.author,
+        );
+
+        let caps = gst::Caps::new_any();
+        let pad_template = gst::PadTemplate::new(
+            "src",
+            gst::PadDirection::Src,
+            gst::PadPresence::Always,
+            &caps,
+        );
+        klass.add_pad_template(pad_template);
+
+        klass.install_properties(&PROPERTIES);
+    }
+
+    fn init(element: &RsBaseSrc, source_info: &SourceInfo) -> Box<BaseSrcImpl> {
+        element.set_blocksize(4096);
+
+        let imp = Self::new(element, source_info);
+        Box::new(imp)
+    }
+
+    fn get_uri(&self, _element: &glib::Object) -> Option<String> {
+        let uri_storage = &self.uri.lock().unwrap();
+        uri_storage.0.as_ref().map(|uri| String::from(uri.as_str()))
+    }
+
+    fn set_uri(&self, element: &glib::Object, uri_str: Option<String>) -> Result<(), glib::Error> {
+        let src = element.clone().dynamic_cast::<RsBaseSrc>().unwrap();
+
         let uri_storage = &mut self.uri.lock().unwrap();
 
-        gst_debug!(self.cat, obj: src, "Setting URI {:?}", uri_str);
+        gst_debug!(self.cat, obj: &src, "Setting URI {:?}", uri_str);
 
         if uri_storage.1 {
-            return Err(UriError::new(
+            return Err(
-                gst::URIError::BadState,
+                UriError::new(gst::URIError::BadState, "Already started".to_string()).into_error(),
-                "Already started".to_string(),
+            );
-            ));
         }
 
         uri_storage.0 = None;
 
         if let Some(uri_str) = uri_str {
-            match Url::parse(uri_str) {
+            match Url::parse(uri_str.as_str()) {
                 Ok(uri) => {
-                    try!((self.uri_validator)(&uri));
+                    try!((self.uri_validator)(&uri).map_err(|e| e.into_error()));
                     uri_storage.0 = Some(uri);
                     Ok(())
                 }
-                Err(err) => Err(UriError::new(
+                Err(err) => Err(
-                    gst::URIError::BadUri,
+                    UriError::new(
-                    format!("Failed to parse URI '{}': {}", uri_str, err),
+                        gst::URIError::BadUri,
-                )),
+                        format!("Failed to parse URI '{}': {}", uri_str, err),
+                    ).into_error(),
+                ),
             }
         } else {
             Ok(())
         }
     }
+}
 
-    fn get_uri(&self, _src: &RsSrcWrapper) -> Option<String> {
+impl ObjectImpl for Source {
-        let uri_storage = &self.uri.lock().unwrap();
+    fn set_property(&self, obj: &glib::Object, id: u32, value: &glib::Value) {
-        uri_storage.0.as_ref().map(|uri| String::from(uri.as_str()))
+        let prop = &PROPERTIES[id as usize];
+
+        match *prop {
+            Property::String("uri", ..) => {
+                self.set_uri(obj, value.get()).unwrap();
+            }
+            _ => unimplemented!(),
+        }
     }
 
-    fn is_seekable(&self, src: &RsSrcWrapper) -> bool {
+    fn get_property(&self, obj: &glib::Object, id: u32) -> Result<glib::Value, ()> {
-        let source_impl = &self.source.lock().unwrap();
+        let prop = &PROPERTIES[id as usize];
-        source_impl.is_seekable(src)
-    }
 
-    fn get_size(&self, src: &RsSrcWrapper) -> u64 {
+        match *prop {
-        let source_impl = &self.source.lock().unwrap();
+            Property::String("uri", ..) => Ok(self.get_uri(obj).to_value()),
-        source_impl.get_size(src).unwrap_or(u64::MAX)
+            _ => unimplemented!(),
+        }
     }
+}
 
-    fn start(&self, src: &RsSrcWrapper) -> bool {
+impl ElementImpl for Source {}
-        gst_debug!(self.cat, obj: src, "Starting");
+
+impl BaseSrcImpl for Source {
+    fn start(&self, element: &gst_base::BaseSrc) -> bool {
+        let src = element.clone().downcast::<RsBaseSrc>().unwrap();
+
+        gst_debug!(self.cat, obj: &src, "Starting");
 
         // Don't keep the URI locked while we call start later
         let uri = match *self.uri.lock().unwrap() {
@@ -134,76 +184,90 @@ impl SourceWrapper {
                 uri.clone()
             }
             (None, _) => {
-                gst_error!(self.cat, obj: src, "No URI given");
+                gst_error!(self.cat, obj: &src, "No URI given");
-                self.post_message(
+                error_msg!(gst::ResourceError::OpenRead, ["No URI given"]).post(&src);
-                    src,
-                    &error_msg!(gst::ResourceError::OpenRead, ["No URI given"]),
-                );
                 return false;
             }
         };
 
-        let source_impl = &mut self.source.lock().unwrap();
+        let source_impl = &mut self.imp.lock().unwrap();
-        match source_impl.start(src, uri) {
+        match source_impl.start(&src, uri) {
             Ok(..) => {
-                gst_trace!(self.cat, obj: src, "Started successfully");
+                gst_trace!(self.cat, obj: &src, "Started successfully");
                 true
             }
             Err(ref msg) => {
-                gst_error!(self.cat, obj: src, "Failed to start: {:?}", msg);
+                gst_error!(self.cat, obj: &src, "Failed to start: {:?}", msg);
 
                 self.uri.lock().unwrap().1 = false;
-                self.post_message(src, msg);
+                msg.post(&src);
                 false
             }
         }
     }
 
-    fn stop(&self, src: &RsSrcWrapper) -> bool {
+    fn stop(&self, element: &gst_base::BaseSrc) -> bool {
-        let source_impl = &mut self.source.lock().unwrap();
+        let src = element.clone().downcast::<RsBaseSrc>().unwrap();
+        let source_impl = &mut self.imp.lock().unwrap();
 
-        gst_debug!(self.cat, obj: src, "Stopping");
+        gst_debug!(self.cat, obj: &src, "Stopping");
 
-        match source_impl.stop(src) {
+        match source_impl.stop(&src) {
             Ok(..) => {
-                gst_trace!(self.cat, obj: src, "Stopped successfully");
+                gst_trace!(self.cat, obj: &src, "Stopped successfully");
                 self.uri.lock().unwrap().1 = false;
                 true
             }
             Err(ref msg) => {
-                gst_error!(self.cat, obj: src, "Failed to stop: {:?}", msg);
+                gst_error!(self.cat, obj: &src, "Failed to stop: {:?}", msg);
 
-                self.post_message(src, msg);
+                msg.post(&src);
                 false
             }
         }
     }
 
+    fn query(&self, element: &gst_base::BaseSrc, query: &mut gst::QueryRef) -> bool {
+        use gst::QueryView;
+
+        match query.view_mut() {
+            QueryView::Scheduling(ref mut q) if self.push_only => {
+                q.set(gst::SCHEDULING_FLAG_SEQUENTIAL, 1, -1, 0);
+                q.add_scheduling_modes(&[gst::PadMode::Push]);
+                return true;
+            }
+            _ => (),
+        }
+
+        element.parent_query(query)
+    }
+
     fn fill(
         &self,
-        src: &RsSrcWrapper,
+        element: &gst_base::BaseSrc,
         offset: u64,
         length: u32,
         buffer: &mut gst::BufferRef,
     ) -> gst::FlowReturn {
-        let source_impl = &mut self.source.lock().unwrap();
+        let src = element.clone().downcast::<RsBaseSrc>().unwrap();
+        let source_impl = &mut self.imp.lock().unwrap();
 
         gst_trace!(
             self.cat,
-            obj: src,
+            obj: &src,
             "Filling buffer {:?} with offset {} and length {}",
             buffer,
             offset,
             length
         );
 
-        match source_impl.fill(src, offset, length, buffer) {
+        match source_impl.fill(&src, offset, length, buffer) {
             Ok(()) => gst::FlowReturn::Ok,
             Err(flow_error) => {
-                gst_error!(self.cat, obj: src, "Failed to fill: {:?}", flow_error);
+                gst_error!(self.cat, obj: &src, "Failed to fill: {:?}", flow_error);
                 match flow_error {
                     FlowError::NotNegotiated(ref msg) | FlowError::Error(ref msg) => {
-                        self.post_message(src, msg)
+                        msg.post(&src);
                     }
                     _ => (),
                 }
@@ -212,154 +276,49 @@ impl SourceWrapper {
         }
     }
 
-    fn seek(&self, src: &RsSrcWrapper, start: u64, stop: Option<u64>) -> bool {
+    fn do_seek(&self, element: &gst_base::BaseSrc, segment: &mut gst::Segment) -> bool {
-        let source_impl = &mut self.source.lock().unwrap();
+        let src = element.clone().downcast::<RsBaseSrc>().unwrap();
+        let source_impl = &mut self.imp.lock().unwrap();
 
-        gst_debug!(self.cat, obj: src, "Seeking to {:?}-{:?}", start, stop);
+        let start = segment.get_start();
-
+        let stop = match segment.get_stop() {
-        match source_impl.seek(src, start, stop) {
+            u64::MAX => None,
-            Ok(..) => true,
+            stop @ _ => Some(stop),
-            Err(ref msg) => {
-                gst_error!(self.cat, obj: src, "Failed to seek {:?}", msg);
-                self.post_message(src, msg);
-                false
-            }
-        }
-    }
-
-    fn post_message(&self, src: &RsSrcWrapper, msg: &ErrorMessage) {
-        msg.post(src);
-    }
-}
-
-unsafe fn source_set_uri(
-    ptr: *mut RsSrc,
-    uri_ptr: *const c_char,
-    cerr: *mut *mut glib_ffi::GError,
-) -> glib_ffi::gboolean {
-    let src: &RsSrcWrapper = &from_glib_borrow(ptr as *mut RsSrc);
-    let wrap = src.get_wrap();
-
-    panic_to_error!(wrap, src, false, {
-        let uri_str = if uri_ptr.is_null() {
-            None
-        } else {
-            Some(CStr::from_ptr(uri_ptr).to_str().unwrap())
         };
 
-        match wrap.set_uri(src, uri_str) {
+        gst_debug!(self.cat, obj: &src, "Seeking to {:?}-{:?}", start, stop);
-            Err(err) => {
+
-                gst_error!(wrap.cat, obj: src, "Failed to set URI {:?}", err);
+        match source_impl.seek(&src, start, stop) {
-                if !cerr.is_null() {
+            Ok(..) => true,
-                    let err = err.into_error();
+            Err(ref msg) => {
-                    *cerr = err.to_glib_full() as *mut _;
+                gst_error!(self.cat, obj: &src, "Failed to seek {:?}", msg);
-                }
+                msg.post(&src);
                 false
             }
-            Ok(_) => true,
         }
-    }).to_glib()
+    }
+
+    fn is_seekable(&self, element: &gst_base::BaseSrc) -> bool {
+        let src = element.clone().downcast::<RsBaseSrc>().unwrap();
+        let source_impl = &self.imp.lock().unwrap();
+        source_impl.is_seekable(&src)
+    }
+
+    fn get_size(&self, element: &gst_base::BaseSrc) -> Option<u64> {
+        let src = element.clone().downcast::<RsBaseSrc>().unwrap();
+        let source_impl = &self.imp.lock().unwrap();
+        source_impl.get_size(&src)
+    }
 }
 
-unsafe fn source_get_uri(ptr: *mut RsSrc) -> *mut c_char {
+impl URIHandlerImpl for Source {
-    let src: &RsSrcWrapper = &from_glib_borrow(ptr as *mut RsSrc);
+    fn get_uri(&self, element: &gst::URIHandler) -> Option<String> {
-    let wrap = src.get_wrap();
+        Source::get_uri(self, &element.clone().upcast())
+    }
 
-    panic_to_error!(wrap, src, None, { wrap.get_uri(src) }).to_glib_full()
+    fn set_uri(&self, element: &gst::URIHandler, uri: Option<String>) -> Result<(), glib::Error> {
-}
+        Source::set_uri(self, &element.clone().upcast(), uri)
-
+    }
-unsafe extern "C" fn source_is_seekable(ptr: *mut gst_base_ffi::GstBaseSrc) -> glib_ffi::gboolean {
-    let src: &RsSrcWrapper = &from_glib_borrow(ptr as *mut RsSrc);
-    let wrap = src.get_wrap();
-
-    panic_to_error!(wrap, src, false, { wrap.is_seekable(src) }).to_glib()
-}
-
-unsafe extern "C" fn source_get_size(
-    ptr: *mut gst_base_ffi::GstBaseSrc,
-    size: *mut u64,
-) -> glib_ffi::gboolean {
-    let src: &RsSrcWrapper = &from_glib_borrow(ptr as *mut RsSrc);
-    let wrap = src.get_wrap();
-
-    panic_to_error!(wrap, src, false, {
-        *size = wrap.get_size(src);
-        true
-    }).to_glib()
-}
-
-unsafe extern "C" fn source_start(ptr: *mut gst_base_ffi::GstBaseSrc) -> glib_ffi::gboolean {
-    let src: &RsSrcWrapper = &from_glib_borrow(ptr as *mut RsSrc);
-    let wrap = src.get_wrap();
-
-    panic_to_error!(wrap, src, false, { wrap.start(src) }).to_glib()
-}
-
-unsafe extern "C" fn source_stop(ptr: *mut gst_base_ffi::GstBaseSrc) -> glib_ffi::gboolean {
-    let src: &RsSrcWrapper = &from_glib_borrow(ptr as *mut RsSrc);
-    let wrap = src.get_wrap();
-
-    panic_to_error!(wrap, src, false, { wrap.stop(src) }).to_glib()
-}
-
-unsafe extern "C" fn source_fill(
-    ptr: *mut gst_base_ffi::GstBaseSrc,
-    offset: u64,
-    length: u32,
-    buffer: *mut gst_ffi::GstBuffer,
-) -> gst_ffi::GstFlowReturn {
-    let src: &RsSrcWrapper = &from_glib_borrow(ptr as *mut RsSrc);
-    let wrap = src.get_wrap();
-    let buffer = gst::BufferRef::from_mut_ptr(buffer);
-
-    panic_to_error!(wrap, src, gst::FlowReturn::Error, {
-        wrap.fill(src, offset, length, buffer)
-    }).to_glib()
-}
-
-unsafe extern "C" fn source_seek(
-    ptr: *mut gst_base_ffi::GstBaseSrc,
-    segment: *mut gst_ffi::GstSegment,
-) -> glib_ffi::gboolean {
-    let src: &RsSrcWrapper = &from_glib_borrow(ptr as *mut RsSrc);
-    let wrap = src.get_wrap();
-
-    let start = (*segment).start;
-    let stop = (*segment).stop;
-
-    panic_to_error!(wrap, src, false, {
-        wrap.seek(src, start, if stop == u64::MAX { None } else { Some(stop) })
-    }).to_glib()
-}
-
-unsafe extern "C" fn source_query(
-    ptr: *mut gst_base_ffi::GstBaseSrc,
-    query_ptr: *mut gst_ffi::GstQuery,
-) -> glib_ffi::gboolean {
-    let src_klass = &**(ptr as *mut *mut RsSrcClass);
-    let source_info = &*src_klass.source_info;
-    let parent_klass = &*(src_klass.parent_vtable as *const gst_base_ffi::GstBaseSrcClass);
-
-    let src: &RsSrcWrapper = &from_glib_borrow(ptr as *mut RsSrc);
-    let wrap = src.get_wrap();
-
-    let query = gst::QueryRef::from_mut_ptr(query_ptr);
-
-    panic_to_error!(wrap, src, false, {
-        use gst::QueryView;
-
-        match query.view_mut() {
-            QueryView::Scheduling(ref mut q) if source_info.push_only => {
-                q.set(gst::SCHEDULING_FLAG_SEQUENTIAL, 1, -1, 0);
-                q.add_scheduling_modes(&[gst::PadMode::Push]);
-                true
-            }
-            _ => parent_klass
-                .query
-                .map(|f| from_glib(f(ptr, query_ptr)))
-                .unwrap_or(false),
-        }
-    }).to_glib()
 }
 
 pub struct SourceInfo {
@@ -369,318 +328,57 @@ pub struct SourceInfo {
     pub classification: String,
     pub author: String,
     pub rank: u32,
-    pub create_instance: fn(&RsSrcWrapper) -> Box<Source>,
+    pub create_instance: fn(&RsBaseSrc) -> Box<SourceImpl>,
     pub protocols: Vec<String>,
     pub push_only: bool,
 }
 
-glib_wrapper! {
+struct SourceStatic {
-    pub struct RsSrcWrapper(Object<RsSrc>): [gst_base::BaseSrc => gst_base_ffi::GstBaseSrc,
+    name: String,
-                                             gst::Element => gst_ffi::GstElement,
+    source_info: SourceInfo,
-                                             gst::Object => gst_ffi::GstObject,
+}
-                                             gst::URIHandler => gst_ffi::GstURIHandler,
-                                            ];
 
-    match fn {
+impl ImplTypeStatic<RsBaseSrc> for SourceStatic {
-        get_type => || rs_src_get_type(),
+    fn get_name(&self) -> &str {
+        self.name.as_str()
+    }
+
+    fn new(&self, element: &RsBaseSrc) -> Box<BaseSrcImpl> {
+        Source::init(element, &self.source_info)
+    }
+
+    fn class_init(&self, klass: &mut RsBaseSrcClass) {
+        Source::class_init(klass, &self.source_info);
+    }
+
+    fn type_init(&self, token: &TypeInitToken, type_: glib::Type) {
+        register_uri_handler(token, type_, self);
     }
 }
 
-impl RsSrcWrapper {
+impl URIHandlerImplStatic<RsBaseSrc> for SourceStatic {
-    fn get_wrap(&self) -> &SourceWrapper {
+    fn get_impl<'a>(&self, imp: &'a Box<BaseSrcImpl>) -> &'a URIHandlerImpl {
-        let stash = self.to_glib_none();
+        imp.downcast_ref::<Source>().unwrap()
-        let src: *mut RsSrc = stash.0;
-
-        unsafe { &*((*src).wrap) }
-    }
-}
-
-#[repr(u32)]
-enum Properties {
-    PropURI = 1u32,
-}
-
-#[repr(C)]
-pub struct RsSrc {
-    parent: gst_base_ffi::GstPushSrc,
-    wrap: *mut SourceWrapper,
-}
-
-#[repr(C)]
-pub struct RsSrcClass {
-    parent_class: gst_base_ffi::GstPushSrcClass,
-    source_info: *const SourceInfo,
-    protocols: *const Vec<*const c_char>,
-    parent_vtable: glib_ffi::gconstpointer,
-}
-
-unsafe fn rs_src_get_type() -> glib_ffi::GType {
-    use std::sync::{Once, ONCE_INIT};
-
-    static mut TYPE: glib_ffi::GType = gobject_ffi::G_TYPE_INVALID;
-    static ONCE: Once = ONCE_INIT;
-
-    ONCE.call_once(|| {
-        let type_info = gobject_ffi::GTypeInfo {
-            class_size: mem::size_of::<RsSrcClass>() as u16,
-            base_init: None,
-            base_finalize: None,
-            class_init: Some(source_class_init),
-            class_finalize: None,
-            class_data: ptr::null_mut(),
-            instance_size: mem::size_of::<RsSrc>() as u16,
-            n_preallocs: 0,
-            instance_init: Some(source_init),
-            value_table: ptr::null(),
-        };
-
-        let type_name = {
-            let mut idx = 0;
-
-            loop {
-                let type_name = CString::new(format!("RsSrc-{}", idx)).unwrap();
-                if gobject_ffi::g_type_from_name(type_name.as_ptr()) == gobject_ffi::G_TYPE_INVALID
-                {
-                    break type_name;
-                }
-                idx += 1;
-            }
-        };
-
-        TYPE = gobject_ffi::g_type_register_static(
-            gst_base_ffi::gst_base_src_get_type(),
-            type_name.as_ptr(),
-            &type_info,
-            gobject_ffi::GTypeFlags::empty(),
-        );
-
-        let iface_info = gobject_ffi::GInterfaceInfo {
-            interface_init: Some(source_uri_handler_init),
-            interface_finalize: None,
-            interface_data: ptr::null_mut(),
-        };
-        gobject_ffi::g_type_add_interface_static(
-            TYPE,
-            gst_ffi::gst_uri_handler_get_type(),
-            &iface_info,
-        );
-    });
-
-    TYPE
-}
-
-unsafe extern "C" fn source_finalize(obj: *mut gobject_ffi::GObject) {
-    let src = &mut *(obj as *mut RsSrc);
-
-    drop(Box::from_raw(src.wrap));
-    src.wrap = ptr::null_mut();
-
-    let src_klass = &**(obj as *const *const RsSrcClass);
-    let parent_klass = &*(src_klass.parent_vtable as *const gobject_ffi::GObjectClass);
-    parent_klass.finalize.map(|f| f(obj));
-}
-
-unsafe extern "C" fn source_set_property(
-    obj: *mut gobject_ffi::GObject,
-    id: u32,
-    value: *mut gobject_ffi::GValue,
-    _pspec: *mut gobject_ffi::GParamSpec,
-) {
-    let src = obj as *mut RsSrc;
-
-    match mem::transmute(id) {
-        Properties::PropURI => {
-            let uri_ptr = gobject_ffi::g_value_get_string(value);
-            source_set_uri(src, uri_ptr, ptr::null_mut());
-        }
-        _ => unreachable!(),
-    }
-}
-
-unsafe extern "C" fn source_get_property(
-    obj: *mut gobject_ffi::GObject,
-    id: u32,
-    value: *mut gobject_ffi::GValue,
-    _pspec: *mut gobject_ffi::GParamSpec,
-) {
-    let src = obj as *mut RsSrc;
-
-    match mem::transmute(id) {
-        Properties::PropURI => {
-            let uri_ptr = source_get_uri(src);
-            gobject_ffi::g_value_take_string(value, uri_ptr);
-        }
-        _ => unreachable!(),
-    }
-}
-
-unsafe extern "C" fn source_sub_class_init(
-    klass: glib_ffi::gpointer,
-    klass_data: glib_ffi::gpointer,
-) {
-    let source_info = &*(klass_data as *const SourceInfo);
-
-    {
-        let element_klass = &mut *(klass as *mut gst_ffi::GstElementClass);
-
-        gst_ffi::gst_element_class_set_metadata(
-            element_klass,
-            source_info.long_name.to_glib_none().0,
-            source_info.classification.to_glib_none().0,
-            source_info.description.to_glib_none().0,
-            source_info.author.to_glib_none().0,
-        );
-
-        // TODO: Methods + source_info.caps
-        let caps = gst::Caps::new_any();
-        let pad_template = gst::PadTemplate::new(
-            "src",
-            gst::PadDirection::Src,
-            gst::PadPresence::Always,
-            &caps,
-        );
-        gst_ffi::gst_element_class_add_pad_template(element_klass, pad_template.to_glib_full());
     }
 
-    {
+    fn get_type(&self) -> gst::URIType {
-        let src_klass = &mut *(klass as *mut RsSrcClass);
+        gst::URIType::Src
-
-        src_klass.source_info = source_info;
-        let mut protocols = Box::new(Vec::with_capacity(source_info.protocols.len()));
-        for p in &source_info.protocols {
-            protocols.push(p.to_glib_full());
-        }
-        protocols.push(ptr::null());
-        src_klass.protocols = Box::into_raw(protocols) as *const Vec<*const c_char>;
-    }
-}
-
-unsafe extern "C" fn source_class_init(klass: glib_ffi::gpointer, _klass_data: glib_ffi::gpointer) {
-    {
-        let gobject_klass = &mut *(klass as *mut gobject_ffi::GObjectClass);
-
-        gobject_klass.set_property = Some(source_set_property);
-        gobject_klass.get_property = Some(source_get_property);
-        gobject_klass.finalize = Some(source_finalize);
-
-        gobject_ffi::g_object_class_install_property(
-            klass as *mut gobject_ffi::GObjectClass,
-            1,
-            gobject_ffi::g_param_spec_string(
-                "uri".to_glib_none().0,
-                "URI".to_glib_none().0,
-                "URI to read from".to_glib_none().0,
-                ptr::null_mut(),
-                gobject_ffi::G_PARAM_READWRITE,
-            ),
-        );
     }
 
-    {
+    fn get_protocols(&self) -> Vec<String> {
-        let basesrc_klass = &mut *(klass as *mut gst_base_ffi::GstBaseSrcClass);
+        self.source_info.protocols.clone()
-
-        basesrc_klass.start = Some(source_start);
-        basesrc_klass.stop = Some(source_stop);
-        basesrc_klass.is_seekable = Some(source_is_seekable);
-        basesrc_klass.get_size = Some(source_get_size);
-        basesrc_klass.fill = Some(source_fill);
-        basesrc_klass.do_seek = Some(source_seek);
-        basesrc_klass.query = Some(source_query);
     }
-
-    {
-        let src_klass = &mut *(klass as *mut RsSrcClass);
-
-        src_klass.parent_vtable = gobject_ffi::g_type_class_peek_parent(klass);
-    }
-}
-
-unsafe extern "C" fn source_init(
-    instance: *mut gobject_ffi::GTypeInstance,
-    klass: glib_ffi::gpointer,
-) {
-    let src = &mut *(instance as *mut RsSrc);
-    let src_klass = &*(klass as *const RsSrcClass);
-    let source_info = &*src_klass.source_info;
-
-    let wrap = Box::new(SourceWrapper::new((source_info.create_instance)(
-        &RsSrcWrapper::from_glib_borrow(instance as *mut _),
-    )));
-    src.wrap = Box::into_raw(wrap);
-
-    let src = &RsSrcWrapper::from_glib_borrow(src as *mut _);
-    src.set_blocksize(4096);
-}
-
-unsafe extern "C" fn source_uri_handler_get_type(_type: glib_ffi::GType) -> gst_ffi::GstURIType {
-    gst::URIType::Src.to_glib()
-}
-
-unsafe extern "C" fn source_uri_handler_get_protocols(
-    type_: glib_ffi::GType,
-) -> *const *const c_char {
-    let klass = gobject_ffi::g_type_class_peek(type_);
-    let src_klass = &*(klass as *const RsSrcClass);
-    (*src_klass.protocols).as_ptr()
-}
-
-unsafe extern "C" fn source_uri_handler_get_uri(
-    uri_handler: *mut gst_ffi::GstURIHandler,
-) -> *mut c_char {
-    source_get_uri(uri_handler as *mut RsSrc)
-}
-
-unsafe extern "C" fn source_uri_handler_set_uri(
-    uri_handler: *mut gst_ffi::GstURIHandler,
-    uri: *const c_char,
-    err: *mut *mut glib_ffi::GError,
-) -> glib_ffi::gboolean {
-    source_set_uri(uri_handler as *mut RsSrc, uri, err)
-}
-
-unsafe extern "C" fn source_uri_handler_init(
-    iface: glib_ffi::gpointer,
-    _iface_data: glib_ffi::gpointer,
-) {
-    let uri_handler_iface = &mut *(iface as *mut gst_ffi::GstURIHandlerInterface);
-
-    uri_handler_iface.get_type = Some(source_uri_handler_get_type);
-    uri_handler_iface.get_protocols = Some(source_uri_handler_get_protocols);
-    uri_handler_iface.get_uri = Some(source_uri_handler_get_uri);
-    uri_handler_iface.set_uri = Some(source_uri_handler_set_uri);
 }
 
 pub fn source_register(plugin: &gst::Plugin, source_info: SourceInfo) {
-    unsafe {
+    let name = source_info.name.clone();
-        let parent_type = rs_src_get_type();
+    let rank = source_info.rank;
-        let type_name = format!("RsSrc-{}", source_info.name);
 
-        let name = source_info.name.clone();
+    let source_static = SourceStatic {
-        let rank = source_info.rank;
+        name: format!("Source-{}", name),
+        source_info: source_info,
+    };
 
-        let source_info = Box::new(source_info);
+    let type_ = register_type(source_static);
-        let source_info_ptr = Box::into_raw(source_info) as glib_ffi::gpointer;
+    gst::Element::register(plugin, &name, rank, type_);
-
-        let type_info = gobject_ffi::GTypeInfo {
-            class_size: mem::size_of::<RsSrcClass>() as u16,
-            base_init: None,
-            base_finalize: None,
-            class_init: Some(source_sub_class_init),
-            class_finalize: None,
-            class_data: source_info_ptr,
-            instance_size: mem::size_of::<RsSrc>() as u16,
-            n_preallocs: 0,
-            instance_init: None,
-            value_table: ptr::null(),
-        };
-
-        let type_ = gobject_ffi::g_type_register_static(
-            parent_type,
-            type_name.to_glib_none().0,
-            &type_info,
-            gobject_ffi::GTypeFlags::empty(),
-        );
-
-        gst::Element::register(plugin, &name, rank, from_glib(type_));
-    }
 }

gst-plugin/src/uri_handler.rs

@@ -5,6 +5,7 @@
 // <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
 // option. This file may not be copied, modified, or distributed
 // except according to those terms.
+use std::ptr;
 
 use mopa;
 
@@ -26,7 +27,7 @@ pub trait URIHandlerImpl: mopa::Any + Send + Sync + 'static {
 }
 
 pub trait URIHandlerImplStatic<T: ObjectType>: Send + Sync + 'static {
-    fn get_impl(&self, imp: &T::ImplType) -> &URIHandlerImpl;
+    fn get_impl<'a>(&self, imp: &'a T::ImplType) -> &'a URIHandlerImpl;
     fn get_type(&self) -> gst::URIType;
     fn get_protocols(&self) -> Vec<String>;
 }
@@ -122,10 +123,11 @@ pub fn register_uri_handler<T: ObjectType, I: URIHandlerImplStatic<T>>(
     imp: &I,
 ) {
     unsafe {
-        let protocols: Vec<_> = imp.get_protocols()
+        let mut protocols: Vec<_> = imp.get_protocols()
             .iter()
             .map(|s| s.to_glib_full())
             .collect();
+        protocols.push(ptr::null());
 
         let imp = imp as &URIHandlerImplStatic<T> as *const URIHandlerImplStatic<T>;
         let interface_static = Box::new(URIHandlerStatic {