.cargo/config.toml: init with pvv-git registry

Duplicate of the LICENSE file

.cargo/config.toml

@@ -0,0 +1,2 @@
+[registries]
+pvv-git = { index = "sparse+https://git.pvv.ntnu.no/api/packages/Grzegorz/cargo/" }

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

@@ -7,9 +7,9 @@ on:
 
 jobs:
   build:
-    runs-on: ubuntu-latest
+    runs-on: debian-latest
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
 
       - name: Install rust toolchain
         uses: dtolnay/rust-toolchain@stable
@@ -18,9 +18,9 @@ jobs:
         run: cargo build --all-features --verbose --release
 
   check:
-    runs-on: ubuntu-latest
+    runs-on: debian-latest
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
 
       - name: Install rust toolchain
         uses: dtolnay/rust-toolchain@stable
@@ -34,9 +34,9 @@ jobs:
         run: cargo clippy --all-features -- --deny warnings
 
   test:
-    runs-on: ubuntu-latest
+    runs-on: debian-latest
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
 
       - name: Install cargo binstall
         uses: cargo-bins/cargo-binstall@main
@@ -88,13 +88,13 @@ jobs:
           target: ${{ gitea.ref_name }}/coverage/
           username: gitea-web
           ssh-key: ${{ secrets.WEB_SYNC_SSH_KEY }}
-          host: bekkalokk.pvv.ntnu.no
+          host: pages.pvv.ntnu.no
-          known-hosts: "bekkalokk.pvv.ntnu.no ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIEI6VSaDrMG8+flg4/AeHlAFIen8RUzWh6URQKqFegSx"
+          known-hosts: "pages.pvv.ntnu.no ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIH2QjfFB+city1SYqltkVqWACfo1j37k+oQQfj13mtgg"
 
   docs:
-    runs-on: ubuntu-latest
+    runs-on: debian-latest
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
 
       - name: Install rust toolchain
         uses: dtolnay/rust-toolchain@stable
@@ -109,5 +109,5 @@ jobs:
           target: ${{ gitea.ref_name }}/docs/
           username: gitea-web
           ssh-key: ${{ secrets.WEB_SYNC_SSH_KEY }}
-          host: bekkalokk.pvv.ntnu.no
+          host: pages.pvv.ntnu.no
-          known-hosts: "bekkalokk.pvv.ntnu.no ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIEI6VSaDrMG8+flg4/AeHlAFIen8RUzWh6URQKqFegSx"
+          known-hosts: "pages.pvv.ntnu.no ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIH2QjfFB+city1SYqltkVqWACfo1j37k+oQQfj13mtgg"

Cargo.toml

@@ -1,32 +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"
-repository = "https://git.pvv.ntnu.no/Projects/mpvipc-async"
+repository = "https://git.pvv.ntnu.no/Grzegorz/mpvipc-async"
-documentation = "https://pages.pvv.ntnu.no/Projects/mpvipc-async/main/docs/mpvipc_async/"
+documentation = "https://pages.pvv.ntnu.no/Grzegorz/mpvipc-async/main/docs/mpvipc_async/"
-edition = "2021"
+edition = "2024"
-rust-version = "1.75"
+rust-version = "1.85.0"
 
 [dependencies]
-serde_json = "1.0.104"
+serde_json = "1.0.148"
-log = "0.4.19"
+log = "0.4.29"
-serde = { version = "1.0.197", features = ["derive"] }
+serde = { version = "1.0.228", features = ["derive"] }
-tokio = { version = "1.37.0", features = ["sync", "macros", "rt", "net"] }
+tokio = { version = "1.48.0", features = ["sync", "macros", "rt", "net"] }
-tokio-util = { version = "0.7.10", features = ["codec"] }
+tokio-util = { version = "0.7.17", features = ["codec"] }
-futures = "0.3.30"
+futures = "0.3.31"
-tokio-stream = { version = "0.1.15", features = ["sync"] }
+tokio-stream = { version = "0.1.17", features = ["sync"] }
-thiserror = "1.0.59"
+thiserror = "2.0.17"
 
 [dev-dependencies]
