feat(cmds): Add support for CLIENT REPLY subcommand and related tests

src/commands/cmd_server.cc

@@ -392,7 +392,7 @@ class CommandClient : public Commander {
  public:
   Status Parse(const std::vector<std::string> &args) override {
     subcommand_ = util::ToLower(args[1]);
-    // subcommand: getname id kill list info setname
+    // subcommand: getname id kill list info setname reply
     if ((subcommand_ == "id" || subcommand_ == "getname" || subcommand_ == "list" || subcommand_ == "info") &&
         args.size() == 2) {
       return Status::OK();
@@ -412,6 +412,17 @@ class CommandClient : public Commander {
       return Status::OK();
     }
 
+    if (subcommand_ == "reply") {
+      if (args.size() != 3) {
+        return {Status::RedisParseErr, errInvalidSyntax};
+      }
+      reply_mode_arg_ = util::ToLower(args[2]);
+      if (reply_mode_arg_ != "on" && reply_mode_arg_ != "off" && reply_mode_arg_ != "skip") {
+        return {Status::RedisParseErr, errInvalidSyntax};
+      }
+      return Status::OK();
+    }
+
     if ((subcommand_ == "kill")) {
       if (args.size() == 2) {
         return {Status::RedisParseErr, errInvalidSyntax};
@@ -464,7 +475,7 @@ class CommandClient : public Commander {
       }
       return Status::OK();
     }
-    return {Status::RedisInvalidCmd, "Syntax error, try CLIENT LIST|INFO|KILL ip:port|GETNAME|SETNAME"};
+    return {Status::RedisInvalidCmd, "Syntax error, try CLIENT LIST|INFO|KILL ip:port|GETNAME|SETNAME|REPLY"};
   }
 
   Status Execute([[maybe_unused]] engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override {
@@ -497,15 +508,26 @@ class CommandClient : public Commander {
           *output = redis::RESP_OK;
       }
       return Status::OK();
+    } else if (subcommand_ == "reply") {
+      if (reply_mode_arg_ == "on") {
+        conn->SetReplyMode(Connection::ReplyMode::ON);
+      } else if (reply_mode_arg_ == "off") {
+        conn->SetReplyMode(Connection::ReplyMode::OFF);
+      } else if (reply_mode_arg_ == "skip") {
+        conn->SetReplyMode(Connection::ReplyMode::SKIP_ONCE_PENDING);
+      }
+      *output = redis::RESP_OK;
+      return Status::OK();
     }
 
-    return {Status::RedisInvalidCmd, "Syntax error, try CLIENT LIST|INFO|KILL ip:port|GETNAME|SETNAME"};
+    return {Status::RedisInvalidCmd, "Syntax error, try CLIENT LIST|INFO|KILL ip:port|GETNAME|SETNAME|REPLY"};
   }
 
  private:
   std::string addr_;
   std::string conn_name_;
   std::string subcommand_;
+  std::string reply_mode_arg_;
   bool skipme_ = false;
   int64_t kill_type_ = 0;
   uint64_t id_ = 0;

src/server/redis_connection.cc

@@ -132,6 +132,17 @@ void Connection::OnEvent(bufferevent *bev, int16_t events) {
 }
 
 void Connection::Reply(const std::string &msg) {
+  if (reply_mode_ == ReplyMode::SKIP_ONCE_PENDING) {
+    reply_mode_ = ReplyMode::SKIP_ONCE_ACTIVE;
+    return;
+  }
+  if (reply_mode_ == ReplyMode::SKIP_ONCE_ACTIVE) {
+    reply_mode_ = ReplyMode::ON;
+    return;
+  }
+  if (reply_mode_ == ReplyMode::OFF) {
+    return;
+  }
   owner_->srv->stats.IncrOutboundBytes(msg.size());
   redis::Reply(bufferevent_get_output(bev_), msg);
 }

src/server/redis_connection.h

@@ -50,6 +50,13 @@ class Connection : public EvbufCallbackBase<Connection> {
     kAsking = 1 << 10,
   };
 
+  enum class ReplyMode {
+    ON,                 // Always reply to every command (default)
+    OFF,                // Never reply to any command
+    SKIP_ONCE_PENDING,  // The next command will NOT send a reply, then switch to SKIP_ONCE_ACTIVE
+    SKIP_ONCE_ACTIVE    // This command does NOT send a reply, then automatically switch back to ON
+  };
+
   explicit Connection(bufferevent *bev, Worker *owner);
   ~Connection();
 
@@ -181,6 +188,10 @@ class Connection : public EvbufCallbackBase<Connection> {
   std::set<std::string> watched_keys;
   std::atomic<bool> watched_keys_modified = false;
 
+  // Reply mode getter/setter
+  void SetReplyMode(ReplyMode mode) { reply_mode_ = mode; }
+  ReplyMode GetReplyMode() const { return reply_mode_; }
+
  private:
   uint64_t id_ = 0;
   std::atomic<int> flags_ = 0;
@@ -215,6 +226,8 @@ class Connection : public EvbufCallbackBase<Connection> {
 
   bool importing_ = false;
   RESP protocol_version_ = RESP::v2;
+
+  ReplyMode reply_mode_ = ReplyMode::ON;
 };
 
 }  // namespace redis

tests/gocase/unit/introspection/introspection_test.go

@@ -265,6 +265,40 @@ func TestIntrospection(t *testing.T) {
 		require.NoError(t, rdb.Do(ctx, "SET", "key", "value").Err())
 		require.EqualValues(t, 1, rdb.Do(ctx, "MOVE", "key", "0").Val())
 	})
+
+	// Test CLIENT REPLY subcommand behaviors
+	t.Run("CLIENT REPLY mode switching", func(t *testing.T) {
+		c := srv.NewTCPClient()
+		defer func() { require.NoError(t, c.Close()) }()
+
+		// Should reply by default
+		require.NoError(t, c.WriteArgs("ECHO", "default"))
+		c.MustReadBulkString(t, "default")
+
+		// Set to OFF, following commands should not reply
+		require.NoError(t, c.WriteArgs("CLIENT", "REPLY", "OFF"))
+		require.NoError(t, c.WriteArgs("ECHO", "off"))
+		// No reply expected here, do not read
+
+		// Set back to ON, commands should reply again
+		require.NoError(t, c.WriteArgs("CLIENT", "REPLY", "ON"))
+		c.MustRead(t, "+OK")
+		require.NoError(t, c.WriteArgs("ECHO", "on"))
+		c.MustReadBulkString(t, "on")
+
+		// Set to SKIP, next command should not reply, then reply resumes
+		require.NoError(t, c.WriteArgs("CLIENT", "REPLY", "SKIP"))
+		// No reply expected here, do not read
+
+		require.NoError(t, c.WriteArgs("ECHO", "skip1"))
+		// No reply expected here, do not read
+
+		require.NoError(t, c.WriteArgs("ECHO", "skip2"))
+		c.MustReadBulkString(t, "skip2")
+
+		require.NoError(t, c.WriteArgs("ECHO", "skip3"))
+		c.MustReadBulkString(t, "skip3")
+	})
 }
 
 func TestMultiServerIntrospection(t *testing.T) {

tests/gocase/util/tcp_client.go

@@ -87,6 +87,16 @@ func (c *TCPClient) MustReadStrings(t testing.TB, s []string) {
 	}
 }
 
+func (c *TCPClient) MustReadBulkString(t testing.TB, s string) {
+	r, err := c.ReadLine()
+	require.NoError(t, err)
+	require.Equal(t, "$"+strconv.Itoa(len(s)), r)
+
+	r, err = c.ReadLine()
+	require.NoError(t, err)
+	require.Equal(t, s, r)
+}
+
 func (c *TCPClient) MustReadStringsWithKey(t testing.TB, key string, s []string) {
 	r, err := c.ReadLine()
 	require.NoError(t, err)