feat: methods for parting channels from events

2 files changed, 68 insertions, 29 deletions

diff --git a/src/betterttv.rs b/src/betterttv.rs
index 0dbd4a3..09f544c 100644
--- a/src/betterttv.rs
+++ b/src/betterttv.rs
@@ -169,8 +169,15 @@ impl BetterTTVWSClient {
     }
 
     pub fn join_channel(&mut self, twitch_id: usize) {
-        if self.awaiting_channels.contains(&twitch_id) || self.joined_channels.contains(&twitch_id)
-        {
+        if self.joined_channels.contains(&twitch_id) {
+            return;
+        }
+
+        self.awaiting_channels.insert(twitch_id);
+    }
+
+    pub fn part_channel(&mut self, twitch_id: usize) {
+        if !self.joined_channels.contains(&twitch_id) {
             return;
         }
 
@@ -254,12 +261,25 @@ impl BetterTTVWSClient {
 
     async fn join_channels(&mut self, stream: &mut WebSocketStream<MaybeTlsStream<TcpStream>>) {
         for id in &self.awaiting_channels {
-            let json = serde_json::json!({
-                "name": "join_channel",
-                "data": {
-                    "name": format!("twitch:{}", id)
-                }
-            });
+            let json = if self.joined_channels.contains(id) {
+                self.joined_channels.remove(id);
+
+                serde_json::json!({
+                    "name": "part_channel",
+                    "data": {
+                        "name": format!("twitch:{}", id)
+                    }
+                })
+            } else {
+                self.joined_channels.insert(*id);
+
+                serde_json::json!({
+                    "name": "join_channel",
+                    "data": {
+                        "name": format!("twitch:{}", id)
+                    }
+                })
+            };
 
             stream
                 .send(Message::Text(
@@ -268,9 +288,7 @@ impl BetterTTVWSClient {
                         .into(),
                 ))
                 .await
-                .expect("Error sending join request");
-
-            self.joined_channels.insert(*id);
+                .expect("Error sending join/part request");
         }
 
         self.awaiting_channels.clear();
diff --git a/src/seventv.rs b/src/seventv.rs
index e2b42ab..eb0829b 100644
--- a/src/seventv.rs
+++ b/src/seventv.rs
@@ -255,7 +255,7 @@ impl SevenTVWSClient {
         stream: &mut WebSocketStream<MaybeTlsStream<TcpStream>>,
     ) -> Result<()> {
         if self.identified {
-            self.join_channels(stream).await;
+            self.listen_emote_sets(stream).await;
         }
 
         tokio::select!(Some(msg) = futures::StreamExt::next(stream) => {
@@ -274,13 +274,20 @@ impl SevenTVWSClient {
         Ok(())
     }
 
-    pub fn join_channel(&mut self, twitch_id: String) {
-        if self.awaiting_channels.contains(&twitch_id) || self.joined_channels.contains(&twitch_id)
-        {
+    pub fn subscribe_emote_set(&mut self, emote_set_id: String) {
+        if self.joined_channels.contains(&emote_set_id) {
             return;
         }
 
-        self.awaiting_channels.insert(twitch_id);
+        self.awaiting_channels.insert(emote_set_id);
+    }
+
+    pub fn unsubscribe_emote_set(&mut self, emote_set_id: String) {
+        if !self.joined_channels.contains(&emote_set_id) {
+            return;
+        }
+
+        self.awaiting_channels.insert(emote_set_id);
     }
 
     async fn process_message(
@@ -299,7 +306,7 @@ impl SevenTVWSClient {
 
                 // unsupported operation
                 if operation_code == 1 {
-                    self.join_channels(stream).await;
+                    self.listen_emote_sets(stream).await;
                     self.identified = true;
                     return;
                 } else if operation_code != 0 {
@@ -405,17 +412,33 @@ impl SevenTVWSClient {
         }
     }
 
-    async fn join_channels(&mut self, stream: &mut WebSocketStream<MaybeTlsStream<TcpStream>>) {
+    async fn listen_emote_sets(&mut self, stream: &mut WebSocketStream<MaybeTlsStream<TcpStream>>) {
         for id in &self.awaiting_channels {
-            let json = serde_json::json!({
-                "op": 35,
-                "d": {
-                    "type": "emote_set.update",
-                    "condition": {
-                        "object_id": id
+            let json = if self.joined_channels.contains(id) {
+                self.joined_channels.remove(id);
+
+                serde_json::json!({
+                    "op": 36,
+                    "d": {
+                        "type": "emote_set.update",
+                        "condition": {
+                            "object_id": id
+                        }
                     }
-                }
-            });
+                })
+            } else {
+                self.joined_channels.insert(id.clone());
+
+                serde_json::json!({
+                    "op": 35,
+                    "d": {
+                        "type": "emote_set.update",
+                        "condition": {
+                            "object_id": id
+                        }
+                    }
+                })
+            };
 
             stream
                 .send(Message::Text(
@@ -424,9 +447,7 @@ impl SevenTVWSClient {
                         .into(),
                 ))
                 .await
-                .expect("Error sending join request");
-
-            self.joined_channels.insert(id.clone());
+                .expect("Error sending join/part request");
         }
 
         self.awaiting_channels.clear();