Merge pull request #4 from adhamsalama/implement-sub-multiple-channels

Implement subscribing to multiple channels

Cargo.toml

@@ -2,7 +2,7 @@
 name = "sider"
 description = "A Multithreaded Redis clone written from scratch in Rust."
 license = "MIT"
-version = "0.1.3"
+version = "0.1.4"
 edition = "2021"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

src/handler.rs

@@ -398,20 +398,21 @@ pub fn handle_subscribe(
     stream: TcpStream,
 ) -> String {
     let mut bus = bus.write().unwrap();
-    let value = bus.get_mut(&req.key);
-    let topic_name = req.key.clone();
-    let topic_name_len = topic_name.len();
-
-    let response = format!("*3\r\n$9\r\nsubscribe\r\n${topic_name_len}\r\n{topic_name}\r\n:1\r\n");
-    // let stream: Arc<&mut TcpStream> = Arc::clone(&stream);
-    match value {
-        Some(streams) => {
-            streams.push(stream);
-            response
-        }
-        None => {
-            bus.insert(topic_name, vec![stream]);
-            response
-        }
+    let topics = &req.value;
+    let mut response = String::new();
+    for topic in topics {
+        let channel = bus.get_mut(topic);
+        let stream = stream.try_clone().unwrap();
+        let topic_name_len = topic.len();
+        response += &format!("*3\r\n$9\r\nsubscribe\r\n${topic_name_len}\r\n{topic}\r\n:1\r\n");
+        match channel {
+            Some(streams) => {
+                streams.push(stream);
+            }
+            None => {
+                bus.insert(topic.to_string(), vec![stream]);
+            }
+        };
     }
+    response
 }

src/parser.rs

@@ -1,3 +1,5 @@
+use std::collections::HashSet;
+
 use crate::{Command, Request};
 
 pub fn parse_resp(s: &String) -> Request {
@@ -38,9 +40,16 @@ pub fn parse_resp(s: &String) -> Request {
         }
     }
     match command {
-        Some(c) => {
+        Some(command) => {
+            if let Command::SUBSCRIBE = command {
+                value.insert(0, key);
+                // Remove duplicates as it's the official Redis behavior
+                let set: HashSet<_> = value.drain(..).collect();
+                value.extend(set);
+                key = String::new();
+            }
             let req = Request {
-                command: c,
+                command,
                 key,
                 value,
             };