fix examples and documentation

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

@@ -63,33 +63,17 @@ jobs:
       - name: Cache dependencies
         uses: Swatinem/rust-cache@v2
 
-      - name: Create necessary directories
-        run: mkdir -p target/test-report
+      - name: Install nextest
+        run: cargo binstall -y cargo-nextest --secure
 
       - name: Run tests
         run: |
-          cargo test --all-features --release --no-fail-fast -- -Zunstable-options --format json --report-time \
-            | tee target/test-report/test-report.json
+          cargo nextest run --all-features --release --no-fail-fast
         env:
+          RUST_LOG: "trace"
           RUSTFLAGS: "-Cinstrument-coverage"
           LLVM_PROFILE_FILE: "target/coverage/%p-%m.profraw"
 
-      - name: Install markdown-test-report
-        run: cargo binstall -y markdown-test-report
-
-      - name: Generate test report
-        run: markdown-test-report target/test-report/test-report.json --output target/test-report/test-report.md
-
-      - name: Upload test report
-        uses: https://git.pvv.ntnu.no/oysteikt/rsync-action@main
-        with:
-          source: target/test-report/test-report.md
-          target: mpvipc/${{ gitea.ref_name }}/
-          username: oysteikt
-          ssh-key: ${{ secrets.OYSTEIKT_GITEA_WEBDOCS_SSH_KEY }}
-          host: microbel.pvv.ntnu.no
-          known-hosts: "microbel.pvv.ntnu.no ecdsa-sha2-nistp256 AAAAE2VjZHNhLXNoYTItbmlzdHAyNTYAAAAIbmlzdHAyNTYAAABBBEq0yasKP0mH6PI6ypmuzPzMnbHELo9k+YB5yW534aKudKZS65YsHJKQ9vapOtmegrn5MQbCCgrshf+/XwZcjbM="
-
       - name: Install grcov
         run: cargo binstall -y grcov
 

Cargo.toml

@@ -1,12 +1,15 @@
 [package]
 name = "mpvipc"
 version = "1.3.0"
-authors = ["Jonas Frei <freijon@pm.me>"]
+authors = [
+  "Jonas Frei <freijon@pm.me>",
+  "h7x4 <h7x4@nani.wtf>"
+]
 description = "A small library which provides bindings to control existing mpv instances through sockets."
 license = "GPL-3.0"
-homepage = "https://gitlab.com/mpv-ipc/mpvipc"
-repository = "https://gitlab.com/mpv-ipc/mpvipc"
-documentation = "https://docs.rs/mpvipc/"
+homepage = "https://git.pvv.ntnu.no/oysteikt/mpvipc"
+repository = "https://git.pvv.ntnu.no/oysteikt/mpvipc"
+documentation = "https://pvv.ntnu.no/~oysteikt/gitea/mpvipc/master/docs/mpvipc/"
 edition = "2021"
 rust-version = "1.75"
 

README.md

@@ -5,46 +5,30 @@
 
 A small library which provides bindings to control existing mpv instances through sockets.
 
-To make use of this library, please make sure mpv is started with the following option:
-`
-$ mpv --input-ipc-server=/tmp/mpv.sock --idle ...
-`
-
 ## Dependencies
 
 - `mpv`
