WIP: add tests for event property parser

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

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

Cargo.toml

@@ -18,6 +18,7 @@ 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"
 
 [dev-dependencies]
 env_logger = "0.10.0"

examples/fetch_state.rs

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

examples/media_player.rs

@@ -1,4 +1,4 @@
-use mpvipc::{Error, Mpv, MpvExt};
+use mpvipc::{MpvError, Mpv, MpvExt};
 
 fn seconds_to_hms(total: f64) -> String {
     let total = total as u64;
@@ -10,7 +10,7 @@ fn seconds_to_hms(total: f64) -> String {
 }
 
 #[tokio::main]
-async fn main() -> Result<(), Error> {
+async fn main() -> Result<(), MpvError> {
     env_logger::init();
 
     let mpv = Mpv::connect("/tmp/mpv.sock").await?;

scripts/coverage.sh

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

src/core_api.rs

@@ -12,7 +12,7 @@ use tokio::{
 use crate::{
     ipc::{MpvIpc, MpvIpcCommand, MpvIpcEvent, MpvIpcResponse},
     message_parser::TypeHandler,
-    Error, ErrorCode, Event,
+    Event, MpvError,
 };
 
 /// All possible commands that can be sent to mpv.
@@ -41,7 +41,7 @@ pub enum MpvCommand {
         to: usize,
     },
     Observe {
-        id: isize,
+        id: usize,
         property: String,
     },
     PlaylistNext,
@@ -59,7 +59,7 @@ pub enum MpvCommand {
         option: SeekOptions,
     },
     Stop,
-    Unobserve(isize),
+    Unobserve(usize),
 }
 
 /// Helper trait to keep track of the string literals that mpv expects.
@@ -134,14 +134,14 @@ impl IntoRawCommandPart for SeekOptions {
 pub trait GetPropertyTypeHandler: Sized {
     // TODO: fix this
     #[allow(async_fn_in_trait)]
-    async fn get_property_generic(instance: &Mpv, property: &str) -> Result<Self, Error>;
+    async fn get_property_generic(instance: &Mpv, property: &str) -> Result<Self, MpvError>;
 }
 
 impl<T> GetPropertyTypeHandler for T
 where
     T: TypeHandler,
 {
-    async fn get_property_generic(instance: &Mpv, property: &str) -> Result<T, Error> {
+    async fn get_property_generic(instance: &Mpv, property: &str) -> Result<T, MpvError> {
         instance
             .get_property_value(property)
             .await
@@ -153,17 +153,22 @@ where
 pub trait SetPropertyTypeHandler<T> {
     // TODO: fix this
     #[allow(async_fn_in_trait)]
-    async fn set_property_generic(instance: &Mpv, property: &str, value: T) -> Result<(), Error>;
+    async fn set_property_generic(instance: &Mpv, property: &str, value: T)
+        -> Result<(), MpvError>;
 }
 
 impl<T> SetPropertyTypeHandler<T> for T
 where
     T: Serialize,
 {
-    async fn set_property_generic(instance: &Mpv, property: &str, value: T) -> Result<(), Error> {
+    async fn set_property_generic(
+        instance: &Mpv,
+        property: &str,
+        value: T,
+    ) -> Result<(), MpvError> {
         let (res_tx, res_rx) = oneshot::channel();
-        let value = serde_json::to_value(value)
+        let value = serde_json::to_value(value).map_err(|why| MpvError::JsonParseError(why))?;
-            .map_err(|why| Error(ErrorCode::JsonParseError(why.to_string())))?;
+
         instance
             .command_sender
             .send((
@@ -171,15 +176,11 @@ where
                 res_tx,
             ))
             .await
-            .map_err(|_| {
+            .map_err(|err| MpvError::InternalConnectionError(err.to_string()))?;
-                Error(ErrorCode::ConnectError(
-                    "Failed to send command".to_string(),
-                ))
-            })?;
 
         match res_rx.await {
             Ok(MpvIpcResponse(response)) => response.map(|_| ()),
-            Err(err) => Err(Error(ErrorCode::ConnectError(err.to_string()))),
+            Err(err) => Err(MpvError::InternalConnectionError(err.to_string())),
         }
     }
 }
@@ -205,18 +206,18 @@ impl fmt::Debug for Mpv {
 }
 
 impl Mpv {
-    pub async fn connect(socket_path: &str) -> Result<Mpv, Error> {
+    pub async fn connect(socket_path: &str) -> Result<Mpv, MpvError> {
         log::debug!("Connecting to mpv socket at {}", socket_path);
 
         let socket = match UnixStream::connect(socket_path).await {
             Ok(stream) => Ok(stream),
-            Err(internal_error) => Err(Error(ErrorCode::ConnectError(internal_error.to_string()))),
+            Err(err) => Err(MpvError::MpvSocketConnectionError(err.to_string())),
         }?;
 
         Self::connect_socket(socket).await
     }
 
-    pub async fn connect_socket(socket: UnixStream) -> Result<Mpv, Error> {
+    pub async fn connect_socket(socket: UnixStream) -> Result<Mpv, MpvError> {
         let (com_tx, com_rx) = mpsc::channel(100);
         let (ev_tx, _) = broadcast::channel(100);
         let ipc = MpvIpc::new(socket, com_rx, ev_tx.clone());
@@ -230,29 +231,24 @@ impl Mpv {
         })
     }
 
-    pub async fn disconnect(&self) -> Result<(), Error> {
+    pub async fn disconnect(&self) -> Result<(), MpvError> {
         let (res_tx, res_rx) = oneshot::channel();
         self.command_sender
             .send((MpvIpcCommand::Exit, res_tx))
             .await
-            .map_err(|_| {
+            .map_err(|err| MpvError::InternalConnectionError(err.to_string()))?;
-                Error(ErrorCode::ConnectError(
+
-                    "Failed to send command".to_string(),
-                ))
-            })?;
         match res_rx.await {
             Ok(MpvIpcResponse(response)) => response.map(|_| ()),
-            Err(err) => Err(Error(ErrorCode::ConnectError(err.to_string()))),
+            Err(err) => Err(MpvError::InternalConnectionError(err.to_string())),
         }
     }
 
-    pub async fn get_event_stream(&self) -> impl futures::Stream<Item = Result<Event, Error>> {
+    pub async fn get_event_stream(&self) -> impl futures::Stream<Item = Result<Event, MpvError>> {
         tokio_stream::wrappers::BroadcastStream::new(self.broadcast_channel.subscribe()).map(
             |event| match event {
                 Ok(event) => crate::event_parser::parse_event(event),
-                Err(_) => Err(Error(ErrorCode::ConnectError(
+                Err(err) => Err(MpvError::InternalConnectionError(err.to_string())),
-                    "Failed to receive event".to_string(),
-                ))),
             },
         )
     }
@@ -264,7 +260,7 @@ impl Mpv {
         &self,
         command: &str,
         args: &[&str],
-    ) -> Result<Option<Value>, Error> {
+    ) -> Result<Option<Value>, MpvError> {
         let command = Vec::from(
             [command]
                 .iter()
@@ -277,15 +273,11 @@ impl Mpv {
         self.command_sender
             .send((MpvIpcCommand::Command(command), res_tx))
             .await
-            .map_err(|_| {
+            .map_err(|err| MpvError::InternalConnectionError(err.to_string()))?;
-                Error(ErrorCode::ConnectError(
-                    "Failed to send command".to_string(),
-                ))
-            })?;
 
         match res_rx.await {
             Ok(MpvIpcResponse(response)) => response,
-            Err(err) => Err(Error(ErrorCode::ConnectError(err.to_string()))),
+            Err(err) => Err(MpvError::InternalConnectionError(err.to_string())),
         }
     }
 
@@ -293,7 +285,7 @@ impl Mpv {
         &self,
         command: &str,
         args: &[&str],
-    ) -> Result<(), Error> {
+    ) -> Result<(), MpvError> {
         self.run_command_raw(command, args).await.map(|_| ())
     }
 
@@ -323,7 +315,7 @@ impl Mpv {
     ///     Ok(())
     /// }
     /// ```
-    pub async fn run_command(&self, command: MpvCommand) -> Result<(), Error> {
+    pub async fn run_command(&self, command: MpvCommand) -> Result<(), MpvError> {
         log::trace!("Running command: {:?}", command);
         let result = match command {
             MpvCommand::LoadFile { file, option } => {
@@ -345,15 +337,11 @@ impl Mpv {
                 self.command_sender
                     .send((MpvIpcCommand::ObserveProperty(id, property), res_tx))
                     .await
-                    .map_err(|_| {
+                    .map_err(|err| MpvError::InternalConnectionError(err.to_string()))?;
-                        Error(ErrorCode::ConnectError(
-                            "Failed to send command".to_string(),
-                        ))
-                    })?;
 
                 match res_rx.await {
                     Ok(MpvIpcResponse(response)) => response.map(|_| ()),
-                    Err(err) => Err(Error(ErrorCode::ConnectError(err.to_string()))),
+                    Err(err) => Err(MpvError::InternalConnectionError(err.to_string())),
                 }
             }
             MpvCommand::PlaylistClear => {
@@ -412,10 +400,11 @@ impl Mpv {
                 self.command_sender
                     .send((MpvIpcCommand::UnobserveProperty(id), res_tx))
                     .await
-                    .unwrap();
+                    .map_err(|err| MpvError::InternalConnectionError(err.to_string()))?;
+
                 match res_rx.await {
                     Ok(MpvIpcResponse(response)) => response.map(|_| ()),
-                    Err(err) => Err(Error(ErrorCode::ConnectError(err.to_string()))),
+                    Err(err) => Err(MpvError::InternalConnectionError(err.to_string())),
                 }
             }
         };
@@ -452,7 +441,7 @@ impl Mpv {
     pub async fn get_property<T: GetPropertyTypeHandler>(
         &self,
         property: &str,
-    ) -> Result<T, Error> {
+    ) -> Result<T, MpvError> {
         T::get_property_generic(self, property).await
     }
 
@@ -475,21 +464,18 @@ impl Mpv {
     ///     Ok(())
     /// }
     /// ```
-    pub async fn get_property_value(&self, property: &str) -> Result<Value, Error> {
+    pub async fn get_property_value(&self, property: &str) -> Result<Value, MpvError> {
         let (res_tx, res_rx) = oneshot::channel();
         self.command_sender
             .send((MpvIpcCommand::GetProperty(property.to_owned()), res_tx))
             .await
-            .map_err(|_| {
+            .map_err(|err| MpvError::InternalConnectionError(err.to_string()))?;
-                Error(ErrorCode::ConnectError(
+
-                    "Failed to send command".to_string(),
-                ))
-            })?;
         match res_rx.await {
             Ok(MpvIpcResponse(response)) => {
-                response.and_then(|value| value.ok_or(Error(ErrorCode::MissingValue)))
+                response.and_then(|value| value.ok_or(MpvError::MissingMpvData))
             }
-            Err(err) => Err(Error(ErrorCode::ConnectError(err.to_string()))),
+            Err(err) => Err(MpvError::InternalConnectionError(err.to_string())),
         }
     }
 
@@ -521,7 +507,7 @@ impl Mpv {
         &self,
         property: &str,
         value: T,
-    ) -> Result<(), Error> {
+    ) -> Result<(), MpvError> {
         T::set_property_generic(self, property, value).await
     }
 }

src/error.rs

@@ -1,72 +1,46 @@
 //! Library specific error messages.
 
-use core::fmt;
+use thiserror::Error;
-use std::fmt::Display;
+use serde_json::{Map, Value};
 
-use serde::{Deserialize, Serialize};
+use crate::MpvDataType;
-
-/// All possible errors that can occur when interacting with mpv.
-#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
-pub enum ErrorCode {
-    MpvError(String),
-    JsonParseError(String),
-    ConnectError(String),
-    JsonContainsUnexptectedType,
-    UnexpectedResult,
-    UnexpectedValue,
-    MissingValue,
-    UnsupportedType,
-    ValueDoesNotContainBool,
-    ValueDoesNotContainF64,
-    ValueDoesNotContainHashMap,
-    ValueDoesNotContainPlaylist,
-    ValueDoesNotContainString,
-    ValueDoesNotContainUsize,
-}
 
 /// Any error that can occur when interacting with mpv.
-#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Error, Debug)]
-pub struct Error(pub ErrorCode);
+pub enum MpvError {
+    #[error("MpvError: {0}")]
+    MpvError(String),
 
-impl Display for Error {
+    #[error("Error communicating over mpv socket: {0}")]
-    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+    MpvSocketConnectionError(String),
-        Display::fmt(&self.0, f)
-    }
-}
 
-impl std::error::Error for Error {}
+    #[error("Internal connection error: {0}")]
+    InternalConnectionError(String),
 
-impl Display for ErrorCode {
+    #[error("JsonParseError: {0}")]
-    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+    JsonParseError(#[from] serde_json::Error),
-        match *self {
+
-            ErrorCode::ConnectError(ref msg) => f.write_str(&format!("ConnectError: {}", msg)),
+    #[error("Mpv sent a value with an unexpected type:\nExpected {expected_type}, received {received:#?}")]
-            ErrorCode::JsonParseError(ref msg) => f.write_str(&format!("JsonParseError: {}", msg)),
+    ValueContainsUnexpectedType {
-            ErrorCode::MpvError(ref msg) => f.write_str(&format!("MpvError: {}", msg)),
+      expected_type: String,
-            ErrorCode::JsonContainsUnexptectedType => {
+      received: Value,
-                f.write_str("Mpv sent a value with an unexpected type")
+    },
-            }
+
-            ErrorCode::UnexpectedResult => f.write_str("Unexpected result received"),
+    #[error("Mpv sent data with an unexpected type:\nExpected {expected_type}, received {received:#?}")]
-            ErrorCode::UnexpectedValue => f.write_str("Unexpected value received"),
+    DataContainsUnexpectedType {
-            ErrorCode::MissingValue => f.write_str("Missing value"),
+      expected_type: String,
-            ErrorCode::UnsupportedType => f.write_str("Unsupported type received"),
+      received: MpvDataType,
-            ErrorCode::ValueDoesNotContainBool => {
+    },
-                f.write_str("The received value is not of type \'std::bool\'")
+
-            }
+    #[error("Missing expected 'data' field in mpv message")]
-            ErrorCode::ValueDoesNotContainF64 => {
+    MissingMpvData,
-                f.write_str("The received value is not of type \'std::f64\'")
+
-            }
+    #[error("Missing key in object:\nExpected {key} in {map:#?}")]
-            ErrorCode::ValueDoesNotContainHashMap => {
+    MissingKeyInObject {
-                f.write_str("The received value is not of type \'std::collections::HashMap\'")
+      key: String,
-            }
+      map: Map<String, Value>,
-            ErrorCode::ValueDoesNotContainPlaylist => {
+    },
-                f.write_str("The received value is not of type \'mpvipc::Playlist\'")
+
-            }
+    #[error("Unknown error: {0}")]
-            ErrorCode::ValueDoesNotContainString => {
+    Other(String),
-                f.write_str("The received value is not of type \'std::string::String\'")
+}
-            }
-            ErrorCode::ValueDoesNotContainUsize => {
-                f.write_str("The received value is not of type \'std::usize\'")
-            }
-        }
-    }
-}

src/event_parser.rs

@@ -5,7 +5,7 @@ use std::str::FromStr;
 use serde::{Deserialize, Serialize};
 use serde_json::{Map, Value};
 
-use crate::{ipc::MpvIpcEvent, message_parser::json_to_value, Error, ErrorCode, MpvDataType};
+use crate::{ipc::MpvIpcEvent, message_parser::json_to_value, MpvDataType, MpvError};
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 #[serde(rename_all = "kebab-case")]
@@ -111,7 +111,7 @@ pub enum Event {
     PropertyChange {
         id: usize,
         name: String,
-        data: MpvDataType,
+        data: Option<MpvDataType>,
     },
     EventQueueOverflow,
     None,
@@ -147,6 +147,37 @@ pub enum Event {
     Unimplemented(Map<String, Value>),
 }
 
+macro_rules! get_key_as {
+    ($as_type:ident, $key:expr, $event:ident) => {{
+        let tmp = $event.get($key).ok_or(MpvError::MissingKeyInObject {
+            key: $key.to_owned(),
+            map: $event.clone(),
+        })?;
+
+        tmp.$as_type()
+            .ok_or(MpvError::ValueContainsUnexpectedType {
+                expected_type: stringify!($as_type).strip_prefix("as_").unwrap().to_owned(),
+                received: tmp.clone(),
+            })?
+    }};
+}
+
+macro_rules! get_optional_key_as {
+    ($as_type:ident, $key:expr, $event:ident) => {{
+        if let Some(tmp) = $event.get($key) {
+            Some(
+                tmp.$as_type()
+                    .ok_or(MpvError::ValueContainsUnexpectedType {
+                        expected_type: stringify!($as_type).strip_prefix("as_").unwrap().to_owned(),
+                        received: tmp.clone(),
+                    })?,
+            )
+        } else {
+            None
+        }
+    }};
+}
+
 // NOTE: I have not been able to test all of these events,
 //       so some of the parsing logic might be incorrect.
 //       In particular, I have not been able to make mpv
@@ -157,19 +188,17 @@ pub enum Event {
 //       If you need this, please open an issue or a PR.
 
 /// Parse a highlevel [`Event`] objects from json.
-#[allow(deprecated)]
+pub(crate) fn parse_event(raw_event: MpvIpcEvent) -> Result<Event, MpvError> {
-pub(crate) fn parse_event(raw_event: MpvIpcEvent) -> Result<Event, Error> {
     let MpvIpcEvent(event) = raw_event;
 
     event
         .as_object()
-        .ok_or(Error(ErrorCode::JsonContainsUnexptectedType))
+        .ok_or(MpvError::ValueContainsUnexpectedType {
+            expected_type: "object".to_owned(),
+            received: event.clone(),
+        })
         .and_then(|event| {
-            let event_name = event
+            let event_name = get_key_as!(as_str, "event", event);
-                .get("event")
-                .ok_or(Error(ErrorCode::MissingValue))?
-                .as_str()
-                .ok_or(Error(ErrorCode::ValueDoesNotContainString))?;
 
             match event_name {
                 "start-file" => parse_start_file(event),
@@ -200,35 +229,20 @@ pub(crate) fn parse_event(raw_event: MpvIpcEvent) -> Result<Event, Error> {
         })
 }
 
-fn parse_start_file(event: &Map<String, Value>) -> Result<Event, Error> {
+fn parse_start_file(event: &Map<String, Value>) -> Result<Event, MpvError> {
-    let playlist_entry_id = event
+    let playlist_entry_id = get_key_as!(as_u64, "playlist_entry_id", event) as usize;
-        .get("playlist_entry_id")
+
-        .ok_or(Error(ErrorCode::MissingValue))?
-        .as_u64()
-        .ok_or(Error(ErrorCode::ValueDoesNotContainUsize))? as usize;
     Ok(Event::StartFile { playlist_entry_id })
 }
 
-fn parse_end_file(event: &Map<String, Value>) -> Result<Event, Error> {
+fn parse_end_file(event: &Map<String, Value>) -> Result<Event, MpvError> {
-    let reason = event
+    let reason = get_key_as!(as_str, "reason", event);
-        .get("reason")
+    let playlist_entry_id = get_key_as!(as_u64, "playlist_entry_id", event) as usize;
-        .ok_or(Error(ErrorCode::MissingValue))?
+    let file_error = get_optional_key_as!(as_str, "file_error", event).map(|s| s.to_string());
-        .as_str()
+    let playlist_insert_id =
-        .ok_or(Error(ErrorCode::ValueDoesNotContainString))?;
+        get_optional_key_as!(as_u64, "playlist_insert_id", event).map(|i| i as usize);
-    let playlist_entry_id = event
+    let playlist_insert_num_entries =
-        .get("playlist_entry_id")
+        get_optional_key_as!(as_u64, "playlist_insert_num_entries", event).map(|i| i as usize);
-        .ok_or(Error(ErrorCode::MissingValue))?
-        .as_u64()
-        .ok_or(Error(ErrorCode::ValueDoesNotContainUsize))? as usize;
-    let file_error = event
-        .get("file_error")
-        .and_then(|v| v.as_str().map(|s| s.to_string()));
-    let playlist_insert_id = event
-        .get("playlist_insert_id")
-        .and_then(|v| v.as_u64().map(|u| u as usize));
-    let playlist_insert_num_entries = event
-        .get("playlist_insert_num_entries")
-        .and_then(|v| v.as_u64().map(|u| u as usize));
 
     Ok(Event::EndFile {
         reason: reason
@@ -241,24 +255,10 @@ fn parse_end_file(event: &Map<String, Value>) -> Result<Event, Error> {
     })
 }
 
-fn parse_log_message(event: &Map<String, Value>) -> Result<Event, Error> {
+fn parse_log_message(event: &Map<String, Value>) -> Result<Event, MpvError> {
-    let prefix = event
+    let prefix = get_key_as!(as_str, "prefix", event).to_owned();
-        .get("prefix")
+    let level = get_key_as!(as_str, "level", event);
-        .ok_or(Error(ErrorCode::MissingValue))?
+    let text = get_key_as!(as_str, "text", event).to_owned();
-        .as_str()
-        .ok_or(Error(ErrorCode::ValueDoesNotContainString))?
-        .to_string();
-    let level = event
-        .get("level")
-        .ok_or(Error(ErrorCode::MissingValue))?
-        .as_str()
-        .ok_or(Error(ErrorCode::ValueDoesNotContainString))?;
-    let text = event
-        .get("text")
-        .ok_or(Error(ErrorCode::MissingValue))?
-        .as_str()
-        .ok_or(Error(ErrorCode::ValueDoesNotContainString))?
-        .to_string();
 
     Ok(Event::LogMessage {
         prefix,
@@ -269,50 +269,34 @@ fn parse_log_message(event: &Map<String, Value>) -> Result<Event, Error> {
     })
 }
 
-fn parse_hook(event: &Map<String, Value>) -> Result<Event, Error> {
+fn parse_hook(event: &Map<String, Value>) -> Result<Event, MpvError> {
-    let hook_id = event
+    let hook_id = get_key_as!(as_u64, "hook_id", event) as usize;
-        .get("hook_id")
-        .ok_or(Error(ErrorCode::MissingValue))?
-        .as_u64()
-        .ok_or(Error(ErrorCode::ValueDoesNotContainUsize))? as usize;
     Ok(Event::Hook { hook_id })
 }
 
-fn parse_client_message(event: &Map<String, Value>) -> Result<Event, Error> {
+fn parse_client_message(event: &Map<String, Value>) -> Result<Event, MpvError> {
-    let args = event
+    let args = get_key_as!(as_array, "args", event)
-        .get("args")
-        .ok_or(Error(ErrorCode::MissingValue))?
-        .as_array()
-        .ok_or(Error(ErrorCode::ValueDoesNotContainString))?
         .iter()
         .map(|arg| {
             arg.as_str()
-                .ok_or(Error(ErrorCode::ValueDoesNotContainString))
+                .ok_or(MpvError::ValueContainsUnexpectedType {
+                    expected_type: "string".to_owned(),
+                    received: arg.clone(),
+                })
                 .map(|s| s.to_string())
         })
-        .collect::<Result<Vec<String>, Error>>()?;
+        .collect::<Result<Vec<String>, MpvError>>()?;
     Ok(Event::ClientMessage { args })
 }
 
-fn parse_property_change(event: &Map<String, Value>) -> Result<Event, Error> {
+fn parse_property_change(event: &Map<String, Value>) -> Result<Event, MpvError> {
-    let id = event
+    let id = get_key_as!(as_u64, "id", event) as usize;
-        .get("id")
+    let property_name = get_key_as!(as_str, "name", event);
-        .ok_or(Error(ErrorCode::MissingValue))?
+    let data = event.get("data").map(|d| json_to_value(d)).transpose()?;
-        .as_u64()
-        .ok_or(Error(ErrorCode::ValueDoesNotContainUsize))? as usize;
-    let property_name = event
-        .get("name")
-        .ok_or(Error(ErrorCode::MissingValue))?
-        .as_str()
-        .ok_or(Error(ErrorCode::ValueDoesNotContainString))?;
-    let data = event
-        .get("data")
-        .ok_or(Error(ErrorCode::MissingValue))?
-        .clone();
 
     Ok(Event::PropertyChange {
         id,
         name: property_name.to_string(),
-        data: json_to_value(&data)?,
+        data: data,
     })
 }

src/event_property_parser.rs

@@ -8,7 +8,7 @@ use std::collections::HashMap;
 
 use serde::{Deserialize, Serialize};
 
-use crate::{Error, ErrorCode, Event, MpvDataType, PlaylistEntry};
+use crate::{Event, MpvDataType, MpvError, PlaylistEntry};
 
 /// All possible properties that can be observed through the event system.
 ///
@@ -34,7 +34,7 @@ pub enum Property {
     Speed(f64),
     Volume(f64),
     Mute(bool),
-    Unknown { name: String, data: MpvDataType },
+    Unknown { name: String, data: Option<MpvDataType> },
 }
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
@@ -46,7 +46,7 @@ pub enum LoopProperty {
 }
 
 /// Parse a highlevel [`Property`] object from json, used for [`Event::PropertyChange`].
-pub fn parse_event_property(event: Event) -> Result<(usize, Property), Error> {
+pub fn parse_event_property(event: Event) -> Result<(usize, Property), MpvError> {
     let (id, name, data) = match event {
         Event::PropertyChange { id, name, data } => (id, name, data),
         // TODO: return proper error
@@ -58,40 +58,71 @@ pub fn parse_event_property(event: Event) -> Result<(usize, Property), Error> {
     match name.as_str() {
         "path" => {
             let path = match data {
-                MpvDataType::String(s) => Some(s),
+                Some(MpvDataType::String(s)) => Some(s),
-                MpvDataType::Null => None,
+                Some(MpvDataType::Null) => None,
-                _ => return Err(Error(ErrorCode::ValueDoesNotContainString)),
+                Some(data) => {
+                    return Err(MpvError::DataContainsUnexpectedType {
+                        expected_type: "String".to_owned(),
+                        received: data,
+                    })
+                }
+                None => {
+                    return Err(MpvError::MissingMpvData);
+                }
             };
             Ok((id, Property::Path(path)))
         }
         "pause" => {
             let pause = match data {
-                MpvDataType::Bool(b) => b,
+                Some(MpvDataType::Bool(b)) => b,
-                _ => return Err(Error(ErrorCode::ValueDoesNotContainBool)),
+                Some(data) => {
+                    return Err(MpvError::DataContainsUnexpectedType {
+                        expected_type: "bool".to_owned(),
+                        received: data,
+                    })
+                }
+                None => {
+                    return Err(MpvError::MissingMpvData);
+                }
             };
             Ok((id, Property::Pause(pause)))
         }
         "playback-time" => {
             let playback_time = match data {
-                MpvDataType::Double(d) => Some(d),
+                Some(MpvDataType::Double(d)) => Some(d),
-                MpvDataType::Null => None,
+                None | Some(MpvDataType::Null) => None,
-                _ => return Err(Error(ErrorCode::ValueDoesNotContainF64)),
+                Some(data) => {
+                    return Err(MpvError::DataContainsUnexpectedType {
+                        expected_type: "f64".to_owned(),
+                        received: data,
+                    })
+                }
             };
             Ok((id, Property::PlaybackTime(playback_time)))
         }
         "duration" => {
             let duration = match data {
-                MpvDataType::Double(d) => Some(d),
+                Some(MpvDataType::Double(d)) => Some(d),
-                MpvDataType::Null => None,
+                None | Some(MpvDataType::Null) => None,
-                _ => return Err(Error(ErrorCode::ValueDoesNotContainF64)),
+                Some(data) => {
+                    return Err(MpvError::DataContainsUnexpectedType {
+                        expected_type: "f64".to_owned(),
+                        received: data,
+                    })
+                }
             };
             Ok((id, Property::Duration(duration)))
         }
         "metadata" => {
             let metadata = match data {
-                MpvDataType::HashMap(m) => Some(m),
+                Some(MpvDataType::HashMap(m)) => Some(m),
-                MpvDataType::Null => None,
+                None | Some(MpvDataType::Null) => None,
-                _ => return Err(Error(ErrorCode::ValueDoesNotContainHashMap)),
+                Some(data) => {
+                    return Err(MpvError::DataContainsUnexpectedType {
+                        expected_type: "HashMap".to_owned(),
+                        received: data,
+                    })
+                }
             };
             Ok((id, Property::Metadata(metadata)))
         }
@@ -105,63 +136,106 @@ pub fn parse_event_property(event: Event) -> Result<(usize, Property), Error> {
         // }
         "playlist-pos" => {
             let playlist_pos = match data {
-                MpvDataType::Usize(u) => Some(u),
+                Some(MpvDataType::Usize(u)) => Some(u),
-                MpvDataType::MinusOne => None,
+                Some(MpvDataType::MinusOne) => None,
-                MpvDataType::Null => None,
+                Some(MpvDataType::Null) => None,
-                _ => return Err(Error(ErrorCode::ValueDoesNotContainUsize)),
+                None => None,
+                Some(data) => {
+                    return Err(MpvError::DataContainsUnexpectedType {
+                        expected_type: "usize or -1".to_owned(),
+                        received: data,
+                    })
+                }
             };
             Ok((id, Property::PlaylistPos(playlist_pos)))
         }
         "loop-file" => {
-            let loop_file = match data {
+            let loop_file = match data.to_owned() {
-                MpvDataType::Usize(n) => LoopProperty::N(n),
+                Some(MpvDataType::Usize(n)) => Some(LoopProperty::N(n)),
-                MpvDataType::Bool(b) => match b {
+                Some(MpvDataType::Bool(b)) => match b {
-                    true => LoopProperty::Inf,
+                    true => Some(LoopProperty::Inf),
-                    false => LoopProperty::No,
+                    false => Some(LoopProperty::No),
                 },
-                MpvDataType::String(s) => match s.as_str() {
+                Some(MpvDataType::String(s)) => match s.as_str() {
-                    "inf" => LoopProperty::Inf,
+                    "inf" => Some(LoopProperty::Inf),
-                    "no" => LoopProperty::No,
+                    _ => None,
-                    _ => return Err(Error(ErrorCode::ValueDoesNotContainString)),
                 },
-                _ => return Err(Error(ErrorCode::ValueDoesNotContainString)),
+                _ => None,
-            };
+            }
+            .ok_or(match data {
+                Some(data) => MpvError::DataContainsUnexpectedType {
+                    expected_type: "'inf', bool, or usize".to_owned(),
+                    received: data,
+                },
+                None => MpvError::MissingMpvData,
+            })?;
             Ok((id, Property::LoopFile(loop_file)))
         }
         "loop-playlist" => {
-            let loop_playlist = match data {
+            let loop_playlist = match data.to_owned() {
-                MpvDataType::Usize(n) => LoopProperty::N(n),
+                Some(MpvDataType::Usize(n)) => Some(LoopProperty::N(n)),
-                MpvDataType::Bool(b) => match b {
+                Some(MpvDataType::Bool(b)) => match b {
-                    true => LoopProperty::Inf,
+                    true => Some(LoopProperty::Inf),
-                    false => LoopProperty::No,
+                    false => Some(LoopProperty::No),
                 },
-                MpvDataType::String(s) => match s.as_str() {
+                Some(MpvDataType::String(s)) => match s.as_str() {
-                    "inf" => LoopProperty::Inf,
+                    "inf" => Some(LoopProperty::Inf),
-                    "no" => LoopProperty::No,
+                    _ => None,
-                    _ => return Err(Error(ErrorCode::ValueDoesNotContainString)),
                 },
-                _ => return Err(Error(ErrorCode::ValueDoesNotContainString)),
+                _ => None,
-            };
+            }
+            .ok_or(match data {
+                Some(data) => MpvError::DataContainsUnexpectedType {
+                    expected_type: "'inf', bool, or usize".to_owned(),
+                    received: data,
+                },
+                None => MpvError::MissingMpvData,
+            })?;
+
             Ok((id, Property::LoopPlaylist(loop_playlist)))
         }
         "speed" => {
             let speed = match data {
-                MpvDataType::Double(d) => d,
+                Some(MpvDataType::Double(d)) => d,
-                _ => return Err(Error(ErrorCode::ValueDoesNotContainF64)),
+                Some(data) => {
+                    return Err(MpvError::DataContainsUnexpectedType {
+                        expected_type: "f64".to_owned(),
+                        received: data,
+                    })
+                }
+                None => {
+                    return Err(MpvError::MissingMpvData);
+                }
             };
             Ok((id, Property::Speed(speed)))
         }
         "volume" => {
             let volume = match data {
-                MpvDataType::Double(d) => d,
+                Some(MpvDataType::Double(d)) => d,
-                _ => return Err(Error(ErrorCode::ValueDoesNotContainF64)),
+                Some(data) => {
+                    return Err(MpvError::DataContainsUnexpectedType {
+                        expected_type: "f64".to_owned(),
+                        received: data,
+                    })
+                }
+                None => {
+                    return Err(MpvError::MissingMpvData);
+                }
             };
             Ok((id, Property::Volume(volume)))
         }
         "mute" => {
             let mute = match data {
-                MpvDataType::Bool(b) => b,
+                Some(MpvDataType::Bool(b)) => b,
-                _ => return Err(Error(ErrorCode::ValueDoesNotContainBool)),
+                Some(data) => {
+                    return Err(MpvError::DataContainsUnexpectedType {
+                        expected_type: "bool".to_owned(),
+                        received: data,
+                    })
+                }
+                None => {
+                    return Err(MpvError::MissingMpvData);
+                }
             };
             Ok((id, Property::Mute(mute)))
         }

src/highlevel_api_extension.rs

@@ -1,7 +1,7 @@
 //! High-level API extension for [`Mpv`].
 
 use crate::{
-    Error, IntoRawCommandPart, Mpv, MpvCommand, MpvDataType, Playlist, PlaylistAddOptions,
+    MpvError, IntoRawCommandPart, Mpv, MpvCommand, MpvDataType, Playlist, PlaylistAddOptions,
     PlaylistEntry, SeekOptions,
 };
 use serde::{Deserialize, Serialize};
@@ -44,58 +44,58 @@ pub enum PlaylistAddTypeOptions {
 // TODO: fix this
 #[allow(async_fn_in_trait)]
 pub trait MpvExt {
-    async fn toggle(&self) -> Result<(), Error>;
+    async fn toggle(&self) -> Result<(), MpvError>;
-    async fn stop(&self) -> Result<(), Error>;
+    async fn stop(&self) -> Result<(), MpvError>;
     async fn set_volume(&self, input_volume: f64, option: NumberChangeOptions)
-        -> Result<(), Error>;
+        -> Result<(), MpvError>;
-    async fn set_speed(&self, input_speed: f64, option: NumberChangeOptions) -> Result<(), Error>;
+    async fn set_speed(&self, input_speed: f64, option: NumberChangeOptions) -> Result<(), MpvError>;
-    async fn set_mute(&self, option: Switch) -> Result<(), Error>;
+    async fn set_mute(&self, option: Switch) -> Result<(), MpvError>;
-    async fn set_loop_playlist(&self, option: Switch) -> Result<(), Error>;
+    async fn set_loop_playlist(&self, option: Switch) -> Result<(), MpvError>;
-    async fn set_loop_file(&self, option: Switch) -> Result<(), Error>;
+    async fn set_loop_file(&self, option: Switch) -> Result<(), MpvError>;
-    async fn seek(&self, seconds: f64, option: SeekOptions) -> Result<(), Error>;
+    async fn seek(&self, seconds: f64, option: SeekOptions) -> Result<(), MpvError>;
-    async fn playlist_shuffle(&self) -> Result<(), Error>;
+    async fn playlist_shuffle(&self) -> Result<(), MpvError>;
-    async fn playlist_remove_id(&self, id: usize) -> Result<(), Error>;
+    async fn playlist_remove_id(&self, id: usize) -> Result<(), MpvError>;
-    async fn playlist_play_next(&self, id: usize) -> Result<(), Error>;
+    async fn playlist_play_next(&self, id: usize) -> Result<(), MpvError>;
-    async fn playlist_play_id(&self, id: usize) -> Result<(), Error>;
+    async fn playlist_play_id(&self, id: usize) -> Result<(), MpvError>;
-    async fn playlist_move_id(&self, from: usize, to: usize) -> Result<(), Error>;
+    async fn playlist_move_id(&self, from: usize, to: usize) -> Result<(), MpvError>;
-    async fn playlist_clear(&self) -> Result<(), Error>;
+    async fn playlist_clear(&self) -> Result<(), MpvError>;
     async fn playlist_add(
         &self,
         file: &str,
         file_type: PlaylistAddTypeOptions,
         option: PlaylistAddOptions,
-    ) -> Result<(), Error>;
+    ) -> Result<(), MpvError>;
-    async fn restart(&self) -> Result<(), Error>;
+    async fn restart(&self) -> Result<(), MpvError>;
-    async fn prev(&self) -> Result<(), Error>;
+    async fn prev(&self) -> Result<(), MpvError>;
-    async fn pause(&self) -> Result<(), Error>;
+    async fn pause(&self) -> Result<(), MpvError>;
-    async fn unobserve_property(&self, id: isize) -> Result<(), Error>;
+    async fn unobserve_property(&self, id: usize) -> Result<(), MpvError>;
-    async fn observe_property(&self, id: isize, property: &str) -> Result<(), Error>;
+    async fn observe_property(&self, id: usize, property: &str) -> Result<(), MpvError>;
-    async fn next(&self) -> Result<(), Error>;
+    async fn next(&self) -> Result<(), MpvError>;
-    async fn kill(&self) -> Result<(), Error>;
+    async fn kill(&self) -> Result<(), MpvError>;
-    async fn get_playlist(&self) -> Result<Playlist, Error>;
+    async fn get_playlist(&self) -> Result<Playlist, MpvError>;
-    async fn get_metadata(&self) -> Result<HashMap<String, MpvDataType>, Error>;
+    async fn get_metadata(&self) -> Result<HashMap<String, MpvDataType>, MpvError>;
 }
 
 impl MpvExt for Mpv {
-    async fn get_metadata(&self) -> Result<HashMap<String, MpvDataType>, Error> {
+    async fn get_metadata(&self) -> Result<HashMap<String, MpvDataType>, MpvError> {
         self.get_property("metadata").await
     }
 
-    async fn get_playlist(&self) -> Result<Playlist, Error> {
+    async fn get_playlist(&self) -> Result<Playlist, MpvError> {
         self.get_property::<Vec<PlaylistEntry>>("playlist")
             .await
             .map(Playlist)
     }
 
-    async fn kill(&self) -> Result<(), Error> {
+    async fn kill(&self) -> Result<(), MpvError> {
         self.run_command(MpvCommand::Quit).await
     }
 
-    async fn next(&self) -> Result<(), Error> {
+    async fn next(&self) -> Result<(), MpvError> {
         self.run_command(MpvCommand::PlaylistNext).await
     }
 
-    async fn observe_property(&self, id: isize, property: &str) -> Result<(), Error> {
+    async fn observe_property(&self, id: usize, property: &str) -> Result<(), MpvError> {
         self.run_command(MpvCommand::Observe {
             id,
             property: property.to_string(),
@@ -103,19 +103,19 @@ impl MpvExt for Mpv {
         .await
     }
 
-    async fn unobserve_property(&self, id: isize) -> Result<(), Error> {
+    async fn unobserve_property(&self, id: usize) -> Result<(), MpvError> {
         self.run_command(MpvCommand::Unobserve(id)).await
     }
 
-    async fn pause(&self) -> Result<(), Error> {
+    async fn pause(&self) -> Result<(), MpvError> {
         self.set_property("pause", true).await
     }
 
-    async fn prev(&self) -> Result<(), Error> {
+    async fn prev(&self) -> Result<(), MpvError> {
         self.run_command(MpvCommand::PlaylistPrev).await
     }
 
-    async fn restart(&self) -> Result<(), Error> {
+    async fn restart(&self) -> Result<(), MpvError> {
         self.run_command(MpvCommand::Seek {
             seconds: 0f64,
             option: SeekOptions::Absolute,
@@ -128,7 +128,7 @@ impl MpvExt for Mpv {
         file: &str,
         file_type: PlaylistAddTypeOptions,
         option: PlaylistAddOptions,
-    ) -> Result<(), Error> {
+    ) -> Result<(), MpvError> {
         match file_type {
             PlaylistAddTypeOptions::File => {
                 self.run_command(MpvCommand::LoadFile {
@@ -148,20 +148,20 @@ impl MpvExt for Mpv {
         }
     }
 
-    async fn playlist_clear(&self) -> Result<(), Error> {
+    async fn playlist_clear(&self) -> Result<(), MpvError> {
         self.run_command(MpvCommand::PlaylistClear).await
     }
 
-    async fn playlist_move_id(&self, from: usize, to: usize) -> Result<(), Error> {
+    async fn playlist_move_id(&self, from: usize, to: usize) -> Result<(), MpvError> {
         self.run_command(MpvCommand::PlaylistMove { from, to })
             .await
     }
 
-    async fn playlist_play_id(&self, id: usize) -> Result<(), Error> {
+    async fn playlist_play_id(&self, id: usize) -> Result<(), MpvError> {
         self.set_property("playlist-pos", id).await
     }
 
-    async fn playlist_play_next(&self, id: usize) -> Result<(), Error> {
+    async fn playlist_play_next(&self, id: usize) -> Result<(), MpvError> {
         match self.get_property::<usize>("playlist-pos").await {
             Ok(current_id) => {
                 self.run_command(MpvCommand::PlaylistMove {
@@ -174,19 +174,19 @@ impl MpvExt for Mpv {
         }
     }
 
-    async fn playlist_remove_id(&self, id: usize) -> Result<(), Error> {
+    async fn playlist_remove_id(&self, id: usize) -> Result<(), MpvError> {
         self.run_command(MpvCommand::PlaylistRemove(id)).await
     }
 
-    async fn playlist_shuffle(&self) -> Result<(), Error> {
+    async fn playlist_shuffle(&self) -> Result<(), MpvError> {
         self.run_command(MpvCommand::PlaylistShuffle).await
     }
 
-    async fn seek(&self, seconds: f64, option: SeekOptions) -> Result<(), Error> {
+    async fn seek(&self, seconds: f64, option: SeekOptions) -> Result<(), MpvError> {
         self.run_command(MpvCommand::Seek { seconds, option }).await
     }
 
-    async fn set_loop_file(&self, option: Switch) -> Result<(), Error> {
+    async fn set_loop_file(&self, option: Switch) -> Result<(), MpvError> {
         let enabled = match option {
             Switch::On => "inf",
             Switch::Off => "no",
@@ -203,7 +203,7 @@ impl MpvExt for Mpv {
         self.set_property("loop-file", enabled).await
     }
 
-    async fn set_loop_playlist(&self, option: Switch) -> Result<(), Error> {
+    async fn set_loop_playlist(&self, option: Switch) -> Result<(), MpvError> {
         let enabled = match option {
             Switch::On => "inf",
             Switch::Off => "no",
@@ -220,7 +220,7 @@ impl MpvExt for Mpv {
         self.set_property("loo-playlist", enabled).await
     }
 
-    async fn set_mute(&self, option: Switch) -> Result<(), Error> {
+    async fn set_mute(&self, option: Switch) -> Result<(), MpvError> {
         let enabled = match option {
             Switch::On => "yes",
             Switch::Off => "no",
@@ -237,7 +237,7 @@ impl MpvExt for Mpv {
         self.set_property("mute", enabled).await
     }
 
-    async fn set_speed(&self, input_speed: f64, option: NumberChangeOptions) -> Result<(), Error> {
+    async fn set_speed(&self, input_speed: f64, option: NumberChangeOptions) -> Result<(), MpvError> {
         match self.get_property::<f64>("speed").await {
             Ok(speed) => match option {
                 NumberChangeOptions::Increase => {
@@ -258,7 +258,7 @@ impl MpvExt for Mpv {
         &self,
         input_volume: f64,
         option: NumberChangeOptions,
-    ) -> Result<(), Error> {
+    ) -> Result<(), MpvError> {
         match self.get_property::<f64>("volume").await {
             Ok(volume) => match option {
                 NumberChangeOptions::Increase => {
@@ -275,11 +275,11 @@ impl MpvExt for Mpv {
         }
     }
 
-    async fn stop(&self) -> Result<(), Error> {
+    async fn stop(&self) -> Result<(), MpvError> {
         self.run_command(MpvCommand::Stop).await
     }
 
-    async fn toggle(&self) -> Result<(), Error> {
+    async fn toggle(&self) -> Result<(), MpvError> {
         self.run_command_raw("cycle", &["pause"]).await.map(|_| ())
     }
 }

src/ipc.rs

@@ -8,7 +8,7 @@ use tokio::{
 };
 use tokio_util::codec::{Framed, LinesCodec};
 
-use crate::{Error, ErrorCode};
+use crate::MpvError;
 
 /// Container for all state that regards communication with the mpv IPC socket
 /// and message passing with [`Mpv`](crate::Mpv) controllers.
@@ -24,14 +24,14 @@ pub(crate) enum MpvIpcCommand {
     Command(Vec<String>),
     GetProperty(String),
     SetProperty(String, Value),
-    ObserveProperty(isize, String),
+    ObserveProperty(usize, String),
-    UnobserveProperty(isize),
+    UnobserveProperty(usize),
     Exit,
 }
 
 /// [`MpvIpc`]'s response to a [`MpvIpcCommand`].
-#[derive(Debug, Clone)]
+#[derive(Debug)]
-pub(crate) struct MpvIpcResponse(pub(crate) Result<Option<Value>, Error>);
+pub(crate) struct MpvIpcResponse(pub(crate) Result<Option<Value>, MpvError>);
 
 /// A deserialized and partially parsed event from mpv.
 #[derive(Debug, Clone)]
@@ -50,28 +50,33 @@ impl MpvIpc {
         }
     }
 
-    pub(crate) async fn send_command(&mut self, command: &[Value]) -> Result<Option<Value>, Error> {
+    pub(crate) async fn send_command(
+        &mut self,
+        command: &[Value],
+    ) -> Result<Option<Value>, MpvError> {
         let ipc_command = json!({ "command": command });
-        let ipc_command_str = serde_json::to_string(&ipc_command)
+        let ipc_command_str =
-            .map_err(|why| Error(ErrorCode::JsonParseError(why.to_string())))?;
+            serde_json::to_string(&ipc_command).map_err(|why| MpvError::JsonParseError(why))?;
 
         log::trace!("Sending command: {}", ipc_command_str);
 
         self.socket
             .send(ipc_command_str)
             .await
-            .map_err(|why| Error(ErrorCode::ConnectError(why.to_string())))?;
+            .map_err(|why| MpvError::MpvSocketConnectionError(why.to_string()))?;
 
         let response = loop {
             let response = self
                 .socket
                 .next()
                 .await
-                .ok_or(Error(ErrorCode::MissingValue))?
+                .ok_or(MpvError::MpvSocketConnectionError(
-                .map_err(|why| Error(ErrorCode::ConnectError(why.to_string())))?;
+                    "Could not receive response from mpv".to_owned(),
+                ))?
+                .map_err(|why| MpvError::MpvSocketConnectionError(why.to_string()))?;
 
             let parsed_response = serde_json::from_str::<Value>(&response)
-                .map_err(|why| Error(ErrorCode::JsonParseError(why.to_string())));
+                .map_err(|why| MpvError::JsonParseError(why));
 
             if parsed_response
                 .as_ref()
@@ -93,7 +98,7 @@ impl MpvIpc {
     pub(crate) async fn get_mpv_property(
         &mut self,
         property: &str,
-    ) -> Result<Option<Value>, Error> {
+    ) -> Result<Option<Value>, MpvError> {
         self.send_command(&[json!("get_property"), json!(property)])
             .await
     }
@@ -102,26 +107,29 @@ impl MpvIpc {
         &mut self,
         property: &str,
         value: Value,
-    ) -> Result<Option<Value>, Error> {
+    ) -> Result<Option<Value>, MpvError> {
         self.send_command(&[json!("set_property"), json!(property), value])
             .await
     }
 
     pub(crate) async fn observe_property(
         &mut self,
-        id: isize,
+        id: usize,
         property: &str,
-    ) -> Result<Option<Value>, Error> {
+    ) -> Result<Option<Value>, MpvError> {
         self.send_command(&[json!("observe_property"), json!(id), json!(property)])
             .await
     }
 
-    pub(crate) async fn unobserve_property(&mut self, id: isize) -> Result<Option<Value>, Error> {
+    pub(crate) async fn unobserve_property(
+        &mut self,
+        id: usize,
+    ) -> Result<Option<Value>, MpvError> {
         self.send_command(&[json!("unobserve_property"), json!(id)])
             .await
     }
 
-    async fn handle_event(&mut self, event: Result<Value, Error>) {
+    async fn handle_event(&mut self, event: Result<Value, MpvError>) {
         match &event {
             Ok(event) => {
                 log::trace!("Parsed event: {:?}", event);
@@ -137,17 +145,17 @@ impl MpvIpc {
         }
     }
 
-    pub(crate) async fn run(mut self) -> Result<(), Error> {
+    pub(crate) async fn run(mut self) -> Result<(), MpvError> {
         loop {
             tokio::select! {
               Some(event) = self.socket.next() => {
                 log::trace!("Got event: {:?}", event);
 
                 let parsed_event = event
-                    .map_err(|why| Error(ErrorCode::ConnectError(why.to_string())))
+                    .map_err(|why| MpvError::MpvSocketConnectionError(why.to_string()))
                     .and_then(|event|
                         serde_json::from_str::<Value>(&event)
-                        .map_err(|why| Error(ErrorCode::JsonParseError(why.to_string()))));
+                        .map_err(|why| MpvError::JsonParseError(why)));
 
                 self.handle_event(parsed_event).await;
               }
@@ -189,20 +197,40 @@ 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>, Error> {
+fn parse_mpv_response_data(value: Value) -> Result<Option<Value>, MpvError> {
     log::trace!("Parsing mpv response data: {:?}", value);
     let result = value
         .as_object()
-        .map(|o| (o.get("error").and_then(|e| e.as_str()), o.get("data")))
+        .ok_or(MpvError::ValueContainsUnexpectedType {
-        .ok_or(Error(ErrorCode::UnexpectedValue))
+            expected_type: "object".to_string(),
+            received: value.clone(),
+        })
+        .and_then(|o| {
+            let error = o
+                .get("error")
+                .ok_or(MpvError::MissingKeyInObject {
+                    key: "error".to_string(),
+                    map: o.clone(),
+                })?
+                .as_str()
+                .ok_or(MpvError::ValueContainsUnexpectedType {
+                    expected_type: "string".to_string(),
+                    received: o.get("error").unwrap().clone(),
+                })?;
+
+            let data = o.get("data");
+
+            Ok((error, data))
+        })
         .and_then(|(error, data)| match error {
-            Some("success") => Ok(data),
+            "success" => Ok(data),
-            Some(e) => Err(Error(ErrorCode::MpvError(e.to_string()))),
+            err => Err(MpvError::MpvError(err.to_string())),
-            None => Err(Error(ErrorCode::UnexpectedValue)),
         });
+
     match &result {
         Ok(v) => log::trace!("Successfully parsed mpv response data: {:?}", v),
         Err(e) => log::trace!("Error parsing mpv response data: {:?}", e),
     }
+
     result.map(|opt| opt.cloned())
 }

src/message_parser.rs

@@ -4,18 +4,21 @@ use std::collections::HashMap;
 
 use serde_json::Value;
 
-use crate::{Error, ErrorCode, MpvDataType, PlaylistEntry};
+use crate::{MpvDataType, MpvError, PlaylistEntry};
 
 pub trait TypeHandler: Sized {
-    fn get_value(value: Value) -> Result<Self, Error>;
+    fn get_value(value: Value) -> Result<Self, MpvError>;
     fn as_string(&self) -> String;
 }
 
 impl TypeHandler for String {
-    fn get_value(value: Value) -> Result<String, Error> {
+    fn get_value(value: Value) -> Result<String, MpvError> {
         value
             .as_str()
-            .ok_or(Error(ErrorCode::ValueDoesNotContainString))
+            .ok_or(MpvError::ValueContainsUnexpectedType {
+                expected_type: "String".to_string(),
+                received: value.clone(),
+            })
             .map(|s| s.to_string())
     }
 
@@ -25,10 +28,13 @@ impl TypeHandler for String {
 }
 
 impl TypeHandler for bool {
-    fn get_value(value: Value) -> Result<bool, Error> {
+    fn get_value(value: Value) -> Result<bool, MpvError> {
         value
             .as_bool()
-            .ok_or(Error(ErrorCode::ValueDoesNotContainBool))
+            .ok_or(MpvError::ValueContainsUnexpectedType {
+                expected_type: "bool".to_string(),
+                received: value.clone(),
+            })
     }
 
     fn as_string(&self) -> String {
@@ -41,10 +47,11 @@ impl TypeHandler for bool {
 }
 
 impl TypeHandler for f64 {
-    fn get_value(value: Value) -> Result<f64, Error> {
+    fn get_value(value: Value) -> Result<f64, MpvError> {
-        value
+        value.as_f64().ok_or(MpvError::ValueContainsUnexpectedType {
-            .as_f64()
+            expected_type: "f64".to_string(),
-            .ok_or(Error(ErrorCode::ValueDoesNotContainF64))
+            received: value.clone(),
+        })
     }
 
     fn as_string(&self) -> String {
@@ -53,11 +60,14 @@ impl TypeHandler for f64 {
 }
 
 impl TypeHandler for usize {
-    fn get_value(value: Value) -> Result<usize, Error> {
+    fn get_value(value: Value) -> Result<usize, MpvError> {
         value
             .as_u64()
             .map(|u| u as usize)
-            .ok_or(Error(ErrorCode::ValueDoesNotContainUsize))
+            .ok_or(MpvError::ValueContainsUnexpectedType {
+                expected_type: "usize".to_string(),
+                received: value.clone(),
+            })
     }
 
     fn as_string(&self) -> String {
@@ -65,12 +75,25 @@ impl TypeHandler for usize {
     }
 }
 
+impl TypeHandler for MpvDataType {
+    fn get_value(value: Value) -> Result<MpvDataType, MpvError> {
+        json_to_value(&value)
+    }
+
+    fn as_string(&self) -> String {
+        format!("{:?}", self)
+    }
+}
+
 impl TypeHandler for HashMap<String, MpvDataType> {
-    fn get_value(value: Value) -> Result<HashMap<String, MpvDataType>, Error> {
+    fn get_value(value: Value) -> Result<HashMap<String, MpvDataType>, MpvError> {
         value
             .as_object()
-            .ok_or(Error(ErrorCode::ValueDoesNotContainHashMap))
+            .ok_or(MpvError::ValueContainsUnexpectedType {
-            .map(json_map_to_hashmap)
+                expected_type: "Map<String, Value>".to_string(),
+                received: value.clone(),
+            })
+            .and_then(json_map_to_hashmap)
     }
 
     fn as_string(&self) -> String {
@@ -79,10 +102,13 @@ impl TypeHandler for HashMap<String, MpvDataType> {
 }
 
 impl TypeHandler for Vec<PlaylistEntry> {
-    fn get_value(value: Value) -> Result<Vec<PlaylistEntry>, Error> {
+    fn get_value(value: Value) -> Result<Vec<PlaylistEntry>, MpvError> {
         value
             .as_array()
-            .ok_or(Error(ErrorCode::ValueDoesNotContainPlaylist))
+            .ok_or(MpvError::ValueContainsUnexpectedType {
+                expected_type: "Array<Value>".to_string(),
+                received: value.clone(),
+            })
             .map(|array| json_array_to_playlist(array))
     }
 
@@ -91,9 +117,9 @@ impl TypeHandler for Vec<PlaylistEntry> {
     }
 }
 
-pub(crate) fn json_to_value(value: &Value) -> Result<MpvDataType, Error> {
+pub(crate) fn json_to_value(value: &Value) -> Result<MpvDataType, MpvError> {
     match value {
-        Value::Array(array) => Ok(MpvDataType::Array(json_array_to_vec(array))),
+        Value::Array(array) => Ok(MpvDataType::Array(json_array_to_vec(array)?)),
         Value::Bool(b) => Ok(MpvDataType::Bool(*b)),
         Value::Number(n) => {
             if n.is_i64() && n.as_i64().unwrap() == -1 {
@@ -103,11 +129,13 @@ pub(crate) fn json_to_value(value: &Value) -> Result<MpvDataType, Error> {
             } else if n.is_f64() {
                 Ok(MpvDataType::Double(n.as_f64().unwrap()))
             } else {
-                // TODO: proper error handling
+                Err(MpvError::ValueContainsUnexpectedType {
-                panic!("Unexpected number type");
+                    expected_type: "i64, u64, or f64".to_string(),
+                    received: value.clone(),
+                })
             }
         }
-        Value::Object(map) => Ok(MpvDataType::HashMap(json_map_to_hashmap(map))),
+        Value::Object(map) => Ok(MpvDataType::HashMap(json_map_to_hashmap(map)?)),
         Value::String(s) => Ok(MpvDataType::String(s.to_string())),
         Value::Null => Ok(MpvDataType::Null),
     }
@@ -115,23 +143,16 @@ pub(crate) fn json_to_value(value: &Value) -> Result<MpvDataType, Error> {
 
 pub(crate) fn json_map_to_hashmap(
     map: &serde_json::map::Map<String, Value>,
-) -> HashMap<String, MpvDataType> {
+) -> Result<HashMap<String, MpvDataType>, MpvError> {
     let mut output_map: HashMap<String, MpvDataType> = HashMap::new();
     for (ref key, value) in map.iter() {
-        // TODO: proper error handling
+        output_map.insert(key.to_string(), json_to_value(value)?);
-        if let Ok(value) = json_to_value(value) {
-            output_map.insert(key.to_string(), value);
-        }
     }
-    output_map
+    Ok(output_map)
 }
 
-pub(crate) fn json_array_to_vec(array: &[Value]) -> Vec<MpvDataType> {
+pub(crate) fn json_array_to_vec(array: &[Value]) -> Result<Vec<MpvDataType>, MpvError> {
-    array
+    array.iter().map(|entry| json_to_value(entry)).collect()
-        .iter()
-        // TODO: proper error handling
-        .filter_map(|entry| json_to_value(entry).ok())
-        .collect()
 }
 
 pub(crate) fn json_array_to_playlist(array: &[Value]) -> Vec<PlaylistEntry> {
@@ -207,7 +228,10 @@ mod test {
             )])),
         );
 
-        assert_eq!(json_map_to_hashmap(json.as_object().unwrap()), expected);
+        match json_map_to_hashmap(json.as_object().unwrap()) {
+            Ok(m) => assert_eq!(m, expected),
+            Err(e) => panic!("{:?}", e),
+        }
     }
 
     #[test]
@@ -246,7 +270,10 @@ mod test {
             )])),
         ];
 
-        assert_eq!(json_array_to_vec(json.as_array().unwrap()), expected);
+        match json_array_to_vec(json.as_array().unwrap()) {
+            Ok(v) => assert_eq!(v, expected),
+            Err(e) => panic!("{:?}", e),
+        }
     }
 
     #[test]

tests/events.rs

@@ -60,8 +60,9 @@ async fn test_observe_event_successful() {
             Err(err) => panic!("{:?}", err),
         };
         match data {
-            MpvDataType::Double(data) => assert_eq!(data, 64.0),
+            Some(MpvDataType::Double(data)) => assert_eq!(data, 64.0),
-            err => panic!("{:?}", err),
+            Some(data) => panic!("Unexpected value: {:?}", data),
+            None => panic!("No data"),
         }
     });
 

tests/get_property.rs

@@ -1,7 +1,7 @@
 use std::{panic, time::Duration};
 
 use futures::{stream::FuturesUnordered, SinkExt, StreamExt};
-use mpvipc::{Error, ErrorCode, Mpv, MpvExt, Playlist, PlaylistEntry};
+use mpvipc::{MpvError, Mpv, MpvExt, Playlist, PlaylistEntry};
 use serde_json::{json, Value};
 use test_log::test;
 use tokio::{net::UnixStream, task::JoinHandle};
@@ -41,12 +41,12 @@ async fn test_get_property_broken_pipe() {
     let mpv = Mpv::connect_socket(server).await.unwrap();
     let maybe_volume = mpv.get_property::<f64>("volume").await;
 
-    assert_eq!(
+    match maybe_volume {
-        maybe_volume,
+        Err(MpvError::MpvSocketConnectionError(err)) => {
-        Err(Error(ErrorCode::ConnectError(
+            assert_eq!(err.to_string(), "Broken pipe (os error 32)");
-            "Broken pipe (os error 32)".to_owned()
+        }
-        )))
+        _ => panic!("Unexpected result: {:?}", maybe_volume),
-    );
+    }
     join_handle.await.unwrap().unwrap();
 }
 
@@ -59,7 +59,16 @@ async fn test_get_property_wrong_type() {
     let mpv = Mpv::connect_socket(server).await.unwrap();
     let maybe_volume = mpv.get_property::<bool>("volume").await;
 
-    assert_eq!(maybe_volume, Err(Error(ErrorCode::ValueDoesNotContainBool)));
+    match maybe_volume {
+        Err(MpvError::ValueContainsUnexpectedType {
+            expected_type,
+            received,
+        }) => {
+            assert_eq!(expected_type, "bool");
+            assert_eq!(received, json!(100.0));
+        }
+        _ => panic!("Unexpected result: {:?}", maybe_volume),
+    }
     join_handle.await.unwrap().unwrap();
 }
 
@@ -72,12 +81,12 @@ async fn test_get_property_error() {
     let mpv = Mpv::connect_socket(server).await.unwrap();
     let maybe_volume = mpv.get_property::<f64>("volume").await;
 
-    assert_eq!(
+    match maybe_volume {
-        maybe_volume,
+        Err(MpvError::MpvError(err)) => {
-        Err(Error(ErrorCode::MpvError(
+            assert_eq!(err, "property unavailable");
-            "property unavailable".to_owned()
+        }
-        )))
+        _ => panic!("Unexpected result: {:?}", maybe_volume),
-    );
+    }
 
     join_handle.await.unwrap().unwrap();
 }
@@ -140,12 +149,12 @@ async fn test_get_property_simultaneous_requests() {
         loop {
             tokio::time::sleep(Duration::from_millis(2)).await;
             let maybe_volume = mpv_clone_3.get_property::<f64>("nonexistent").await;
-            assert_eq!(
+            match maybe_volume {
-                maybe_volume,
+                Err(MpvError::MpvError(err)) => {
-                Err(Error(ErrorCode::MpvError(
+                    assert_eq!(err, "property unavailable");
-                    "property unavailable".to_owned()
+                }
-                )))
+                _ => panic!("Unexpected result: {:?}", maybe_volume),
-            );
+            }
         }
     });
 

tests/integration.rs

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

tests/integration/event_property_parser.rs

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

tests/integration/misc.rs

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

tests/integration/mod.rs

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

tests/integration/util.rs

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

tests/mod.rs

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

tests/set_property.rs

@@ -1,7 +1,7 @@
 use std::{panic, time::Duration};
 
 use futures::{stream::FuturesUnordered, SinkExt, StreamExt};
-use mpvipc::{Error, ErrorCode, Mpv, MpvExt, Playlist, PlaylistEntry};
+use mpvipc::{MpvError, Mpv, MpvExt, Playlist, PlaylistEntry};
 use serde_json::{json, Value};
 use test_log::test;
 use tokio::{net::UnixStream, task::JoinHandle};
@@ -41,12 +41,12 @@ async fn test_set_property_broken_pipe() {
     let mpv = Mpv::connect_socket(server).await.unwrap();
     let maybe_set_volume = mpv.set_property("volume", 64.0).await;
 
-    assert_eq!(
+    match maybe_set_volume {
-        maybe_set_volume,
+        Err(MpvError::MpvSocketConnectionError(err)) => {
-        Err(Error(ErrorCode::ConnectError(
+            assert_eq!(err.to_string(), "Broken pipe (os error 32)");
-            "Broken pipe (os error 32)".to_owned()
+        }
-        )))
+        _ => panic!("Unexpected result: {:?}", maybe_set_volume),
-    );
+    }
     join_handle.await.unwrap().unwrap();
 }
 
@@ -59,12 +59,12 @@ async fn test_set_property_wrong_type() {
     let mpv = Mpv::connect_socket(server).await.unwrap();
     let maybe_volume = mpv.set_property::<bool>("volume", true).await;
 
-    assert_eq!(
+    match maybe_volume {
-        maybe_volume,
+        Err(MpvError::MpvError(err)) => {
-        Err(Error(ErrorCode::MpvError(
+            assert_eq!(err, "unsupported format for accessing property");
-            "unsupported format for accessing property".to_owned()
+        }
-        )))
+        _ => panic!("Unexpected result: {:?}", maybe_volume),
-    );
+    }
     join_handle.await.unwrap().unwrap();
 }
 
@@ -77,10 +77,12 @@ async fn test_get_property_error() {
     let mpv = Mpv::connect_socket(server).await.unwrap();
     let maybe_volume = mpv.set_property("nonexistent", true).await;
 
-    assert_eq!(
+    match maybe_volume {
-        maybe_volume,
+        Err(MpvError::MpvError(err)) => {
-        Err(Error(ErrorCode::MpvError("property not found".to_owned())))
+            assert_eq!(err, "property not found");
-    );
+        }
+        _ => panic!("Unexpected result: {:?}", maybe_volume),
+    }
 
     join_handle.await.unwrap().unwrap();
 }
@@ -127,7 +129,10 @@ async fn test_set_property_simultaneous_requests() {
     let mpv_poller_1 = tokio::spawn(async move {
         loop {
             let status = mpv_clone_1.set_property("volume", 100).await;
-            assert_eq!(status, Ok(()));
+            match status {
+                Ok(()) => {},
+                _ => panic!("Unexpected result: {:?}", status),
+            }
         }
     });
 
@@ -136,7 +141,10 @@ async fn test_set_property_simultaneous_requests() {
         loop {
             tokio::time::sleep(Duration::from_millis(1)).await;
             let status = mpv_clone_2.set_property("pause", false).await;
-            assert_eq!(status, Ok(()));
+            match status {
+                Ok(()) => {},
+                _ => panic!("Unexpected result: {:?}", status),
+            }
         }
     });
 
@@ -145,10 +153,12 @@ async fn test_set_property_simultaneous_requests() {
         loop {
             tokio::time::sleep(Duration::from_millis(2)).await;
             let maybe_volume = mpv_clone_3.set_property("nonexistent", "a").await;
-            assert_eq!(
+            match maybe_volume {
-                maybe_volume,
+                Err(MpvError::MpvError(err)) => {
-                Err(Error(ErrorCode::MpvError("property not found".to_owned())))
+                    assert_eq!(err, "property not found");
-            );
+                }
+                _ => panic!("Unexpected result: {:?}", maybe_volume),
+            }
         }
     });