Adjust to new changes from August

Cargo.toml

@@ -48,3 +48,7 @@ tls = ["tokio-tungstenite/rustls-tls"]
 [[example]]
 name = "events"
 required-features = ["events"]
+
+[[test]]
+name = "general"
+required-features = ["events"]

src/client/config.rs

@@ -2,7 +2,7 @@ use serde::{de::DeserializeOwned, Serialize};
 
 use super::Client;
 use crate::{
-    requests::{RequestType, SetProfileParameter, SetVideoSettings},
+    requests::{Realm, RequestType, SetPersistentData, SetProfileParameter, SetVideoSettings},
     responses, Error, Result,
 };
 
@@ -12,6 +12,22 @@ pub struct Config<'a> {
 }
 
 impl<'a> Config<'a> {
+    pub async fn get_persistent_data(
+        &self,
+        realm: Realm,
+        slot_name: &str,
+    ) -> Result<serde_json::Value> {
+        self.client
+            .send_message(RequestType::GetPersistentData { realm, slot_name })
+            .await
+    }
+
+    pub async fn set_persistent_data(&self, data: SetPersistentData<'_>) -> Result<()> {
+        self.client
+            .send_message(RequestType::SetPersistentData(data))
+            .await
+    }
+
     pub async fn get_scene_collection_list(&self) -> Result<responses::SceneCollections> {
         self.client
             .send_message(RequestType::GetSceneCollectionList)
@@ -26,6 +42,14 @@ impl<'a> Config<'a> {
             .await
     }
 
+    pub async fn create_scene_collection(&self, scene_collection_name: &str) -> Result<()> {
+        self.client
+            .send_message(RequestType::CreateSceneCollection {
+                scene_collection_name,
+            })
+            .await
+    }
+
     pub async fn get_profile_list(&self) -> Result<responses::Profiles> {
         self.client.send_message(RequestType::GetProfileList).await
     }
@@ -36,6 +60,18 @@ impl<'a> Config<'a> {
             .await
     }
 
+    pub async fn create_profile(&self, profile_name: &str) -> Result<()> {
+        self.client
+            .send_message(RequestType::CreateProfile { profile_name })
+            .await
+    }
+
+    pub async fn remove_profile(&self, profile_name: &str) -> Result<()> {
+        self.client
+            .send_message(RequestType::RemoveProfile { profile_name })
+            .await
+    }
+
     pub async fn get_profile_parameter(
         &self,
         parameter_category: &str,

src/client/general.rs

@@ -27,6 +27,10 @@ impl<'a> General<'a> {
             .await
     }
 
+    pub async fn get_stats(&self) -> Result<responses::Stats> {
+        self.client.send_message(RequestType::GetStats).await
+    }
+
     pub async fn get_hotkey_list(&self) -> Result<Vec<String>> {
         self.client
             .send_message::<responses::Hotkeys>(RequestType::GetHotkeyList)

src/client/mod.rs

@@ -35,8 +35,8 @@ pub use self::{
 #[cfg(feature = "events")]
 use crate::events::Event;
 use crate::{
-    requests::{ClientRequest, RequestType},
-    responses::{ServerMessage, Status},
+    requests::{ClientRequest, Identify, Request, RequestType},
+    responses::{Hello, Identified, RequestResponse, ServerMessage, Status},
     Error, Result,
 };
 
@@ -217,11 +217,11 @@ impl Client {
                         .map_err(InnerError::DeserializeMessage)?;
 
                     match message {
-                        ServerMessage::RequestResponse {
+                        ServerMessage::RequestResponse(RequestResponse {
                             request_id,
                             request_status,
                             response_data,
-                        } => {
+                        }) => {
                             let request_id = request_id
                                 .parse()
                                 .map_err(|e| InnerError::InvalidRequestId(e, request_id))?;
@@ -304,10 +304,11 @@ impl Client {
         T: DeserializeOwned,
     {
         let id = self.id_counter.fetch_add(1, Ordering::SeqCst);
-        let req = ClientRequest::Request {
-            request_id: &id.to_string(),
+        let id_str = id.to_string();
+        let req = ClientRequest::Request(Request {
+            request_id: &id_str,
             ty: req,
-        };
+        });
         let json = serde_json::to_string(&req).map_err(Error::SerializeMessage)?;
 
         let (tx, rx) = oneshot::channel();
@@ -474,22 +475,22 @@ async fn handshake(
     }
 
     match read_message(read).await? {
-        ServerMessage::Hello {
+        ServerMessage::Hello(Hello {
             obs_web_socket_version: _,
             rpc_version,
             authentication,
-        } => {
+        }) => {
             let authentication = authentication.zip(password).map(|(auth, password)| {
                 create_auth_response(&auth.challenge, &auth.salt, password)
             });
 
-            let req = serde_json::to_string(&ClientRequest::Identify {
+            let req = serde_json::to_string(&ClientRequest::Identify(Identify {
                 rpc_version,
                 authentication,
                 ignore_invalid_messages: false,
                 ignore_non_fatal_request_checks: false,
                 event_subscriptions: None,
-            })
+            }))
             .map_err(HandshakeError::SerializeMessage)?;
 
             write
@@ -501,9 +502,9 @@ async fn handshake(
     }
 
     match read_message(read).await? {
-        ServerMessage::Identified {
+        ServerMessage::Identified(Identified {
             negotiated_rpc_version,
-        } => {
+        }) => {
             debug!("identified with RPC version {}", negotiated_rpc_version);
         }
         _ => return Err(HandshakeError::NoIdentified),

src/events.rs

@@ -34,6 +34,7 @@ pub enum Event {
     // --------------------------------
     // General
     // --------------------------------
+    CustomEvent(serde_json::Value),
     ExitStarted,
     #[serde(rename_all = "camelCase")]
     StudioModeStateChanged {
@@ -276,6 +277,5 @@ pub struct BasicSceneItem {
 #[serde(rename_all = "camelCase")]
 pub struct Scene {
     scene_name: String,
-    scene_index: u32,
     is_group: bool,
 }

src/requests/mod.rs

@@ -2,39 +2,75 @@
 
 use std::path::Path;
 
-use serde::Serialize;
+use serde::{ser::SerializeStruct, Serialize};
 use serde_with::skip_serializing_none;
 
+pub(crate) enum ClientRequest<'a> {
+    Identify(Identify),
+    Reidentify(Reidentify),
+    Request(Request<'a>),
+    RequestBatch(RequestBatch<'a>),
+}
+
+impl<'a> Serialize for ClientRequest<'a> {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        fn write_state<S>(serializer: S, op: u8, d: &impl Serialize) -> Result<S::Ok, S::Error>
+        where
+            S: serde::Serializer,
+        {
+            let mut state = serializer.serialize_struct("ClientRequest", 2)?;
+            state.serialize_field("op", &op)?;
+            state.serialize_field("d", d)?;
+            state.end()
+        }
+
+        match self {
+            Self::Identify(value) => write_state(serializer, 1, value),
+            Self::Reidentify(value) => write_state(serializer, 3, value),
+            Self::Request(value) => write_state(serializer, 6, value),
+            Self::RequestBatch(value) => write_state(serializer, 8, value),
+        }
+    }
+}
+
 #[skip_serializing_none]
 #[derive(Serialize)]
-#[serde(tag = "messageType")]
-pub(crate) enum ClientRequest<'a> {
-    #[serde(rename_all = "camelCase")]
-    Identify {
-        rpc_version: u32,
-        authentication: Option<String>,
-        ignore_invalid_messages: bool,
-        ignore_non_fatal_request_checks: bool,
-        event_subscriptions: Option<u32>,
-    },
-    #[serde(rename_all = "camelCase")]
-    Reidentify {
-        ignore_invalid_messages: bool,
-        ignore_non_fatal_request_checks: bool,
-        event_subscriptions: Option<u32>,
-    },
-    #[serde(rename_all = "camelCase")]
-    Request {
-        request_id: &'a str,
-        #[serde(flatten)]
-        ty: RequestType<'a>,
-    },
-    #[serde(rename_all = "camelCase")]
-    RequestBatch {
-        request_id: &'a str,
-        halt_on_failure: Option<bool>,
-        requests: &'a [RequestType<'a>],
-    },
+#[serde(rename_all = "camelCase")]
+pub(crate) struct Identify {
+    pub rpc_version: u32,
+    pub authentication: Option<String>,
+    pub ignore_invalid_messages: bool,
+    pub ignore_non_fatal_request_checks: bool,
+    pub event_subscriptions: Option<u32>,
+}
+
+#[skip_serializing_none]
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct Reidentify {
+    pub ignore_invalid_messages: bool,
+    pub ignore_non_fatal_request_checks: bool,
+    pub event_subscriptions: Option<u32>,
+}
+
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct Request<'a> {
+    pub request_id: &'a str,
+    #[serde(flatten)]
+    pub ty: RequestType<'a>,
+}
+
+#[skip_serializing_none]
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct RequestBatch<'a> {
+    pub request_id: &'a str,
+    pub halt_on_failure: Option<bool>,
+    pub requests: &'a [RequestType<'a>],
 }
 
 #[derive(Serialize)]
@@ -43,17 +79,35 @@ pub(crate) enum RequestType<'a> {
     // --------------------------------
     // Config
     // --------------------------------
+    #[serde(rename_all = "camelCase")]
+    GetPersistentData {
+        realm: Realm,
+        slot_name: &'a str,
+    },
+    SetPersistentData(SetPersistentData<'a>),
     GetSceneCollectionList,
     #[serde(rename_all = "camelCase")]
     SetCurrentSceneCollection {
         scene_collection_name: &'a str,
     },
+    #[serde(rename_all = "camelCase")]
+    CreateSceneCollection {
+        scene_collection_name: &'a str,
+    },
     GetProfileList,
     #[serde(rename_all = "camelCase")]
     SetCurrentProfile {
         profile_name: &'a str,
     },
     #[serde(rename_all = "camelCase")]
+    CreateProfile {
+        profile_name: &'a str,
+    },
+    #[serde(rename_all = "camelCase")]
+    RemoveProfile {
+        profile_name: &'a str,
+    },
+    #[serde(rename_all = "camelCase")]
     GetProfileParameter {
         parameter_category: &'a str,
         parameter_name: &'a str,
@@ -75,6 +129,7 @@ pub(crate) enum RequestType<'a> {
     BroadcastCustomEvent {
         event_data: serde_json::Value,
     },
+    GetStats,
     GetHotkeyList,
     #[serde(rename_all = "camelCase")]
     TriggerHotkeyByName {
@@ -183,6 +238,22 @@ pub(crate) enum RequestType<'a> {
     StopStream,
 }
 
+#[derive(Clone, Copy, Serialize)]
+pub enum Realm {
+    #[serde(rename = "OBS_WEBSOCKET_DATA_REALM_GLOBAL")]
+    Global,
+    #[serde(rename = "OBS_WEBSOCKET_DATA_REALM_PROFILE")]
+    Profile,
+}
+
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct SetPersistentData<'a> {
+    pub realm: Realm,
+    pub slot_name: &'a str,
+    pub slot_value: &'a serde_json::Value,
+}
+
 #[derive(Default, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct SetProfileParameter<'a> {

src/responses/mod.rs

@@ -2,46 +2,112 @@
 
 use chrono::Duration;
 pub use semver::Version as SemVerVersion;
-use serde::Deserialize;
+use serde::{de, Deserialize, Deserializer};
 use serde_repr::Deserialize_repr;
 
-#[derive(Debug, Deserialize)]
-#[serde(tag = "messageType")]
+#[derive(Debug)]
 pub(crate) enum ServerMessage {
     /// First message sent from the server immediately on client connection. Contains authentication
     /// information if auth is required. Also contains RPC version for version negotiation.
-    #[serde(rename_all = "camelCase")]
-    Hello {
-        obs_web_socket_version: SemVerVersion,
-        /// Version number which gets incremented on each **breaking change** to the obs-websocket
-        /// protocol.
-        rpc_version: u32,
-        authentication: Option<Authentication>,
-    },
+    Hello(Hello),
     /// The identify request was received and validated, and the connection is now ready for normal
     /// operation.
-    #[serde(rename_all = "camelCase")]
-    Identified {
-        /// The RPC version to be used.
-        negotiated_rpc_version: u32,
-    },
+    Identified(Identified),
     /// An event coming from OBS has occurred. Eg scene switched, source muted.
     #[cfg(feature = "events")]
-    #[serde(rename_all = "camelCase")]
     Event(crate::events::Event),
+    #[cfg(not(feature = "events"))]
+    Event,
     /// `obs-websocket` is responding to a request coming from a client.
-    #[serde(rename_all = "camelCase")]
-    RequestResponse {
-        request_id: String,
-        request_status: Status,
-        #[serde(default)]
-        response_data: serde_json::Value,
-    },
-    #[serde(rename_all = "camelCase")]
-    RequestBatchResponse {
-        request_id: String,
-        results: Vec<serde_json::Value>,
-    },
+    RequestResponse(RequestResponse),
+    RequestBatchResponse(RequestBatchResponse),
+}
+
+impl<'de> Deserialize<'de> for ServerMessage {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        #[derive(Deserialize)]
+        struct RawServerMessage {
+            #[serde(rename = "op")]
+            op_code: OpCode,
+            #[serde(rename = "d")]
+            data: serde_json::Value,
+        }
+
+        #[derive(Deserialize_repr)]
+        #[repr(u8)]
+        enum OpCode {
+            Hello = 0,
+            Identified = 2,
+            Event = 5,
+            RequestResponse = 7,
+            RequestBatchResponse = 9,
+        }
+
+        let raw = RawServerMessage::deserialize(deserializer)?;
+
+        Ok(match raw.op_code {
+            OpCode::Hello => {
+                ServerMessage::Hello(serde_json::from_value(raw.data).map_err(de::Error::custom)?)
+            }
+            OpCode::Identified => ServerMessage::Identified(
+                serde_json::from_value(raw.data).map_err(de::Error::custom)?,
+            ),
+            OpCode::Event => {
+                #[cfg(feature = "events")]
+                {
+                    ServerMessage::Event(
+                        serde_json::from_value(raw.data).map_err(de::Error::custom)?,
+                    )
+                }
+                #[cfg(not(feature = "events"))]
+                {
+                    ServerMessage::Event
+                }
+            }
+            OpCode::RequestResponse => ServerMessage::RequestResponse(
+                serde_json::from_value(raw.data).map_err(de::Error::custom)?,
+            ),
+            OpCode::RequestBatchResponse => ServerMessage::RequestBatchResponse(
+                serde_json::from_value(raw.data).map_err(de::Error::custom)?,
+            ),
+        })
+    }
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct Hello {
+    pub obs_web_socket_version: SemVerVersion,
+    /// Version number which gets incremented on each **breaking change** to the obs-websocket
+    /// protocol.
+    pub rpc_version: u32,
+    pub authentication: Option<Authentication>,
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct Identified {
+    /// The RPC version to be used.
+    pub negotiated_rpc_version: u32,
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct RequestResponse {
+    pub request_id: String,
+    pub request_status: Status,
+    #[serde(default)]
+    pub response_data: serde_json::Value,
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct RequestBatchResponse {
+    pub request_id: String,
+    pub results: Vec<serde_json::Value>,
 }
 
 #[derive(Debug, Deserialize)]
@@ -153,6 +219,18 @@ pub enum StatusCode {
     PropertyNotFound = 618,
     /// The specififed key (OBS_KEY_*) was not found.
     KeyNotFound = 619,
+    /// The specified data realm (OBS_WEBSOCKET_DATA_REALM_*) was not found.
+    DataRealmNotFound = 620,
+    /// The scene collection already exists.
+    SceneCollectionAlreadyExists = 621,
+    /// There are not enough scene collections to perform the action.
+    NotEnoughSceneCollections = 622,
+    /// The profile already exists.
+    ProfileAlreadyExists = 623,
+    /// There are not enough profiles to perform the action.
+    NotEnoughProfiles = 624,
+    /// There are not enough scenes to perform the action.
+    NotEnoughScenes = 625,
     /// Processing the request failed unexpectedly.
     RequestProcessingFailed = 700,
     /// Starting the Output failed.
@@ -222,6 +300,13 @@ pub struct Version {
     pub supported_image_formats: Vec<String>,
 }
 
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct Stats {
+    pub web_socket_session_incoming_messages: u64,
+    pub web_socket_session_outgoing_messages: u64,
+}
+
 #[derive(Debug, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub(crate) struct Hotkeys {
@@ -298,7 +383,6 @@ pub struct Scenes {
 #[serde(rename_all = "camelCase")]
 pub struct Scene {
     pub scene_name: String,
-    pub scene_index: u64,
     pub is_group: bool,
 }
 
@@ -331,8 +415,12 @@ pub(crate) struct ImageData {
 #[serde(rename_all = "camelCase")]
 pub struct StreamStatus {
     pub output_active: bool,
+    pub output_reconnecting: bool,
     #[serde(deserialize_with = "crate::de::duration_timecode")]
     pub output_timecode: Duration,
     #[serde(deserialize_with = "crate::de::duration_nanos")]
     pub output_duration: Duration,
+    pub output_bytes: u64,
+    pub output_skipped_frames: u32,
+    pub output_total_frames: u32,
 }

tests/common/mod.rs

@@ -29,7 +29,10 @@ pub async fn new_client() -> Result<Client> {
     });
 
     let host = env::var("OBS_HOST").unwrap_or_else(|_| "localhost".to_owned());
-    let client = Client::connect(host, 4444, env::var("OBS_PASSWORD").ok()).await?;
+    let port = env::var("OBS_PORT")
+        .map(|p| p.parse())
+        .unwrap_or(Ok(4444))?;
+    let client = Client::connect(host, port, env::var("OBS_PASSWORD").ok()).await?;
 
     ensure_obs_setup(&client).await?;
 
@@ -159,11 +162,13 @@ fn is_required_profile(profile: &str) -> bool {
     profile == TEST_PROFILE
 }
 
-#[allow(unused_macros)]
+#[macro_export]
 macro_rules! wait_for {
     ($expression:expr, $pattern:pat) => {
-        while let Some(Event { ty, .. }) = $expression.next().await {
-            if matches!(ty, $pattern) {
+        use futures_util::stream::StreamExt;
+
+        while let Some(event) = $expression.next().await {
+            if matches!(event, $pattern) {
                 break;
             }
         }

tests/config.rs

@@ -4,7 +4,7 @@ use std::time::Duration;
 
 use anyhow::Result;
 use obws::{
-    requests::SetProfileParameter,
+    requests::{Realm, SetPersistentData, SetProfileParameter},
     responses::{Profiles, SceneCollections},
 };
 use tokio::time;
@@ -16,6 +16,17 @@ async fn main() -> Result<()> {
     let client = common::new_client().await?;
     let client = client.config();
 
+    client
+        .set_persistent_data(SetPersistentData {
+            realm: Realm::Profile,
+            slot_name: "obws-test",
+            slot_value: &true.into(),
+        })
+        .await?;
+    client
+        .get_persistent_data(Realm::Profile, "obws-test")
+        .await?;
+
     let SceneCollections {
         current_scene_collection_name,
         scene_collections,
@@ -24,7 +35,7 @@ async fn main() -> Result<()> {
         .iter()
         .find(|sc| *sc != &current_scene_collection_name)
         .unwrap();
-    client.set_current_scene_collection(&other).await?;
+    client.set_current_scene_collection(other).await?;
     time::sleep(Duration::from_secs(1)).await;
     client
         .set_current_scene_collection(&current_scene_collection_name)
@@ -39,10 +50,12 @@ async fn main() -> Result<()> {
         .iter()
         .find(|p| *p != &current_profile_name)
         .unwrap();
-    client.set_current_profile(&other).await?;
+    client.set_current_profile(other).await?;
     time::sleep(Duration::from_secs(1)).await;
     client.set_current_profile(&current_profile_name).await?;
     time::sleep(Duration::from_secs(1)).await;
+    client.create_profile("OBWS-TEST-New-Profile").await?;
+    client.remove_profile("OBWS-TEST-New-Profile").await?;
 
     // Currently broken in obs-websocket
     // client.get_profile_parameter("General", "Name").await?;

tests/general.rs

@@ -1,7 +1,7 @@
 #![cfg(feature = "test-integration")]
 
 use anyhow::Result;
-use obws::requests::KeyModifiers;
+use obws::{events::Event, requests::KeyModifiers};
 use serde::Serialize;
 
 mod common;
@@ -9,12 +9,17 @@ mod common;
 #[tokio::test]
 async fn main() -> Result<()> {
     let client = common::new_client().await?;
+    let events = client.events()?;
     let client = client.general();
 
+    tokio::pin!(events);
+
     client.get_version().await?;
     client
         .broadcast_custom_event(&CustomEvent { hello: "world!" })
         .await?;
+    wait_for!(events, Event::CustomEvent(_));
+    client.get_stats().await?;
 
     client.get_hotkey_list().await?;
     client.trigger_hotkey_by_name("ReplayBuffer.Save").await?;