event_parser: add some unit tests

Duplicate of the LICENSE file

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

@@ -7,37 +7,26 @@ on:
 
 jobs:
   build:
-    runs-on: ubuntu-latest-personal
+    runs-on: debian-latest
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v6
 
-      - name: Install latest nightly toolchain
-        uses: actions-rs/toolchain@v1
-        with:
-            toolchain: nightly
-            override: true
-
-      - name: Cache dependencies
-        uses: Swatinem/rust-cache@v2
+      - name: Install rust toolchain
+        uses: dtolnay/rust-toolchain@stable
 
       - name: Build
         run: cargo build --all-features --verbose --release
 
   check:
-    runs-on: ubuntu-latest-personal
+    runs-on: debian-latest
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v6
 
-      - name: Install latest nightly toolchain
-        uses: actions-rs/toolchain@v1
+      - name: Install rust toolchain
+        uses: dtolnay/rust-toolchain@stable
         with:
-            toolchain: nightly
-            override: true
             components: rustfmt, clippy
 
-      - name: Cache dependencies
-        uses: Swatinem/rust-cache@v2
-
       - name: Check code format
         run: cargo fmt -- --check
 
@@ -45,30 +34,27 @@ jobs:
         run: cargo clippy --all-features -- --deny warnings
 
   test:
-    runs-on: ubuntu-latest-personal
+    runs-on: debian-latest
     steps:
-      - uses: actions/checkout@v3
-      - uses: cargo-bins/cargo-binstall@main
+      - uses: actions/checkout@v6
+
+      - name: Install cargo binstall
+        uses: cargo-bins/cargo-binstall@main
 
       - name: Install mpv
         run: apt-get update && apt-get install -y mpv
 
-      - name: Install latest nightly toolchain
-        uses: actions-rs/toolchain@v1
+      - name: Install rust toolchain
+        uses: dtolnay/rust-toolchain@nightly
         with:
-            toolchain: nightly
-            override: true
             components: llvm-tools-preview
 
-      - name: Cache dependencies
-        uses: Swatinem/rust-cache@v2
-
       - name: Install nextest
         run: cargo binstall -y cargo-nextest --secure
 
       - name: Run tests
         run: |
-          cargo nextest run --all-features --release --no-fail-fast
+          cargo nextest run --all-features --release --no-fail-fast --test-threads=1
         env:
           RUST_LOG: "trace"
           RUSTFLAGS: "-Cinstrument-coverage"
@@ -96,38 +82,32 @@ jobs:
             target/coverage/
 
       - name: Upload test report
-        uses: https://git.pvv.ntnu.no/oysteikt/rsync-action@main
+        uses: https://git.pvv.ntnu.no/Projects/rsync-action@v1
         with:
           source: target/coverage/html/
-          target: mpvipc-async/${{ gitea.ref_name }}/coverage/
-          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="
+          target: ${{ gitea.ref_name }}/coverage/
+          username: gitea-web
+          ssh-key: ${{ secrets.WEB_SYNC_SSH_KEY }}
+          host: pages.pvv.ntnu.no
+          known-hosts: "pages.pvv.ntnu.no ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIH2QjfFB+city1SYqltkVqWACfo1j37k+oQQfj13mtgg"
 
   docs:
-    runs-on: ubuntu-latest-personal
+    runs-on: debian-latest
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v6
 
-      - name: Install latest nightly toolchain
-        uses: actions-rs/toolchain@v1
-        with:
-            toolchain: nightly
-            override: true
-
-      - name: Cache dependencies
-        uses: Swatinem/rust-cache@v2
+      - name: Install rust toolchain
+        uses: dtolnay/rust-toolchain@stable
 
       - name: Build docs
         run: cargo doc --all-features --document-private-items --release
 
       - name: Transfer files
-        uses: https://git.pvv.ntnu.no/oysteikt/rsync-action@main
+        uses: https://git.pvv.ntnu.no/Projects/rsync-action@v1
         with:
           source: target/doc/
-          target: mpvipc-async/${{ gitea.ref_name }}/docs/
-          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="
+          target: ${{ gitea.ref_name }}/docs/
+          username: gitea-web
+          ssh-key: ${{ secrets.WEB_SYNC_SSH_KEY }}
+          host: pages.pvv.ntnu.no
+          known-hosts: "pages.pvv.ntnu.no ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIH2QjfFB+city1SYqltkVqWACfo1j37k+oQQfj13mtgg"

Cargo.toml

@@ -1,33 +1,32 @@
 [package]
 name = "mpvipc-async"
-version = "0.1.0"
+version = "0.2.0"
 authors = [
   "Jonas Frei <freijon@pm.me>",
   "Øystein Tveit <oysteikt@pvv.ntnu.no>"
 ]
 description = "A small library which provides bindings to control existing mpv instances through sockets."
 license = "GPL-3.0"
-homepage = "https://git.pvv.ntnu.no/oysteikt/mpvipc-async"
-repository = "https://git.pvv.ntnu.no/oysteikt/mpvipc-async"
-documentation = "https://pvv.ntnu.no/~oysteikt/gitea/mpvipc-async/master/docs/mpvipc-async/"
-edition = "2021"
-rust-version = "1.75"
+repository = "https://git.pvv.ntnu.no/Grzegorz/mpvipc-async"
+documentation = "https://pages.pvv.ntnu.no/Grzegorz/mpvipc-async/main/docs/mpvipc_async/"
+edition = "2024"
+rust-version = "1.85.0"
 
 [dependencies]
-serde_json = "1.0.104"
-log = "0.4.19"
-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"] }
-thiserror = "1.0.59"
+serde_json = "1.0.148"
+log = "0.4.29"
+serde = { version = "1.0.228", features = ["derive"] }
+tokio = { version = "1.48.0", features = ["sync", "macros", "rt", "net"] }
+tokio-util = { version = "0.7.17", features = ["codec"] }
+futures = "0.3.31"
+tokio-stream = { version = "0.1.17", features = ["sync"] }
+thiserror = "2.0.17"
 
 [dev-dependencies]
-env_logger = "0.10.0"
-test-log = "0.2.15"
-tokio = { version = "1.37.0", features = ["rt-multi-thread", "time", "process"] }
-uuid = { version = "1.8.0", features = ["v4"] }
+env_logger = "0.11.8"
+test-log = "0.2.19"
+tokio = { version = "1.48.0", features = ["rt-multi-thread", "time", "process"] }
+uuid = { version = "1.19.0", features = ["v4"] }
 
 [lib]
 doctest = false

LICENSE

@@ -1,7 +1,7 @@
                     GNU GENERAL PUBLIC LICENSE
                        Version 3, 29 June 2007
 
- Copyright (C) 2007 Free Software Foundation, Inc. <http://fsf.org/>
+ Copyright (C) 2007 Free Software Foundation, Inc. <https://fsf.org/>
  Everyone is permitted to copy and distribute verbatim copies
  of this license document, but changing it is not allowed.
 
@@ -645,7 +645,7 @@ the "copyright" line and a pointer to where the full notice is found.
     GNU General Public License for more details.
 
     You should have received a copy of the GNU General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+    along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
 Also add information on how to contact you by electronic and paper mail.
 
