gst-plugin-file: port filesink to new subclass architecture

gst-plugin-file/Cargo.toml

@@ -8,9 +8,8 @@ license = "MIT/Apache-2.0"
 [dependencies]
 url = "1.1"
 glib = { git = "https://github.com/gtk-rs/glib" }
-gst-plugin = { path="../gst-plugin" }
-gst-plugin-simple = { path="../gst-plugin-simple" }
-gstreamer = { git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs" }
+gstreamer = { git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs", features = ["subclassing"] }
+gstreamer-base = { git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs", features = ["subclassing"] }
 
 [lib]
 name = "gstrsfile"

gst-plugin-file/src/file_location.rs

@@ -0,0 +1,141 @@
+use glib;
+use gst;
+use url::Url;
+
+use std::convert::AsRef;
+use std::fmt;
+use std::ops::Deref;
+use std::path::{Path, PathBuf};
+
+#[cfg(target_os = "windows")]
+const WIN_EXT_PATH_PREFIX: &str = "\\\\?\\";
+#[cfg(target_os = "windows")]
+const WIN_EXT_PATH_PREFIX_LEN: usize = 4;
+
+#[derive(Debug)]
+pub(super) struct FileLocation(PathBuf);
+
+impl FileLocation {
+    pub(super) fn try_from_path_str(path_str: String) -> Result<Self, glib::Error> {
+        FileLocation::try_from(PathBuf::from(path_str))
+    }
+
+    pub(super) fn try_from_uri_str(uri_str: &str) -> Result<Self, glib::Error> {
+        match Url::parse(uri_str) {
+            Ok(url) => {
+                if url.scheme() != "file" {
+                    return Err(gst::Error::new(
+                        gst::URIError::UnsupportedProtocol,
+                        format!("Unsupported URI {}", uri_str).as_str(),
+                    ));
+                }
+
+                let path = url.to_file_path().or_else(|_| {
+                    Err(gst::Error::new(
+                        gst::URIError::BadUri,
+                        format!("Unsupported URI {}", uri_str).as_str(),
+                    ))
+                })?;
+
+                FileLocation::try_from(path)
+            }
+            Err(err) => Err(gst::Error::new(
+                gst::URIError::BadUri,
+                format!("Couldn't parse URI {}: {}", uri_str, err.to_string()).as_str(),
+            )),
+        }
+    }
+
+    fn try_from(location: PathBuf) -> Result<Self, glib::Error> {
+        let location_str = location.to_str().ok_or_else(|| {
+            gst::Error::new(
+                gst::LibraryError::Failed,
+                format!("Invalid path {:?}", location).as_str(),
+            )
+        })?;
+
+        let file_name = location.file_name().ok_or_else(|| {
+            gst::Error::new(
+                gst::LibraryError::Failed,
+                format!("Expected a path with a filename, got {}", location_str,).as_str(),
+            )
+        })?;
+
+        // The filename might not exist yet, so check the parent only.
+        // Note: `location` contains a filename, so its parent can't be `None`
+        let mut parent_dir = location
+            .parent()
+            .expect("FileSink::set_location `location` with filename but without a parent")
+            .to_owned();
+        if parent_dir.is_relative() && parent_dir.components().next() == None {
+            // `location` only contains the filename
+            // need to specify "." for `canonicalize` to resolve the actual path
+            parent_dir = PathBuf::from(".");
+        }
+
+        let parent_canonical = parent_dir.canonicalize().map_err(|err| {
+            gst::Error::new(
+                gst::LibraryError::Failed,
+                format!(
+                    "Could not resolve path {}: {}",
+                    location_str,
+                    err.to_string(),
+                )
+                .as_str(),
+            )
+        })?;
+
+        #[cfg(target_os = "windows")]
+        let parent_canonical = {
+            let has_prefix = parent_canonical
+                .to_str()
+                .unwrap() // already checked above
+                .starts_with(WIN_EXT_PATH_PREFIX);
+            if has_prefix {
+                // Remove the "extended length path" prefix
+                // for compatibility with applications which can't deal with it.
+                // See https://doc.rust-lang.org/std/fs/fn.canonicalize.html
+                let parent_canonical_str = parent_canonical.to_str().unwrap();
+                PathBuf::from(&parent_canonical_str[WIN_EXT_PATH_PREFIX_LEN..])
+            } else {
+                parent_canonical
+            }
+        };
+
+        let location_canonical = parent_canonical.join(file_name);
+        Url::from_file_path(&location_canonical)
+            .map_err(|_| {
+                gst::Error::new(
+                    gst::LibraryError::Failed,
+                    format!("Could not resolve path to URL {}", location_str).as_str(),
+                )
+            })
+            .map(|_| FileLocation(location_canonical))
+    }
+
+    fn to_str(&self) -> &str {
+        self.0
+            .to_str()
+            .expect("FileLocation: couldn't get `&str` from internal `PathBuf`")
+    }
+}
+
+impl AsRef<Path> for FileLocation {
+    fn as_ref(&self) -> &Path {
+        self.0.as_ref()
+    }
+}
+
+impl Deref for FileLocation {
+    type Target = Path;
+
+    fn deref(&self) -> &Path {
+        self.0.as_ref()
+    }
+}
+
+impl fmt::Display for FileLocation {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(f, "{}", self.to_str())
+    }
+}

gst-plugin-file/src/filesink.rs

@@ -1,5 +1,6 @@
 // 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
@@ -7,151 +8,359 @@
 // option. This file may not be copied, modified, or distributed
 // except according to those terms.
 
+use glib;
+use glib::subclass;
+use glib::subclass::prelude::*;
+use gst;
+use gst::prelude::*;
+use gst::subclass::prelude::*;
+use gst_base;
+use gst_base::subclass::prelude::*;
+
 use std::fs::File;
+use std::io::Write;
+use std::sync::Mutex;
+
 use url::Url;
 
-use std::io::Write;
+use file_location::FileLocation;
 
-use gst_plugin::error::*;
-use gst_plugin_simple::error::*;
-use gst_plugin_simple::sink::*;
-use gst_plugin_simple::UriValidator;
-
-use gst;
+const DEFAULT_LOCATION: Option<FileLocation> = None;
 
 #[derive(Debug)]
-enum StreamingState {
+struct Settings {
+    location: Option<FileLocation>,
+}
+
+impl Default for Settings {
+    fn default() -> Self {
+        Settings {
+            location: DEFAULT_LOCATION,
+        }
+    }
+}
+
+const LOCATION_PROP: &str = "location";
+
+static PROPERTIES: [subclass::Property; 1] = [subclass::Property(LOCATION_PROP, || {
+    glib::ParamSpec::string(
+        LOCATION_PROP,
+        "File Location",
+        "Location of the file to write",
+        None,
+        glib::ParamFlags::READWRITE,
+    )
+})];
+
+enum State {
     Stopped,
     Started { file: File, position: u64 },
 }
 
-#[derive(Debug)]
+impl Default for State {
+    fn default() -> State {
+        State::Stopped
+    }
+}
+
 pub struct FileSink {
-    streaming_state: StreamingState,
     cat: gst::DebugCategory,
+    settings: Mutex<Settings>,
+    state: Mutex<State>,
 }
 
 impl FileSink {
-    pub fn new(_sink: &BaseSink) -> FileSink {
-        FileSink {
-            streaming_state: StreamingState::Stopped,
-            cat: gst::DebugCategory::new(
-                "rsfilesink",
-                gst::DebugColorFlags::empty(),
-                "Rust file source",
-            ),
-        }
-    }
-
-    pub fn new_boxed(sink: &BaseSink) -> Box<SinkImpl> {
-        Box::new(FileSink::new(sink))
-    }
-}
-
-fn validate_uri(uri: &Url) -> Result<(), UriError> {
-    let _ = try!(uri.to_file_path().or_else(|_| Err(UriError::new(
-        gst::URIError::UnsupportedProtocol,
-        format!("Unsupported file URI '{}'", uri.as_str()),
-    ))));
-    Ok(())
-}
-
-impl SinkImpl for FileSink {
-    fn uri_validator(&self) -> Box<UriValidator> {
-        Box::new(validate_uri)
-    }
-
-    fn start(&mut self, sink: &BaseSink, uri: Url) -> Result<(), gst::ErrorMessage> {
-        if let StreamingState::Started { .. } = self.streaming_state {
-            return Err(gst_error_msg!(
+    fn set_location(
+        &self,
+        element: &gst_base::BaseSink,
+        location: Option<FileLocation>,
+    ) -> Result<(), glib::Error> {
+        let state = self.state.lock().unwrap();
+        if let State::Started { .. } = *state {
+            return Err(gst::Error::new(
                 gst::LibraryError::Failed,
-                ["Sink already started"]
+                format!(
+                    "Changing the `{}` property on a started `filesink` is not supported",
+                    LOCATION_PROP,
+                )
+                .as_str(),
             ));
         }
 
-        let location = try!(uri.to_file_path().or_else(|_| {
-            gst_error!(
+        let mut settings = self.settings.lock().unwrap();
+        settings.location = match location {
+            Some(location) => {
+                match settings.location {
+                    Some(ref location_cur) => {
+                        gst_info!(
                             self.cat,
-                obj: sink,
-                "Unsupported file URI '{}'",
-                uri.as_str()
+                            obj: element,
+                            "Changing `{}` from {:?} to {}",
+                            LOCATION_PROP,
+                            location_cur,
+                            location,
                         );
-            Err(gst_error_msg!(
-                gst::LibraryError::Failed,
-                ["Unsupported file URI '{}'", uri.as_str()]
-            ))
-        }));
-
-        let file = try!(File::create(location.as_path()).or_else(|err| {
-            gst_error!(
-                self.cat,
-                obj: sink,
-                "Could not open file for writing: {}",
-                err.to_string()
-            );
-            Err(gst_error_msg!(
-                gst::ResourceError::OpenWrite,
-                [
-                    "Could not open file for writing '{}': {}",
-                    location.to_str().unwrap_or("Non-UTF8 path"),
-                    err.to_string(),
-                ]
-            ))
-        }));
-
-        gst_debug!(self.cat, obj: sink, "Opened file {:?}", file);
-
-        self.streaming_state = StreamingState::Started { file, position: 0 };
-
-        Ok(())
                     }
-
-    fn stop(&mut self, _sink: &BaseSink) -> Result<(), gst::ErrorMessage> {
-        self.streaming_state = StreamingState::Stopped;
-
-        Ok(())
-    }
-
-    fn render(&mut self, sink: &BaseSink, buffer: &gst::BufferRef) -> Result<(), FlowError> {
-        let cat = self.cat;
-        let streaming_state = &mut self.streaming_state;
-
-        gst_trace!(cat, obj: sink, "Rendering {:?}", buffer);
-
-        let (file, position) = match *streaming_state {
-            StreamingState::Started {
-                ref mut file,
-                ref mut position,
-            } => (file, position),
-            StreamingState::Stopped => {
-                return Err(FlowError::Error(gst_error_msg!(
-                    gst::LibraryError::Failed,
-                    ["Not started yet"]
-                )));
-            }
-        };
-
-        let map = match buffer.map_readable() {
                     None => {
-                return Err(FlowError::Error(gst_error_msg!(
-                    gst::LibraryError::Failed,
-                    ["Failed to map buffer"]
-                )));
+                        gst_info!(
+                            self.cat,
+                            obj: element,
+                            "Setting `{}` to {}",
+                            LOCATION_PROP,
+                            location,
+                        );
+                    }
+                }
+                Some(location)
+            }
+            None => {
+                gst_info!(
+                    self.cat,
+                    obj: element,
+                    "Resetting `{}` to None",
+                    LOCATION_PROP
+                );
+                None
             }
-            Some(map) => map,
         };
-        let data = map.as_slice();
-
-        try!(file.write_all(data).or_else(|err| {
-            gst_error!(cat, obj: sink, "Failed to write: {}", err);
-            Err(FlowError::Error(gst_error_msg!(
-                gst::ResourceError::Write,
-                ["Failed to write: {}", err]
-            )))
-        }));
-
-        *position += data.len() as u64;
 
         Ok(())
     }
 }
+
+impl ObjectSubclass for FileSink {
+    const NAME: &'static str = "RsFileSink";
+    type ParentType = gst_base::BaseSink;
+    type Instance = gst::subclass::ElementInstanceStruct<Self>;
+    type Class = subclass::simple::ClassStruct<Self>;
+
+    glib_object_subclass!();
+
+    fn new() -> Self {
+        Self {
+            cat: gst::DebugCategory::new("rsfilesink", gst::DebugColorFlags::empty(), "File Sink"),
+            settings: Mutex::new(Default::default()),
+            state: Mutex::new(Default::default()),
+        }
+    }
+
+    fn type_init(type_: &subclass::InitializingType<Self>) {
+        gst::subclass::uri_handler::register(type_);
+    }
+
+    fn class_init(klass: &mut subclass::simple::ClassStruct<Self>) {
+        klass.set_metadata(
+            "File Sink",
+            "Sink/File",
+            "Write stream to a file",
+            "François Laignel <fengalin@free.fr>, Luis de Bethencourt <luisbg@osg.samsung.com>",
+        );
+
+        let caps = gst::Caps::new_any();
+        let sink_pad_template = gst::PadTemplate::new(
+            "sink",
+            gst::PadDirection::Sink,
+            gst::PadPresence::Always,
+            &caps,
+        );
+        klass.add_pad_template(sink_pad_template);
+
+        klass.install_properties(&PROPERTIES);
+    }
+}
+
+impl ObjectImpl for FileSink {
+    glib_object_impl!();
+
+    fn set_property(&self, obj: &glib::Object, id: usize, value: &glib::Value) {
+        let prop = &PROPERTIES[id];
+        match *prop {
+            subclass::Property(LOCATION_PROP, ..) => {
+                let element = obj.downcast_ref::<gst_base::BaseSink>().unwrap();
+
+                let res = match value.get::<String>() {
+                    Some(location) => FileLocation::try_from_path_str(location)
+                        .and_then(|file_location| self.set_location(&element, Some(file_location))),
+                    None => self.set_location(&element, None),
+                };
+
+                if let Err(err) = res {
+                    gst_error!(
+                        self.cat,
+                        obj: element,
+                        "Failed to set property `{}`: {}",
+                        LOCATION_PROP,
+                        err
+                    );
+                }
+            }
+            _ => unimplemented!(),
+        };
+    }
+
+    fn get_property(&self, _obj: &glib::Object, id: usize) -> Result<glib::Value, ()> {
+        let prop = &PROPERTIES[id];
+        match *prop {
+            subclass::Property(LOCATION_PROP, ..) => {
+                let settings = self.settings.lock().unwrap();
+                let location = settings
+                    .location
+                    .as_ref()
+                    .map(|location| location.to_string());
+
+                Ok(location.to_value())
+            }
+            _ => unimplemented!(),
+        }
+    }
+}
+
+impl ElementImpl for FileSink {}
+
+impl BaseSinkImpl for FileSink {
+    fn start(&self, element: &gst_base::BaseSink) -> bool {
+        let mut state = self.state.lock().unwrap();
+        if let State::Started { .. } = *state {
+            gst_element_info!(
+                element,
+                gst::CoreError::StateChange,
+                ["FileSink already started"]
+            );
+            return false;
+        }
+
+        let settings = self.settings.lock().unwrap();
+        let location = match settings.location {
+            Some(ref location) => location,
+            None => {
+                gst_element_error!(
+                    element,
+                    gst::ResourceError::Settings,
+                    ["File location is not defined"]
+                );
+                return false;
+            }
+        };
+
+        let file = match File::create(location) {
+            Ok(file) => file,
+            Err(err) => {
+                gst_element_error!(
+                    element,
+                    gst::ResourceError::OpenWrite,
+                    [
+                        "Could not open file {} for writing: {}",
+                        location,
+                        err.to_string(),
+                    ]
+                );
+                return false;
+            }
+        };
+        gst_debug!(self.cat, obj: element, "Opened file {:?}", file);
+
+        *state = State::Started { file, position: 0 };
+        gst_info!(self.cat, obj: element, "Started");
+
+        true
+    }
+
+    fn stop(&self, element: &gst_base::BaseSink) -> bool {
+        let mut state = self.state.lock().unwrap();
+        if let State::Stopped = *state {
+            gst_element_warning!(
+                element,
+                gst::CoreError::StateChange,
+                ["FileSink not started"]
+            );
+            return false;
+        }
+
+        *state = State::Stopped;
+        gst_info!(self.cat, obj: element, "Stopped");
+
+        true
+    }
+
+    // TODO: implement seek in BYTES format
+
+    fn render(&self, element: &gst_base::BaseSink, buffer: &gst::BufferRef) -> gst::FlowReturn {
+        let mut state = self.state.lock().unwrap();
+        let (file, position) = match *state {
+            State::Started {
+                ref mut file,
+                ref mut position,
+            } => (file, position),
+            State::Stopped => {
+                gst_element_error!(element, gst::CoreError::Failed, ["Not started yet"]);
+                return gst::FlowReturn::Error;
+            }
+        };
+
+        gst_trace!(self.cat, obj: element, "Rendering {:?}", buffer);
+        let map = match buffer.map_readable() {
+            None => {
+                gst_element_error!(element, gst::CoreError::Failed, ["Failed to map buffer"]);
+                return gst::FlowReturn::Error;
+            }
+            Some(map) => map,
+        };
+
+        match file.write_all(map.as_ref()) {
+            Ok(()) => {
+                *position += map.len() as u64;
+
+                gst::FlowReturn::Ok
+            }
+            Err(err) => {
+                gst_element_error!(
+                    element,
+                    gst::ResourceError::Write,
+                    ["Failed to write buffer: {}", err]
+                );
+
+                gst::FlowReturn::Error
+            }
+        }
+    }
+}
+
+impl URIHandlerImpl for FileSink {
+    fn get_uri(&self, _element: &gst::URIHandler) -> Option<String> {
+        let settings = self.settings.lock().unwrap();
+
+        // Conversion to Url already checked while building the `FileLocation`
+        settings.location.as_ref().map(|location| {
+            Url::from_file_path(location)
+                .expect("FileSink::get_uri couldn't build `Url` from `location`")
+                .into_string()
+        })
+    }
+
+    fn set_uri(&self, element: &gst::URIHandler, uri: Option<String>) -> 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
+        let uri = uri.filter(|uri| uri != "file://");
+
+        let file_location = match uri {
+            Some(uri) => Some(FileLocation::try_from_uri_str(&uri)?),
+            None => None,
+        };
+
+        self.set_location(&element, file_location)
+    }
+
+    fn get_uri_type() -> gst::URIType {
+        gst::URIType::Sink
+    }
+
+    fn get_protocols() -> Vec<String> {
+        vec!["file".to_string()]
+    }
+}
+
+pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
+    gst::Element::register(plugin, "rsfilesink", 256 + 100, FileSink::get_type())
+}

gst-plugin-file/src/filesrc.rs

@@ -1,4 +1,5 @@
 // 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
@@ -6,187 +7,424 @@
 // option. This file may not be copied, modified, or distributed
 // except according to those terms.
 
+use glib;
+use glib::subclass;
+use glib::subclass::prelude::*;
+use gst;
+use gst::prelude::*;
+use gst::subclass::prelude::*;
+use gst_base;
+use gst_base::subclass::prelude::*;
+
 use std::fs::File;
 use std::io::{Read, Seek, SeekFrom};
-use std::u64;
+use std::sync::Mutex;
+
 use url::Url;
 
-use gst_plugin::error::*;
-use gst_plugin_simple::error::*;
-use gst_plugin_simple::source::*;
-use gst_plugin_simple::UriValidator;
+use file_location::FileLocation;
 
-use gst;
+const DEFAULT_LOCATION: Option<FileLocation> = None;
 
 #[derive(Debug)]
-enum StreamingState {
+struct Settings {
+    location: Option<FileLocation>,
+}
+
+impl Default for Settings {
+    fn default() -> Self {
+        Settings {
+            location: DEFAULT_LOCATION,
+        }
+    }
+}
+
+const LOCATION_PROP: &str = "location";
+
+static PROPERTIES: [subclass::Property; 1] = [subclass::Property(LOCATION_PROP, || {
+    glib::ParamSpec::string(
+        LOCATION_PROP,
+        "File Location",
+        "Location of the file to read from",
+        None,
+        glib::ParamFlags::READWRITE,
+    )
+})];
+
+enum State {
     Stopped,
     Started { file: File, position: u64 },
 }
 
-#[derive(Debug)]
+impl Default for State {
+    fn default() -> State {
+        State::Stopped
+    }
+}
+
 pub struct FileSrc {
-    streaming_state: StreamingState,
     cat: gst::DebugCategory,
+    settings: Mutex<Settings>,
+    state: Mutex<State>,
 }
 
 impl FileSrc {
-    pub fn new(_src: &BaseSrc) -> FileSrc {
-        FileSrc {
-            streaming_state: StreamingState::Stopped,
-            cat: gst::DebugCategory::new(
-                "rsfilesrc",
-                gst::DebugColorFlags::empty(),
-                "Rust file source",
-            ),
-        }
+    fn set_location(
+        &self,
+        element: &gst_base::BaseSrc,
+        location: Option<FileLocation>,
+    ) -> Result<(), glib::Error> {
+        let state = self.state.lock().unwrap();
+        if let State::Started { .. } = *state {
+            return Err(gst::Error::new(
+                gst::LibraryError::Failed,
+                format!(
+                    "Changing the `{}` property on a started `filesrc` is not supported",
+                    LOCATION_PROP,
+                )
+                .as_str(),
+            ));
         }
 
-    pub fn new_boxed(src: &BaseSrc) -> Box<SourceImpl> {
-        Box::new(FileSrc::new(src))
+        let mut settings = self.settings.lock().unwrap();
+        settings.location = match location {
+            Some(location) => {
+                if !location.exists() {
+                    return Err(gst::Error::new(
+                        gst::LibraryError::Failed,
+                        format!("{} doesn't exist", location).as_str(),
+                    ));
                 }
-}
 
-fn validate_uri(uri: &Url) -> Result<(), UriError> {
-    let _ = try!(uri.to_file_path().or_else(|_| Err(UriError::new(
-        gst::URIError::UnsupportedProtocol,
-        format!("Unsupported file URI '{}'", uri.as_str()),
-    ))));
+                if !location.is_file() {
+                    return Err(gst::Error::new(
+                        gst::LibraryError::Failed,
+                        format!("{} is not a file", location).as_str(),
+                    ));
+                }
+
+                match settings.location {
+                    Some(ref location_cur) => {
+                        gst_info!(
+                            self.cat,
+                            obj: element,
+                            "Changing `{}` from {:?} to {}",
+                            LOCATION_PROP,
+                            location_cur,
+                            location,
+                        );
+                    }
+                    None => {
+                        gst_info!(
+                            self.cat,
+                            obj: element,
+                            "Setting `{}` to {}",
+                            LOCATION_PROP,
+                            location,
+                        );
+                    }
+                }
+                Some(location)
+            }
+            None => {
+                gst_info!(
+                    self.cat,
+                    obj: element,
+                    "Resetting `{}` to None",
+                    LOCATION_PROP
+                );
+                None
+            }
+        };
+
         Ok(())
+    }
 }
 
-impl SourceImpl for FileSrc {
-    fn uri_validator(&self) -> Box<UriValidator> {
-        Box::new(validate_uri)
+impl ObjectSubclass for FileSrc {
+    const NAME: &'static str = "RsFileSrc";
+    type ParentType = gst_base::BaseSrc;
+    type Instance = gst::subclass::ElementInstanceStruct<Self>;
+    type Class = subclass::simple::ClassStruct<Self>;
+
+    glib_object_subclass!();
+
+    fn new() -> Self {
+        Self {
+            cat: gst::DebugCategory::new("rsfilesrc", gst::DebugColorFlags::empty(), "File Source"),
+            settings: Mutex::new(Default::default()),
+            state: Mutex::new(Default::default()),
+        }
     }
 
-    fn is_seekable(&self, _src: &BaseSrc) -> bool {
+    fn type_init(type_: &subclass::InitializingType<Self>) {
+        gst::subclass::uri_handler::register(type_);
+    }
+
+    fn class_init(klass: &mut subclass::simple::ClassStruct<Self>) {
+        klass.set_metadata(
+            "File Source",
+            "Source/File",
+            "Read stream from a file",
+            "François Laignel <fengalin@free.fr>, Sebastian Dröge <sebastian@centricular.com>",
+        );
+
+        let caps = gst::Caps::new_any();
+        let src_pad_template = gst::PadTemplate::new(
+            "src",
+            gst::PadDirection::Src,
+            gst::PadPresence::Always,
+            &caps,
+        );
+        klass.add_pad_template(src_pad_template);
+
+        klass.install_properties(&PROPERTIES);
+    }
+}
+
+impl ObjectImpl for FileSrc {
+    glib_object_impl!();
+
+    fn set_property(&self, obj: &glib::Object, id: usize, value: &glib::Value) {
+        let prop = &PROPERTIES[id];
+        match *prop {
+            subclass::Property(LOCATION_PROP, ..) => {
+                let element = obj.downcast_ref::<gst_base::BaseSrc>().unwrap();
+
+                let res = match value.get::<String>() {
+                    Some(location) => FileLocation::try_from_path_str(location)
+                        .and_then(|file_location| self.set_location(&element, Some(file_location))),
+                    None => self.set_location(&element, None),
+                };
+
+                if let Err(err) = res {
+                    gst_error!(
+                        self.cat,
+                        obj: element,
+                        "Failed to set property `{}`: {}",
+                        LOCATION_PROP,
+                        err
+                    );
+                }
+            }
+            _ => unimplemented!(),
+        };
+    }
+
+    fn get_property(&self, _obj: &glib::Object, id: usize) -> Result<glib::Value, ()> {
+        let prop = &PROPERTIES[id];
+        match *prop {
+            subclass::Property(LOCATION_PROP, ..) => {
+                let settings = self.settings.lock().unwrap();
+                let location = settings
+                    .location
+                    .as_ref()
+                    .map(|location| location.to_string());
+
+                Ok(location.to_value())
+            }
+            _ => unimplemented!(),
+        }
+    }
+}
+
+impl ElementImpl for FileSrc {}
+
+impl BaseSrcImpl for FileSrc {
+    fn is_seekable(&self, _src: &gst_base::BaseSrc) -> bool {
         true
     }
 
-    fn get_size(&self, _src: &BaseSrc) -> Option<u64> {
-        if let StreamingState::Started { ref file, .. } = self.streaming_state {
+    fn get_size(&self, _src: &gst_base::BaseSrc) -> Option<u64> {
+        let state = self.state.lock().unwrap();
+        if let State::Started { ref file, .. } = *state {
             file.metadata().ok().map(|m| m.len())
         } else {
             None
         }
     }
 
-    fn start(&mut self, src: &BaseSrc, uri: Url) -> Result<(), gst::ErrorMessage> {
-        if let StreamingState::Started { .. } = self.streaming_state {
-            return Err(gst_error_msg!(
-                gst::LibraryError::Failed,
-                ["Source already started"]
-            ));
+    fn start(&self, element: &gst_base::BaseSrc) -> bool {
+        let mut state = self.state.lock().unwrap();
+        if let State::Started { .. } = *state {
+            gst_element_info!(
+                element,
+                gst::CoreError::StateChange,
+                ["FileSrc already started"]
+            );
+            return false;
         }
 
-        let location = try!(uri.to_file_path().or_else(|_| {
-            gst_error!(
-                self.cat,
-                obj: src,
-                "Unsupported file URI '{}'",
-                uri.as_str()
+        let settings = self.settings.lock().unwrap();
+        let location = match settings.location {
+            Some(ref location) => location,
+            None => {
+                gst_element_error!(
+                    element,
+                    gst::CoreError::StateChange,
+                    ["File location is not defined"]
                 );
-            Err(gst_error_msg!(
-                gst::LibraryError::Failed,
-                ["Unsupported file URI '{}'", uri.as_str()]
-            ))
-        }));
+                return false;
+            }
+        };
 
-        let file = try!(File::open(location.as_path()).or_else(|err| {
-            gst_error!(
-                self.cat,
-                obj: src,
-                "Could not open file for reading: {}",
-                err.to_string()
-            );
-            Err(gst_error_msg!(
+        let file = match File::open(location) {
+            Ok(file) => file,
+            Err(err) => {
+                gst_element_error!(
+                    element,
                     gst::ResourceError::OpenRead,
                     [
-                    "Could not open file for reading '{}': {}",
-                    location.to_str().unwrap_or("Non-UTF8 path"),
+                        "Could not open file {} for reading: {}",
+                        location,
                         err.to_string(),
                     ]
-            ))
-        }));
+                );
+                return false;
+            }
+        };
 
-        gst_debug!(self.cat, obj: src, "Opened file {:?}", file);
+        gst_debug!(self.cat, obj: element, "Opened file {:?}", file);
 
-        self.streaming_state = StreamingState::Started { file, position: 0 };
+        *state = State::Started { file, position: 0 };
 
-        Ok(())
+        gst_info!(self.cat, obj: element, "Started");
+
+        true
     }
 
-    fn stop(&mut self, _src: &BaseSrc) -> Result<(), gst::ErrorMessage> {
-        self.streaming_state = StreamingState::Stopped;
-
-        Ok(())
+    fn stop(&self, element: &gst_base::BaseSrc) -> bool {
+        let mut state = self.state.lock().unwrap();
+        if let State::Stopped = *state {
+            gst_element_warning!(
+                element,
+                gst::CoreError::StateChange,
+                ["FileSink not started"]
+            );
+            return false;
         }
 
-    fn fill(
-        &mut self,
-        src: &BaseSrc,
+        *state = State::Stopped;
+
+        gst_info!(self.cat, obj: element, "Stopped");
+
+        true
+    }
+
+    fn create(
+        &self,
+        element: &gst_base::BaseSrc,
         offset: u64,
-        _: u32,
-        buffer: &mut gst::BufferRef,
-    ) -> Result<(), FlowError> {
-        let cat = self.cat;
-        let streaming_state = &mut self.streaming_state;
+        length: u32,
+    ) -> Result<gst::Buffer, gst::FlowError> {
+        let mut state = self.state.lock().unwrap();
 
-        let (file, position) = match *streaming_state {
-            StreamingState::Started {
+        let (file, position) = match *state {
+            State::Started {
                 ref mut file,
                 ref mut position,
             } => (file, position),
-            StreamingState::Stopped => {
-                return Err(FlowError::Error(gst_error_msg!(
-                    gst::LibraryError::Failed,
-                    ["Not started yet"]
-                )));
+            State::Stopped => {
+                gst_element_error!(element, gst::CoreError::Failed, ["Not started yet"]);
+                return Err(gst::FlowError::Error);
             }
         };
 
         if *position != offset {
-            try!(file.seek(SeekFrom::Start(offset)).or_else(|err| {
-                gst_error!(cat, obj: src, "Failed to seek to {}: {:?}", offset, err);
-                Err(FlowError::Error(gst_error_msg!(
-                    gst::ResourceError::Seek,
+            file.seek(SeekFrom::Start(offset)).map_err(|err| {
+                gst_element_error!(
+                    element,
+                    gst::LibraryError::Failed,
                     ["Failed to seek to {}: {}", offset, err.to_string()]
-                )))
-            }));
+                );
+                gst::FlowError::Error
+            })?;
+
             *position = offset;
         }
 
-        let size = {
-            let mut map = match buffer.map_writable() {
+        let mut buffer = match gst::Buffer::with_size(length as usize) {
+            Some(buffer) => buffer,
             None => {
-                    return Err(FlowError::Error(gst_error_msg!(
+                gst_element_error!(
+                    element,
                     gst::LibraryError::Failed,
-                        ["Failed to map buffer"]
-                    )));
+                    ["Failed to allocate buffer"]
+                );
+                return Err(gst::FlowError::Error);
             }
-                Some(map) => map,
         };
 
-            let data = map.as_mut_slice();
+        {
+            let buffer = buffer.make_mut();
+            let size = {
+                let mut map = match buffer.map_writable() {
+                    Some(map) => map,
+                    None => {
+                        gst_element_error!(
+                            element,
+                            gst::LibraryError::Failed,
+                            ["Failed to map buffer"]
+                        );
+                        return Err(gst::FlowError::Error);
+                    }
+                };
 
-            try!(file.read(data).or_else(|err| {
-                gst_error!(cat, obj: src, "Failed to read: {:?}", err);
-                Err(FlowError::Error(gst_error_msg!(
-                    gst::ResourceError::Read,
+                file.read(map.as_mut()).map_err(|err| {
+                    gst_element_error!(
+                        element,
+                        gst::LibraryError::Failed,
                         ["Failed to read at {}: {}", offset, err.to_string()]
-                )))
-            }))
+                    );
+                    gst::FlowError::Error
+                })?
             };
 
             *position += size as u64;
 
             buffer.set_size(size);
-
-        Ok(())
         }
 
-    fn seek(&mut self, _src: &BaseSrc, _: u64, _: Option<u64>) -> Result<(), gst::ErrorMessage> {
-        Ok(())
+        Ok(buffer)
     }
 }
+
+impl URIHandlerImpl for FileSrc {
+    fn get_uri(&self, _element: &gst::URIHandler) -> Option<String> {
+        let settings = self.settings.lock().unwrap();
+
+        // Conversion to Url already checked while building the `FileLocation`
+        settings.location.as_ref().map(|location| {
+            Url::from_file_path(location)
+                .expect("FileSrc::get_uri couldn't build `Url` from `location`")
+                .into_string()
+        })
+    }
+
+    fn set_uri(&self, element: &gst::URIHandler, uri: Option<String>) -> 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
+        let uri = uri.filter(|uri| uri != "file://");
+
+        let file_location = match uri {
+            Some(uri) => Some(FileLocation::try_from_uri_str(&uri)?),
+            None => None,
+        };
+
+        self.set_location(&element, file_location)
+    }
+
+    fn get_uri_type() -> gst::URIType {
+        gst::URIType::Sink
+    }
+
+    fn get_protocols() -> Vec<String> {
+        vec!["file".to_string()]
+    }
+}
+
+pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
+    gst::Element::register(plugin, "rsfilesrc", 256 + 100, FileSrc::get_type())
+}

gst-plugin-file/src/lib.rs

@@ -8,64 +8,31 @@
 
 #![crate_type = "cdylib"]
 
+#[macro_use]
 extern crate glib;
 #[macro_use]
-extern crate gst_plugin;
-extern crate gst_plugin_simple;
-#[macro_use]
 extern crate gstreamer as gst;
+extern crate gstreamer_base as gst_base;
 extern crate url;
 
-use gst_plugin_simple::sink::*;
-use gst_plugin_simple::source::*;
-
+mod file_location;
 mod filesink;
 mod filesrc;
 
-use filesink::FileSink;
-use filesrc::FileSrc;
-
 fn plugin_init(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
-    source_register(
-        plugin,
-        SourceInfo {
-            name: "rsfilesrc".into(),
-            long_name: "File Source".into(),
-            description: "Reads local files".into(),
-            classification: "Source/File".into(),
-            author: "Sebastian Dröge <sebastian@centricular.com>".into(),
-            rank: 256 + 100,
-            create_instance: FileSrc::new_boxed,
-            protocols: vec!["file".into()],
-            push_only: false,
-        },
-    )?;
-
-    sink_register(
-        plugin,
-        SinkInfo {
-            name: "rsfilesink".into(),
-            long_name: "File Sink".into(),
-            description: "Writes to local files".into(),
-            classification: "Sink/File".into(),
-            author: "Luis de Bethencourt <luisbg@osg.samsung.com>".into(),
-            rank: 256 + 100,
-            create_instance: FileSink::new_boxed,
-            protocols: vec!["file".into()],
-        },
-    )?;
-
+    filesink::register(plugin)?;
+    filesrc::register(plugin)?;
     Ok(())
 }
 
-plugin_define!(
-    b"rsfile\0",
-    b"Rust File Plugin\0",
+gst_plugin_define!(
+    "rsfile",
+    "Rust File Plugin",
     plugin_init,
-    b"1.0\0",
-    b"MIT/X11\0",
-    b"rsfile\0",
-    b"rsfile\0",
-    b"https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs\0",
-    b"2016-12-08\0"
+    "1.0",
+    "MIT/X11",
+    "rsfile",
+    "rsfile",
+    "https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs",
+    "2016-12-08"
 );