threadshare: Run everything through rustfmt again

gst-plugin-threadshare/src/appsrc.rs

@@ -459,10 +459,7 @@ impl AppSrc {
 
         let mut state = self.state.lock().unwrap();
 
-        let io_context = IOContext::new(
+        let io_context = IOContext::new(&settings.context, settings.context_wait).map_err(|err| {
-            &settings.context,
-            settings.context_wait,
-        ).map_err(|err| {
             gst_error_msg!(
                 gst::ResourceError::OpenRead,
                 ["Failed to create IO context: {}", err]

gst-plugin-threadshare/src/dataqueue.rs

@@ -236,7 +236,8 @@ impl DataQueue {
         gst_debug!(DATA_QUEUE_CAT, obj: &inner.element, "Clearing queue");
         for item in inner.queue.drain(..) {
             if let DataQueueItem::Event(event) = item {
-                if event.is_sticky() && event.get_type() != gst::EventType::Segment
+                if event.is_sticky()
+                    && event.get_type() != gst::EventType::Segment
                     && event.get_type() != gst::EventType::Eos
                 {
                     let _ = src_pad.store_sticky_event(&event);

gst-plugin-threadshare/src/iocontext.rs

@@ -27,8 +27,8 @@ use futures::stream::futures_unordered::FuturesUnordered;
 use futures::sync::oneshot;
 use futures::{Future, Stream};
 use tokio::reactor;
-use tokio_timer::timer;
 use tokio_current_thread;
+use tokio_timer::timer;
 
 use gst;
 
@@ -84,7 +84,12 @@ impl IOContextRunner {
         (runtime_handle, shutdown)
     }
 
-    fn run(&mut self, wait: u32, reactor: reactor::Reactor, sender: mpsc::Sender<tokio_current_thread::Handle>) {
+    fn run(
+        &mut self,
+        wait: u32,
+        reactor: reactor::Reactor,
+        sender: mpsc::Sender<tokio_current_thread::Handle>,
+    ) {
         use std::time;
         let wait = time::Duration::from_millis(wait as u64);
 

gst-plugin-threadshare/src/lib.rs

@@ -29,9 +29,9 @@ extern crate gstreamer as gst;
 
 extern crate futures;
 extern crate tokio;
+extern crate tokio_current_thread;
 extern crate tokio_executor;
 extern crate tokio_reactor;
-extern crate tokio_current_thread;
 extern crate tokio_timer;
 
 extern crate either;

gst-plugin-threadshare/src/proxy.rs

@@ -136,15 +136,13 @@ static PROPERTIES_SRC: [Property; 6] = [
     ),
 ];
 
-static PROPERTIES_SINK: [Property; 1] = [
+static PROPERTIES_SINK: [Property; 1] = [Property::String(
-    Property::String(
+    "proxy-context",
-        "proxy-context",
+    "Proxy Context",
-        "Proxy Context",
+    "Context name of the proxy to share with",
-        "Context name of the proxy to share with",
+    Some(DEFAULT_PROXY_CONTEXT),
-        Some(DEFAULT_PROXY_CONTEXT),
+    PropertyMutability::ReadWrite,
-        PropertyMutability::ReadWrite,
+)];
-    ),
-];
 
 // TODO: Refactor into a Sender and Receiver instead of the have_ booleans
 
@@ -826,11 +824,7 @@ impl ProxySrc {
         fallback: G,
         f: F,
     ) -> T {
-        let element = parent
+        let element = parent.as_ref().unwrap().downcast_ref::<Element>().unwrap();
-            .as_ref()
-            .unwrap()
-            .downcast_ref::<Element>()
-            .unwrap();
         let src = element.get_impl().downcast_ref::<ProxySrc>().unwrap();
         element.catch_panic(fallback, |element| f(src, element))
     }
@@ -1065,10 +1059,7 @@ impl ProxySrc {
 
         let mut state = self.state.lock().unwrap();
 
-        let io_context = IOContext::new(
+        let io_context = IOContext::new(&settings.context, settings.context_wait).map_err(|err| {
-            &settings.context,
-            settings.context_wait,
-        ).map_err(|err| {
             gst_error_msg!(
                 gst::ResourceError::OpenRead,
                 ["Failed to create IO context: {}", err]

gst-plugin-threadshare/src/queue.rs

@@ -565,12 +565,14 @@ impl Queue {
 
                 let (flags, min, max, align) = new_query.get_result();
                 q.set(flags, min, max, align);
-                q.add_scheduling_modes(&new_query
+                q.add_scheduling_modes(
-                    .get_scheduling_modes()
+                    &new_query
-                    .iter()
+                        .get_scheduling_modes()
-                    .cloned()
+                        .iter()
-                    .filter(|m| m != &gst::PadMode::Pull)
+                        .cloned()
-                    .collect::<Vec<_>>());
+                        .filter(|m| m != &gst::PadMode::Pull)
+                        .collect::<Vec<_>>(),
+                );
                 gst_log!(self.cat, obj: pad, "Returning {:?}", q.get_mut_query());
                 return true;
             }
@@ -691,10 +693,7 @@ impl Queue {
 
         let mut state = self.state.lock().unwrap();
 
-        let io_context = IOContext::new(
+        let io_context = IOContext::new(&settings.context, settings.context_wait).map_err(|err| {
-            &settings.context,
-            settings.context_wait,
-        ).map_err(|err| {
             gst_error_msg!(
                 gst::ResourceError::OpenRead,
                 ["Failed to create IO context: {}", err]

gst-plugin-threadshare/src/tcpclientsrc.rs

@@ -443,10 +443,7 @@ impl TcpClientSrc {
 
         let mut state = self.state.lock().unwrap();
 
-        let io_context = IOContext::new(
+        let io_context = IOContext::new(&settings.context, settings.context_wait).map_err(|err| {
-            &settings.context,
-            settings.context_wait,
-        ).map_err(|err| {
             gst_error_msg!(
                 gst::ResourceError::OpenRead,
                 ["Failed to create IO context: {}", err]

gst-plugin-threadshare/src/udpsrc.rs

@@ -430,10 +430,7 @@ impl UdpSrc {
 
         let mut state = self.state.lock().unwrap();
 
-        let io_context = IOContext::new(
+        let io_context = IOContext::new(&settings.context, settings.context_wait).map_err(|err| {
-            &settings.context,
-            settings.context_wait,
-        ).map_err(|err| {
             gst_error_msg!(
                 gst::ResourceError::OpenRead,
                 ["Failed to create IO context: {}", err]
@@ -522,12 +519,13 @@ impl UdpSrc {
             )
         })?;
 
-        let socket = net::UdpSocket::from_std(socket, io_context.reactor_handle()).map_err(|err| {
+        let socket =
-            gst_error_msg!(
+            net::UdpSocket::from_std(socket, io_context.reactor_handle()).map_err(|err| {
-                gst::ResourceError::OpenRead,
+                gst_error_msg!(
-                ["Failed to setup socket for tokio: {}", err]
+                    gst::ResourceError::OpenRead,
-            )
+                    ["Failed to setup socket for tokio: {}", err]
-        })?;
+                )
+            })?;
 
         if addr.is_multicast() {
             // TODO: Multicast interface configuration, going to be tricky

gst-plugin-threadshare/tests/proxy.rs

@@ -72,12 +72,8 @@ fn test_push() {
     proxysrc.link(&appsink).unwrap();
 
     fakesrc.set_property("num-buffers", &3i32).unwrap();
-    proxysink
+    proxysink.set_property("proxy-context", &"test").unwrap();
-        .set_property("proxy-context", &"test")
+    proxysrc.set_property("proxy-context", &"test").unwrap();
-        .unwrap();
-    proxysrc
-        .set_property("proxy-context", &"test")
-        .unwrap();
 
     appsink.set_property("emit-signals", &true).unwrap();
 

gst-plugin-threadshare/tests/udpsrc.rs

@@ -68,9 +68,7 @@ fn test_push() {
 
     let caps = gst::Caps::new_simple("foo/bar", &[]);
     udpsrc.set_property("caps", &caps).unwrap();
-    udpsrc
+    udpsrc.set_property("port", &(5000 as u32)).unwrap();
-        .set_property("port", &(5000 as u32))
-        .unwrap();
 
     appsink.set_property("emit-signals", &true).unwrap();