@@ -664,12 +664,11 @@ might be different; for a GUI interface, you would use an "about box".
   You should also get your employer (if you work as a programmer) or school,
 if any, to sign a "copyright disclaimer" for the program, if necessary.
 For more information on this, and how to apply and follow the GNU GPL, see
-<http://www.gnu.org/licenses/>.
+<https://www.gnu.org/licenses/>.
 
   The GNU General Public License does not permit incorporating your program
 into proprietary programs.  If your program is a subroutine library, you
 may consider it more useful to permit linking proprietary applications with
 the library.  If this is what you want to do, use the GNU Lesser General
 Public License instead of this License.  But first, please read
-<http://www.gnu.org/philosophy/why-not-lgpl.html>.
-
+<https://www.gnu.org/licenses/why-not-lgpl.html>.

README.md

@@ -1,11 +1,11 @@
-[![Coverage](https://pvv.ntnu.no/~oysteikt/gitea/mpvipc-async/main/coverage/badges/for_the_badge.svg)](https://pvv.ntnu.no/~oysteikt/gitea/mpvipc-async/main/coverage/src/)
-[![Docs](https://img.shields.io/badge/docs-blue?style=for-the-badge&logo=rust)](https://pvv.ntnu.no/~oysteikt/gitea/mpvipc-async/main/docs/mpvipc_async/)
+[![Coverage](https://pages.pvv.ntnu.no/Grzegorz/mpvipc-async/main/coverage/badges/for_the_badge.svg)](https://pages.pvv.ntnu.no/Grzegorz/mpvipc-async/main/coverage/src/)
+[![Docs](https://img.shields.io/badge/docs-blue?style=for-the-badge&logo=rust)](https://pages.pvv.ntnu.no/Grzegorz/mpvipc-async/main/docs/mpvipc_async/)
 
 # mpvipc-async
 
-> **NOTE:** This is a fork of [gitlab.com/mpv-ipc/mpvipc](https://gitlab.com/mpv-ipc/mpvipc), which introduces a lot of changes to be able to use the library asynchronously with [tokio](https://github.com/tokio-rs/tokio).
-
----
+> [!NOTE]
+> This is a fork of [gitlab.com/mpv-ipc/mpvipc](https://gitlab.com/mpv-ipc/mpvipc).
+> The fork adds support for use in asynchronous contexts.
 
 A small library which provides bindings to control existing mpv instances through sockets.
 
@@ -34,4 +34,6 @@ async fn main() -> Result<(), MpvError> {
     let paused: bool = mpv.get_property("pause").await?;
     mpv.set_property("pause", !paused).await.expect("Error pausing");
 }
-```
+```
+
+[You can find more examples in the `examples` directory](./examples)

examples/mpv_event_stream.rs

@@ -1,5 +1,5 @@
 use futures::StreamExt;
-use mpvipc_async::{parse_property, Event, Mpv, MpvDataType, MpvError, MpvExt, Property};
+use mpvipc_async::{Event, Mpv, MpvDataType, MpvError, MpvExt, Property, parse_property};
 
 fn seconds_to_hms(total: f64) -> String {
     let total = total as u64;

flake.lock

@@ -1,33 +1,12 @@
 {
   "nodes": {
-    "fenix": {
-      "inputs": {
-        "nixpkgs": [
-          "nixpkgs"
-        ],
-        "rust-analyzer-src": "rust-analyzer-src"
-      },
-      "locked": {
-        "lastModified": 1713421495,
-        "narHash": "sha256-5vVF9W1tJT+WdfpWAEG76KywktKDAW/71mVmNHEHjac=",
-        "owner": "nix-community",
-        "repo": "fenix",
-        "rev": "fd47b1f9404fae02a4f38bd9f4b12bad7833c96b",
-        "type": "github"
-      },
-      "original": {
-        "owner": "nix-community",
-        "repo": "fenix",
-        "type": "github"
-      }
-    },
     "nixpkgs": {
       "locked": {
-        "lastModified": 1713248628,
-        "narHash": "sha256-NLznXB5AOnniUtZsyy/aPWOk8ussTuePp2acb9U+ISA=",
+        "lastModified": 1767116409,
+        "narHash": "sha256-5vKw92l1GyTnjoLzEagJy5V5mDFck72LiQWZSOnSicw=",
         "owner": "NixOS",
         "repo": "nixpkgs",
-        "rev": "5672bc9dbf9d88246ddab5ac454e82318d094bb8",
+        "rev": "cad22e7d996aea55ecab064e84834289143e44a0",
         "type": "github"
       },
       "original": {
@@ -39,24 +18,27 @@
     },
     "root": {
       "inputs": {
-        "fenix": "fenix",
-        "nixpkgs": "nixpkgs"
+        "nixpkgs": "nixpkgs",
+        "rust-overlay": "rust-overlay"
       }
     },
-    "rust-analyzer-src": {
-      "flake": false,
+    "rust-overlay": {
+      "inputs": {
+        "nixpkgs": [
+          "nixpkgs"
+        ]
+      },
       "locked": {
-        "lastModified": 1713373173,
-        "narHash": "sha256-octd9BFY9G/Gbr4KfwK4itZp4Lx+qvJeRRcYnN+dEH8=",
-        "owner": "rust-lang",
-        "repo": "rust-analyzer",
-        "rev": "46702ffc1a02a2ac153f1d1ce619ec917af8f3a6",
+        "lastModified": 1767322002,
+        "narHash": "sha256-yHKXXw2OWfIFsyTjduB4EyFwR0SYYF0hK8xI9z4NIn0=",
+        "owner": "oxalica",
+        "repo": "rust-overlay",
+        "rev": "03c6e38661c02a27ca006a284813afdc461e9f7e",
         "type": "github"
       },
       "original": {
-        "owner": "rust-lang",
-        "ref": "nightly",
-        "repo": "rust-analyzer",
+        "owner": "oxalica",
+        "repo": "rust-overlay",
         "type": "github"
       }
     }

flake.nix

@@ -1,11 +1,12 @@
 {
   inputs = {
     nixpkgs.url = "github:NixOS/nixpkgs/nixos-unstable";
-    fenix.url = "github:nix-community/fenix";
-    fenix.inputs.nixpkgs.follows = "nixpkgs";
+
+    rust-overlay.url = "github:oxalica/rust-overlay";
+    rust-overlay.inputs.nixpkgs.follows = "nixpkgs";
   };
 
-  outputs = { self, nixpkgs, fenix }@inputs:
+  outputs = { self, nixpkgs, rust-overlay }@inputs:
   let
     systems = [
       "x86_64-linux"
@@ -14,25 +15,29 @@
       "aarch64-darwin"
     ];
     forAllSystems = f: nixpkgs.lib.genAttrs systems (system: let
-      toolchain = fenix.packages.${system}.complete;
       pkgs = import nixpkgs {
         inherit system;
         overlays = [
-          (_: super: let pkgs = fenix.inputs.nixpkgs.legacyPackages.${system}; in fenix.overlays.default pkgs pkgs)
+          (import rust-overlay)
         ];
       };
+
+      rust-bin = rust-overlay.lib.mkRustBin { } pkgs.buildPackages;
+      toolchain = rust-bin.stable.latest.default.override {
+        extensions = [ "rust-src" ];
+      };
     in f system pkgs toolchain);
   in {
     devShell = forAllSystems (system: pkgs: toolchain: pkgs.mkShell {
-      packages = [
-        (toolchain.withComponents [
-          "cargo" "rustc" "rustfmt" "clippy" "llvm-tools"
-        ])
-        pkgs.mpv
-        pkgs.grcov
-        pkgs.cargo-nextest
+      packages = with pkgs; [
+        toolchain
+        mpv
+        grcov
+        cargo-nextest
+        cargo-edit
       ];
-      RUST_SRC_PATH = "${toolchain.rust-src}/lib/rustlib/src/rust/library";
+
+      env.RUST_SRC_PATH = "${toolchain}/lib/rustlib/src/rust/library";
     });
   };
 }

rustfmt.toml

@@ -0,0 +1 @@
+style_edition = "2024"

src/core_api.rs

@@ -10,9 +10,9 @@ use tokio::{
 };
 
 use crate::{
+    Event, MpvError,
     ipc::{MpvIpc, MpvIpcCommand, MpvIpcEvent, MpvIpcResponse},
     message_parser::TypeHandler,
-    Event, MpvError,
 };
 
 /// All possible commands that can be sent to mpv.
@@ -27,39 +27,65 @@ use crate::{
 /// the upstream list of commands.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub enum MpvCommand {
+    /// Load the given file or URL and play it.
     LoadFile {
         file: String,
         option: PlaylistAddOptions,
     },
+
+    /// Load the given playlist file or URL.
     LoadList {
         file: String,
         option: PlaylistAddOptions,
     },
+
+    /// Clear the playlist, except for the currently playing file.
     PlaylistClear,
-    PlaylistMove {
-        from: usize,
-        to: usize,
-    },
-    Observe {
-        id: usize,
-        property: String,
-    },
+
+    ///Move the playlist entry at `from`, so that it takes the place of the entry `to`.
+    /// (Paradoxically, the moved playlist entry will not have the index value `to` after moving
+    /// if `from` was lower than `to`, because `to` refers to the target entry, not the index
+    /// the entry will have after moving.)
+    PlaylistMove { from: usize, to: usize },
+
+    /// Observe a property. This will start triggering [`Event::PropertyChange`] events
+    /// in the event stream whenever the specific property changes.
+    /// You can use [`Mpv::get_event_stream`] to get the stream.
+    Observe { id: u64, property: String },
+
+    /// Skip to the next entry in the playlist.
     PlaylistNext,
+
+    /// Skip to the previous entry in the playlist.
     PlaylistPrev,
+
+    /// Remove an entry from the playlist by its position in the playlist.
     PlaylistRemove(usize),
+
+    /// Shuffle the playlist
     PlaylistShuffle,
+
+    /// Exit the player
     Quit,
+
+    /// Send a message to all clients, and pass it the following list of arguments.
+    /// What this message means, how many arguments it takes, and what the arguments
+    /// mean is fully up to the receiver and the sender.
     ScriptMessage(Vec<String>),
-    ScriptMessageTo {
-        target: String,
-        args: Vec<String>,
-    },
-    Seek {
-        seconds: f64,
-        option: SeekOptions,
-    },
+
+    /// Same as [`MpvCommand::ScriptMessage`], but send the message to a specific target.
+    ScriptMessageTo { target: String, args: Vec<String> },
+
+    /// Change the playback position.
+    Seek { seconds: f64, option: SeekOptions },
+
+    /// Stop the current playback, and clear the playlist.
+    /// This esentially resets the entire player state without exiting mpv.
     Stop,
-    Unobserve(usize),
+
+    /// Unobserve all properties registered with the given id.
+    /// See [`MpvCommand::Observe`] for more context.
+    Unobserve(u64),
 }
 
 /// Helper trait to keep track of the string literals that mpv expects.
@@ -69,6 +95,7 @@ pub(crate) trait IntoRawCommandPart {
 
 /// Generic data type representing all possible data types that mpv can return.
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(untagged)]
 pub enum MpvDataType {
     Array(Vec<MpvDataType>),
     Bool(bool),
@@ -82,7 +109,7 @@ pub enum MpvDataType {
 }
 
 /// A mpv playlist.
-#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
 pub struct Playlist(pub Vec<PlaylistEntry>);
 
 /// A single entry in the mpv playlist.
@@ -135,7 +162,7 @@ pub trait GetPropertyTypeHandler: Sized {
     // TODO: fix this
     #[allow(async_fn_in_trait)]
     async fn get_property_generic(instance: &Mpv, property: &str)
-        -> Result<Option<Self>, MpvError>;
+    -> Result<Option<Self>, MpvError>;
 }
 
 impl<T> GetPropertyTypeHandler for T
@@ -158,7 +185,7 @@ pub trait SetPropertyTypeHandler<T> {
     // TODO: fix this
     #[allow(async_fn_in_trait)]
     async fn set_property_generic(instance: &Mpv, property: &str, value: T)
-        -> Result<(), MpvError>;
+    -> Result<(), MpvError>;
 }
 
 impl<T> SetPropertyTypeHandler<T> for T

src/error.rs

@@ -8,8 +8,11 @@ use crate::{MpvDataType, Property};
 /// Any error that can occur when interacting with mpv.
 #[derive(Error, Debug)]
 pub enum MpvError {
-    #[error("MpvError: {0}")]
-    MpvError(String),
+    #[error("Mpv returned error in response to command: {message}\nCommand: {command:#?}")]
+    MpvError {
+        command: Vec<Value>,
+        message: String,
+    },
 
     #[error("Error communicating over mpv socket: {0}")]
     MpvSocketConnectionError(String),
@@ -20,7 +23,9 @@ pub enum MpvError {
     #[error("JsonParseError: {0}")]
     JsonParseError(#[from] serde_json::Error),
 
-    #[error("Mpv sent a value with an unexpected type:\nExpected {expected_type}, received {received:#?}")]
+    #[error(
+        "Mpv sent a value with an unexpected type:\nExpected {expected_type}, received {received:#?}"
+    )]
     ValueContainsUnexpectedType {
         expected_type: String,
         received: Value,
@@ -53,7 +58,16 @@ pub enum MpvError {
 impl PartialEq for MpvError {
     fn eq(&self, other: &Self) -> bool {
         match (self, other) {
-            (Self::MpvError(l0), Self::MpvError(r0)) => l0 == r0,
+            (
+                Self::MpvError {
+                    command: l_command,
+                    message: l_message,
+                },
+                Self::MpvError {
+                    command: r_command,
+                    message: r_message,
+                },
+            ) => l_command == r_command && l_message == r_message,
             (Self::MpvSocketConnectionError(l0), Self::MpvSocketConnectionError(r0)) => l0 == r0,
             (Self::InternalConnectionError(l0), Self::InternalConnectionError(r0)) => l0 == r0,
             (Self::JsonParseError(l0), Self::JsonParseError(r0)) => {

src/event_parser.rs

@@ -5,17 +5,39 @@ use std::str::FromStr;
 use serde::{Deserialize, Serialize};
 use serde_json::{Map, Value};
 
-use crate::{ipc::MpvIpcEvent, message_parser::json_to_value, MpvDataType, MpvError};
+use crate::{MpvDataType, MpvError, ipc::MpvIpcEvent, message_parser::json_to_value};
 
+/// Reason behind the `MPV_EVENT_END_FILE` event.
+///
+/// Ref: <https://mpv.io/manual/stable/#command-interface-mpv-event-end-file>
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 #[serde(rename_all = "kebab-case")]
 pub enum EventEndFileReason {
+    /// The file has ended. This can (but doesn't have to) include
+    /// incomplete files or broken network connections under circumstances.
     Eof,
+
+    /// Playback was ended by a command.
     Stop,
+
+    /// Playback was ended by sending the quit command.
     Quit,
+
+    /// An error happened. In this case, an `error` field is present with the error string.
     Error,
+
+    /// Happens with playlists and similar. For details, see
+    /// [`MPV_END_FILE_REASON_REDIRECT`](https://github.com/mpv-player/mpv/blob/72efbfd009a2b3259055133d74b88c81b1115ae1/include/mpv/client.h#L1493)
+    /// in the C API.
     Redirect,
+
+    /// Unknown. Normally doesn't happen, unless the Lua API is out of sync
+    /// with the C API. (Likewise, it could happen that your script gets reason
+    /// strings that did not exist yet at the time your script was written.)
     Unknown,
+
+    /// A catch-all enum variant in case `mpvipc-async` has not implemented the
+    /// returned error yet.
     Unimplemented(String),
 }
 
@@ -34,6 +56,11 @@ impl FromStr for EventEndFileReason {
     }
 }
 
+/// The log level of a log message event.
+///
+/// Ref:
+/// - <https://mpv.io/manual/stable/#command-interface-mpv-event-log-message>
+/// - <https://mpv.io/manual/stable/#mp-msg-functions>
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 #[serde(rename_all = "kebab-case")]
 pub enum EventLogMessageLevel {
@@ -44,6 +71,9 @@ pub enum EventLogMessageLevel {
     Verbose,
     Debug,
     Trace,
+
+    /// A catch-all enum variant in case `mpvipc-async` has not implemented the
+    /// returned log-level yet.
     Unimplemented(String),
 }
 
@@ -109,7 +139,7 @@ pub enum Event {
     VideoReconfig,
     AudioReconfig,
     PropertyChange {
-        id: usize,
+        id: Option<u64>,
         name: String,
         data: Option<MpvDataType>,
     },
@@ -164,16 +194,16 @@ macro_rules! get_key_as {
 
 macro_rules! get_optional_key_as {
     ($as_type:ident, $key:expr, $event:ident) => {{
-        if let Some(tmp) = $event.get($key) {
-            Some(
+        match $event.get($key) {
+            Some(Value::Null) => None,
+            Some(tmp) => Some(
                 tmp.$as_type()
                     .ok_or(MpvError::ValueContainsUnexpectedType {
                         expected_type: stringify!($as_type).strip_prefix("as_").unwrap().to_owned(),
                         received: tmp.clone(),
                     })?,
-            )
-        } else {
-            None
+            ),
+            None => None,
         }
     }};
 }
@@ -296,7 +326,7 @@ 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 id = get_optional_key_as!(as_u64, "id", event);
     let property_name = get_key_as!(as_str, "name", event);
     let data = event.get("data").map(json_to_value).transpose()?;
 
@@ -306,3 +336,195 @@ fn parse_property_change(event: &Map<String, Value>) -> Result<Event, MpvError>
         data,
     })
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::ipc::MpvIpcEvent;
+    use serde_json::json;
+
+    #[test]
+    fn test_parse_simple_events() {
+        let simple_events = vec![
+            (json!({"event": "file-loaded"}), Event::FileLoaded),
+            (json!({"event": "seek"}), Event::Seek),
+            (json!({"event": "playback-restart"}), Event::PlaybackRestart),
+            (json!({"event": "shutdown"}), Event::Shutdown),
+            (json!({"event": "video-reconfig"}), Event::VideoReconfig),
+            (json!({"event": "audio-reconfig"}), Event::AudioReconfig),
+            (json!({"event": "tick"}), Event::Tick),
+            (json!({"event": "idle"}), Event::Idle),
+            (json!({"event": "tracks-changed"}), Event::TracksChanged),
+            (json!({"event": "track-switched"}), Event::TrackSwitched),
+            (json!({"event": "pause"}), Event::Pause),
+            (json!({"event": "unpause"}), Event::Unpause),
+            (json!({"event": "metadata-update"}), Event::MetadataUpdate),
+            (json!({"event": "chapter-change"}), Event::ChapterChange),
+        ];
+
+        for (raw_event_json, expected_event) in simple_events {
+            let raw_event = MpvIpcEvent(raw_event_json);
+            let event = parse_event(raw_event).unwrap();
+            assert_eq!(event, expected_event);
+        }
+    }
+
+    #[test]
+    fn test_parse_start_file_event() {
+        let raw_event = MpvIpcEvent(json!({
+            "event": "start-file",
+            "playlist_entry_id": 1
+        }));
+
+        let event = parse_event(raw_event).unwrap();
+
+        assert_eq!(
+            event,
+            Event::StartFile {
+                playlist_entry_id: 1
+            }
+        );
+    }
+
+    #[test]
+    fn test_parse_end_file_event() {
+        let raw_event = MpvIpcEvent(json!({
+            "event": "end-file",
+            "reason": "eof",
+            "playlist_entry_id": 2,
+            "file_error": null,
+            "playlist_insert_id": 3,
+            "playlist_insert_num_entries": 5
+        }));
+        let event = parse_event(raw_event).unwrap();
+        assert_eq!(
+            event,
+            Event::EndFile {
+                reason: EventEndFileReason::Eof,
+                playlist_entry_id: 2,
+                file_error: None,
+                playlist_insert_id: Some(3),
+                playlist_insert_num_entries: Some(5)
+            }
+        );
+
+        let raw_event_with_error = MpvIpcEvent(json!({
+            "event": "end-file",
+            "reason": "error",
+            "playlist_entry_id": 4,
+            "file_error": "File not found",
+        }));
+        let event_with_error = parse_event(raw_event_with_error).unwrap();
+        assert_eq!(
+            event_with_error,
+            Event::EndFile {
+                reason: EventEndFileReason::Error,
+                playlist_entry_id: 4,
+                file_error: Some("File not found".to_string()),
+                playlist_insert_id: None,
+                playlist_insert_num_entries: None,
+            }
+        );
+
+        let raw_event_unimplemented = MpvIpcEvent(json!({
+            "event": "end-file",
+            "reason": "unknown-reason",
+            "playlist_entry_id": 5
+        }));
+        let event_unimplemented = parse_event(raw_event_unimplemented).unwrap();
+        assert_eq!(
+            event_unimplemented,
+            Event::EndFile {
+                reason: EventEndFileReason::Unimplemented("unknown-reason".to_string()),
+                playlist_entry_id: 5,
+                file_error: None,
+                playlist_insert_id: None,
+                playlist_insert_num_entries: None,
+            }
+        );
+    }
+
+    #[test]
+    fn test_parse_log_message_event() {
+        let raw_event = MpvIpcEvent(json!({
+            "event": "log-message",
+            "prefix": "mpv",
+            "level": "info",
+            "text": "This is a log message"
+        }));
+        let event = parse_event(raw_event).unwrap();
+        assert_eq!(
+            event,
+            Event::LogMessage {
+                prefix: "mpv".to_string(),
+                level: EventLogMessageLevel::Info,
+                text: "This is a log message".to_string(),
+            }
+        );
+    }
+
+    #[test]
+    fn test_parse_hook_event() {
+        let raw_event = MpvIpcEvent(json!({
+            "event": "hook",
+            "hook_id": 42
+        }));
+        let event = parse_event(raw_event).unwrap();
+        assert_eq!(event, Event::Hook { hook_id: 42 });
+    }
+
+    #[test]
+    fn test_parse_client_message_event() {
+        let raw_event = MpvIpcEvent(json!({
+            "event": "client-message",
+            "args": ["arg1", "arg2", "arg3"]
+        }));
+        let event = parse_event(raw_event).unwrap();
+        assert_eq!(
+            event,
+            Event::ClientMessage {
+                args: vec!["arg1".to_string(), "arg2".to_string(), "arg3".to_string()]
+            }
+        );
+    }
+
+    #[test]
+    fn test_parse_property_change_event() {
+        let raw_event = MpvIpcEvent(json!({
+            "event": "property-change",
+            "id": 1,
+            "name": "pause",
+            "data": true
+        }));
+        let event = parse_event(raw_event).unwrap();
+        assert_eq!(
+            event,
+            Event::PropertyChange {
+                id: Some(1),
+                name: "pause".to_string(),
+                data: Some(MpvDataType::Bool(true)),
+            }
+        );
+    }
+
+    #[test]
+    fn test_parse_unimplemented_event() {
+        let raw_event = MpvIpcEvent(json!({
+            "event": "some-unimplemented-event",
+            "some_key": "some_value"
+        }));
+        let event = parse_event(raw_event).unwrap();
+        assert_eq!(
+            event,
+            Event::Unimplemented(
+                json!({
+                    "event": "some-unimplemented-event",
+                    "some_key": "some_value"
+                })
+                .as_object()
+                .unwrap()
+                .to_owned()
+            )
+        );
+    }
+}

src/highlevel_api_extension.rs

@@ -1,8 +1,8 @@
 //! High-level API extension for [`Mpv`].
 
 use crate::{
-    parse_property, IntoRawCommandPart, LoopProperty, Mpv, MpvCommand, MpvDataType, MpvError,
-    Playlist, PlaylistAddOptions, Property, SeekOptions,
+    IntoRawCommandPart, LoopProperty, Mpv, MpvCommand, MpvDataType, MpvError, Playlist,
+    PlaylistAddOptions, Property, SeekOptions, parse_property,
 };
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
@@ -88,11 +88,11 @@ pub trait MpvExt {
 
     /// 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>;
+    async fn observe_property(&self, id: u64, 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>;
+    async fn unobserve_property(&self, id: u64) -> Result<(), MpvError>;
 
     /// Skip to the next entry in the playlist.
     async fn next(&self) -> Result<(), MpvError>;
@@ -259,7 +259,7 @@ impl MpvExt for Mpv {
         self.run_command(MpvCommand::PlaylistPrev).await
     }
 
-    async fn observe_property(&self, id: usize, property: &str) -> Result<(), MpvError> {
+    async fn observe_property(&self, id: u64, property: &str) -> Result<(), MpvError> {
         self.run_command(MpvCommand::Observe {
             id,
             property: property.to_string(),
@@ -267,7 +267,7 @@ impl MpvExt for Mpv {
         .await
     }
 
-    async fn unobserve_property(&self, id: usize) -> Result<(), MpvError> {
+    async fn unobserve_property(&self, id: u64) -> Result<(), MpvError> {
         self.run_command(MpvCommand::Unobserve(id)).await
     }
 
@@ -323,9 +323,9 @@ impl MpvExt for Mpv {
             Switch::Off => "yes",
             Switch::Toggle => {
                 if self.is_playing().await? {
-                    "no"
-                } else {
                     "yes"
+                } else {
+                    "no"
                 }
             }
         };

src/ipc.rs

@@ -1,7 +1,7 @@
 //! IPC handling thread/task. Handles communication between [`Mpv`](crate::Mpv) instances and mpv's unix socket
 
 use futures::{SinkExt, StreamExt};
-use serde_json::{json, Value};
+use serde_json::{Value, json};
 use tokio::{
     net::UnixStream,
     sync::{broadcast, mpsc, oneshot},
@@ -24,8 +24,8 @@ pub(crate) enum MpvIpcCommand {
     Command(Vec<String>),
     GetProperty(String),
     SetProperty(String, Value),
-    ObserveProperty(usize, String),
-    UnobserveProperty(usize),
+    ObserveProperty(u64, String),
+    UnobserveProperty(u64),
     Exit,
 }
 
@@ -92,7 +92,7 @@ impl MpvIpc {
 
         log::trace!("Received response: {:?}", response);
 
-        parse_mpv_response_data(response?)
+        parse_mpv_response_data(response?, command)
     }
 
     pub(crate) async fn get_mpv_property(
@@ -114,17 +114,14 @@ impl MpvIpc {
 
     pub(crate) async fn observe_property(
         &mut self,
-        id: usize,
+        id: u64,
         property: &str,
     ) -> Result<Option<Value>, MpvError> {
         self.send_command(&[json!("observe_property"), json!(id), json!(property)])
             .await
     }
 
-    pub(crate) async fn unobserve_property(
-        &mut self,
-        id: usize,
-    ) -> Result<Option<Value>, MpvError> {
+    pub(crate) async fn unobserve_property(&mut self, id: u64) -> Result<Option<Value>, MpvError> {
         self.send_command(&[json!("unobserve_property"), json!(id)])
             .await
     }
@@ -197,7 +194,7 @@ impl MpvIpc {
 /// This function does the most basic JSON parsing and error handling
 /// for status codes and errors that all responses from mpv are
 /// expected to contain.
-fn parse_mpv_response_data(value: Value) -> Result<Option<Value>, MpvError> {
+fn parse_mpv_response_data(value: Value, command: &[Value]) -> Result<Option<Value>, MpvError> {
     log::trace!("Parsing mpv response data: {:?}", value);
     let result = value
         .as_object()
@@ -225,7 +222,10 @@ fn parse_mpv_response_data(value: Value) -> Result<Option<Value>, MpvError> {
         .and_then(|(error, data)| match error {
             "success" => Ok(data),
             "property unavailable" => Ok(None),
-            err => Err(MpvError::MpvError(err.to_string())),
+            err => Err(MpvError::MpvError {
+                command: command.to_owned(),
+                message: err.to_string(),
+            }),
         });
 
     match &result {

src/message_parser.rs

@@ -164,7 +164,7 @@ fn json_map_to_playlist_entry(
             return Err(MpvError::ValueContainsUnexpectedType {
                 expected_type: "String".to_owned(),
                 received: data.clone(),
-            })
+            });
         }
         None => return Err(MpvError::MissingMpvData),
     };
@@ -174,7 +174,7 @@ fn json_map_to_playlist_entry(
             return Err(MpvError::ValueContainsUnexpectedType {
                 expected_type: "String".to_owned(),
                 received: data.clone(),
-            })
+            });
         }
         None => None,
     };
@@ -184,7 +184,7 @@ fn json_map_to_playlist_entry(
             return Err(MpvError::ValueContainsUnexpectedType {
                 expected_type: "bool".to_owned(),
                 received: data.clone(),
-            })
+            });
         }
         None => false,
     };

src/property_parser.rs

@@ -1,5 +1,5 @@
 //! JSON parsing logic for properties returned by
-//! [[`Event::PropertyChange`], and used internally in `MpvExt`
+//! [`Event::PropertyChange`], and used internally in `MpvExt`
 //! to parse the response from `Mpv::get_property()`.
 //!
 //! This module is used to parse the json data from the `data` field of
@@ -46,11 +46,15 @@ pub enum Property {
     },
 }
 
+/// Loop mode used by mpv for files and playlists.
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 #[serde(rename_all = "kebab-case")]
 pub enum LoopProperty {
+    /// Loop N times
     N(usize),
+    /// Loop infinitely
     Inf,
+    /// Disable looping
     No,
 }
 
@@ -68,7 +72,7 @@ pub fn parse_property(name: &str, data: Option<MpvDataType>) -> Result<Property,
                     return Err(MpvError::DataContainsUnexpectedType {
                         expected_type: "String".to_owned(),
                         received: data,
-                    })
+                    });
                 }
                 None => {
                     return Err(MpvError::MissingMpvData);
@@ -83,7 +87,7 @@ pub fn parse_property(name: &str, data: Option<MpvDataType>) -> Result<Property,
                     return Err(MpvError::DataContainsUnexpectedType {
                         expected_type: "bool".to_owned(),
                         received: data,
-                    })
+                    });
                 }
                 None => {
                     return Err(MpvError::MissingMpvData);
@@ -99,7 +103,7 @@ pub fn parse_property(name: &str, data: Option<MpvDataType>) -> Result<Property,
                     return Err(MpvError::DataContainsUnexpectedType {
                         expected_type: "f64".to_owned(),
                         received: data,
-                    })
+                    });
                 }
             };
             Ok(Property::PlaybackTime(playback_time))
@@ -112,7 +116,7 @@ pub fn parse_property(name: &str, data: Option<MpvDataType>) -> Result<Property,
                     return Err(MpvError::DataContainsUnexpectedType {
                         expected_type: "f64".to_owned(),
                         received: data,
-                    })
+                    });
                 }
             };
             Ok(Property::Duration(duration))
@@ -125,7 +129,7 @@ pub fn parse_property(name: &str, data: Option<MpvDataType>) -> Result<Property,
                     return Err(MpvError::DataContainsUnexpectedType {
                         expected_type: "HashMap".to_owned(),
                         received: data,
-                    })
+                    });
                 }
             };
             Ok(Property::Metadata(metadata))
@@ -138,7 +142,7 @@ pub fn parse_property(name: &str, data: Option<MpvDataType>) -> Result<Property,
                     return Err(MpvError::DataContainsUnexpectedType {
                         expected_type: "Array".to_owned(),
                         received: data,
-                    })
+                    });
                 }
             };
             Ok(Property::Playlist(playlist))
@@ -153,7 +157,7 @@ pub fn parse_property(name: &str, data: Option<MpvDataType>) -> Result<Property,
                     return Err(MpvError::DataContainsUnexpectedType {
                         expected_type: "usize or -1".to_owned(),
                         received: data,
-                    })
+                    });
                 }
             };
             Ok(Property::PlaylistPos(playlist_pos))
@@ -210,7 +214,7 @@ pub fn parse_property(name: &str, data: Option<MpvDataType>) -> Result<Property,
                     return Err(MpvError::DataContainsUnexpectedType {
                         expected_type: "f64".to_owned(),
                         received: data,
-                    })
+                    });
                 }
                 None => None,
             };
@@ -224,7 +228,7 @@ pub fn parse_property(name: &str, data: Option<MpvDataType>) -> Result<Property,
                     return Err(MpvError::DataContainsUnexpectedType {
                         expected_type: "f64".to_owned(),
                         received: data,
-                    })
+                    });
                 }
                 None => None,
             };
