From 02466c45a6ccf7efcd04f7a94f6fc07e51a41e1a Mon Sep 17 00:00:00 2001
From: Jyotinder <jyotindrsingh@gmail.com>
Date: Mon, 21 Oct 2024 15:43:16 +0530
Subject: [PATCH] allow reactivity features to be toggled with flag

---
 config/config.go               |  2 ++
 dice.toml                      |  2 +-
 internal/server/resp/server.go | 13 ++++++++-----
 main.go                        | 12 ++++++++++--
 4 files changed, 21 insertions(+), 8 deletions(-)

diff --git a/config/config.go b/config/config.go
index 4291408b6..ae13c386d 100644
--- a/config/config.go
+++ b/config/config.go
@@ -52,6 +52,8 @@ var (
 	KeysLimit = DefaultKeysLimit
 
 	EnableProfiling = false
+
+	EnableWatch = true
 )
 
 type Config struct {
diff --git a/dice.toml b/dice.toml
index 337170cfa..be63922b7 100644
--- a/dice.toml
+++ b/dice.toml
@@ -24,7 +24,7 @@ Enabled = true
 Port = 8379
 
 [Performance]
-WatchChanBufSize = 20000
+WatchChanBufSize = 20000000
 ShardCronFrequency = 1000000000
 MultiplexerPollTimeout = 100000000
 MaxClients = 20000
diff --git a/internal/server/resp/server.go b/internal/server/resp/server.go
index 145337476..fd0a76583 100644
--- a/internal/server/resp/server.go
+++ b/internal/server/resp/server.go
@@ -75,12 +75,15 @@ func (s *Server) Run(ctx context.Context) (err error) {
 	errChan := make(chan error, 1)
 	wg := &sync.WaitGroup{}
 
-	wg.Add(2)
-	go func() {
-		defer wg.Done()
-		s.watchManager.Run(ctx, s.cmdWatchChan)
-	}()
+	if s.cmdWatchChan != nil {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			s.watchManager.Run(ctx, s.cmdWatchChan)
+		}()
+	}
 
+	wg.Add(1)
 	go func(wg *sync.WaitGroup) {
 		defer wg.Done()
 		if err := s.AcceptConnectionRequests(ctx, wg); err != nil {
diff --git a/main.go b/main.go
index 893513591..f997be6c4 100644
--- a/main.go
+++ b/main.go
@@ -40,6 +40,8 @@ func init() {
 		"This flag controls the number of keys each shard holds at startup. You can multiply this number with the "+
 		"total number of shard threads to estimate how much memory will be required at system start up.")
 	flag.BoolVar(&config.EnableProfiling, "enable-profiling", false, "enable profiling for the dicedb server")
+	flag.BoolVar(&config.EnableWatch, "enable-watch", false, "enable reactivity features which power the .WATCH commands")
+
 	flag.Parse()
 
 	config.SetupConfig()
@@ -60,8 +62,14 @@ func main() {
 	sigs := make(chan os.Signal, 1)
 	signal.Notify(sigs, syscall.SIGTERM, syscall.SIGINT)
 
-	queryWatchChan := make(chan dstore.QueryWatchEvent, config.DiceConfig.Performance.WatchChanBufSize)
-	cmdWatchChan := make(chan dstore.CmdWatchEvent, config.DiceConfig.Memory.KeysLimit)
+	var queryWatchChan chan dstore.QueryWatchEvent = nil
+	var cmdWatchChan chan dstore.CmdWatchEvent = nil
+
+	if config.EnableWatch {
+		queryWatchChan = make(chan dstore.QueryWatchEvent, config.DiceConfig.Performance.WatchChanBufSize)
+		cmdWatchChan = make(chan dstore.CmdWatchEvent, config.DiceConfig.Performance.WatchChanBufSize)
+	}
+
 	var serverErrCh chan error
 
 	// Get the number of available CPU cores on the machine using runtime.NumCPU().