-- `cargo` (makedep)
-
-## Install
-
-- [Cargo](https://crates.io/crates/mpvipc)
-
-You can use this package with cargo.
+- `cargo` (make dependency)
+- `cargo-nextest` (test depencency)
+- `grcov` (test depencency)
 
 ## Example
 
 Make sure mpv is started with the following option:
-`
-$ mpv --input-ipc-server=/tmp/mpv.sock --idle
-`
 
-Here is a small code example which connects to the socket /tmp/mpv.sock and toggles playback.
+```bash
+$ mpv --input-ipc-server=/tmp/mpv.sock --idle
+```
+
+Here is a small code example which connects to the socket `/tmp/mpv.sock` and toggles playback.
 
 ```rust
-extern crate mpvipc;
-
 use mpvipc::*;
-use std::sync::mpsc::channel;
 
-fn main() {
-    let mpv = Mpv::connect("/tmp/mpv.sock").unwrap();
-    let paused: bool = mpv.get_property("pause").unwrap();
+#[tokio::main]
+async fn main() -> Result<(), MpvError> {
+    let mpv = Mpv::connect("/tmp/mpv.sock").await?;
+    let paused: bool = mpv.get_property("pause").await?;
     mpv.set_property("pause", !paused).expect("Error pausing");
 }
 ```
-
-For a more extensive example and proof of concept, see project [mpvc](https://gitlab.com/mpv-ipc/mpvc).
-
-## Bugs / Ideas
-
-Check out the [Issue Tracker](https://gitlab.com/mpv-ipc/mpvipc/issues)

examples/fetch_state.rs

@@ -1,15 +1,19 @@
-use mpvipc::{MpvError, Mpv, MpvExt};
+use mpvipc::{Mpv, MpvError, MpvExt};
 
 #[tokio::main]
 async fn main() -> Result<(), MpvError> {
     env_logger::init();
 
     let mpv = Mpv::connect("/tmp/mpv.sock").await?;
+
     let meta = mpv.get_metadata().await?;
     println!("metadata: {:?}", meta);
+
     let playlist = mpv.get_playlist().await?;
     println!("playlist: {:?}", playlist);
+
     let playback_time: f64 = mpv.get_property("playback-time").await?;
     println!("playback-time: {}", playback_time);
+
     Ok(())
 }

examples/media_player.rs

@@ -1,4 +1,5 @@
-use mpvipc::{MpvError, Mpv, MpvExt};
+use futures::StreamExt;
+use mpvipc::{parse_event_property, Event, Mpv, MpvDataType, MpvError, MpvExt, Property};
 
 fn seconds_to_hms(total: f64) -> String {
     let total = total as u64;
@@ -14,55 +15,49 @@ async fn main() -> Result<(), MpvError> {
     env_logger::init();
 
     let mpv = Mpv::connect("/tmp/mpv.sock").await?;
-    let pause = false;
-    let playback_time = std::f64::NAN;
-    let duration = std::f64::NAN;
+
     mpv.observe_property(1, "path").await?;
     mpv.observe_property(2, "pause").await?;
     mpv.observe_property(3, "playback-time").await?;
     mpv.observe_property(4, "duration").await?;
     mpv.observe_property(5, "metadata").await?;
-    loop {
-        // TODO:
-        //     let event = mpv.event_listen()?;
-        //     match event {
-        //         Event::PropertyChange { id: _, property } => match property {
-        //             Property::Path(Some(value)) => println!("\nPlaying: {}", value),
-        //             Property::Path(None) => (),
-        //             Property::Pause(value) => pause = value,
-        //             Property::PlaybackTime(Some(value)) => playback_time = value,
-        //             Property::PlaybackTime(None) => playback_time = std::f64::NAN,
-        //             Property::Duration(Some(value)) => duration = value,
-        //             Property::Duration(None) => duration = std::f64::NAN,
-        //             Property::Metadata(Some(value)) => {
-        //                 println!("File tags:");
-        //                 if let Some(MpvDataType::String(value)) = value.get("ARTIST") {
-        //                     println!(" Artist: {}", value);
-        //                 }
-        //                 if let Some(MpvDataType::String(value)) = value.get("ALBUM") {
-        //                     println!(" Album: {}", value);
-        //                 }
-        //                 if let Some(MpvDataType::String(value)) = value.get("TITLE") {
-        //                     println!(" Title: {}", value);
-        //                 }
-        //                 if let Some(MpvDataType::String(value)) = value.get("TRACK") {
-        //                     println!(" Track: {}", value);
-        //                 }
-        //             }
-        //             Property::Metadata(None) => (),
-        //             Property::Unknown { name: _, data: _ } => (),
-        //         },
-        //         Event::Shutdown => return Ok(()),
-        //         Event::Unimplemented => panic!("Unimplemented event"),
-        //         _ => (),
-        //     }
-        //     print!(
-        //         "{}{} / {} ({:.0}%)\r",
-        //         if pause { "(Paused) " } else { "" },
-        //         seconds_to_hms(playback_time),
-        //         seconds_to_hms(duration),
-        //         100. * playback_time / duration
-        //     );
-        //     io::stdout().flush().unwrap();
+
+    let mut events = mpv.get_event_stream().await;
+    while let Some(Ok(event)) = events.next().await {
+        match event {
+            mpvipc::Event::PropertyChange { .. } => match parse_event_property(event)? {
+                (1, Property::Path(Some(value))) => println!("\nPlaying: {}", value),
+                (2, Property::Pause(value)) => {
+                    println!("Pause: {}", value);
+                }
+                (3, Property::PlaybackTime(Some(value))) => {
+                    println!("Playback time: {}", seconds_to_hms(value));
+                }
+                (4, Property::Duration(Some(value))) => {
+                    println!("Duration: {}", seconds_to_hms(value));
+                }
+                (5, Property::Metadata(Some(value))) => {
+                    println!("File tags:");
+                    if let Some(MpvDataType::String(value)) = value.get("ARTIST") {
+                        println!(" Artist: {}", value);
+                    }
+                    if let Some(MpvDataType::String(value)) = value.get("ALBUM") {
+                        println!(" Album: {}", value);
+                    }
+                    if let Some(MpvDataType::String(value)) = value.get("TITLE") {
+                        println!(" Title: {}", value);
+                    }
+                    if let Some(MpvDataType::String(value)) = value.get("TRACK") {
+                        println!(" Track: {}", value);
+                    }
+                }
+                _ => (),
+            },
+            Event::Shutdown => return Ok(()),
+            Event::Unimplemented(_) => panic!("Unimplemented event"),
+            _ => (),
+        }
     }
+
+    Ok(())
 }

scripts/coverage.sh

@@ -3,7 +3,7 @@ rm -rf target/coverage || true
 mkdir -p target/coverage
 
 echo "Running tests"
-RUST_LOG=mpvipc=trace RUSTFLAGS="-Cinstrument-coverage" LLVM_PROFILE_FILE="target/coverage/%p-%m.profraw" cargo test --all-features --release --no-fail-fast
+RUST_LOG=mpvipc=trace RUSTFLAGS="-Cinstrument-coverage" LLVM_PROFILE_FILE="target/coverage/%p-%m.profraw" cargo nextest run --all-features --release --no-fail-fast
 
 echo "Generating coverage report"
 grcov \

src/core_api.rs

@@ -167,7 +167,7 @@ where
         value: T,
     ) -> Result<(), MpvError> {
         let (res_tx, res_rx) = oneshot::channel();
-        let value = serde_json::to_value(value).map_err(|why| MpvError::JsonParseError(why))?;
+        let value = serde_json::to_value(value).map_err(MpvError::JsonParseError)?;
 
         instance
             .command_sender
@@ -199,6 +199,7 @@ pub struct Mpv {
     broadcast_channel: broadcast::Sender<MpvIpcEvent>,
 }
 
+// TODO: Can we somehow provide a more useful Debug implementation?
 impl fmt::Debug for Mpv {
     fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
         fmt.debug_struct("Mpv").finish()
@@ -206,6 +207,8 @@ impl fmt::Debug for Mpv {
 }
 
 impl Mpv {
+    /// Connect to a unix socket, hosted by mpv, at the given path.
+    /// This is the inteded way of creating a new [`Mpv`] instance.
     pub async fn connect(socket_path: &str) -> Result<Mpv, MpvError> {
         log::debug!("Connecting to mpv socket at {}", socket_path);
 
@@ -217,6 +220,10 @@ impl Mpv {
         Self::connect_socket(socket).await
     }
 
+    /// Connect to an existing [`UnixStream`].
+    /// This is an alternative to [`Mpv::connect`], if you already have a [`UnixStream`] available.
+    ///
+    /// Internally, this is used for testing purposes.
     pub async fn connect_socket(socket: UnixStream) -> Result<Mpv, MpvError> {
         let (com_tx, com_rx) = mpsc::channel(100);
         let (ev_tx, _) = broadcast::channel(100);
@@ -231,6 +238,11 @@ impl Mpv {
         })
     }
 
+    /// Disconnect from the mpv socket.
+    ///
+    /// Note that this will also kill communication for all other clones of this instance.
+    /// It will not kill the mpv process itself - for that you should use [`MpvCommand::Quit`]
+    /// or run [`MpvExt::kill`](crate::MpvExt::kill).
     pub async fn disconnect(&self) -> Result<(), MpvError> {
         let (res_tx, res_rx) = oneshot::channel();
         self.command_sender
@@ -244,6 +256,10 @@ impl Mpv {
         }
     }
 
+    /// Create a new stream, providing [`Event`]s from mpv.
+    ///
+    /// This is intended to be used with [`MpvCommand::Observe`] and [`MpvCommand::Unobserve`]
+    /// (or [`MpvExt::observe_property`] and [`MpvExt::unobserve_property`] respectively).
     pub async fn get_event_stream(&self) -> impl futures::Stream<Item = Result<Event, MpvError>> {
         tokio_stream::wrappers::BroadcastStream::new(self.broadcast_channel.subscribe()).map(
             |event| match event {
@@ -255,7 +271,7 @@ impl Mpv {
 
     /// Run a custom command.
     /// This should only be used if the desired command is not implemented
-    /// with [MpvCommand].
+    /// with [`MpvCommand`].
     pub async fn run_command_raw(
         &self,
         command: &str,
@@ -281,6 +297,7 @@ impl Mpv {
         }
     }
 
+    /// Helper function to ignore the return value of a command, and only check for errors.
     async fn run_command_raw_ignore_value(
         &self,
         command: &str,
@@ -300,18 +317,20 @@ impl Mpv {
     ///
     /// # Example
     /// ```
-    /// use mpvipc::{Mpv, Error};
-    /// fn main() -> Result<(), Error> {
-    ///     let mpv = Mpv::connect("/tmp/mpvsocket")?;
+    /// use mpvipc::{Mpv, MpvError};
+    ///
+    /// #[tokio::main]
+    /// async fn main() -> Result<(), MpvError> {
+    ///     let mpv = Mpv::connect("/tmp/mpvsocket").await?;
     ///
     ///     //Run command 'playlist-shuffle' which takes no arguments
-    ///     mpv.run_command(MpvCommand::PlaylistShuffle)?;
+    ///     mpv.run_command(MpvCommand::PlaylistShuffle).await?;
     ///
     ///     //Run command 'seek' which in this case takes two arguments
     ///     mpv.run_command(MpvCommand::Seek {
     ///         seconds: 0f64,
     ///         option: SeekOptions::Absolute,
-    ///     })?;
+    ///     }).await?;
     ///     Ok(())
     /// }
     /// ```
@@ -430,9 +449,11 @@ impl Mpv {
     ///
     /// # Example
     /// ```
-    /// use mpvipc::{Mpv, Error};
-    /// async fn main() -> Result<(), Error> {
-    ///     let mpv = Mpv::connect("/tmp/mpvsocket")?;
+    /// use mpvipc::{Mpv, MpvError};
+    ///
+    /// #[tokio::main]
+    /// async fn main() -> Result<(), MpvError> {
+    ///     let mpv = Mpv::connect("/tmp/mpvsocket").await?;
     ///     let paused: bool = mpv.get_property("pause").await?;
     ///     let title: String = mpv.get_property("media-title").await?;
     ///     Ok(())
@@ -457,10 +478,12 @@ impl Mpv {
     /// # Example
     ///
     /// ```
-    /// use mpvipc::{Mpv, Error};
-    /// fn main() -> Result<(), Error> {
-    ///     let mpv = Mpv::connect("/tmp/mpvsocket")?;
-    ///     let title = mpv.get_property_string("media-title")?;
+    /// use mpvipc::{Mpv, MpvError};
+    ///
+    /// #[tokio::main]
+    /// async fn main() -> Result<(), MpvError> {
+    ///     let mpv = Mpv::connect("/tmp/mpvsocket").await?;
+    ///     let title = mpv.get_property_string("media-title").await?;
     ///     Ok(())
     /// }
     /// ```
@@ -496,9 +519,9 @@ impl Mpv {
     ///
     /// # Example
     /// ```
-    /// use mpvipc::{Mpv, Error};
-    /// fn async main() -> Result<(), Error> {
-    ///     let mpv = Mpv::connect("/tmp/mpvsocket")?;
+    /// use mpvipc::{Mpv, MpvError};
+    /// async fn main() -> Result<(), MpvError> {
+    ///     let mpv = Mpv::connect("/tmp/mpvsocket").await?;
     ///     mpv.set_property("pause", true).await?;
     ///     Ok(())
     /// }

src/error.rs

@@ -1,7 +1,7 @@
 //! Library specific error messages.
 
-use thiserror::Error;
 use serde_json::{Map, Value};
+use thiserror::Error;
 
 use crate::MpvDataType;
 
@@ -22,14 +22,16 @@ pub enum MpvError {
 
     #[error("Mpv sent a value with an unexpected type:\nExpected {expected_type}, received {received:#?}")]
     ValueContainsUnexpectedType {
-      expected_type: String,
-      received: Value,
+        expected_type: String,
+        received: Value,
     },
 
-    #[error("Mpv sent data with an unexpected type:\nExpected {expected_type}, received {received:#?}")]
+    #[error(
+        "Mpv sent data with an unexpected type:\nExpected {expected_type}, received {received:#?}"
+    )]
     DataContainsUnexpectedType {
-      expected_type: String,
-      received: MpvDataType,
+        expected_type: String,
+        received: MpvDataType,
     },
 
     #[error("Missing expected 'data' field in mpv message")]
@@ -37,8 +39,8 @@ pub enum MpvError {
 
     #[error("Missing key in object:\nExpected {key} in {map:#?}")]
     MissingKeyInObject {
-      key: String,
-      map: Map<String, Value>,
+        key: String,
+        map: Map<String, Value>,
     },
 
     #[error("Unknown error: {0}")]

src/event_parser.rs

@@ -292,11 +292,11 @@ fn parse_client_message(event: &Map<String, Value>) -> Result<Event, MpvError> {
 fn parse_property_change(event: &Map<String, Value>) -> Result<Event, MpvError> {
     let id = get_key_as!(as_u64, "id", event) as usize;
     let property_name = get_key_as!(as_str, "name", event);
-    let data = event.get("data").map(|d| json_to_value(d)).transpose()?;
+    let data = event.get("data").map(json_to_value).transpose()?;
 
     Ok(Event::PropertyChange {
         id,
         name: property_name.to_string(),
-        data: data,
+        data,
     })
 }

src/event_property_parser.rs

@@ -34,7 +34,10 @@ pub enum Property {
     Speed(f64),
     Volume(f64),
     Mute(bool),
-    Unknown { name: String, data: Option<MpvDataType> },
+    Unknown {
+        name: String,
+        data: Option<MpvDataType>,
+    },
 }
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]

src/highlevel_api_extension.rs

@@ -1,7 +1,7 @@
 //! High-level API extension for [`Mpv`].
 
 use crate::{
-    MpvError, IntoRawCommandPart, Mpv, MpvCommand, MpvDataType, Playlist, PlaylistAddOptions,
+    IntoRawCommandPart, Mpv, MpvCommand, MpvDataType, MpvError, Playlist, PlaylistAddOptions,
     PlaylistEntry, SeekOptions,
 };
 use serde::{Deserialize, Serialize};
@@ -44,35 +44,98 @@ pub enum PlaylistAddTypeOptions {
 // TODO: fix this
 #[allow(async_fn_in_trait)]
 pub trait MpvExt {
-    async fn toggle(&self) -> Result<(), MpvError>;
+    /// Stop the player completely (as opposed to pausing),
+    /// removing the pointer to the current video.
     async fn stop(&self) -> Result<(), MpvError>;
-    async fn set_volume(&self, input_volume: f64, option: NumberChangeOptions)
-        -> Result<(), MpvError>;
-    async fn set_speed(&self, input_speed: f64, option: NumberChangeOptions) -> Result<(), MpvError>;
+
+    /// Set the volume of the player.
+    async fn set_volume(
+        &self,
+        input_volume: f64,
+        option: NumberChangeOptions,
+    ) -> Result<(), MpvError>;
+
+    /// Set the playback speed of the player.
+    async fn set_speed(
+        &self,
+        input_speed: f64,
+        option: NumberChangeOptions,
+    ) -> Result<(), MpvError>;
+
+    /// Toggle/set the pause state of the player.
+    async fn set_playback(&self, option: Switch) -> Result<(), MpvError>;
+
+    /// Toggle/set the mute state of the player.
     async fn set_mute(&self, option: Switch) -> Result<(), MpvError>;
+
+    /// Toggle/set whether the player should loop the current playlist.
     async fn set_loop_playlist(&self, option: Switch) -> Result<(), MpvError>;
+
+    /// Toggle/set whether the player should loop the current video.
     async fn set_loop_file(&self, option: Switch) -> Result<(), MpvError>;
+
+    /// Seek to a specific position in the current video.
     async fn seek(&self, seconds: f64, option: SeekOptions) -> Result<(), MpvError>;
+
+    /// Shuffle the current playlist.
     async fn playlist_shuffle(&self) -> Result<(), MpvError>;
+
+    /// Remove an entry from the playlist.
     async fn playlist_remove_id(&self, id: usize) -> Result<(), MpvError>;
+
+    /// Play the next entry in the playlist.
     async fn playlist_play_next(&self, id: usize) -> Result<(), MpvError>;
+
+    /// Play a specific entry in the playlist.
     async fn playlist_play_id(&self, id: usize) -> Result<(), MpvError>;
+
+    /// Move an entry in the playlist.
+    ///
+    /// The `from` parameter is the current position of the entry, and the `to` parameter is the new position.
+    /// Mpv will then move the entry from the `from` position to the `to` position,
+    /// shifting after `to` one number up. Paradoxically, that means that moving an entry further down the list
+    /// will result in a final position that is one less than the `to` parameter.
     async fn playlist_move_id(&self, from: usize, to: usize) -> Result<(), MpvError>;
+
+    /// Remove all entries from the playlist.
     async fn playlist_clear(&self) -> Result<(), MpvError>;
+
+    /// Add a file or playlist to the playlist.
     async fn playlist_add(
         &self,
         file: &str,
         file_type: PlaylistAddTypeOptions,
         option: PlaylistAddOptions,
     ) -> Result<(), MpvError>;
+
+    /// Start the current video from the beginning.
     async fn restart(&self) -> Result<(), MpvError>;
+
+    /// Play the previous entry in the playlist.
     async fn prev(&self) -> Result<(), MpvError>;
-    async fn pause(&self) -> Result<(), MpvError>;
-    async fn unobserve_property(&self, id: usize) -> Result<(), MpvError>;
+
+    /// Notify mpv to send events whenever a property changes.
+    /// See [`Mpv::get_event_stream`] and [`Property`](crate::Property) for more information.
     async fn observe_property(&self, id: usize, property: &str) -> Result<(), MpvError>;
+
+    /// Stop observing a property.
+    /// See [`Mpv::get_event_stream`] and [`Property`](crate::Property) for more information.
+    async fn unobserve_property(&self, id: usize) -> Result<(), MpvError>;
+
+    /// Skip to the next entry in the playlist.
     async fn next(&self) -> Result<(), MpvError>;
+
+    /// Stop mpv completely, and kill the process.
+    ///
+    /// Note that this is different than forcefully killing the process using
+    /// as handle to a subprocess, it will only send a command to mpv to ask
+    /// it to exit itself. If mpv is stuck, it may not respond to this command.
     async fn kill(&self) -> Result<(), MpvError>;
+
+    /// Get a list of all entries in the playlist.
     async fn get_playlist(&self) -> Result<Playlist, MpvError>;
+
+    /// Get metadata about the current video.
     async fn get_metadata(&self) -> Result<HashMap<String, MpvDataType>, MpvError>;
 }
 
@@ -107,8 +170,21 @@ impl MpvExt for Mpv {
         self.run_command(MpvCommand::Unobserve(id)).await
     }
 
-    async fn pause(&self) -> Result<(), MpvError> {
-        self.set_property("pause", true).await
+    async fn set_playback(&self, option: Switch) -> Result<(), MpvError> {
+        let enabled = match option {
+            Switch::On => "yes",
+            Switch::Off => "no",
+            Switch::Toggle => {
+                self.get_property::<String>("pause")
+                    .await
+                    .map(|s| match s.as_str() {
+                        "yes" => "no",
+                        "no" => "yes",
+                        _ => "no",
+                    })?
+            }
+        };
+        self.set_property("pause", enabled).await
     }
 
     async fn prev(&self) -> Result<(), MpvError> {
@@ -237,7 +313,11 @@ impl MpvExt for Mpv {
         self.set_property("mute", enabled).await
     }
 
-    async fn set_speed(&self, input_speed: f64, option: NumberChangeOptions) -> Result<(), MpvError> {
+    async fn set_speed(
+        &self,
+        input_speed: f64,
+        option: NumberChangeOptions,
+    ) -> Result<(), MpvError> {
         match self.get_property::<f64>("speed").await {
             Ok(speed) => match option {
                 NumberChangeOptions::Increase => {
@@ -278,8 +358,4 @@ impl MpvExt for Mpv {
     async fn stop(&self) -> Result<(), MpvError> {
         self.run_command(MpvCommand::Stop).await
     }
-
-    async fn toggle(&self) -> Result<(), MpvError> {
-        self.run_command_raw("cycle", &["pause"]).await.map(|_| ())
-    }
 }

src/ipc.rs

@@ -56,7 +56,7 @@ impl MpvIpc {
     ) -> Result<Option<Value>, MpvError> {
         let ipc_command = json!({ "command": command });
         let ipc_command_str =
-            serde_json::to_string(&ipc_command).map_err(|why| MpvError::JsonParseError(why))?;
+            serde_json::to_string(&ipc_command).map_err(MpvError::JsonParseError)?;
 
         log::trace!("Sending command: {}", ipc_command_str);
 
@@ -75,8 +75,8 @@ impl MpvIpc {
                 ))?
                 .map_err(|why| MpvError::MpvSocketConnectionError(why.to_string()))?;
 
-            let parsed_response = serde_json::from_str::<Value>(&response)
-                .map_err(|why| MpvError::JsonParseError(why));
+            let parsed_response =
+                serde_json::from_str::<Value>(&response).map_err(MpvError::JsonParseError);
 
             if parsed_response
                 .as_ref()
@@ -155,7 +155,7 @@ impl MpvIpc {
                     .map_err(|why| MpvError::MpvSocketConnectionError(why.to_string()))
                     .and_then(|event|
                         serde_json::from_str::<Value>(&event)
-                        .map_err(|why| MpvError::JsonParseError(why)));
+                        .map_err(MpvError::JsonParseError));
 
                 self.handle_event(parsed_event).await;
               }

src/message_parser.rs

@@ -152,7 +152,7 @@ pub(crate) fn json_map_to_hashmap(
 }
 
 pub(crate) fn json_array_to_vec(array: &[Value]) -> Result<Vec<MpvDataType>, MpvError> {
-    array.iter().map(|entry| json_to_value(entry)).collect()
+    array.iter().map(json_to_value).collect()
 }
 
 pub(crate) fn json_array_to_playlist(array: &[Value]) -> Vec<PlaylistEntry> {

tests/integration.rs

@@ -1,99 +1,8 @@
-use mpvipc::{MpvError, Mpv, MpvExt};
-use std::path::Path;
-use tokio::{
-    process::{Child, Command},
-    time::{sleep, timeout, Duration},
-};
+// mod event_property_parser {
+//   include!("integration/event_property_parser.rs")
+// }
+// mod util;
+// mod misc;
+mod integration_tests;
 
-#[cfg(target_family = "unix")]
-async fn spawn_headless_mpv() -> Result<(Child, Mpv), MpvError> {
-    let socket_path_str = format!("/tmp/mpv-ipc-{}", uuid::Uuid::new_v4());
-    let socket_path = Path::new(&socket_path_str);
-
-    let process_handle = Command::new("mpv")
-        .arg("--no-config")
-        .arg("--idle")
-        .arg("--no-video")
-        .arg("--no-audio")
-        .arg(format!(
-            "--input-ipc-server={}",
-            &socket_path.to_str().unwrap()
-        ))
-        .spawn()
-        .expect("Failed to start mpv");
-
-    if timeout(Duration::from_millis(500), async {
-        while !&socket_path.exists() {
-            sleep(Duration::from_millis(10)).await;
-        }
-    })
-    .await
-    .is_err()
-    {
-        panic!("Failed to create mpv socket at {:?}", &socket_path);
-    }
-
-    let mpv = Mpv::connect(socket_path.to_str().unwrap()).await.unwrap();
-    Ok((process_handle, mpv))
-}
-
-#[tokio::test]
-#[cfg(target_family = "unix")]
-async fn test_get_mpv_version() {
-    let (mut proc, mpv) = spawn_headless_mpv().await.unwrap();
-    let version: String = mpv.get_property("mpv-version").await.unwrap();
-    assert!(version.starts_with("mpv"));
-
-    mpv.kill().await.unwrap();
-    proc.kill().await.unwrap();
-}
-
-#[tokio::test]
-#[cfg(target_family = "unix")]
-async fn test_set_property() {
-    let (mut proc, mpv) = spawn_headless_mpv().await.unwrap();
-    mpv.set_property("pause", true).await.unwrap();
-    let paused: bool = mpv.get_property("pause").await.unwrap();
-    assert!(paused);
-
-    mpv.kill().await.unwrap();
-    proc.kill().await.unwrap();
-}
-
-#[tokio::test]
-#[cfg(target_family = "unix")]
-async fn test_events() {
-    use futures::stream::StreamExt;
-
-    let (mut proc, mpv) = spawn_headless_mpv().await.unwrap();
-
-    mpv.observe_property(1337, "pause").await.unwrap();
-
-    let mut events = mpv.get_event_stream().await;
-    let event_checking_thread = tokio::spawn(async move {
-        loop {
-            let event = events.next().await.unwrap().unwrap();
-            if let (1337, property) = mpvipc::parse_event_property(event).unwrap() {
-                assert_eq!(property, mpvipc::Property::Pause(true));
-                break;
-            }
-        }
-    });
-
-    tokio::time::sleep(Duration::from_millis(10)).await;
-
-    mpv.set_property("pause", true).await.unwrap();
-
-    if tokio::time::timeout(
-        tokio::time::Duration::from_millis(500),
-        event_checking_thread,
-    )
-    .await
-    .is_err()
-    {
-        panic!("Event checking thread timed out");
-    }
-
-    mpv.kill().await.unwrap();
-    proc.kill().await.unwrap();
-}
+// use util::*;

tests/integration_tests/event_property_parser.rs

@@ -0,0 +1,216 @@
+use futures::{stream::StreamExt, Stream};
+use mpvipc::{parse_event_property, Event, Mpv, MpvError, MpvExt};
+use thiserror::Error;
+use tokio::time::sleep;
+use tokio::time::{timeout, Duration};
+
+use test_log::test;
+
+use super::*;
+
+const MPV_CHANNEL_ID: usize = 1337;
+
+#[derive(Error, Debug)]
+enum PropertyCheckingThreadError {
+    #[error("Unexpected property: {0:?}")]
+    UnexpectedPropertyError(mpvipc::Property),
+
+    #[error(transparent)]
+    MpvError(#[from] MpvError),
+}
+
+fn create_interruptable_event_property_checking_thread<T>(
+    mut events: impl Stream<Item = Result<Event, MpvError>> + Unpin + Send + 'static,
+    on_property: T,
+) -> (
+    tokio::task::JoinHandle<Result<(), PropertyCheckingThreadError>>,
+    tokio_util::sync::CancellationToken,
+)
+where
+    T: Fn(mpvipc::Property) -> bool + Send + 'static,
+{
+    let cancellation_token = tokio_util::sync::CancellationToken::new();
+    let cancellation_token_clone = cancellation_token.clone();
+    let handle = tokio::spawn(async move {
+        loop {
+            tokio::select! {
+                event = events.next() => {
+                    match event {
+                        Some(Ok(event)) => {
+                            match event {
+                                Event::PropertyChange { id: MPV_CHANNEL_ID, .. } => {
+                                    let property = parse_event_property(event).unwrap().1;
+                                    if !on_property(property.clone()) {
+                                        return Err(PropertyCheckingThreadError::UnexpectedPropertyError(property))
+                                    }
+                                }
+                                _ => {
+                                    log::trace!("Received unrelated event, ignoring: {:?}", event);
+                                }
+                            }
+                        }
+                        Some(Err(err)) => return Err(err.into()),
+                        None => return Ok(()),
+                    }
+                }
+                _ = cancellation_token_clone.cancelled() => return Ok(()),
+            }
+        }
+    });
+
+    (handle, cancellation_token)
+}
+
+async fn graceful_shutdown(
+    cancellation_token: tokio_util::sync::CancellationToken,
+    handle: tokio::task::JoinHandle<Result<(), PropertyCheckingThreadError>>,
+    mpv: Mpv,
+    mut proc: tokio::process::Child,
+) -> Result<(), MpvError> {
+    cancellation_token.cancel();
+
+    match timeout(Duration::from_millis(500), handle).await {
+        Ok(Ok(Ok(()))) => {}
+        Ok(Ok(Err(err))) => match err {
+            PropertyCheckingThreadError::UnexpectedPropertyError(property) => {
+                return Err(MpvError::Other(format!(
+                    "Unexpected property: {:?}",
+                    property
+                )));
+            }
+            PropertyCheckingThreadError::MpvError(err) => return Err(err),
+        },
+        Ok(Err(_)) => {
+            return Err(MpvError::InternalConnectionError(
+                "Event checking thread timed out".to_owned(),
+            ));
+        }
+        Err(_) => {
+            return Err(MpvError::InternalConnectionError(
+                "Event checking thread panicked".to_owned(),
+            ));
+        }
+    }
+
+    mpv.kill().await?;
+    proc.wait().await.map_err(|err| {
+        MpvError::InternalConnectionError(format!(
+            "Failed to wait for mpv process to exit: {}",
+            err
+        ))
+    })?;
+
+    Ok(())
+}
+
+#[test(tokio::test)]
+#[cfg(target_family = "unix")]
+async fn test_highlevel_event_pause() -> Result<(), MpvError> {
+    let (proc, mpv) = spawn_headless_mpv().await?;
+
+    mpv.observe_property(MPV_CHANNEL_ID, "pause").await?;
+
+    let events = mpv.get_event_stream().await;
+    let (handle, cancellation_token) =
+        create_interruptable_event_property_checking_thread(events, |property| match property {
+            mpvipc::Property::Pause(_) => {
+                log::debug!("{:?}", property);
+                true
+            }
+            _ => false,
+        });
+
+    sleep(Duration::from_millis(5)).await;
+    mpv.set_property("pause", false).await?;
+    sleep(Duration::from_millis(5)).await;
+    mpv.set_property("pause", true).await?;
+    sleep(Duration::from_millis(5)).await;
+
+    graceful_shutdown(cancellation_token, handle, mpv, proc).await?;
+
+    Ok(())
+}
+
+#[test(tokio::test)]
+#[cfg(target_family = "unix")]
+async fn test_highlevel_event_volume() -> Result<(), MpvError> {
+    let (proc, mpv) = spawn_headless_mpv().await?;
+
+    mpv.observe_property(1337, "volume").await?;
+    let events = mpv.get_event_stream().await;
+    let (handle, cancellation_token) =
+        create_interruptable_event_property_checking_thread(events, |property| match property {
+            mpvipc::Property::Volume(_) => {
+                log::trace!("{:?}", property);
+                true
+            }
+            _ => false,
+        });
+
+    sleep(Duration::from_millis(5)).await;
+    mpv.set_property("volume", 100.0).await?;
+    sleep(Duration::from_millis(5)).await;
+    mpv.set_property("volume", 40).await?;
+    sleep(Duration::from_millis(5)).await;
+    mpv.set_property("volume", 0.0).await?;
+    sleep(Duration::from_millis(5)).await;
+
+    graceful_shutdown(cancellation_token, handle, mpv, proc).await?;
+
+    Ok(())
+}
+
+#[test(tokio::test)]
+#[cfg(target_family = "unix")]
+async fn test_highlevel_event_mute() -> Result<(), MpvError> {
+    let (proc, mpv) = spawn_headless_mpv().await?;
+
+    mpv.observe_property(1337, "mute").await?;
+    let events = mpv.get_event_stream().await;
+    let (handle, cancellation_token) =
+        create_interruptable_event_property_checking_thread(events, |property| match property {
+            mpvipc::Property::Mute(_) => {
+                log::trace!("{:?}", property);
+                true
+            }
+            _ => false,
+        });
+
+    sleep(Duration::from_millis(5)).await;
+    mpv.set_property("mute", true).await?;
+    sleep(Duration::from_millis(5)).await;
+    mpv.set_property("mute", false).await?;
+    sleep(Duration::from_millis(5)).await;
+
+    graceful_shutdown(cancellation_token, handle, mpv, proc).await?;
+
+    Ok(())
+}
+
+#[test(tokio::test)]
+#[cfg(target_family = "unix")]
+async fn test_highlevel_event_duration() -> Result<(), MpvError> {
+    let (proc, mpv) = spawn_headless_mpv().await?;
+
+    mpv.observe_property(1337, "duration").await?;
+
+    let events = mpv.get_event_stream().await;
+    let (handle, cancellation_token) =
+        create_interruptable_event_property_checking_thread(events, |property| match property {
+            mpvipc::Property::Duration(_) => {
+                log::trace!("{:?}", property);
+                true
+            }
+            _ => false,
+        });
+
+    sleep(Duration::from_millis(5)).await;
+    mpv.set_property("pause", true).await?;
+    sleep(Duration::from_millis(5)).await;
+    mpv.set_property("pause", false).await?;
+    sleep(Duration::from_millis(5)).await;
+
+    graceful_shutdown(cancellation_token, handle, mpv, proc).await?;
+
+    Ok(())
+}

tests/integration_tests/misc.rs

@@ -0,0 +1,26 @@
+use mpvipc::MpvExt;
+
+use super::*;
+
+#[tokio::test]
+#[cfg(target_family = "unix")]
+async fn test_get_mpv_version() {
+    let (mut proc, mpv) = spawn_headless_mpv().await.unwrap();
+    let version: String = mpv.get_property("mpv-version").await.unwrap();
+    assert!(version.starts_with("mpv"));
+
+    mpv.kill().await.unwrap();
+    proc.kill().await.unwrap();
+}
+
+#[tokio::test]
+#[cfg(target_family = "unix")]
+async fn test_set_property() {
+    let (mut proc, mpv) = spawn_headless_mpv().await.unwrap();
+    mpv.set_property("pause", true).await.unwrap();
+    let paused: bool = mpv.get_property("pause").await.unwrap();
+    assert!(paused);
+
+    mpv.kill().await.unwrap();
+    proc.kill().await.unwrap();
+}

tests/integration_tests/mod.rs

@@ -0,0 +1,5 @@
+mod event_property_parser;
+mod misc;
+mod util;
+
+use util::*;

tests/integration_tests/util.rs

@@ -0,0 +1,43 @@
+use std::{path::Path, time::Duration};
+
+use mpvipc::{Mpv, MpvError};
+use tokio::{
+    process::{Child, Command},
+    time::{sleep, timeout},
+};
+
+#[cfg(target_family = "unix")]
+pub async fn spawn_headless_mpv() -> Result<(Child, Mpv), MpvError> {
+    let socket_path_str = format!("/tmp/mpv-ipc-{}", uuid::Uuid::new_v4());
+    let socket_path = Path::new(&socket_path_str);
+
+    // TODO: Verify that `mpv` exists in `PATH``
+    let process_handle = Command::new("mpv")
+        .arg("--no-config")
+        .arg("--idle")
+        .arg("--no-video")
+        .arg("--no-audio")
+        .arg(format!(
+            "--input-ipc-server={}",
+            &socket_path.to_str().unwrap()
+        ))
+        .kill_on_drop(true)
+        .spawn()
+        .expect("Failed to start mpv");
+
+    timeout(Duration::from_millis(500), async {
+        while !&socket_path.exists() {
+            sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .map_err(|_| {
+        MpvError::MpvSocketConnectionError(format!(
+            "Failed to create mpv socket at {:?}, timed out waiting for socket file to be created",
+            &socket_path
+        ))
+    })?;
+
+    let mpv = Mpv::connect(socket_path.to_str().unwrap()).await?;
+    Ok((process_handle, mpv))
+}

tests/mock_socket.rs

@@ -0,0 +1 @@
+mod mock_socket_tests;

tests/mock_socket_tests/get_property.rs

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

tests/mock_socket_tests/mod.rs

@@ -0,0 +1,3 @@
+mod events;
+mod get_property;
+mod set_property;

tests/mock_socket_tests/set_property.rs

@@ -1,7 +1,7 @@
 use std::{panic, time::Duration};
 
 use futures::{stream::FuturesUnordered, SinkExt, StreamExt};
-use mpvipc::{MpvError, Mpv, MpvExt, Playlist, PlaylistEntry};
+use mpvipc::{Mpv, MpvError, MpvExt, Playlist, PlaylistEntry};
 use serde_json::{json, Value};
 use test_log::test;
 use tokio::{net::UnixStream, task::JoinHandle};
@@ -130,7 +130,7 @@ async fn test_set_property_simultaneous_requests() {
         loop {
             let status = mpv_clone_1.set_property("volume", 100).await;
             match status {
-                Ok(()) => {},
+                Ok(()) => {}
                 _ => panic!("Unexpected result: {:?}", status),
             }
         }
@@ -142,7 +142,7 @@ async fn test_set_property_simultaneous_requests() {
             tokio::time::sleep(Duration::from_millis(1)).await;
             let status = mpv_clone_2.set_property("pause", false).await;
             match status {
-                Ok(()) => {},
+                Ok(()) => {}
                 _ => panic!("Unexpected result: {:?}", status),
             }
         }