@@ -237,7 +241,7 @@ pub fn parse_property(name: &str, data: Option<MpvDataType>) -> Result<Property,
                     return Err(MpvError::DataContainsUnexpectedType {
                         expected_type: "f64".to_owned(),
                         received: data,
-                    })
+                    });
                 }
                 None => {
                     return Err(MpvError::MissingMpvData);
@@ -252,7 +256,7 @@ pub fn parse_property(name: &str, data: Option<MpvDataType>) -> Result<Property,
                     return Err(MpvError::DataContainsUnexpectedType {
                         expected_type: "f64".to_owned(),
                         received: data,
-                    })
+                    });
                 }
                 None => {
                     return Err(MpvError::MissingMpvData);
@@ -267,7 +271,7 @@ pub fn parse_property(name: &str, data: Option<MpvDataType>) -> Result<Property,
                     return Err(MpvError::DataContainsUnexpectedType {
                         expected_type: "bool".to_owned(),
                         received: data,
-                    })
+                    });
                 }
                 None => {
                     return Err(MpvError::MissingMpvData);
@@ -282,7 +286,7 @@ pub fn parse_property(name: &str, data: Option<MpvDataType>) -> Result<Property,
                     return Err(MpvError::DataContainsUnexpectedType {
                         expected_type: "bool".to_owned(),
                         received: data,
-                    })
+                    });
                 }
                 None => true,
             };
