feat(Server):support Verbatim strings resp type, using it for CLIENT LIST and INFO commands (#2264)

fixes #2242 #2253

Reference:
The definition of Redis verbatim strings: https://redis.io/docs/reference/protocol-spec/#verbatim-strings

"For example, the Redis command INFO outputs a report that includes newlines. When using RESP3, redis-cli displays it correctly because it is sent as a Verbatim String reply (with its three bytes being "txt"). When using RESP2, however, the redis-cli is hard-coded to look for the INFO command to ensure its correct display to the user."

---------

Signed-off-by: Yue Li <61070669+theyueli@users.noreply.github.com>

src/facade/reply_builder.cc

@@ -328,6 +328,28 @@ void RedisReplyBuilder::SendBulkString(std::string_view str) {
   return Send(v, ABSL_ARRAYSIZE(v));
 }
 
+void RedisReplyBuilder::SendVerbatimString(std::string_view str, VerbatimFormat format) {
+  if (!is_resp3_)
+    return SendBulkString(str);
+
+  char tmp[absl::numbers_internal::kFastToBufferSize + 7];
+  tmp[0] = '=';
+  // + 4 because format is three byte, and need to be followed by a ":"
+  char* next = absl::numbers_internal::FastIntToBuffer(uint32_t(str.size() + 4), tmp + 1);
+  *next++ = '\r';
+  *next++ = '\n';
+
+  DCHECK(format <= VerbatimFormat::MARKDOWN);
+  if (format == VerbatimFormat::TXT)
+    strcpy(next, "txt:");
+  else if (format == VerbatimFormat::MARKDOWN)
+    strcpy(next, "mkd:");
+  next += 4;
+  std::string_view lenpref{tmp, size_t(next - tmp)};
+  iovec v[3] = {IoVec(lenpref), IoVec(str), IoVec(kCRLF)};
+  return Send(v, ABSL_ARRAYSIZE(v));
+}
+
 void RedisReplyBuilder::SendLong(long num) {
   string str = absl::StrCat(":", num, kCRLF);
   SendRaw(str);

src/facade/reply_builder.h

@@ -212,6 +212,8 @@ class RedisReplyBuilder : public SinkReplyBuilder {
  public:
   enum CollectionType { ARRAY, SET, MAP, PUSH };
 
+  enum VerbatimFormat { TXT, MARKDOWN };
+
   using StrSpan = std::variant<absl::Span<const std::string>, absl::Span<const std::string_view>>;
 
   RedisReplyBuilder(::io::Sink* stream);
@@ -238,6 +240,7 @@ class RedisReplyBuilder : public SinkReplyBuilder {
   void SendSimpleString(std::string_view str) override;
 
   virtual void SendBulkString(std::string_view str);
+  virtual void SendVerbatimString(std::string_view str, VerbatimFormat format = TXT);
   virtual void SendScoredArray(const std::vector<std::pair<std::string, double>>& arr,
                                bool with_scores);
 

src/facade/reply_builder_test.cc

@@ -913,6 +913,26 @@ TEST_F(RedisReplyBuilderTest, FormatDouble) {
   EXPECT_STREQ("1e-23", format(1e-23));
 }
 
+TEST_F(RedisReplyBuilderTest, VerbatimString) {
+  // test resp3
+  std::string str = "A simple string!";
+
+  builder_->SetResp3(true);
+  builder_->SendVerbatimString(str, RedisReplyBuilder::VerbatimFormat::TXT);
+  ASSERT_TRUE(builder_->err_count().empty());
+  ASSERT_EQ(TakePayload(), "=20\r\ntxt:A simple string!\r\n") << "Resp3 VerbatimString TXT failed.";
+
+  builder_->SetResp3(true);
+  builder_->SendVerbatimString(str, RedisReplyBuilder::VerbatimFormat::MARKDOWN);
+  ASSERT_TRUE(builder_->err_count().empty());
+  ASSERT_EQ(TakePayload(), "=20\r\nmkd:A simple string!\r\n") << "Resp3 VerbatimString TXT failed.";
+
+  builder_->SetResp3(false);
+  builder_->SendVerbatimString(str);
+  ASSERT_TRUE(builder_->err_count().empty());
+  ASSERT_EQ(TakePayload(), "$16\r\nA simple string!\r\n") << "Resp3 VerbatimString TXT failed.";
+}
+
 static void BM_FormatDouble(benchmark::State& state) {
   vector<double> values;
   char buf[64];

src/server/server_family.cc

@@ -1277,7 +1277,7 @@ void ServerFamily::ClientList(CmdArgList args, ConnectionContext* cntx) {
   string result = absl::StrJoin(client_info, "\n");
   result.append("\n");
   auto* rb = static_cast<RedisReplyBuilder*>(cntx->reply_builder());
-  return rb->SendBulkString(result);
+  return rb->SendVerbatimString(result);
 }
 
 void ServerFamily::ClientPause(CmdArgList args, ConnectionContext* cntx) {
@@ -1827,7 +1827,7 @@ void ServerFamily::Info(CmdArgList args, ConnectionContext* cntx) {
     append("cluster_enabled", ClusterConfig::IsEnabledOrEmulated());
   }
   auto* rb = static_cast<RedisReplyBuilder*>(cntx->reply_builder());
-  rb->SendBulkString(info);
+  rb->SendVerbatimString(info);
 }
 
 void ServerFamily::Hello(CmdArgList args, ConnectionContext* cntx) {