Properly report error / error messages and make the trait APIs more Rust-like

src/error.rs

@@ -0,0 +1,248 @@
+//  Copyright (C) 2016 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 St, Fifth Floor,
+//  Boston, MA 02110-1301, USA.
+//
+//
+use libc::c_char;
+use std::os::raw::c_void;
+use std::ffi::CString;
+use std::ptr;
+use std::error::Error;
+use std::fmt::{Display, Formatter};
+use std::fmt::Error as FmtError;
+use std::borrow::Cow;
+
+use utils::*;
+
+macro_rules! error_msg(
+// Plain strings
+    ($err:expr, ($msg:expr), [$dbg:expr]) =>  {
+        ErrorMessage::new(&$err, Some(From::from($msg)),
+                          Some(From::from($dbg)),
+                          file!(), module_path!(), line!())
+    };
+    ($err:expr, ($msg:expr)) => {
+        ErrorMessage::new(&$err, Some(From::from($msg)),
+                          None,
+                          file!(), module_path!(), line!())
+    };
+    ($err:expr, [$dbg:expr]) => {
+        ErrorMessage::new(&$err, None,
+                          Some(From::from($dbg)),
+                          file!(), module_path!(), line!())
+    };
+
+// Format strings
+    ($err:expr, ($($msg:tt)*), [$($dbg:tt)*]) =>  { {
+        ErrorMessage::new(&$err, Some(From::from(format!($($msg)*))),
+                          From::from(Some(format!($($dbg)*))),
+                          file!(), module_path!(), line!())
+    }};
+    ($err:expr, ($($msg:tt)*)) =>  { {
+        ErrorMessage::new(&$err, Some(From::from(format!($($msg)*))),
+                          None,
+                          file!(), module_path!(), line!())
+    }};
+
+    ($err:expr, [$($dbg:tt)*]) =>  { {
+        ErrorMessage::new(&$err, None,
+                          Some(From::from(format!($($dbg)*))),
+                          file!(), module_path!(), line!())
+    }};
+);
+
+pub trait ToGError {
+    fn to_gerror(&self) -> (u32, i32);
+}
+
+pub fn gst_library_error_domain() -> u32 {
+    extern "C" {
+        fn gst_library_error_quark() -> u32;
+    }
+
+    unsafe { gst_library_error_quark() }
+}
+
+pub fn gst_resource_error_domain() -> u32 {
+    extern "C" {
+        fn gst_resource_error_quark() -> u32;
+    }
+
+    unsafe { gst_resource_error_quark() }
+}
+
+#[derive(Debug)]
+pub struct ErrorMessage {
+    pub error_domain: u32,
+    pub error_code: i32,
+    pub message: Option<String>,
+    pub debug: Option<String>,
+    pub filename: &'static str,
+    pub function: &'static str,
+    pub line: u32,
+}
+
+impl ErrorMessage {
+    pub fn new<T: ToGError>(error: &T,
+                            message: Option<Cow<str>>,
+                            debug: Option<Cow<str>>,
+                            filename: &'static str,
+                            function: &'static str,
+                            line: u32)
+                            -> ErrorMessage {
+        let (gdomain, gcode) = error.to_gerror();
+
+        ErrorMessage {
+            error_domain: gdomain,
+            error_code: gcode,
+            message: message.map(|m| m.into_owned()),
+            debug: debug.map(|d| d.into_owned()),
+            filename: filename,
+            function: function,
+            line: line,
+        }
+    }
+}
+
+#[derive(Debug)]
+pub enum FlowError {
+    NotLinked,
+    Flushing,
+    Eos,
+    NotNegotiated(ErrorMessage),
+    Error(ErrorMessage),
+}
+
+impl FlowError {
+    pub fn to_native(&self) -> GstFlowReturn {
+        match *self {
+            FlowError::NotLinked => GstFlowReturn::NotLinked,
+            FlowError::Flushing => GstFlowReturn::Flushing,
+            FlowError::Eos => GstFlowReturn::Eos,
+            FlowError::NotNegotiated(..) => GstFlowReturn::NotNegotiated,
+            FlowError::Error(..) => GstFlowReturn::Error,
+        }
+    }
+}
+
+impl Display for FlowError {
+    fn fmt(&self, f: &mut Formatter) -> Result<(), FmtError> {
+        match *self {
+            FlowError::NotLinked | FlowError::Flushing | FlowError::Eos => {
+                f.write_str(self.description())
+            }
+            FlowError::NotNegotiated(ref m) => {
+                f.write_fmt(format_args!("{}: {} ({})",
+                                         self.description(),
+                                         m.message.as_ref().map_or("None", |s| s.as_str()),
+                                         m.debug.as_ref().map_or("None", |s| s.as_str())))
+            }
+            FlowError::Error(ref m) => {
+                f.write_fmt(format_args!("{}: {} ({})",
+                                         self.description(),
+                                         m.message.as_ref().map_or("None", |s| s.as_str()),
+                                         m.debug.as_ref().map_or("None", |s| s.as_str())))
+            }
+        }
+    }
+}
+
+impl Error for FlowError {
+    fn description(&self) -> &str {
+        match *self {
+            FlowError::NotLinked => "Not Linked",
+            FlowError::Flushing => "Flushing",
+            FlowError::Eos => "Eos",
+            FlowError::NotNegotiated(..) => "Not Negotiated",
+            FlowError::Error(..) => "Error",
+        }
+    }
+}
+
+#[derive(Debug)]
+pub enum UriErrorKind {
+    UnsupportedProtocol = 0,
+    BadUri,
+    BadState,
+    BadReference,
+}
+
+#[derive(Debug)]
+pub struct UriError {
+    error_kind: UriErrorKind,
+    message: Option<String>,
+}
+
+impl UriError {
+    pub fn new(error_kind: UriErrorKind, message: Option<String>) -> UriError {
+        UriError {
+            error_kind: error_kind,
+            message: message,
+        }
+    }
+
+    pub fn message(&self) -> &Option<String> {
+        &self.message
+    }
+
+    pub fn kind(&self) -> &UriErrorKind {
+        &self.error_kind
+    }
+
+    pub unsafe fn into_gerror(self, err: *mut c_void) {
+        extern "C" {
+            fn g_set_error_literal(err: *mut c_void,
+                                   domain: u32,
+                                   code: i32,
+                                   message: *const c_char);
+            fn gst_uri_error_quark() -> u32;
+        }
+
+
+        if let Some(msg) = self.message {
+            let cmsg = CString::new(msg.as_str()).unwrap();
+            g_set_error_literal(err,
+                                gst_uri_error_quark(),
+                                self.error_kind as i32,
+                                cmsg.as_ptr());
+        } else {
+            g_set_error_literal(err,
+                                gst_uri_error_quark(),
+                                self.error_kind as i32,
+                                ptr::null());
+        }
+    }
+}
+
+impl Display for UriError {
+    fn fmt(&self, f: &mut Formatter) -> Result<(), FmtError> {
+        match self.message {
+            None => f.write_str(self.description()),
+            Some(ref message) => f.write_fmt(format_args!("{}: {}", self.description(), message)),
+        }
+    }
+}
+
+impl Error for UriError {
+    fn description(&self) -> &str {
+        match self.error_kind {
+            UriErrorKind::UnsupportedProtocol => "Unsupported protocol",
+            UriErrorKind::BadUri => "Bad URI",
+            UriErrorKind::BadState => "Bad State",
+            UriErrorKind::BadReference => "Bad Reference",
+        }
+    }
+}