@@ -305,7 +309,7 @@ fn mpv_data_to_playlist_entry(
             return Err(MpvError::DataContainsUnexpectedType {
                 expected_type: "String".to_owned(),
                 received: data.clone(),
-            })
+            });
         }
         None => return Err(MpvError::MissingMpvData),
     };
@@ -315,7 +319,7 @@ fn mpv_data_to_playlist_entry(
             return Err(MpvError::DataContainsUnexpectedType {
                 expected_type: "String".to_owned(),
                 received: data.clone(),
-            })
+            });
         }
         None => None,
     };
@@ -325,7 +329,7 @@ fn mpv_data_to_playlist_entry(
             return Err(MpvError::DataContainsUnexpectedType {
                 expected_type: "bool".to_owned(),
                 received: data.clone(),
-            })
+            });
         }
         None => false,
     };

tests/integration_tests/event_property_parser.rs

@@ -1,116 +1,11 @@
-use futures::{stream::StreamExt, Stream};
-use mpvipc_async::{parse_property, Event, Mpv, MpvError, MpvExt, Property};
-use thiserror::Error;
-use tokio::time::sleep;
-use tokio::time::{timeout, Duration};
-
 use test_log::test;
+use tokio::time::Duration;
+use tokio::time::sleep;
+
+use mpvipc_async::{MpvError, MpvExt, Property};
 
 use super::*;
 
