Changed listen functions not to use Sender,Reader but to return a Event-Result. Other minor improvements.

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mpvipc"
-version = "1.1.3"
+version = "1.1.4"
 authors = ["Jonas Frei <freijon@gmail.com>"]
 description = "A small library which provides bindings to control existing mpv instances through sockets."
 license = "GPL-3.0"

src/ipc.rs

@@ -3,7 +3,6 @@ use std::collections::HashMap;
 use std::io::BufReader;
 use std::io::prelude::*;
 use std::iter::Iterator;
-use std::sync::mpsc::Sender;
 use super::*;
 
 #[derive(Debug)]
@@ -216,13 +215,16 @@ pub fn get_mpv_property_string(instance: &Mpv, property: &str) -> Result<String,
     }
 }
 
-pub fn set_mpv_property<T: TypeHandler>(instance: &Mpv,
+pub fn set_mpv_property<T: TypeHandler>(
+    instance: &Mpv,
     property: &str,
-                                        value: T)
+    value: T,
-                                        -> Result<(), Error> {
+) -> Result<(), Error> {
-    let ipc_string = format!("{{ \"command\": [\"set_property\", \"{}\", {}] }}\n",
+    let ipc_string = format!(
+        "{{ \"command\": [\"set_property\", \"{}\", {}] }}\n",
         property,
-                             value.as_string());
+        value.as_string()
+    );
     match serde_json::from_str::<Value>(&send_command_sync(instance, &ipc_string)) {
         Ok(_) => Ok(()),
         Err(why) => Err(Error(ErrorCode::JsonParseError(why.to_string()))),
@@ -255,9 +257,11 @@ pub fn run_mpv_command(instance: &Mpv, command: &str, args: &[&str]) -> Result<(
 }
 
 pub fn observe_mpv_property(instance: &Mpv, id: &usize, property: &str) -> Result<(), Error> {
-    let ipc_string = format!("{{ \"command\": [\"observe_property\", {}, \"{}\"] }}\n",
+    let ipc_string = format!(
+        "{{ \"command\": [\"observe_property\", {}, \"{}\"] }}\n",
         id,
-                             property);
+        property
+    );
     match serde_json::from_str::<Value>(&send_command_sync(instance, &ipc_string)) {
         Ok(feedback) => {
             if let Value::String(ref error) = feedback["error"] {
@@ -274,19 +278,10 @@ pub fn observe_mpv_property(instance: &Mpv, id: &usize, property: &str) -> Resul
     }
 }
 
-/// #Description
+pub fn listen(instance: &mut Mpv) -> Result<Event, Error> {
-///
-/// Listens on socket <socket> for events and prints them in real-time to stdout.
-/// This function contains an infinite-loop which keeps the application open indefinitely.
-///
-/// #Example
-/// ```
-/// listen("/tmp/mpvsocket");
-/// ```
-pub fn listen(instance: &Mpv, tx: &Sender<Event>) -> Result<(), Error> {
     let mut response = String::new();
-    let mut reader = BufReader::new(&instance.0);
+    instance.reader.read_line(&mut response).unwrap();
-    reader.read_line(&mut response).unwrap();
+    response = response.trim_right().to_string();
     match serde_json::from_str::<Value>(&response) {
         Ok(e) => {
             if let Value::String(ref name) = e["event"] {
@@ -389,8 +384,8 @@ pub fn listen(instance: &Mpv, tx: &Sender<Event>) -> Result<(), Error> {
                                 data = MpvDataType::HashMap(json_map_to_hashmap(m));
                             }
 
-                            _ => {
+                            Value::Null => {
-                                unimplemented!();
+                                data = MpvDataType::Null;
                             }
                         }
 
@@ -400,24 +395,26 @@ pub fn listen(instance: &Mpv, tx: &Sender<Event>) -> Result<(), Error> {
                         event = Event::Unimplemented;
                     }
                 };
-                tx.send(event).unwrap();
+                return Ok(event);
             }
         }
         Err(why) => return Err(Error(ErrorCode::JsonParseError(why.to_string()))),
     }
-    Ok(())
+    unreachable!();
 }
 
-pub fn listen_raw(instance: &Mpv, tx: &Sender<String>) {
+pub fn listen_raw(instance: &mut Mpv) -> String {
     let mut response = String::new();
-    let mut reader = BufReader::new(&instance.0);
+    instance.reader.read_line(&mut response).unwrap();
-    reader.read_line(&mut response).unwrap();
+    response.trim_right().to_string()
-    tx.send(response.clone()).unwrap();
+    // let mut stream = &instance.0;
-    response.clear();
+    // let mut buffer = [0; 32];
+    // stream.read(&mut buffer[..]).unwrap();
+    // String::from_utf8_lossy(&buffer).into_owned()
 }
 
 fn send_command_sync(instance: &Mpv, command: &str) -> String {
-    let mut stream = &instance.0;
+    let mut stream = &instance.stream;
     match stream.write_all(command.as_bytes()) {
         Err(why) => panic!("Error: Could not write to socket: {}", why),
         Ok(_) => {
@@ -439,16 +436,20 @@ fn json_map_to_hashmap(map: &serde_json::map::Map<String, Value>) -> HashMap<Str
     for (ref key, ref value) in map.iter() {
         match **value {
             Value::Array(ref array) => {
-                output_map.insert(key.to_string(),
+                output_map.insert(
-                                  MpvDataType::Array(json_array_to_vec(array)));
+                    key.to_string(),
+                    MpvDataType::Array(json_array_to_vec(array)),
+                );
             }
             Value::Bool(ref b) => {
                 output_map.insert(key.to_string(), MpvDataType::Bool(*b));
             }
             Value::Number(ref n) => {
                 if n.is_u64() {
-                    output_map.insert(key.to_string(),
+                    output_map.insert(
-                                      MpvDataType::Usize(n.as_u64().unwrap() as usize));
+                        key.to_string(),
+                        MpvDataType::Usize(n.as_u64().unwrap() as usize),
+                    );
                 } else if n.is_f64() {
                     output_map.insert(key.to_string(), MpvDataType::Double(n.as_f64().unwrap()));
                 } else {
@@ -459,8 +460,10 @@ fn json_map_to_hashmap(map: &serde_json::map::Map<String, Value>) -> HashMap<Str
                 output_map.insert(key.to_string(), MpvDataType::String(s.to_string()));
             }
             Value::Object(ref m) => {
-                output_map.insert(key.to_string(),
+                output_map.insert(
-                                  MpvDataType::HashMap(json_map_to_hashmap(m)));
+                    key.to_string(),
+                    MpvDataType::HashMap(json_map_to_hashmap(m)),
+                );
             }
             Value::Null => {
                 unimplemented!();

src/lib.rs

@@ -7,7 +7,7 @@ use ipc::*;
 use std::collections::HashMap;
 use std::fmt::{self, Display};
 use std::os::unix::net::UnixStream;
-use std::sync::mpsc::Sender;
+use std::io::{Read, BufReader};
 
 #[derive(Debug)]
 pub enum Event {
@@ -50,13 +50,14 @@ pub enum Event {
 
 #[derive(Debug)]
 pub enum MpvDataType {
-    Bool(bool),
-    String(String),
-    Double(f64),
-    Usize(usize),
-    HashMap(HashMap<String, MpvDataType>),
     Array(Vec<MpvDataType>),
+    Bool(bool),
+    Double(f64),
+    HashMap(HashMap<String, MpvDataType>),
+    Null,
     Playlist(Playlist),
+    String(String),
+    Usize(usize),
 }
 
 pub enum NumberChangeOptions {
@@ -106,7 +107,10 @@ pub enum ErrorCode {
     ValueDoesNotContainUsize,
 }
 
-pub struct Mpv(UnixStream);
+pub struct Mpv {
+    stream: UnixStream,
+    reader: BufReader<UnixStream>,
+}
 #[derive(Debug)]
 pub struct Playlist(pub Vec<PlaylistEntry>);
 #[derive(Debug)]
@@ -114,19 +118,27 @@ pub struct Error(pub ErrorCode);
 
 impl Drop for Mpv {
     fn drop(&mut self) {
-        self.0
+        self.disconnect();
-            .shutdown(std::net::Shutdown::Both)
-            .expect("stream shutdown");
     }
 }
 
 impl Clone for Mpv {
     fn clone(&self) -> Self {
-        Mpv(self.0.try_clone().expect("cloning UnixStream"))
+        let stream = self.stream.try_clone().expect("cloning UnixStream");
+        let cloned_stream = stream.try_clone().expect("cloning UnixStream");
+        Mpv {
+            stream,
+            reader: BufReader::new(cloned_stream),
+        }
     }
 
     fn clone_from(&mut self, source: &Self) {
-        *self = Mpv(source.0.try_clone().expect("cloning UnixStream"));
+        let stream = source.stream.try_clone().expect("cloning UnixStream");
+        let cloned_stream = stream.try_clone().expect("cloning UnixStream");
+        *self = Mpv {
+            stream,
+            reader: BufReader::new(cloned_stream),
+        }
     }
 }
 
@@ -157,7 +169,9 @@ impl Display for ErrorCode {
                 f.write_str("The received value is not of type \'std::f64\'")
             }
             ErrorCode::ValueDoesNotContainHashMap => {
-                f.write_str("The received value is not of type \'std::collections::HashMap\'")
+                f.write_str(
+                    "The received value is not of type \'std::collections::HashMap\'",
+                )
             }
             ErrorCode::ValueDoesNotContainPlaylist => {
                 f.write_str("The received value is not of type \'mpvipc::Playlist\'")
@@ -207,9 +221,10 @@ impl GetPropertyTypeHandler for Vec<PlaylistEntry> {
 }
 
 impl GetPropertyTypeHandler for HashMap<String, MpvDataType> {
-    fn get_property_generic(instance: &Mpv,
+    fn get_property_generic(
-                            property: &str)
+        instance: &Mpv,
-                            -> Result<HashMap<String, MpvDataType>, Error> {
+        property: &str,
+    ) -> Result<HashMap<String, MpvDataType>, Error> {
         get_mpv_property::<HashMap<String, MpvDataType>>(instance, property)
     }
 }
@@ -245,11 +260,32 @@ impl SetPropertyTypeHandler<usize> for usize {
 impl Mpv {
     pub fn connect(socket: &str) -> Result<Mpv, Error> {
         match UnixStream::connect(socket) {
-            Ok(stream) => Ok(Mpv(stream)),
+            Ok(stream) => {
+                let cloned_stream = stream.try_clone().expect("cloning UnixStream");
+                return Ok(Mpv {
+                    stream,
+                    reader: BufReader::new(cloned_stream),
+                });
+            }
             Err(internal_error) => Err(Error(ErrorCode::ConnectError(internal_error.to_string()))),
         }
     }
 
+    pub fn disconnect(&self) {
+        let mut stream = &self.stream;
+        stream.shutdown(std::net::Shutdown::Both).expect(
+            "socket disconnect",
+        );
+        let mut buffer = [0; 32];
+        for _ in 0..stream.bytes().count() {
+            stream.read(&mut buffer[..]).unwrap();
+        }
+    }
+
+    pub fn get_stream_ref(&self) -> &UnixStream {
+        &self.stream
+    }
+
     pub fn get_metadata(&self) -> Result<HashMap<String, MpvDataType>, Error> {
         match get_mpv_property(self, "metadata") {
             Ok(map) => Ok(map),
@@ -315,28 +351,23 @@ impl Mpv {
 
     /// #Description
     ///
-    /// Listens for mpv events and triggers the channel once an event has been received.
+    /// Waits until an mpv event occurs and returns the Event.
-    ///
-    /// ##Input arguments
-    ///
-    /// - **tx** A reference to the sender halve of the channel
     ///
     /// #Example
     ///
     /// ```
-    /// let (tx, rx) = std::sync::mpsc::channel();
+    /// let mpv = Mpv::connect("/tmp/mpvsocket").unwrap();
     /// loop {
-    ///     mpv.event_listen(&tx);
+    ///     let event = mpv.event_listen().unwrap();
-    ///     let event = rx.recv().unwrap();
     ///     println!("{:?}", event);
     /// }
     /// ```
-    pub fn event_listen(&self, tx: &Sender<Event>) -> Result<(), Error> {
+    pub fn event_listen(&mut self) -> Result<Event, Error> {
-        listen(self, tx)
+        listen(self)
     }
 
-    pub fn event_listen_raw(&self, tx: &Sender<String>) {
+    pub fn event_listen_raw(&mut self) -> String {
-        listen_raw(self, tx);
+        listen_raw(self)
     }
 
     pub fn next(&self) -> Result<(), Error> {
@@ -382,11 +413,12 @@ impl Mpv {
         run_mpv_command(self, command, args)
     }
 
-    pub fn playlist_add(&self,
+    pub fn playlist_add(
+        &self,
         file: &str,
         file_type: PlaylistAddTypeOptions,
-                        option: PlaylistAddOptions)
+        option: PlaylistAddOptions,
-                        -> Result<(), Error> {
+    ) -> Result<(), Error> {
         match file_type {
             PlaylistAddTypeOptions::File => {
                 match option {
@@ -422,9 +454,7 @@ impl Mpv {
     }
 
     pub fn playlist_move_id(&self, from: usize, to: usize) -> Result<(), Error> {
-        run_mpv_command(self,
+        run_mpv_command(self, "playlist-move", &[&from.to_string(), &to.to_string()])
-                        "playlist-remove",
-                        &[&from.to_string(), &to.to_string()])
     }
 
     pub fn playlist_play_id(&self, id: usize) -> Result<(), Error> {
@@ -434,9 +464,11 @@ impl Mpv {
     pub fn playlist_play_next(&self, id: usize) -> Result<(), Error> {
         match get_mpv_property::<usize>(self, "playlist-pos") {
             Ok(current_id) => {
-                run_mpv_command(self,
+                run_mpv_command(
+                    self,
                     "playlist-move",
-                                &[&id.to_string(), &(current_id + 1).to_string()])
+                    &[&id.to_string(), &(current_id + 1).to_string()],
+                )
             }
             Err(msg) => Err(msg),
         }
@@ -552,10 +584,11 @@ impl Mpv {
     /// let mpv = Mpv::connect("/tmp/mpvsocket").unwrap();
     /// mpv.set_property("pause", true);
     /// ```
-    pub fn set_property<T: SetPropertyTypeHandler<T>>(&self,
+    pub fn set_property<T: SetPropertyTypeHandler<T>>(
+        &self,
         property: &str,
-                                                      value: T)
+        value: T,
-                                                      -> Result<(), Error> {
+    ) -> Result<(), Error> {
         T::set_property_generic(self, property, value)
     }