setup gitea action workflows

.gitea/workflows/build-and-test.yml

@@ -0,0 +1,64 @@
+name: "Build and test"
+on:
+  push:
+    branches:
+      - master
+  pull_request:
+
+jobs:
+  build:
+    runs-on: ubuntu-latest-personal
+    steps:
+      - uses: actions/checkout@v3
+
+      - name: Install latest nightly toolchain
+        uses: actions-rs/toolchain@v1
+        with:
+            toolchain: nightly
+            override: true
+            components: rustfmt, clippy
+
+      - name: Cache dependencies
+        uses: Swatinem/rust-cache@v2
+
+      - name: Build
+        run: cargo build --all-features --verbose
+
+  check:
+    runs-on: ubuntu-latest-personal
+    steps:
+      - uses: actions/checkout@v3
+
+      - name: Install latest nightly toolchain
+        uses: actions-rs/toolchain@v1
+        with:
+            toolchain: nightly
+            override: true
+            components: rustfmt, clippy
+
+      - name: Cache dependencies
+        uses: Swatinem/rust-cache@v2
+
+      - name: Check code format
+        run: cargo fmt -- --check
+
+      - name: Check clippy
+        run: cargo clippy --all-features -- --deny warnings
+
+  test:
+    runs-on: ubuntu-latest-personal
+    steps:
+      - uses: actions/checkout@v3
+
+      - name: Install latest nightly toolchain
+        uses: actions-rs/toolchain@v1
+        with:
+            toolchain: nightly
+            override: true
+            components: rustfmt, clippy
+
+      - name: Cache dependencies
+        uses: Swatinem/rust-cache@v2
+
+      - name: Run tests
+        run: cargo test --all-features --verbose

Cargo.toml

@@ -17,6 +17,7 @@ serde = { version = "1.0.197", features = ["derive"] }
 tokio = { version = "1.37.0", features = ["sync", "macros", "rt", "net"] }
 tokio-util = { version = "0.7.10", features = ["codec"] }
 futures = "0.3.30"
+tokio-stream = { version = "0.1.15", features = ["sync"] }
 
 [dev-dependencies]
 env_logger = "0.10.0"

examples/fetch_state.rs

@@ -1,5 +1,5 @@
 use env_logger;