-const MPV_CHANNEL_ID: usize = 1337;
-
-#[derive(Error, Debug)]
-enum PropertyCheckingThreadError {
-    #[error("Unexpected property: {0:?}")]
-    UnexpectedPropertyError(Property),
-
-    #[error(transparent)]
-    MpvError(#[from] MpvError),
-}
-
-/// This function will create an ongoing tokio task that collects [`Event::PropertyChange`] events,
-/// and parses them into [`Property`]s. It will then run the property through the provided
-/// closure, and return an error if the closure returns false.
-///
-/// The returned cancellation token can be used to stop the task.
-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(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, name, data } => {
-                                    let property = parse_property(&name, data).unwrap();
-                                    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)
-}
-
-/// This helper function will gracefully shut down both the event checking thread and the mpv process.
-/// It will also return an error if the event checking thread happened to panic, or if it times out
-/// The timeout is hardcoded to 500ms.
-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 correct parsing of different values of the "pause" property
 #[test(tokio::test)]
 #[cfg(target_family = "unix")]
@@ -119,15 +14,16 @@ async fn test_highlevel_event_pause() -> Result<(), MpvError> {
 
     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 {
+    let (handle, cancellation_token) = create_interruptable_event_property_checking_thread(
+        mpv.clone(),
+        |property| match property {
             Property::Pause(_) => {
                 log::debug!("{:?}", property);
                 true
             }
             _ => false,
-        });
+        },
+    );
 
     sleep(Duration::from_millis(5)).await;
     mpv.set_property("pause", false).await?;
