added pubsub command registrycommand-registry

4 files changed, 71 insertions, 7 deletions

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