-use mpvipc::{Error as MpvError, Mpv};
+use mpvipc::{Error as MpvError, Mpv, MpvExt};
 
 #[tokio::main]
 async fn main() -> Result<(), MpvError> {

examples/media_player.rs

@@ -1,5 +1,5 @@
 use env_logger;
-use mpvipc::{Error, Event, Mpv, MpvDataType, Property};
+use mpvipc::{Error, Event, Mpv, MpvExt, MpvDataType, Property};
 use std::io::{self, Write};
 
 fn seconds_to_hms(total: f64) -> String {

src/api.rs

@@ -1,12 +1,16 @@
+use futures::StreamExt;
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
 use std::{
     collections::HashMap,
     fmt::{self, Display},
 };
-use tokio::{net::UnixStream, sync::oneshot};
+use tokio::{
+    net::UnixStream,
+    sync::{broadcast, mpsc, oneshot},
+};
 
-use crate::ipc::{MpvIpc, MpvIpcCommand, MpvIpcResponse};
+use crate::ipc::{MpvIpc, MpvIpcCommand, MpvIpcEvent, MpvIpcResponse};
 use crate::message_parser::TypeHandler;
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -79,7 +83,7 @@ pub enum MpvCommand {
     Unobserve(isize),
 }
 
-trait IntoRawCommandPart {
+pub(crate) trait IntoRawCommandPart {
     fn into_raw_command_part(self) -> String;
 }
 
@@ -95,23 +99,6 @@ pub enum MpvDataType {
     Usize(usize),
 }
 
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub enum NumberChangeOptions {
-    Absolute,
-    Increase,
-    Decrease,
-}
-
-impl IntoRawCommandPart for NumberChangeOptions {
-    fn into_raw_command_part(self) -> String {
-        match self {
-            NumberChangeOptions::Absolute => "absolute".to_string(),
-            NumberChangeOptions::Increase => "increase".to_string(),
-            NumberChangeOptions::Decrease => "decrease".to_string(),
-        }
-    }
-}
-
 #[derive(Debug, Clone, Copy, Serialize, Deserialize)]
 pub enum PlaylistAddOptions {
     Replace,
@@ -152,13 +139,6 @@ impl IntoRawCommandPart for SeekOptions {
     }
 }
 
-#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
-pub enum Switch {
-    On,
-    Off,
-    Toggle,
-}
-
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub enum ErrorCode {
     MpvError(String),
@@ -288,7 +268,8 @@ impl Display for ErrorCode {
 
 #[derive(Clone)]
 pub struct Mpv {
-    command_sender: tokio::sync::mpsc::Sender<(MpvIpcCommand, oneshot::Sender<MpvIpcResponse>)>,
+    command_sender: mpsc::Sender<(MpvIpcCommand, oneshot::Sender<MpvIpcResponse>)>,
+    broadcast_channel: broadcast::Sender<MpvIpcEvent>,
 }
 
 impl fmt::Debug for Mpv {
@@ -310,14 +291,16 @@ impl Mpv {
     }
 
     pub async fn connect_socket(socket: UnixStream) -> Result<Mpv, Error> {
-        let (com_tx, com_rx) = tokio::sync::mpsc::channel(100);
+        let (com_tx, com_rx) = mpsc::channel(100);
-        let ipc = MpvIpc::new(socket, com_rx);
+        let (ev_tx, _) = broadcast::channel(100);
+        let ipc = MpvIpc::new(socket, com_rx, ev_tx.clone());
 
         log::debug!("Starting IPC handler");
         tokio::spawn(ipc.run());
 
         Ok(Mpv {
             command_sender: com_tx,
+            broadcast_channel: ev_tx,
         })
     }
 
@@ -337,18 +320,117 @@ impl Mpv {
         }
     }
 
-    // pub fn get_stream_ref(&self) -> &UnixStream {
+    pub async fn get_event_stream(&self) -> impl futures::Stream<Item = Result<Event, Error>> {
-    //     &self.stream
+        tokio_stream::wrappers::BroadcastStream::new(self.broadcast_channel.subscribe())
-    // }
+        .map(|event| {
-
+          match event {
-    pub async fn get_metadata(&self) -> Result<HashMap<String, MpvDataType>, Error> {
+            Ok(event) => Mpv::map_event(event),
-        self.get_property("metadata").await
+            Err(_) => Err(Error(ErrorCode::ConnectError("Failed to receive event".to_string()))),
+          }
+        })
     }
 
-    pub async fn get_playlist(&self) -> Result<Playlist, Error> {
+    fn map_event(raw_event: MpvIpcEvent) -> Result<Event, Error> {
-        self.get_property::<Vec<PlaylistEntry>>("playlist")
+        let MpvIpcEvent(event) = raw_event;
-            .await
+
-            .map(|entries| Playlist(entries))
+        event
+            .as_object()
+            .ok_or(Error(ErrorCode::JsonContainsUnexptectedType))
+            .and_then(|event| {
+                let event_name = event
+                    .get("event")
+                    .ok_or(Error(ErrorCode::MissingValue))?
+                    .as_str()
+                    .ok_or(Error(ErrorCode::ValueDoesNotContainString))?;
+
+                match event_name {
+                    "shutdown" => Ok(Event::Shutdown),
+                    "start-file" => Ok(Event::StartFile),
+                    "end-file" => Ok(Event::EndFile),
+                    "file-loaded" => Ok(Event::FileLoaded),
+                    "tracks-changed" => Ok(Event::TracksChanged),
+                    "track-switched" => Ok(Event::TrackSwitched),
+                    "idle" => Ok(Event::Idle),
+                    "pause" => Ok(Event::Pause),
+                    "unpause" => Ok(Event::Unpause),
+                    "tick" => Ok(Event::Tick),
+                    "video-reconfig" => Ok(Event::VideoReconfig),
+                    "audio-reconfig" => Ok(Event::AudioReconfig),
+                    "metadata-update" => Ok(Event::MetadataUpdate),
+                    "seek" => Ok(Event::Seek),
+                    "playback-restart" => Ok(Event::PlaybackRestart),
+                    "property-change" => {
+                        let id = event
+                            .get("id")
+                            .ok_or(Error(ErrorCode::MissingValue))?
+                            .as_u64()
+                            .ok_or(Error(ErrorCode::ValueDoesNotContainUsize))?
+                            as usize;
+                        let property_name = event
+                            .get("name")
+                            .ok_or(Error(ErrorCode::MissingValue))?
+                            .as_str()
+                            .ok_or(Error(ErrorCode::ValueDoesNotContainString))?;
+
+                        match property_name {
+                          "path" => {
+                            let path = event
+                                .get("data")
+                                .ok_or(Error(ErrorCode::MissingValue))?
+                                .as_str()
+                                .map(|s| s.to_string());
+                            Ok(Event::PropertyChange {
+                                id,
+                                property: Property::Path(path),
+                            })
+                          }
+                          "pause" => {
+                            let pause = event
+                                .get("data")
+                                .ok_or(Error(ErrorCode::MissingValue))?
+                                .as_bool()
+                                .ok_or(Error(ErrorCode::ValueDoesNotContainBool))?;
+                            Ok(Event::PropertyChange {
+                                id,
+                                property: Property::Pause(pause),
+                            })
+                          }
+                          // TODO: missing cases
+                          _ => {
+                            let data = event
+                                .get("data")
+                                .ok_or(Error(ErrorCode::MissingValue))?
+                                .clone();
+                            Ok(Event::PropertyChange {
+                                id,
+                                property: Property::Unknown {
+                                    name: property_name.to_string(),
+                                    // TODO: fix
+                                    data: MpvDataType::Double(data.as_f64().unwrap_or(0.0)),
+                                },
+                            })
+                          }
+                        }
+                    }
+                    "chapter-change" => Ok(Event::ChapterChange),
+                    "client-message" => {
+                        let args = event
+                            .get("args")
+                            .ok_or(Error(ErrorCode::MissingValue))?
+                            .as_array()
+                            .ok_or(Error(ErrorCode::ValueDoesNotContainString))?
+                            .iter()
+                            .map(|arg| {
+                                arg.as_str()
+                                    .ok_or(Error(ErrorCode::ValueDoesNotContainString))
+                                    .map(|s| s.to_string())
+                            })
+                            .collect::<Result<Vec<String>, Error>>()?;
+                        Ok(Event::ClientMessage { args })
+                    }
+                    _ => Ok(Event::Unimplemented),
+                }
+            })
     }
 
     /// # Description
@@ -414,70 +496,13 @@ impl Mpv {
                 ))
             })?;
         match res_rx.await {
-            Ok(MpvIpcResponse(response)) => response.and_then(|value| {
+            Ok(MpvIpcResponse(response)) => {
-                value.ok_or(Error(ErrorCode::MissingValue))
+                response.and_then(|value| value.ok_or(Error(ErrorCode::MissingValue)))
-            }),
+            }
             Err(err) => Err(Error(ErrorCode::ConnectError(err.to_string()))),
         }
     }
 
-    pub async fn kill(&self) -> Result<(), Error> {
-        self.run_command(MpvCommand::Quit).await
-    }
-
-    /// # Description
-    ///
-    /// Waits until an mpv event occurs and returns the Event.
-    ///
-    /// # Example
-    ///
-    /// ```ignore
-    /// let mut mpv = Mpv::connect("/tmp/mpvsocket")?;
-    /// loop {
-    ///     let event = mpv.event_listen()?;
-    ///     println!("{:?}", event);
-    /// }
-    /// ```
-    // pub fn event_listen(&mut self) -> Result<Event, Error> {
-    //     listen(self)
-    // }
-
-    // pub fn event_listen_raw(&mut self) -> String {
-    //     listen_raw(self)
-    // }
-
-    pub async fn next(&self) -> Result<(), Error> {
-        self.run_command(MpvCommand::PlaylistNext).await
-    }
-
-    pub async fn observe_property(&self, id: isize, property: &str) -> Result<(), Error> {
-        self.run_command(MpvCommand::Observe {
-            id,
-            property: property.to_string(),
-        })
-        .await
-    }
-
-    pub async fn unobserve_property(&self, id: isize) -> Result<(), Error> {
-        self.run_command(MpvCommand::Unobserve(id)).await
-    }
-
-    pub async fn pause(&self) -> Result<(), Error> {
-        self.set_property("pause", true).await
-    }
-
-    pub async fn prev(&self) -> Result<(), Error> {
-        self.run_command(MpvCommand::PlaylistPrev).await
-    }
-
-    pub async fn restart(&self) -> Result<(), Error> {
-        self.run_command(MpvCommand::Seek {
-            seconds: 0f64,
-            option: SeekOptions::Absolute,
-        })
-        .await
-    }
-
     /// # Description
     ///
     /// Runs mpv commands. The arguments are passed as a String-Vector reference:
@@ -607,7 +632,11 @@ impl Mpv {
     /// Run a custom command.
     /// This should only be used if the desired command is not implemented
     /// with [MpvCommand].
-    pub async fn run_command_raw(&self, command: &str, args: &[&str]) -> Result<Option<Value>, Error> {
+    pub async fn run_command_raw(
+        &self,
+        command: &str,
+        args: &[&str],
+    ) -> Result<Option<Value>, Error> {
         let command = Vec::from(
             [command]
                 .iter()
@@ -640,120 +669,6 @@ impl Mpv {
         self.run_command_raw(command, args).await.map(|_| ())
     }
 
-    pub async fn playlist_add(
-        &self,
-        file: &str,
-        file_type: PlaylistAddTypeOptions,
-        option: PlaylistAddOptions,
-    ) -> Result<(), Error> {
-        match file_type {
-            PlaylistAddTypeOptions::File => {
-                self.run_command(MpvCommand::LoadFile {
-                    file: file.to_string(),
-                    option,
-                })
-                .await
-            }
-
-            PlaylistAddTypeOptions::Playlist => {
-                self.run_command(MpvCommand::LoadList {
-                    file: file.to_string(),
-                    option,
-                })
-                .await
-            }
-        }
-    }
-
-    pub async fn playlist_clear(&self) -> Result<(), Error> {
-        self.run_command(MpvCommand::PlaylistClear).await
-    }
-
-    pub async fn playlist_move_id(&self, from: usize, to: usize) -> Result<(), Error> {
-        self.run_command(MpvCommand::PlaylistMove { from, to })
-            .await
-    }
-
-    pub async fn playlist_play_id(&self, id: usize) -> Result<(), Error> {
-        self.set_property("playlist-pos", id).await
-    }
-
-    pub async fn playlist_play_next(&self, id: usize) -> Result<(), Error> {
-        match self.get_property::<usize>("playlist-pos").await {
-            Ok(current_id) => {
-                self.run_command(MpvCommand::PlaylistMove {
-                    from: id,
-                    to: current_id + 1,
-                })
-                .await
-            }
-            Err(msg) => Err(msg),
-        }
-    }
-
-    pub async fn playlist_remove_id(&self, id: usize) -> Result<(), Error> {
-        self.run_command(MpvCommand::PlaylistRemove(id)).await
-    }
-
-    pub async fn playlist_shuffle(&self) -> Result<(), Error> {
-        self.run_command(MpvCommand::PlaylistShuffle).await
-    }
-
-    pub async fn seek(&self, seconds: f64, option: SeekOptions) -> Result<(), Error> {
-        self.run_command(MpvCommand::Seek { seconds, option }).await
-    }
-
-    pub async fn set_loop_file(&self, option: Switch) -> Result<(), Error> {
-        let enabled = match option {
-            Switch::On => "inf",
-            Switch::Off => "no",
-            Switch::Toggle => {
-                self.get_property::<String>("loop-file")
-                    .await
-                    .map(|s| match s.as_str() {
-                        "inf" => "no",
-                        "no" => "inf",
-                        _ => "no",
-                    })?
-            }
-        };
-        self.set_property("loop-file", enabled).await
-    }
-
-    pub async fn set_loop_playlist(&self, option: Switch) -> Result<(), Error> {
-        let enabled = match option {
-            Switch::On => "inf",
-            Switch::Off => "no",
-            Switch::Toggle => {
-                self.get_property::<String>("loop-playlist")
-                    .await
-                    .map(|s| match s.as_str() {
-                        "inf" => "no",
-                        "no" => "inf",
-                        _ => "no",
-                    })?
-            }
-        };
-        self.set_property("loo-playlist", enabled).await
-    }
-
-    pub async fn set_mute(&self, option: Switch) -> Result<(), Error> {
-        let enabled = match option {
-            Switch::On => "yes",
-            Switch::Off => "no",
-            Switch::Toggle => {
-                self.get_property::<String>("mute")
-                    .await
-                    .map(|s| match s.as_str() {
-                        "yes" => "no",
-                        "no" => "yes",
-                        _ => "no",
-                    })?
-            }
-        };
-        self.set_property("mute", enabled).await
-    }
-
     /// # Description
     ///
     /// Sets the mpv property _<property>_ to _<value>_.
@@ -785,54 +700,4 @@ impl Mpv {
     ) -> Result<(), Error> {
         T::set_property_generic(self, property, value).await
     }
-
-    pub async fn set_speed(
-        &self,
-        input_speed: f64,
-        option: NumberChangeOptions,
-    ) -> Result<(), Error> {
-        match self.get_property::<f64>("speed").await {
-            Ok(speed) => match option {
-                NumberChangeOptions::Increase => {
-                    self.set_property("speed", speed + input_speed).await
-                }
-
-                NumberChangeOptions::Decrease => {
-                    self.set_property("speed", speed - input_speed).await
-                }
-
-                NumberChangeOptions::Absolute => self.set_property("speed", input_speed).await,
-            },
-            Err(msg) => Err(msg),
-        }
-    }
-
-    pub async fn set_volume(
-        &self,
-        input_volume: f64,
-        option: NumberChangeOptions,
-    ) -> Result<(), Error> {
-        match self.get_property::<f64>("volume").await {
-            Ok(volume) => match option {
-                NumberChangeOptions::Increase => {
-                    self.set_property("volume", volume + input_volume).await
-                }
-
-                NumberChangeOptions::Decrease => {
-                    self.set_property("volume", volume - input_volume).await
-                }
-
-                NumberChangeOptions::Absolute => self.set_property("volume", input_volume).await,
-            },
-            Err(msg) => Err(msg),
-        }
-    }
-
-    pub async fn stop(&self) -> Result<(), Error> {
-        self.run_command(MpvCommand::Stop).await
-    }
-
-    pub async fn toggle(&self) -> Result<(), Error> {
-        self.run_command_raw("cycle", &["pause"]).await.map(|_| ())
-    }
 }

src/api_extension.rs

@@ -0,0 +1,269 @@
+use crate::{Error, IntoRawCommandPart, Mpv, MpvCommand, MpvDataType, Playlist, PlaylistAddOptions, PlaylistAddTypeOptions, PlaylistEntry, SeekOptions};
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum NumberChangeOptions {
+    Absolute,
+    Increase,
+    Decrease,
+}
+
+impl IntoRawCommandPart for NumberChangeOptions {
+    fn into_raw_command_part(self) -> String {
+        match self {
+            NumberChangeOptions::Absolute => "absolute".to_string(),
+            NumberChangeOptions::Increase => "increase".to_string(),
+            NumberChangeOptions::Decrease => "decrease".to_string(),
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
+pub enum Switch {
+    On,
+    Off,
+    Toggle,
+}
+
+// TODO: fix this
+#[allow(async_fn_in_trait)]
+pub trait MpvExt {
+    async fn toggle(&self) -> Result<(), Error>;
+    async fn stop(&self) -> Result<(), Error>;
+    async fn set_volume(&self, input_volume: f64, option: NumberChangeOptions) -> Result<(), Error>;
+    async fn set_speed(&self, input_speed: f64, option: NumberChangeOptions) -> Result<(), Error>;
+    async fn set_mute(&self, option: Switch) -> Result<(), Error>;
+    async fn set_loop_playlist(&self, option: Switch) -> Result<(), Error>;
+    async fn set_loop_file(&self, option: Switch) -> Result<(), Error>;
+    async fn seek(&self, seconds: f64, option: SeekOptions) -> Result<(), Error>;
+    async fn playlist_shuffle(&self) -> Result<(), Error>;
+    async fn playlist_remove_id(&self, id: usize) -> Result<(), Error>;
+    async fn playlist_play_next(&self, id: usize) -> Result<(), Error>;
+    async fn playlist_play_id(&self, id: usize) -> Result<(), Error>;
+    async fn playlist_move_id(&self, from: usize, to: usize) -> Result<(), Error>;
+    async fn playlist_clear(&self) -> Result<(), Error>;
+    async fn playlist_add(&self, file: &str, file_type: PlaylistAddTypeOptions, option: PlaylistAddOptions) -> Result<(), Error>;
+    async fn restart(&self) -> Result<(), Error>;
+    async fn prev(&self) -> Result<(), Error>;
+    async fn pause(&self) -> Result<(), Error>;
+    async fn unobserve_property(&self, id: isize) -> Result<(), Error>;
+    async fn observe_property(&self, id: isize, property: &str) -> Result<(), Error>;
+    async fn next(&self) -> Result<(), Error>;
+    async fn kill(&self) -> Result<(), Error>;
+    async fn get_playlist(&self) -> Result<Playlist, Error>;
+    async fn get_metadata(&self) -> Result<HashMap<String, MpvDataType>, Error>;
+
+}
+
+impl MpvExt for Mpv {
+    async fn get_metadata(&self) -> Result<HashMap<String, MpvDataType>, Error> {
+        self.get_property("metadata").await
+    }
+
+    async fn get_playlist(&self) -> Result<Playlist, Error> {
+        self.get_property::<Vec<PlaylistEntry>>("playlist")
+            .await
+            .map(|entries| Playlist(entries))
+    }
+
+    async fn kill(&self) -> Result<(), Error> {
+        self.run_command(MpvCommand::Quit).await
+    }
+
+    async fn next(&self) -> Result<(), Error> {
+        self.run_command(MpvCommand::PlaylistNext).await
+    }
+
+    async fn observe_property(&self, id: isize, property: &str) -> Result<(), Error> {
+        self.run_command(MpvCommand::Observe {
+            id,
+            property: property.to_string(),
+        })
+        .await
+    }
+
+    async fn unobserve_property(&self, id: isize) -> Result<(), Error> {
+        self.run_command(MpvCommand::Unobserve(id)).await
+    }
+
+    async fn pause(&self) -> Result<(), Error> {
+        self.set_property("pause", true).await
+    }
+
+    async fn prev(&self) -> Result<(), Error> {
+        self.run_command(MpvCommand::PlaylistPrev).await
+    }
+
+    async fn restart(&self) -> Result<(), Error> {
+        self.run_command(MpvCommand::Seek {
+            seconds: 0f64,
+            option: SeekOptions::Absolute,
+        })
+        .await
+    }
+
+    async fn playlist_add(
+        &self,
+        file: &str,
+        file_type: PlaylistAddTypeOptions,
+        option: PlaylistAddOptions,
+    ) -> Result<(), Error> {
+        match file_type {
+            PlaylistAddTypeOptions::File => {
+                self.run_command(MpvCommand::LoadFile {
+                    file: file.to_string(),
+                    option,
+                })
+                .await
+            }
+
+            PlaylistAddTypeOptions::Playlist => {
+                self.run_command(MpvCommand::LoadList {
+                    file: file.to_string(),
+                    option,
+                })
+                .await
+            }
+        }
+    }
+
+    async fn playlist_clear(&self) -> Result<(), Error> {
+        self.run_command(MpvCommand::PlaylistClear).await
+    }
+
+    async fn playlist_move_id(&self, from: usize, to: usize) -> Result<(), Error> {
+        self.run_command(MpvCommand::PlaylistMove { from, to })
+            .await
+    }
+
+    async fn playlist_play_id(&self, id: usize) -> Result<(), Error> {
+        self.set_property("playlist-pos", id).await
+    }
+
+    async fn playlist_play_next(&self, id: usize) -> Result<(), Error> {
+        match self.get_property::<usize>("playlist-pos").await {
+            Ok(current_id) => {
+                self.run_command(MpvCommand::PlaylistMove {
+                    from: id,
+                    to: current_id + 1,
+                })
+                .await
+            }
+            Err(msg) => Err(msg),
+        }
+    }
+
+    async fn playlist_remove_id(&self, id: usize) -> Result<(), Error> {
+        self.run_command(MpvCommand::PlaylistRemove(id)).await
+    }
+
+    async fn playlist_shuffle(&self) -> Result<(), Error> {
+        self.run_command(MpvCommand::PlaylistShuffle).await
+    }
+
+    async fn seek(&self, seconds: f64, option: SeekOptions) -> Result<(), Error> {
+        self.run_command(MpvCommand::Seek { seconds, option }).await
+    }
+
+    async fn set_loop_file(&self, option: Switch) -> Result<(), Error> {
+        let enabled = match option {
+            Switch::On => "inf",
+            Switch::Off => "no",
+            Switch::Toggle => {
+                self.get_property::<String>("loop-file")
+                    .await
+                    .map(|s| match s.as_str() {
+                        "inf" => "no",
+                        "no" => "inf",
+                        _ => "no",
+                    })?
+            }
+        };
+        self.set_property("loop-file", enabled).await
+    }
+
+    async fn set_loop_playlist(&self, option: Switch) -> Result<(), Error> {
+        let enabled = match option {
+            Switch::On => "inf",
+            Switch::Off => "no",
+            Switch::Toggle => {
+                self.get_property::<String>("loop-playlist")
+                    .await
+                    .map(|s| match s.as_str() {
+                        "inf" => "no",
+                        "no" => "inf",
+                        _ => "no",
+                    })?
+            }
+        };
+        self.set_property("loo-playlist", enabled).await
+    }
+
+    async fn set_mute(&self, option: Switch) -> Result<(), Error> {
+        let enabled = match option {
+            Switch::On => "yes",
+            Switch::Off => "no",
+            Switch::Toggle => {
+                self.get_property::<String>("mute")
+                    .await
+                    .map(|s| match s.as_str() {
+                        "yes" => "no",
+                        "no" => "yes",
+                        _ => "no",
+                    })?
+            }
+        };
+        self.set_property("mute", enabled).await
+    }
+
+    async fn set_speed(
+        &self,
+        input_speed: f64,
+        option: NumberChangeOptions,
+    ) -> Result<(), Error> {
+        match self.get_property::<f64>("speed").await {
+            Ok(speed) => match option {
+                NumberChangeOptions::Increase => {
+                    self.set_property("speed", speed + input_speed).await
+                }
+
+                NumberChangeOptions::Decrease => {
+                    self.set_property("speed", speed - input_speed).await
+                }
+
+                NumberChangeOptions::Absolute => self.set_property("speed", input_speed).await,
+            },
+            Err(msg) => Err(msg),
+        }
+    }
+
+    async fn set_volume(
+        &self,
+        input_volume: f64,
+        option: NumberChangeOptions,
+    ) -> Result<(), Error> {
+        match self.get_property::<f64>("volume").await {
+            Ok(volume) => match option {
+                NumberChangeOptions::Increase => {
+                    self.set_property("volume", volume + input_volume).await
+                }
+
+                NumberChangeOptions::Decrease => {
+                    self.set_property("volume", volume - input_volume).await
+                }
+
+                NumberChangeOptions::Absolute => self.set_property("volume", input_volume).await,
+            },
+            Err(msg) => Err(msg),
+        }
+    }
+
+    async fn stop(&self) -> Result<(), Error> {
+        self.run_command(MpvCommand::Stop).await
+    }
+
+    async fn toggle(&self) -> Result<(), Error> {
+        self.run_command_raw("cycle", &["pause"]).await.map(|_| ())
+    }
+}

src/ipc.rs

@@ -3,14 +3,15 @@ use futures::{SinkExt, StreamExt};
 use serde_json::{json, Value};
 use std::mem;
 use tokio::net::UnixStream;
-use tokio::sync::mpsc::Receiver;
+use tokio::sync::mpsc;
-use tokio::sync::{oneshot, Mutex};
+use tokio::sync::{broadcast, oneshot, Mutex};
-use tokio_util::codec::{Framed, LinesCodec};
+use tokio_util::codec::{Framed, LinesCodec, LinesCodecError};
 
 pub(crate) struct MpvIpc {
     socket: Framed<UnixStream, LinesCodec>,
-    command_channel: Receiver<(MpvIpcCommand, oneshot::Sender<MpvIpcResponse>)>,
+    command_channel: mpsc::Receiver<(MpvIpcCommand, oneshot::Sender<MpvIpcResponse>)>,
     socket_lock: Mutex<()>,
+    event_channel: broadcast::Sender<MpvIpcEvent>,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -26,19 +27,24 @@ pub(crate) enum MpvIpcCommand {
 #[derive(Debug, Clone)]
 pub(crate) struct MpvIpcResponse(pub(crate) Result<Option<Value>, Error>);
 
+#[derive(Debug, Clone)]
+pub(crate) struct MpvIpcEvent(pub(crate) Value);
+
 impl MpvIpc {
     pub(crate) fn new(
         socket: UnixStream,
-        command_channel: Receiver<(MpvIpcCommand, oneshot::Sender<MpvIpcResponse>)>,
+        command_channel: mpsc::Receiver<(MpvIpcCommand, oneshot::Sender<MpvIpcResponse>)>,
+        event_channel: broadcast::Sender<MpvIpcEvent>,
     ) -> Self {
         MpvIpc {
             socket: Framed::new(socket, LinesCodec::new()),
             command_channel,
             socket_lock: Mutex::new(()),
+            event_channel,
         }
     }
 
-    pub(crate) async fn send_command(&mut self, command: &[&str]) -> Result<Option<Value>, Error> {
+    pub(crate) async fn send_command(&mut self, command: &[Value]) -> Result<Option<Value>, Error> {
         let lock = self.socket_lock.lock().await;
         // START CRITICAL SECTION
         let ipc_command = json!({ "command": command });
@@ -67,11 +73,14 @@ impl MpvIpc {
         serde_json::from_str::<Value>(&response)
             .map_err(|why| Error(ErrorCode::JsonParseError(why.to_string())))
             .and_then(parse_mpv_response_data)
-
     }
 
-    pub(crate) async fn get_mpv_property(&mut self, property: &str) -> Result<Option<Value>, Error> {
+    pub(crate) async fn get_mpv_property(
-        self.send_command(&["get_property", property]).await
+        &mut self,
+        property: &str,
+    ) -> Result<Option<Value>, Error> {
+        self.send_command(&[json!("get_property"), json!(property)])
+            .await
     }
 
     pub(crate) async fn set_mpv_property(
@@ -79,11 +88,11 @@ impl MpvIpc {
         property: &str,
         value: Value,
     ) -> Result<Option<Value>, Error> {
-        let str_value = match &value {
+        // let str_value = match &value {
-          Value::String(s) => s,
+        //     Value::String(s) => s,
-          v => &serde_json::to_string(&v).unwrap()
+        //     v => &serde_json::to_string(&v).unwrap(),
-        };
+        // };
-        self.send_command(&["set_property", property, &str_value])
+        self.send_command(&[json!("set_property"), json!(property), value])
             .await
     }
 
@@ -92,27 +101,52 @@ impl MpvIpc {
         id: isize,
         property: &str,
     ) -> Result<Option<Value>, Error> {
-        self.send_command(&["observe_property", &id.to_string(), property])
+        self.send_command(&[json!("observe_property"), json!(id), json!(property)])
             .await
     }
 
     pub(crate) async fn unobserve_property(&mut self, id: isize) -> Result<Option<Value>, Error> {
-        self.send_command(&["unobserve_property", &id.to_string()])
+        self.send_command(&[json!("unobserve_property"), json!(id)])
             .await
     }
 
+    async fn handle_event(&mut self, event: Result<String, LinesCodecError>) {
+        let parsed_event = event
+            .as_ref()
+            .map_err(|why| Error(ErrorCode::ConnectError(why.to_string())))
+            .and_then(|event| {
+                serde_json::from_str::<Value>(&event)
+                    .map_err(|why| Error(ErrorCode::JsonParseError(why.to_string())))
+            });
+
+        match parsed_event {
+            Ok(event) => {
+                log::trace!("Parsed event: {:?}", event);
+                if let Err(broadcast::error::SendError(_)) =
+                    self.event_channel.send(MpvIpcEvent(event.to_owned()))
+                {
+                    log::trace!("Failed to send event to channel, ignoring");
+                }
+            }
+            Err(e) => {
+                log::trace!("Error parsing event, ignoring:\n  {:?}\n  {:?}", event, e);
+            }
+        }
+    }
+
     pub(crate) async fn run(mut self) -> Result<(), Error> {
         loop {
             tokio::select! {
               Some(event) = self.socket.next() => {
-                log::trace!("Handling event: {:?}", serde_json::from_str::<Value>(&event.unwrap()).unwrap());
+                log::trace!("Got event: {:?}", event);
-                // TODO: handle event
+                // TODO: error handling
+                self.handle_event(event).await;
               }
               Some((cmd, tx)) = self.command_channel.recv() => {
                   log::trace!("Handling command: {:?}", cmd);
                   match cmd {
                       MpvIpcCommand::Command(command) => {
-                          let refs = command.iter().map(|s| s.as_str()).collect::<Vec<&str>>();
+                          let refs = command.iter().map(|s| json!(s)).collect::<Vec<Value>>();
                           let response = self.send_command(refs.as_slice()).await;
                           tx.send(MpvIpcResponse(response)).unwrap()
                       }

src/lib.rs

@@ -1,5 +1,7 @@
 mod api;
 mod ipc;
 mod message_parser;
+mod api_extension;
 
 pub use api::*;
+pub use api_extension::*;

tests/events.rs

@@ -0,0 +1,79 @@
+use std::panic;
+
+use futures::{stream::StreamExt, SinkExt};
+use mpvipc::{Mpv, MpvExt, MpvDataType, Property};
+use serde_json::json;
+use test_log::test;
+use tokio::{net::UnixStream, task::JoinHandle};
+use tokio_util::codec::{Framed, LinesCodec, LinesCodecError};
+
+use mpvipc::Event;
+
+fn test_socket(
+    answers: Vec<(bool, String)>,
+) -> (UnixStream, JoinHandle<Result<(), LinesCodecError>>) {
+    let (socket, server) = UnixStream::pair().unwrap();
+    let join_handle = tokio::spawn(async move {
+        let mut framed = Framed::new(socket, LinesCodec::new());
+        for (unsolicited, answer) in answers {
+            if !unsolicited {
+                framed.next().await;
+            }
+            framed.send(answer).await?;
+        }
+        Ok(())
+    });
+
+    (server, join_handle)
+}
+
+#[test(tokio::test)]
+async fn test_observe_event_successful() {
+    let (server, join_handle) = test_socket(vec![
+        (
+            false,
+            json!({ "request_id": 0, "error": "success" }).to_string(),
+        ),
+        (
+            false,
+            json!({ "request_id": 0, "error": "success" }).to_string(),
+        ),
+        (
+            true,
+            json!({ "data": 64.0, "event": "property-change", "id": 1, "name": "volume" })
+                .to_string(),
+        ),
+    ]);
+
+    let mpv = Mpv::connect_socket(server).await.unwrap();
+
+    mpv.observe_property(1, "volume").await.unwrap();
+
+    let mpv2 = mpv.clone();
+    tokio::spawn(async move {
+        let event = mpv2.get_event_stream().await.next().await.unwrap().unwrap();
+
+        let property = match event {
+            Event::PropertyChange { id, property } => {
+                assert_eq!(id, 1);
+                property
+            }
+            err => panic!("{:?}", err),
+        };
+        let data = match property {
+            Property::Unknown { name, data } => {
+                assert_eq!(name, "volume");
+                data
+            }
+            err => panic!("{:?}", err),
+        };
+        match data {
+            MpvDataType::Double(data) => assert_eq!(data, 64.0),
+            err => panic!("{:?}", err),
+        }
+    });
+
+    mpv.set_property("volume", 64.0).await.unwrap();
+
+    join_handle.await.unwrap().unwrap();
+}

tests/get_property.rs

@@ -1,7 +1,7 @@
 use std::{panic, time::Duration};
 
 use futures::{stream::FuturesUnordered, SinkExt, StreamExt};
-use mpvipc::{Error, ErrorCode, Mpv, Playlist, PlaylistEntry};
+use mpvipc::{Error, ErrorCode, Mpv, MpvExt, Playlist, PlaylistEntry};
 use serde_json::{json, Value};
 use test_log::test;
 use tokio::{net::UnixStream, task::JoinHandle};

tests/set_property.rs

@@ -1,7 +1,7 @@
 use std::{panic, time::Duration};
 
 use futures::{stream::FuturesUnordered, SinkExt, StreamExt};
-use mpvipc::{Error, ErrorCode, Mpv, Playlist, PlaylistEntry};
+use mpvipc::{Error, ErrorCode, Mpv, MpvExt, Playlist, PlaylistEntry};
 use serde_json::{json, Value};
 use test_log::test;
 use tokio::{net::UnixStream, task::JoinHandle};