5 files changed, 79 insertions, 9 deletions

diff --git a/main.go b/main.go
index 9e48eb0..36a41b5 100644
--- a/main.go
+++ b/main.go
@@ -86,7 +86,13 @@ func handleConnection(conn net.Conn) {
 
 		command := strings.ToUpper(args[0])
 
-		conn.Write(core.SerializeRESP(commands.DispatchBaseCommand(db,command,args)))
-		
+		if cmd , ok := commands.PubSubRegistry[command]; ok {
+			commands.DispatchPubSub(conn, &hub, cmd, args)
+		} else if cmd, ok := commands.BaseRegistry[command]; ok {
+				res := commands.DispatchBaseCommand(db, cmd, args)
+				conn.Write(core.SerializeRESP(res))
+		} else {
+			conn.Write(core.SerializeRESP(fmt.Errorf("ERR unknown command '%s'", command)))
+		}		
 	}
 }
\ No newline at end of file
diff --git a/pkg/commands/command-registry.go b/pkg/commands/command-registry.go
index 068fe7c..19c8acc 100644
--- a/pkg/commands/command-registry.go
+++ b/pkg/commands/command-registry.go
@@ -2,7 +2,9 @@ package commands
 
 import (
 	"errors"
+	"net"
 	"redisClone/pkg/core"
+	"redisClone/pkg/pubsub"
 )
 
 var dbInstance core.RedisDB
@@ -27,17 +29,13 @@ func RenameKeys(args []string) []string {
 	return []string{args[1], args[2]}
 }
 
-type CommandDef struct{
+type BaseCommand struct{
 	MinArgs int
 	ExtractKeys func(args []string)[]string
 	Execute func(args []string, getShard func(k string) *core.Shard) interface{}
 }
 
-func DispatchBaseCommand(db core.RedisDB ,commandName string, args []string) interface{} {
-    cmd, exists := Registry[commandName]
-    if !exists {
-        return nil
-    }
+func DispatchBaseCommand(db core.RedisDB ,cmd BaseCommand, args []string) interface{} {
 
     if len(args) < cmd.MinArgs {
         return errors.New("wrong number of arguments")
@@ -50,7 +48,23 @@ func DispatchBaseCommand(db core.RedisDB ,commandName string, args []string) int
     return cmd.Execute(args, db.GetShard)
 }
 
-var Registry = map[string]CommandDef{
+type PubSubCommand struct{
+    MinArgs int
+    Execute func(hub *pubsub.Hub, conn net.Conn, args []string) interface{}
+}
+
+func DispatchPubSub(conn net.Conn, h *pubsub.Hub, cmd PubSubCommand, args []string) {
+	if len(args) < cmd.MinArgs {
+		conn.Write(core.SerializeRESP(errors.New("ERR wrong number of arguments")))
+		return
+	}
+	if res := cmd.Execute(h, conn, args); res != nil {
+		conn.Write(core.SerializeRESP(res))
+	}
+}
+
+
+var BaseRegistry = map[string]BaseCommand{
     "GET": {
         MinArgs:     2,
         ExtractKeys: SingleKey,
@@ -171,4 +185,10 @@ var Registry = map[string]CommandDef{
         ExtractKeys: NoKeys,
         Execute:     Ping,
     },
+}
+
+var PubSubRegistry = map[string]PubSubCommand{
+	"SUBSCRIBE":   {MinArgs: 2, Execute: Subscribe},
+	"UNSUBSCRIBE": {MinArgs: 1, Execute: Unsubscribe},
+	"PUBLISH":     {MinArgs: 3, Execute: Publish},
 }
\ No newline at end of file
diff --git a/pkg/commands/publish.go b/pkg/commands/publish.go
new file mode 100644
index 0000000..5959d57
--- /dev/null
+++ b/pkg/commands/publish.go
@@ -0,0 +1,12 @@
+package commands
+
+import (
+	"net"
+	"redisClone/pkg/pubsub"
+)
+
+func Publish(hub *pubsub.Hub, conn net.Conn, args []string) interface{}{
+	channel := args[1]
+	msg := args[2]
+	return hub.Publish(channel, msg)
+}
\ No newline at end of file
diff --git a/pkg/commands/subscribe.go b/pkg/commands/subscribe.go
new file mode 100644
index 0000000..06dee13
--- /dev/null
+++ b/pkg/commands/subscribe.go
@@ -0,0 +1,13 @@
+package commands
+
+import (
+	"net"
+	"redisClone/pkg/pubsub"
+)
+
+func Subscribe(hub *pubsub.Hub, conn net.Conn, args []string) interface{}{
+	for _, channel := range args[1:]{
+		hub.Subscribe(conn,channel)
+	}
+	return nil
+}
\ No newline at end of file
diff --git a/pkg/commands/unsubscribe.go b/pkg/commands/unsubscribe.go
new file mode 100644
index 0000000..9ca502d
--- /dev/null
+++ b/pkg/commands/unsubscribe.go
@@ -0,0 +1,19 @@
+package commands
+
+import (
+	"net"
+	"redisClone/pkg/core"
+	"redisClone/pkg/pubsub"
+)
+
+func Unsubscribe(hub *pubsub.Hub, conn net.Conn, args []string) interface{}{
+	channels := args[1:]
+	results := hub.Unsubscribe(conn, channels)
+
+	for channel, remainingCount := range results {
+		resp := core.SerializeRESP([]interface{}{"unsubscribe", channel, remainingCount})
+		conn.Write(resp)
+	}
+
+	return nil
+}
\ No newline at end of file