-env_logger = "0.10.0"
+env_logger = "0.11.8"
-test-log = "0.2.15"
+test-log = "0.2.19"
-tokio = { version = "1.37.0", features = ["rt-multi-thread", "time", "process"] }
+tokio = { version = "1.48.0", features = ["rt-multi-thread", "time", "process"] }
-uuid = { version = "1.8.0", features = ["v4"] }
+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,10 +1,11 @@
-[![Coverage](https://pages.pvv.ntnu.no/Projects/mpvipc-async/main/coverage/badges/for_the_badge.svg)](https://pages.pvv.ntnu.no/Projects/mpvipc-async/main/coverage/src/)
+[![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/Projects/mpvipc-async/main/docs/mpvipc_async/)
+[![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,3 +35,5 @@ async fn main() -> Result<(), MpvError> {
     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,
+        "lastModified": 1767116409,
-        "narHash": "sha256-NLznXB5AOnniUtZsyy/aPWOk8ussTuePp2acb9U+ISA=",
+        "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": {
+    "rust-overlay": {
-      "flake": false,
+      "inputs": {
+        "nixpkgs": [
+          "nixpkgs"
+        ]
+      },
       "locked": {
-        "lastModified": 1713373173,
+        "lastModified": 1767322002,
-        "narHash": "sha256-octd9BFY9G/Gbr4KfwK4itZp4Lx+qvJeRRcYnN+dEH8=",
+        "narHash": "sha256-yHKXXw2OWfIFsyTjduB4EyFwR0SYYF0hK8xI9z4NIn0=",
-        "owner": "rust-lang",
+        "owner": "oxalica",
-        "repo": "rust-analyzer",
+        "repo": "rust-overlay",
-        "rev": "46702ffc1a02a2ac153f1d1ce619ec917af8f3a6",
+        "rev": "03c6e38661c02a27ca006a284813afdc461e9f7e",
         "type": "github"
       },
       "original": {
-        "owner": "rust-lang",
+        "owner": "oxalica",
-        "ref": "nightly",
+        "repo": "rust-overlay",
-        "repo": "rust-analyzer",
         "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 = [
+      packages = with pkgs; [
-        (toolchain.withComponents [
+        toolchain
-          "cargo" "rustc" "rustfmt" "clippy" "llvm-tools"
+        mpv
-        ])
+        grcov
-        pkgs.mpv
+        cargo-nextest
-        pkgs.grcov
+        cargo-edit
-        pkgs.cargo-nextest
       ];
-      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.
@@ -95,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),
@@ -161,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
@@ -184,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

@@ -23,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,

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: u64,
+        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) {
+        match $event.get($key) {
-            Some(
+            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);
+    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,
+    IntoRawCommandPart, LoopProperty, Mpv, MpvCommand, MpvDataType, MpvError, Playlist,
-    Playlist, PlaylistAddOptions, Property, SeekOptions,
+    PlaylistAddOptions, Property, SeekOptions, parse_property,
 };
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
@@ -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},

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

@@ -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: u64 = 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(
-    let (handle, cancellation_token) =
+        mpv.clone(),
-        create_interruptable_event_property_checking_thread(events, |property| match property {
+        |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?;
+    mpv.observe_property(MPV_CHANNEL_ID, "volume").await?;
-    let events = mpv.get_event_stream().await;
+    let (handle, cancellation_token) = create_interruptable_event_property_checking_thread(
-    let (handle, cancellation_token) =
+        mpv.clone(),
-        create_interruptable_event_property_checking_thread(events, |property| match property {
+        |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?;
+    mpv.observe_property(MPV_CHANNEL_ID, "mute").await?;
-    let events = mpv.get_event_stream().await;
+    let (handle, cancellation_token) = create_interruptable_event_property_checking_thread(
-    let (handle, cancellation_token) =
+        mpv.clone(),
-        create_interruptable_event_property_checking_thread(events, |property| match property {
+        |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(
-    let (handle, cancellation_token) =
+        mpv.clone(),
-        create_interruptable_event_property_checking_thread(events, |property| match property {
+        |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::*;
 
@@ -60,3 +65,45 @@ async fn test_get_nonexistent_property() -> Result<(), MpvError> {
 
     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> {
@@ -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};
@@ -196,18 +196,20 @@ async fn test_get_playlist() -> Result<(), MpvError> {
         },
     ]);
 
-    let (server, join_handle) = test_socket(vec![json!({
+    let (server, join_handle) = test_socket(vec![
-      "data": expected.0.iter().map(|entry| {
         json!({
-          "filename": entry.filename,
+          "data": expected.0.iter().map(|entry| {
-          "title": entry.title,
+            json!({
-          "current": entry.current
+              "filename": entry.filename,
+              "title": entry.title,
+              "current": entry.current
+            })
+          }).collect::<Vec<Value>>(),
+          "request_id": 0,
+          "error": "success"
         })
-      }).collect::<Vec<Value>>(),
+        .to_string(),
-      "request_id": 0,
+    ]);
-      "error": "success"
-    })
-    .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};