src/lib.rs

@@ -24,6 +24,8 @@ extern crate hyper;
 
 #[macro_use]
 pub mod utils;
+#[macro_use]
+pub mod error;
 pub mod rssource;
 pub mod rssink;
 pub mod rsfilesrc;

src/rsfilesink.rs

@@ -22,10 +22,48 @@ use url::Url;
 
 use std::io::Write;
 use std::sync::Mutex;
+use std::convert::From;
 
-use utils::*;
+use error::*;
 use rssink::*;
 
+macro_rules! error_msg(
+// Format strings
+    ($err:expr, ($($msg:tt)*), [$($dbg:tt)*]) =>  { {
+        ErrorMessage::new(&$err, Some(From::from(format!($($msg)*))),
+                          From::from(Some(format!($($dbg)*))),
+                          file!(), module_path!(), line!())
+    }};
+    ($err:expr, ($($msg:tt)*)) =>  { {
+        ErrorMessage::new(&$err, Some(From::from(format!($($msg)*))),
+                          None,
+                          file!(), module_path!(), line!())
+    }};
+
+    ($err:expr, [$($dbg:tt)*]) =>  { {
+        ErrorMessage::new(&$err, None,
+                          Some(From::from(format!($($dbg)*))),
+                          file!(), module_path!(), line!())
+    }};
+
+// Plain strings
+    ($err:expr, ($msg:expr), [$dbg:expr]) =>  {
+        ErrorMessage::new(&$err, Some(From::from($msg)),
+                          Some(From::from($dbg)),
+                          file!(), module_path!(), line!())
+    };
+    ($err:expr, ($msg:expr)) => {
+        ErrorMessage::new(&$err, Some(From::from($msg)),
+                          None,
+                          file!(), module_path!(), line!())
+    };
+    ($err:expr, [$dbg:expr]) => {
+        ErrorMessage::new(&$err, None,
+                          Some(From::from($dbg)),
+                          file!(), module_path!(), line!())
+    };
+);
+
 #[derive(Debug)]
 struct Settings {
     location: Option<PathBuf>,
@@ -69,23 +107,16 @@ impl Sink for FileSink {
     fn set_uri(&self, uri: Option<Url>) -> Result<(), UriError> {
         let location = &mut self.settings.lock().unwrap().location;
 
+        *location = None;
         match uri {
-            None => {
-                *location = None;
-                Ok(())
-            }
+            None => Ok(()),
             Some(ref uri) => {
-                match uri.to_file_path().ok() {
-                    Some(p) => {
-                        *location = Some(p);
-                        Ok(())
-                    }
-                    None => {
-                        *location = None;
+                *location = Some(try!(uri.to_file_path()
+                    .or_else(|_| {
                         Err(UriError::new(UriErrorKind::UnsupportedProtocol,
                                           Some(format!("Unsupported file URI '{}'", uri.as_str()))))
-                    }
-                }
+                    })));
+                Ok(())
             }
         }
     }
@@ -97,59 +128,55 @@ impl Sink for FileSink {
             .and_then(|i| i) // join()
     }
 
-    fn start(&self) -> bool {
+    fn start(&self) -> Result<(), ErrorMessage> {
         let location = &self.settings.lock().unwrap().location;
         let mut streaming_state = self.streaming_state.lock().unwrap();
 
         if let StreamingState::Started { .. } = *streaming_state {
-            return false;
+            return Err(error_msg!(SinkError::Failure, ["Sink already started"]));
         }
 
-        match *location {
-            None => false,
-            Some(ref location) => {
-                match File::create(location.as_path()) {
-                    Ok(file) => {
-                        *streaming_state = StreamingState::Started {
-                            file: file,
-                            position: 0,
-                        };
-                        true
-                    }
-                    Err(err) => {
-                        println_err!("Could not open file for writing '{}': {}",
-                                     location.to_str().unwrap_or("Non-UTF8 path"),
-                                     err.to_string());
-                        false
-                    }
-                }
-            }
-        }
+        let location = &try!(location.as_ref().ok_or_else(|| {
+            error_msg!(SinkError::Failure, ["No URI provided"])
+        }));
+
+        let file = try!(File::create(location.as_path()).or_else(|err| {
+            Err(error_msg!(SinkError::OpenFailed,
+                           ["Could not open file for writing '{}': {}",
+                            location.to_str().unwrap_or("Non-UTF8 path"),
+                            err.to_string()]))
+        }));
+
+        *streaming_state = StreamingState::Started {
+            file: file,
+            position: 0,
+        };
+
+        Ok(())
     }
 
-    fn stop(&self) -> bool {
+    fn stop(&self) -> Result<(), ErrorMessage> {
         let mut streaming_state = self.streaming_state.lock().unwrap();
         *streaming_state = StreamingState::Stopped;
 
-        true
+        Ok(())
     }
 
-    fn render(&self, data: &[u8]) -> GstFlowReturn {
+    fn render(&self, data: &[u8]) -> Result<(), FlowError> {
         let mut streaming_state = self.streaming_state.lock().unwrap();
 
-        if let StreamingState::Started { ref mut file, ref mut position } = *streaming_state {
-            match file.write_all(data) {
-                Ok(_) => {
-                    *position += data.len() as u64;
-                    GstFlowReturn::Ok
-                }
-                Err(err) => {
-                    println_err!("Failed to write: {}", err);
-                    GstFlowReturn::Error
-                }
+        let (file, position) = match *streaming_state {
+            StreamingState::Started { ref mut file, ref mut position } => (file, position),
+            StreamingState::Stopped => {
+                return Err(FlowError::Error(error_msg!(SinkError::Failure, ["Not started yet"])));
             }
-        } else {
-            GstFlowReturn::Error
-        }
+        };
+
+        try!(file.write_all(data).or_else(|err| {
+            Err(FlowError::Error(error_msg!(SinkError::WriteFailed, ["Failed to write: {}", err])))
+        }));
+
+        *position += data.len() as u64;
+        Ok(())
     }
 }

src/rsfilesrc.rs

@@ -22,9 +22,7 @@ use std::path::PathBuf;
 use std::sync::Mutex;
 use url::Url;
 
-use std::io::Write;
-
-use utils::*;
+use error::*;
 use rssource::*;
 
 #[derive(Debug)]
@@ -76,17 +74,12 @@ impl Source for FileSrc {
                 Ok(())
             }
             Some(ref uri) => {
-                match uri.to_file_path().ok() {
-                    Some(p) => {
-                        *location = Some(p);
-                        Ok(())
-                    }
-                    None => {
-                        *location = None;
+                *location = Some(try!(uri.to_file_path()
+                    .or_else(|_| {
                         Err(UriError::new(UriErrorKind::UnsupportedProtocol,
                                           Some(format!("Unsupported file URI '{}'", uri.as_str()))))
-                    }
-                }
+                    })));
+                Ok(())
             }
         }
     }
@@ -115,75 +108,71 @@ impl Source for FileSrc {
         }
     }
 
-    fn start(&self) -> bool {
+    fn start(&self) -> Result<(), ErrorMessage> {
         let location = &self.settings.lock().unwrap().location;
         let mut streaming_state = self.streaming_state.lock().unwrap();
 
         if let StreamingState::Started { .. } = *streaming_state {
-            return false;
+            return Err(error_msg!(SourceError::Failure, ["Source already started"]));
         }
 
-        match *location {
-            None => false,
-            Some(ref location) => {
-                match File::open(location.as_path()) {
-                    Ok(file) => {
-                        *streaming_state = StreamingState::Started {
-                            file: file,
-                            position: 0,
-                        };
-                        true
-                    }
-                    Err(err) => {
-                        println_err!("Could not open file for writing '{}': {}",
-                                     location.to_str().unwrap_or("Non-UTF8 path"),
-                                     err.to_string());
-                        false
-                    }
-                }
-            }
-        }
+        let location = &try!(location.as_ref()
+            .ok_or_else(|| error_msg!(SourceError::Failure, ["No URI provided"])));
+
+        let file = try!(File::open(location.as_path()).or_else(|err| {
+            Err(error_msg!(SourceError::OpenFailed,
+                           ["Could not open file for reading '{}': {}",
+                            location.to_str().unwrap_or("Non-UTF8 path"),
+                            err.to_string()]))
+        }));
+
+        *streaming_state = StreamingState::Started {
+            file: file,
+            position: 0,
+        };
+
+        Ok(())
     }
 
-    fn stop(&self) -> bool {
+    fn stop(&self) -> Result<(), ErrorMessage> {
         let mut streaming_state = self.streaming_state.lock().unwrap();
         *streaming_state = StreamingState::Stopped;
 
-        true
+        Ok(())
     }
 
-    fn fill(&self, offset: u64, data: &mut [u8]) -> Result<usize, GstFlowReturn> {
+    fn fill(&self, offset: u64, data: &mut [u8]) -> Result<usize, FlowError> {
         let mut streaming_state = self.streaming_state.lock().unwrap();
 
-        if let StreamingState::Started { ref mut file, ref mut position } = *streaming_state {
-            if *position != offset {
-                match file.seek(SeekFrom::Start(offset)) {
-                    Ok(_) => {
-                        *position = offset;
-                    }
-                    Err(err) => {
-                        println_err!("Failed to seek to {}: {}", offset, err.to_string());
-                        return Err(GstFlowReturn::Error);
-                    }
-                }
-            }
 
-            match file.read(data) {
-                Ok(size) => {
-                    *position += size as u64;
-                    Ok(size)
-                }
-                Err(err) => {
-                    println_err!("Failed to read at {}: {}", offset, err.to_string());
-                    Err(GstFlowReturn::Error)
-                }
+        let (file, position) = match *streaming_state {
+            StreamingState::Started { ref mut file, ref mut position } => (file, position),
+            StreamingState::Stopped => {
+                return Err(FlowError::Error(error_msg!(SourceError::Failure, ["Not started yet"])));
             }
-        } else {
-            Err(GstFlowReturn::Error)
+        };
+
+        if *position != offset {
+            try!(file.seek(SeekFrom::Start(offset)).or_else(|err| {
+                Err(FlowError::Error(error_msg!(SourceError::SeekFailed,
+                                                ["Failed to seek to {}: {}",
+                                                 offset,
+                                                 err.to_string()])))
+            }));
+            *position = offset;
         }
+
+        let size = try!(file.read(data).or_else(|err| {
+            Err(FlowError::Error(error_msg!(SourceError::ReadFailed,
+                                            ["Failed to read at {}: {}", offset, err.to_string()])))
+        }));
+
+        *position += size as u64;
+
+        Ok(size)
     }
 
-    fn do_seek(&self, _: u64, _: u64) -> bool {
-        true
+    fn do_seek(&self, _: u64, _: u64) -> Result<(), ErrorMessage> {
+        Ok(())
     }
 }

src/rshttpsrc.rs

@@ -23,10 +23,9 @@ use hyper::header::{ContentLength, ContentRange, ContentRangeSpec, Range, ByteRa
 use hyper::client::Client;
 use hyper::client::response::Response;
 
-use std::io::Write;
 use std::sync::Mutex;
 
-use utils::*;
+use error::*;
 use rssource::*;
 
 #[derive(Debug)]
@@ -72,71 +71,70 @@ impl HttpSrc {
         Box::new(HttpSrc::new(controller))
     }
 
-    fn do_request(&self, start: u64, stop: u64) -> StreamingState {
+    fn do_request(&self, start: u64, stop: u64) -> Result<StreamingState, ErrorMessage> {
         let url = &self.settings.lock().unwrap().url;
 
-        match *url {
-            None => StreamingState::Stopped,
-            Some(ref url) => {
-                let mut req = self.client.get(url.clone());
+        let url = try!(url.as_ref()
+            .ok_or_else(|| error_msg!(SourceError::Failure, ["No URI provided"])));
 
-                if start != 0 || stop != u64::MAX {
-                    req = if stop == u64::MAX {
-                        req.header(Range::Bytes(vec![ByteRangeSpec::AllFrom(start)]))
-                    } else {
-                        req.header(Range::Bytes(vec![ByteRangeSpec::FromTo(start, stop - 1)]))
-                    };
-                }
 
-                match req.send() {
-                    Ok(response) => {
-                        if response.status.is_success() {
-                            let size = if let Some(&ContentLength(content_length)) =
-                                              response.headers.get() {
-                                content_length + start
-                            } else {
-                                u64::MAX
-                            };
-                            let accept_byte_ranges = if let Some(&AcceptRanges(ref ranges)) =
-                                                            response.headers.get() {
-                                ranges.iter().any(|u| *u == RangeUnit::Bytes)
-                            } else {
-                                false
-                            };
+        let mut req = self.client.get(url.clone());
 
-                            let seekable = size != u64::MAX && accept_byte_ranges;
-
-                            let position = if let Some(&ContentRange(ContentRangeSpec::Bytes{range: Some((range_start, _)), ..})) = response.headers.get() {
-                                range_start
-                            } else {
-                                start
-                            };
-
-                            if position != start {
-                                println_err!("Failed to seek to {}: Got {}", start, position);
-                                StreamingState::Stopped
-                            } else {
-                                StreamingState::Started {
-                                    response: response,
-                                    seekable: seekable,
-                                    position: 0,
-                                    size: size,
-                                    start: start,
-                                    stop: stop,
-                                }
-                            }
-                        } else {
-                            println_err!("Failed to fetch {}: {}", url, response.status);
-                            StreamingState::Stopped
-                        }
-                    }
-                    Err(err) => {
-                        println_err!("Failed to fetch {}: {}", url, err.to_string());
-                        StreamingState::Stopped
-                    }
-                }
-            }
+        if start != 0 || stop != u64::MAX {
+            req = if stop == u64::MAX {
+                req.header(Range::Bytes(vec![ByteRangeSpec::AllFrom(start)]))
+            } else {
+                req.header(Range::Bytes(vec![ByteRangeSpec::FromTo(start, stop - 1)]))
+            };
         }
+
+        let response = try!(req.send().or_else(|err| {
+            Err(error_msg!(SourceError::ReadFailed,
+                           ["Failed to fetch {}: {}", url, err.to_string()]))
+        }));
+
+        if !response.status.is_success() {
+            return Err(error_msg!(SourceError::ReadFailed,
+                                  ["Failed to fetch {}: {}", url, response.status]));
+        }
+
+        let size = if let Some(&ContentLength(content_length)) = response.headers.get() {
+            content_length + start
+        } else {
+            u64::MAX
+        };
+
+        let accept_byte_ranges = if let Some(&AcceptRanges(ref ranges)) = response.headers
+            .get() {
+            ranges.iter().any(|u| *u == RangeUnit::Bytes)
+        } else {
+            false
+        };
+
+        let seekable = size != u64::MAX && accept_byte_ranges;
+
+        let position = if let Some(&ContentRange(ContentRangeSpec::Bytes { range: Some((range_start,
+                                                                                 _)),
+                                                                           .. })) = response.headers
+            .get() {
+            range_start
+        } else {
+            start
+        };
+
+        if position != start {
+            return Err(error_msg!(SourceError::SeekFailed,
+                                  ["Failed to seek to {}: Got {}", start, position]));
+        }
+
+        Ok(StreamingState::Started {
+            response: response,
+            seekable: seekable,
+            position: 0,
+            size: size,
+            start: start,
+            stop: stop,
+        })
     }
 }
 
@@ -154,14 +152,14 @@ impl Source for HttpSrc {
                 Ok(())
             }
             Some(uri) => {
-                if uri.scheme() == "http" || uri.scheme() == "https" {
-                    *url = Some(uri);
-                    Ok(())
-                } else {
+                if uri.scheme() != "http" && uri.scheme() != "https" {
                     *url = None;
-                    Err(UriError::new(UriErrorKind::UnsupportedProtocol,
-                                      Some(format!("Unsupported URI '{}'", uri.as_str()))))
+                    return Err(UriError::new(UriErrorKind::UnsupportedProtocol,
+                                             Some(format!("Unsupported URI '{}'", uri.as_str()))));
                 }
+
+                *url = Some(uri);
+                Ok(())
             }
         }
     }
@@ -188,70 +186,65 @@ impl Source for HttpSrc {
         }
     }
 
-    fn start(&self) -> bool {
-        let mut streaming_state = self.streaming_state.lock().unwrap();
-        *streaming_state = self.do_request(0, u64::MAX);
-
-        if let StreamingState::Stopped = *streaming_state {
-            false
-        } else {
-            true
-        }
-    }
-
-    fn stop(&self) -> bool {
+    fn start(&self) -> Result<(), ErrorMessage> {
         let mut streaming_state = self.streaming_state.lock().unwrap();
         *streaming_state = StreamingState::Stopped;
 
-        true
+        let new_state = try!(self.do_request(0, u64::MAX));
+
+        *streaming_state = new_state;
+        Ok(())
     }
 
-    fn do_seek(&self, start: u64, stop: u64) -> bool {
+    fn stop(&self) -> Result<(), ErrorMessage> {
         let mut streaming_state = self.streaming_state.lock().unwrap();
-        *streaming_state = self.do_request(start, stop);
+        *streaming_state = StreamingState::Stopped;
 
-        if let StreamingState::Stopped = *streaming_state {
-            false
-        } else {
-            true
-        }
+        Ok(())
     }
 
-    fn fill(&self, offset: u64, data: &mut [u8]) -> Result<usize, GstFlowReturn> {
+    fn do_seek(&self, start: u64, stop: u64) -> Result<(), ErrorMessage> {
         let mut streaming_state = self.streaming_state.lock().unwrap();
+        *streaming_state = StreamingState::Stopped;
 
-        if let StreamingState::Stopped = *streaming_state {
-            return Err(GstFlowReturn::Error);
-        }
+        let new_state = try!(self.do_request(start, stop));
+
+        *streaming_state = new_state;
+        Ok(())
+    }
+
+    fn fill(&self, offset: u64, data: &mut [u8]) -> Result<usize, FlowError> {
+        let mut streaming_state = self.streaming_state.lock().unwrap();
 
         if let StreamingState::Started { position, stop, .. } = *streaming_state {
             if position != offset {
-                *streaming_state = self.do_request(offset, stop);
-                if let StreamingState::Stopped = *streaming_state {
-                    println_err!("Failed to seek to {}", offset);
-                    return Err(GstFlowReturn::Error);
-                }
+                *streaming_state = StreamingState::Stopped;
+                let new_state = try!(self.do_request(offset, stop)
+                    .or_else(|err| Err(FlowError::Error(err))));
+
+                *streaming_state = new_state;
             }
         }
 
-        if let StreamingState::Started { ref mut response, ref mut position, .. } =
-               *streaming_state {
-            match response.read(data) {
-                Ok(size) => {
-                    if size == 0 {
-                        return Err(GstFlowReturn::Eos);
-                    }
-
-                    *position += size as u64;
-                    Ok(size)
-                }
-                Err(err) => {
-                    println_err!("Failed to read at {}: {}", offset, err.to_string());
-                    Err(GstFlowReturn::Error)
-                }
+        let (response, position) = match *streaming_state {
+            StreamingState::Started { ref mut response, ref mut position, .. } => {
+                (response, position)
             }
-        } else {
-            Err(GstFlowReturn::Error)
+            StreamingState::Stopped => {
+                return Err(FlowError::Error(error_msg!(SourceError::Failure, ["Not started yet"])));
+            }
+        };
+
+        let size = try!(response.read(data).or_else(|err| {
+            Err(FlowError::Error(error_msg!(SourceError::ReadFailed,
+                                            ["Failed to read at {}: {}", offset, err.to_string()])))
+        }));
+
+        if size == 0 {
+            return Err(FlowError::Eos);
         }
+
+        *position += size as u64;
+        Ok(size)
     }
 }

src/rssink.c

@@ -246,6 +246,15 @@ gst_rs_sink_uri_handler_init (gpointer g_iface, gpointer iface_data)
   iface->set_uri = gst_rs_sink_uri_set_uri;
 }
 
+void
+gst_rs_sink_error (GstRsSink * sink, GQuark error_domain, gint error_code,
+    const gchar * message, const gchar * debug, const gchar * file,
+    const gchar * function, guint line)
+{
+  gst_element_message_full (GST_ELEMENT (sink), GST_MESSAGE_ERROR, error_domain,
+      error_code, g_strdup (message), g_strdup (debug), file, function, line);
+}
+
 gboolean
 gst_rs_sink_plugin_init (GstPlugin * plugin)
 {

src/rssink.rs

@@ -25,6 +25,28 @@ use std::ptr;
 use url::Url;
 
 use utils::*;
+use error::*;
+
+#[derive(Debug)]
+pub enum SinkError {
+    Failure,
+    OpenFailed,
+    NotFound,
+    WriteFailed,
+    SeekFailed,
+}
+
+impl ToGError for SinkError {
+    fn to_gerror(&self) -> (u32, i32) {
+        match *self {
+            SinkError::Failure => (gst_library_error_domain(), 1),
+            SinkError::OpenFailed => (gst_resource_error_domain(), 6),
+            SinkError::NotFound => (gst_resource_error_domain(), 3),
+            SinkError::WriteFailed => (gst_resource_error_domain(), 10),
+            SinkError::SeekFailed => (gst_resource_error_domain(), 11),
+        }
+    }
+}
 
 #[derive(Debug)]
 pub struct SinkController {
@@ -35,6 +57,50 @@ impl SinkController {
     fn new(sink: *mut c_void) -> SinkController {
         SinkController { sink: sink }
     }
+
+    pub fn error(&self, error: &ErrorMessage) {
+        extern "C" {
+            fn gst_rs_sink_error(sink: *mut c_void,
+                                 error_domain: u32,
+                                 error_code: i32,
+                                 message: *const c_char,
+                                 debug: *const c_char,
+                                 filename: *const c_char,
+                                 function: *const c_char,
+                                 line: u32);
+        }
+
+        let ErrorMessage { error_domain,
+                           error_code,
+                           ref message,
+                           ref debug,
+                           filename,
+                           function,
+                           line } = *error;
+
+        let message_cstr = message.as_ref().map(|m| CString::new(m.as_bytes()).unwrap());
+        let message_ptr = message_cstr.as_ref().map_or(ptr::null(), |m| m.as_ptr());
+
+        let debug_cstr = debug.as_ref().map(|m| CString::new(m.as_bytes()).unwrap());
+        let debug_ptr = debug_cstr.as_ref().map_or(ptr::null(), |m| m.as_ptr());
+
+        let file_cstr = CString::new(filename.as_bytes()).unwrap();
+        let file_ptr = file_cstr.as_ptr();
+
+        let function_cstr = CString::new(function.as_bytes()).unwrap();
+        let function_ptr = function_cstr.as_ptr();
+
+        unsafe {
+            gst_rs_sink_error(self.sink,
+                              error_domain,
+                              error_code,
+                              message_ptr,
+                              debug_ptr,
+                              file_ptr,
+                              function_ptr,
+                              line);
+        }
+    }
 }
 
 pub trait Sink: Sync + Send {
@@ -45,9 +111,9 @@ pub trait Sink: Sync + Send {
     fn get_uri(&self) -> Option<Url>;
 
     // Called from the streaming thread only
-    fn start(&self) -> bool;
-    fn stop(&self) -> bool;
-    fn render(&self, data: &[u8]) -> GstFlowReturn;
+    fn start(&self) -> Result<(), ErrorMessage>;
+    fn stop(&self) -> Result<(), ErrorMessage>;
+    fn render(&self, data: &[u8]) -> Result<(), FlowError>;
 }
 
 #[no_mangle]
@@ -117,19 +183,41 @@ pub unsafe extern "C" fn sink_render(ptr: *mut Box<Sink>,
     let sink: &mut Box<Sink> = &mut *ptr;
     let data = slice::from_raw_parts(data_ptr, data_len);
 
-    sink.render(data)
+    match sink.render(data) {
+        Ok(..) => GstFlowReturn::Ok,
+        Err(flow_error) => {
+            match flow_error {
+                FlowError::NotNegotiated(ref msg) |
+                FlowError::Error(ref msg) => sink.get_controller().error(msg),
+                _ => (),
+            }
+            flow_error.to_native()
+        }
+    }
 }
 
 #[no_mangle]
 pub unsafe extern "C" fn sink_start(ptr: *mut Box<Sink>) -> GBoolean {
     let sink: &mut Box<Sink> = &mut *ptr;
 
-    GBoolean::from_bool(sink.start())
+    match sink.start() {
+        Ok(..) => GBoolean::True,
+        Err(ref msg) => {
+            sink.get_controller().error(msg);
+            GBoolean::False
+        }
+    }
 }
 
 #[no_mangle]
 pub unsafe extern "C" fn sink_stop(ptr: *mut Box<Sink>) -> GBoolean {
     let sink: &mut Box<Sink> = &mut *ptr;
 
-    GBoolean::from_bool(sink.stop())
+    match sink.stop() {
+        Ok(..) => GBoolean::True,
+        Err(ref msg) => {
+            sink.get_controller().error(msg);
+            GBoolean::False
+        }
+    }
 }

src/rssource.c

@@ -291,6 +291,15 @@ gst_rs_src_uri_handler_init (gpointer g_iface, gpointer iface_data)
   iface->set_uri = gst_rs_src_uri_set_uri;
 }
 
+void
+gst_rs_source_error (GstRsSrc * src, GQuark error_domain, gint error_code,
+    const gchar * message, const gchar * debug, const gchar * file,
+    const gchar * function, guint line)
+{
+  gst_element_message_full (GST_ELEMENT (src), GST_MESSAGE_ERROR, error_domain,
+      error_code, g_strdup (message), g_strdup (debug), file, function, line);
+}
+
 gboolean
 gst_rs_source_plugin_init (GstPlugin * plugin)
 {

src/rssource.rs

@@ -24,6 +24,28 @@ use std::ptr;
 use url::Url;
 
 use utils::*;
+use error::*;
+
+#[derive(Debug)]
+pub enum SourceError {
+    Failure,
+    OpenFailed,
+    NotFound,
+    ReadFailed,
+    SeekFailed,
+}
+
+impl ToGError for SourceError {
+    fn to_gerror(&self) -> (u32, i32) {
+        match *self {
+            SourceError::Failure => (gst_library_error_domain(), 1),
+            SourceError::OpenFailed => (gst_resource_error_domain(), 5),
+            SourceError::NotFound => (gst_resource_error_domain(), 3),
+            SourceError::ReadFailed => (gst_resource_error_domain(), 9),
+            SourceError::SeekFailed => (gst_resource_error_domain(), 11),
+        }
+    }
+}
 
 #[derive(Debug)]
 pub struct SourceController {
@@ -34,6 +56,50 @@ impl SourceController {
     fn new(source: *mut c_void) -> SourceController {
         SourceController { source: source }
     }
+
+    pub fn error(&self, error: &ErrorMessage) {
+        extern "C" {
+            fn gst_rs_source_error(source: *mut c_void,
+                                   error_domain: u32,
+                                   error_code: i32,
+                                   message: *const c_char,
+                                   debug: *const c_char,
+                                   filename: *const c_char,
+                                   function: *const c_char,
+                                   line: u32);
+        }
+
+        let ErrorMessage { error_domain,
+                           error_code,
+                           ref message,
+                           ref debug,
+                           filename,
+                           function,
+                           line } = *error;
+
+        let message_cstr = message.as_ref().map(|m| CString::new(m.as_bytes()).unwrap());
+        let message_ptr = message_cstr.as_ref().map_or(ptr::null(), |m| m.as_ptr());
+
+        let debug_cstr = debug.as_ref().map(|m| CString::new(m.as_bytes()).unwrap());
+        let debug_ptr = debug_cstr.as_ref().map_or(ptr::null(), |m| m.as_ptr());
+
+        let file_cstr = CString::new(filename.as_bytes()).unwrap();
+        let file_ptr = file_cstr.as_ptr();
+
+        let function_cstr = CString::new(function.as_bytes()).unwrap();
+        let function_ptr = function_cstr.as_ptr();
+
+        unsafe {
+            gst_rs_source_error(self.source,
+                                error_domain,
+                                error_code,
+                                message_ptr,
+                                debug_ptr,
+                                file_ptr,
+                                function_ptr,
+                                line);
+        }
+    }
 }
 
 pub trait Source: Sync + Send {
@@ -47,10 +113,10 @@ pub trait Source: Sync + Send {
     fn is_seekable(&self) -> bool;
 
     // Called from the streaming thread only
-    fn start(&self) -> bool;
-    fn stop(&self) -> bool;
-    fn fill(&self, offset: u64, data: &mut [u8]) -> Result<usize, GstFlowReturn>;
-    fn do_seek(&self, start: u64, stop: u64) -> bool;
+    fn start(&self) -> Result<(), ErrorMessage>;
+    fn stop(&self) -> Result<(), ErrorMessage>;
+    fn fill(&self, offset: u64, data: &mut [u8]) -> Result<usize, FlowError>;
+    fn do_seek(&self, start: u64, stop: u64) -> Result<(), ErrorMessage>;
     fn get_size(&self) -> u64;
 }
 
@@ -129,7 +195,14 @@ pub unsafe extern "C" fn source_fill(ptr: *mut Box<Source>,
             *data_len = actual_len;
             GstFlowReturn::Ok
         }
-        Err(ret) => ret,
+        Err(flow_error) => {
+            match flow_error {
+                FlowError::NotNegotiated(ref msg) |
+                FlowError::Error(ref msg) => source.get_controller().error(msg),
+                _ => (),
+            }
+            flow_error.to_native()
+        }
     }
 }
 
@@ -144,14 +217,26 @@ pub unsafe extern "C" fn source_get_size(ptr: *const Box<Source>) -> u64 {
 pub unsafe extern "C" fn source_start(ptr: *mut Box<Source>) -> GBoolean {
     let source: &mut Box<Source> = &mut *ptr;
 
-    GBoolean::from_bool(source.start())
+    match source.start() {
+        Ok(..) => GBoolean::True,
+        Err(ref msg) => {
+            source.get_controller().error(msg);
+            GBoolean::False
+        }
+    }
 }
 
 #[no_mangle]
 pub unsafe extern "C" fn source_stop(ptr: *mut Box<Source>) -> GBoolean {
     let source: &mut Box<Source> = &mut *ptr;
 
-    GBoolean::from_bool(source.stop())
+    match source.stop() {
+        Ok(..) => GBoolean::True,
+        Err(ref msg) => {
+            source.get_controller().error(msg);
+            GBoolean::False
+        }
+    }
 }
 
 #[no_mangle]
@@ -165,5 +250,11 @@ pub unsafe extern "C" fn source_is_seekable(ptr: *const Box<Source>) -> GBoolean
 pub unsafe extern "C" fn source_do_seek(ptr: *mut Box<Source>, start: u64, stop: u64) -> GBoolean {
     let source: &mut Box<Source> = &mut *ptr;
 
-    GBoolean::from_bool(source.do_seek(start, stop))
+    match source.do_seek(start, stop) {
+        Ok(..) => GBoolean::True,
+        Err(ref msg) => {
+            source.get_controller().error(msg);
+            GBoolean::False
+        }
+    }
 }

src/utils.rs

@@ -17,21 +17,7 @@
 //
 //
 use libc::c_char;
-use std::os::raw::c_void;
 use std::ffi::CString;
-use std::ptr;
-use std::error::Error;
-use std::fmt::{Display, Formatter};
-use std::fmt::Error as FmtError;
-
-#[macro_export]
-macro_rules! println_err(
-    ($($arg:tt)*) => { {
-        if let Err(_) = writeln!(&mut ::std::io::stderr(), $($arg)*) {
-// Ignore when writing fails
-        };
-    } }
-);
 
 #[repr(C)]
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
@@ -61,74 +47,3 @@ impl GBoolean {
 pub unsafe extern "C" fn cstring_drop(ptr: *mut c_char) {
     CString::from_raw(ptr);
 }
-
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
-pub enum UriErrorKind {
-    UnsupportedProtocol = 0,
-    BadUri,
-    BadState,
-    BadReference,
-}
-
-#[derive(Debug)]
-pub struct UriError {
-    error_kind: UriErrorKind,
-    message: Option<String>,
-}
-
-extern "C" {
-    fn g_set_error_literal(err: *mut c_void, domain: u32, code: i32, message: *const c_char);
-    fn gst_uri_error_quark() -> u32;
-}
-
-impl UriError {
-    pub fn new(error_kind: UriErrorKind, message: Option<String>) -> UriError {
-        UriError {
-            error_kind: error_kind,
-            message: message,
-        }
-    }
-
-    pub fn message(&self) -> &Option<String> {
-        &self.message
-    }
-
-    pub fn kind(&self) -> UriErrorKind {
-        self.error_kind
-    }
-
-    pub unsafe fn into_gerror(self, err: *mut c_void) {
-        if let Some(msg) = self.message {
-            let cmsg = CString::new(msg.as_str()).unwrap();
-            g_set_error_literal(err,
-                                gst_uri_error_quark(),
-                                self.error_kind as i32,
-                                cmsg.as_ptr());
-        } else {
-            g_set_error_literal(err,
-                                gst_uri_error_quark(),
-                                self.error_kind as i32,
-                                ptr::null());
-        }
-    }
-}
-
-impl Display for UriError {
-    fn fmt(&self, f: &mut Formatter) -> Result<(), FmtError> {
-        match self.message {
-            None => f.write_str(self.description()),
-            Some(ref message) => f.write_fmt(format_args!("{}: {}", self.description(), message)),
-        }
-    }
-}
-
-impl Error for UriError {
-    fn description(&self) -> &str {
-        match self.error_kind {
-            UriErrorKind::UnsupportedProtocol => "Unsupported protocol",
-            UriErrorKind::BadUri => "Bad URI",
-            UriErrorKind::BadState => "Bad State",
-            UriErrorKind::BadReference => "Bad Reference",
-        }
-    }
-}