@@ -146,16 +42,17 @@ async fn test_highlevel_event_pause() -> Result<(), MpvError> {
 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 {
+    mpv.observe_property(MPV_CHANNEL_ID, "volume").await?;
+    let (handle, cancellation_token) = create_interruptable_event_property_checking_thread(
+        mpv.clone(),
+        |property| match property {
             Property::Volume(_) => {
                 log::trace!("{:?}", property);
                 true
             }
             _ => false,
-        });
+        },
+    );
 
     sleep(Duration::from_millis(5)).await;
     mpv.set_property("volume", 100.0).await?;
@@ -176,16 +73,17 @@ async fn test_highlevel_event_volume() -> Result<(), MpvError> {
 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 {
+    mpv.observe_property(MPV_CHANNEL_ID, "mute").await?;
+    let (handle, cancellation_token) = create_interruptable_event_property_checking_thread(
+        mpv.clone(),
+        |property| match property {
             Property::Mute(_) => {
                 log::trace!("{:?}", property);
                 true
             }
             _ => false,
-        });
+        },
+    );
 
     sleep(Duration::from_millis(5)).await;
     mpv.set_property("mute", true).await?;
@@ -204,17 +102,18 @@ async fn test_highlevel_event_mute() -> Result<(), MpvError> {
 async fn test_highlevel_event_duration() -> Result<(), MpvError> {
     let (proc, mpv) = spawn_headless_mpv().await?;
 
-    mpv.observe_property(1337, "duration").await?;
+    mpv.observe_property(MPV_CHANNEL_ID, "duration").await?;
 
-    let events = mpv.get_event_stream().await;
-    let (handle, cancellation_token) =
-        create_interruptable_event_property_checking_thread(events, |property| match property {
+    let (handle, cancellation_token) = create_interruptable_event_property_checking_thread(
+        mpv.clone(),
+        |property| match property {
             Property::Duration(_) => {
                 log::trace!("{:?}", property);
                 true
             }
             _ => false,
-        });
+        },
+    );
 
     sleep(Duration::from_millis(5)).await;
     mpv.set_property("pause", true).await?;

tests/integration_tests/misc.rs

@@ -1,4 +1,9 @@
-use mpvipc_async::{MpvError, MpvExt};
+use std::time::Duration;
+
+use test_log::test;
+use tokio::time::sleep;
+
+use mpvipc_async::{MpvError, MpvExt, Property};
 
 use super::*;
 
@@ -47,13 +52,58 @@ async fn test_get_unavailable_property() -> Result<(), MpvError> {
 async fn test_get_nonexistent_property() -> Result<(), MpvError> {
     let (mut proc, mpv) = spawn_headless_mpv().await.unwrap();
     let nonexistent = mpv.get_property::<f64>("nonexistent").await;
-    assert_eq!(
-        nonexistent,
-        Err(MpvError::MpvError("property not found".to_string()))
-    );
+
+    match nonexistent {
+        Err(MpvError::MpvError { message, .. }) => {
+            assert_eq!(message, "property not found");
+        }
+        _ => panic!("Unexpected result: {:?}", nonexistent),
+    }
 
     mpv.kill().await.unwrap();
     proc.kill().await.unwrap();
 
     Ok(())
 }
+
+#[test(tokio::test)]
+#[cfg(target_family = "unix")]
+async fn test_unobserve_property() -> Result<(), MpvError> {
+    let (proc, mpv) = spawn_headless_mpv().await?;
+
+    mpv.observe_property(MPV_CHANNEL_ID, "pause").await?;
+
+    let (handle, cancellation_token) = create_interruptable_event_property_checking_thread(
+        mpv.clone(),
+        |property| match property {
+            Property::Pause(_) => {
+                log::debug!("{:?}", property);
+                true
+            }
+            _ => false,
+        },
+    );
+
+    sleep(Duration::from_millis(5)).await;
+    mpv.set_property("pause", true).await?;
+    sleep(Duration::from_millis(5)).await;
+
+    cancellation_token.cancel();
+    check_property_thread_result(handle).await?;
+
+    mpv.unobserve_property(MPV_CHANNEL_ID).await?;
+
+    let (handle, cancellation_token) =
+        create_interruptable_event_property_checking_thread(mpv.clone(), |_property| {
+            // We should not receive any properties after unobserving
+            false
+        });
+
+    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/util.rs

@@ -1,10 +1,13 @@
 use std::{path::Path, time::Duration};
 
-use mpvipc_async::{Mpv, MpvError};
+use thiserror::Error;
 use tokio::{
     process::{Child, Command},
     time::{sleep, timeout},
 };
+use tokio_stream::StreamExt;
+
+use mpvipc_async::{Event, Mpv, MpvError, MpvExt, Property, parse_property};
 
 #[cfg(target_family = "unix")]
 pub async fn spawn_headless_mpv() -> Result<(Child, Mpv), MpvError> {
@@ -25,7 +28,7 @@ pub async fn spawn_headless_mpv() -> Result<(Child, Mpv), MpvError> {
         .spawn()
         .expect("Failed to start mpv");
 
-    timeout(Duration::from_millis(500), async {
+    timeout(Duration::from_millis(1000), async {
         while !&socket_path.exists() {
             sleep(Duration::from_millis(10)).await;
         }
@@ -41,3 +44,107 @@ pub async fn spawn_headless_mpv() -> Result<(Child, Mpv), MpvError> {
     let mpv = Mpv::connect(socket_path.to_str().unwrap()).await?;
     Ok((process_handle, mpv))
 }
+
+/// The channel ID used for property observation in tests
+pub const MPV_CHANNEL_ID: u64 = 1337;
+
+#[derive(Error, Debug)]
+pub enum PropertyCheckingThreadError {
+    #[error("Unexpected property: {0:?}")]
+    UnexpectedPropertyError(Property),
+
+    #[error(transparent)]
+    MpvError(#[from] MpvError),
+}
+
+/// This function will create an ongoing tokio task that collects [`Event::PropertyChange`] events,
+/// and parses them into [`Property`]s. It will then run the property through the provided
+/// closure, and return an error if the closure returns false.
+///
+/// The returned cancellation token can be used to stop the task.
+pub fn create_interruptable_event_property_checking_thread<T>(
+    mpv: Mpv,
+    on_property: T,
+) -> (
+    tokio::task::JoinHandle<Result<(), PropertyCheckingThreadError>>,
+    tokio_util::sync::CancellationToken,
+)
+where
+    T: Fn(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 {
+        let mut events = mpv.get_event_stream().await;
+
+        loop {
+            tokio::select! {
+                event = events.next() => {
+                    match event {
+                        Some(Ok(event)) => {
+                            match event {
+                                Event::PropertyChange { id: Some(MPV_CHANNEL_ID), name, data } => {
+                                    let property = parse_property(&name, data).unwrap();
+                                    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)
+}
+
+pub async fn check_property_thread_result(
+    handle: tokio::task::JoinHandle<Result<(), PropertyCheckingThreadError>>,
+) -> Result<(), MpvError> {
+    timeout(Duration::from_millis(500), handle)
+        .await
+        .map_err(|_| {
+            MpvError::InternalConnectionError("Event checking thread timed out".to_owned())
+        })?
+        .map_err(|_| {
+            MpvError::InternalConnectionError("Event checking thread panicked".to_owned())
+        })?
+        .map_err(|err| match err {
+            PropertyCheckingThreadError::UnexpectedPropertyError(property) => {
+                MpvError::Other(format!("Unexpected property: {:?}", property))
+            }
+            PropertyCheckingThreadError::MpvError(err) => err,
+        })
+}
+
+/// This helper function will gracefully shut down both the event checking thread and the mpv process.
+/// It will also return an error if the event checking thread happened to panic, or if it times out
+/// The timeout is hardcoded to 500ms.
+pub 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();
+
+    check_property_thread_result(handle).await?;
+
+    mpv.kill().await?;
+    proc.wait().await.map_err(|err| {
+        MpvError::InternalConnectionError(format!(
+            "Failed to wait for mpv process to exit: {}",
+            err
+        ))
+    })?;
+
+    Ok(())
+}

tests/mock_socket_tests/events.rs

@@ -1,4 +1,4 @@
-use futures::{stream::StreamExt, SinkExt};
+use futures::{SinkExt, stream::StreamExt};
 use mpvipc_async::{Event, Mpv, MpvDataType, MpvExt};
 use serde_json::json;
 use test_log::test;
@@ -52,7 +52,7 @@ async fn test_observe_event_successful() {
         assert_eq!(
             event,
             Event::PropertyChange {
-                id: 1,
+                id: Some(1),
                 name: "volume".to_string(),
                 data: Some(MpvDataType::Double(64.0))
             }

tests/mock_socket_tests/get_property.rs

@@ -1,8 +1,8 @@
 use std::{panic, time::Duration};
 
-use futures::{stream::FuturesUnordered, SinkExt, StreamExt};
+use futures::{SinkExt, StreamExt, stream::FuturesUnordered};
 use mpvipc_async::{Mpv, MpvError, MpvExt, Playlist, PlaylistEntry};
-use serde_json::{json, Value};
+use serde_json::{Value, json};
 use test_log::test;
 use tokio::{net::UnixStream, task::JoinHandle};
 use tokio_util::codec::{Framed, LinesCodec, LinesCodecError};
@@ -151,8 +151,8 @@ async fn test_get_property_simultaneous_requests() {
             tokio::time::sleep(Duration::from_millis(2)).await;
             let maybe_volume = mpv_clone_3.get_property::<f64>("nonexistent").await;
             match maybe_volume {
-                Err(MpvError::MpvError(err)) => {
-                    assert_eq!(err, "property not found");
+                Err(MpvError::MpvError { message, .. }) => {
+                    assert_eq!(message, "property not found");
                 }
                 _ => panic!("Unexpected result: {:?}", maybe_volume),
             }
@@ -196,18 +196,20 @@ async fn test_get_playlist() -> Result<(), MpvError> {
         },
     ]);
 
-    let (server, join_handle) = test_socket(vec![json!({
-      "data": expected.0.iter().map(|entry| {
+    let (server, join_handle) = test_socket(vec![
         json!({
-          "filename": entry.filename,
-          "title": entry.title,
-          "current": entry.current
+          "data": expected.0.iter().map(|entry| {
+            json!({
+              "filename": entry.filename,
+              "title": entry.title,
+              "current": entry.current
+            })
+          }).collect::<Vec<Value>>(),
+          "request_id": 0,
+          "error": "success"
         })
-      }).collect::<Vec<Value>>(),
-      "request_id": 0,
-      "error": "success"
-    })
-    .to_string()]);
+        .to_string(),
+    ]);
 
     let mpv = Mpv::connect_socket(server).await?;
     let playlist = mpv.get_playlist().await?;

tests/mock_socket_tests/set_property.rs

@@ -1,8 +1,8 @@
 use std::{panic, time::Duration};
 
-use futures::{stream::FuturesUnordered, SinkExt, StreamExt};
+use futures::{SinkExt, StreamExt, stream::FuturesUnordered};
 use mpvipc_async::{Mpv, MpvError};
-use serde_json::{json, Value};
+use serde_json::{Value, json};
 use test_log::test;
 use tokio::{net::UnixStream, task::JoinHandle};
 use tokio_util::codec::{Framed, LinesCodec, LinesCodecError};
@@ -63,12 +63,12 @@ async fn test_set_property_wrong_type() -> Result<(), MpvError> {
     let mpv = Mpv::connect_socket(server).await?;
     let maybe_volume = mpv.set_property::<bool>("volume", true).await;
 
-    assert_eq!(
-        maybe_volume,
-        Err(MpvError::MpvError(
-            "unsupported format for accessing property".to_string()
-        ))
-    );
+    match maybe_volume {
+        Err(MpvError::MpvError { message, .. }) => {
+            assert_eq!(message, "unsupported format for accessing property");
+        }
+        _ => panic!("Unexpected result: {:?}", maybe_volume),
+    }
 
     join_handle.await.unwrap().unwrap();
 
@@ -84,10 +84,12 @@ async fn test_get_property_error() -> Result<(), MpvError> {
     let mpv = Mpv::connect_socket(server).await?;
     let maybe_volume = mpv.set_property("nonexistent", true).await;
 
-    assert_eq!(
-        maybe_volume,
-        Err(MpvError::MpvError("property not found".to_string()))
-    );
+    match maybe_volume {
+        Err(MpvError::MpvError { message, .. }) => {
+            assert_eq!(message, "property not found");
+        }
+        _ => panic!("Unexpected result: {:?}", maybe_volume),
+    }
 
     join_handle.await.unwrap().unwrap();
 
@@ -161,8 +163,8 @@ async fn test_set_property_simultaneous_requests() {
             tokio::time::sleep(Duration::from_millis(2)).await;
             let maybe_volume = mpv_clone_3.set_property("nonexistent", "a").await;
             match maybe_volume {
-                Err(MpvError::MpvError(err)) => {
-                    assert_eq!(err, "property not found");
+                Err(MpvError::MpvError { message, .. }) => {
+                    assert_eq!(message, "property not found");
                 }
                 _ => panic!("Unexpected result: {:?}", maybe_volume),
             }