add inline redis protocol support

src/brpc/redis_command.cpp

@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+#include <cctype>
 #include <limits>
 
 #include "butil/logging.h"
@@ -373,13 +374,65 @@ size_t RedisCommandParser::ParsedArgsSize() {
 ParseError RedisCommandParser::Consume(butil::IOBuf& buf,
                                        std::vector<butil::StringPiece>* args,
                                        butil::Arena* arena) {
-    const char* pfc = (const char*)buf.fetch1();
+    const auto pfc = static_cast<const char *>(buf.fetch1());
     if (pfc == NULL) {
         return PARSE_ERROR_NOT_ENOUGH_DATA;
     }
     // '*' stands for array "*<size>\r\n<sub-reply1><sub-reply2>..."
     if (!_parsing_array && *pfc != '*') {
-        return PARSE_ERROR_TRY_OTHERS;
+        if (!std::isalpha(static_cast<unsigned char>(*pfc))) {
+            return PARSE_ERROR_TRY_OTHERS;
+        }
+        const size_t buf_size = buf.size();
+        const auto copy_str = static_cast<char *>(arena->allocate(buf_size + 1));
+        buf.copy_to(copy_str, buf_size);
+        if (*copy_str == ' ') {
+            return PARSE_ERROR_ABSOLUTELY_WRONG;
+        }
+        copy_str[buf_size] = '\0';
+        const size_t crlf_pos = butil::StringPiece(copy_str, buf_size).find("\r\n");
+        if (crlf_pos == butil::StringPiece::npos) {  // not enough data
+            return PARSE_ERROR_NOT_ENOUGH_DATA;
+        }
+        args->clear();
+        size_t offset = 0;
+        while (offset < crlf_pos && copy_str[offset] != ' ') {
+            ++offset;
+        }
+        const auto first_arg = static_cast<char*>(arena->allocate(offset));
+        memcpy(first_arg, copy_str, offset);
+        for (size_t i = 0; i < offset; ++i) {
+            first_arg[i] = tolower(first_arg[i]);
+        }
+        args->push_back(butil::StringPiece(first_arg, offset));
+        if (offset == crlf_pos) {
+            // only one argument, directly return
+            buf.pop_front(crlf_pos + 2);
+            return PARSE_OK;
+        }
+        size_t arg_start_pos = ++offset;
+
+        for (; offset < crlf_pos; ++offset) {
+            if (copy_str[offset] != ' ') {
+                continue;
+            }
+            const auto arg_length = offset - arg_start_pos;
+            const auto arg = static_cast<char *>(arena->allocate(arg_length));
+            memcpy(arg, copy_str + arg_start_pos, arg_length);
+            args->push_back(butil::StringPiece(arg, arg_length));
+            arg_start_pos = ++offset;
+        }
+
+        if (arg_start_pos < crlf_pos) {
+            // process the last argument
+            const auto arg_length = crlf_pos - arg_start_pos;
+            const auto arg = static_cast<char *>(arena->allocate(arg_length));
+            memcpy(arg, copy_str + arg_start_pos, arg_length);
+            args->push_back(butil::StringPiece(arg, arg_length));
+        }
+
+        buf.pop_front(crlf_pos + 2);
+        return PARSE_OK;
     }
     // '$' stands for bulk string "$<length>\r\n<string>\r\n"
     if (_parsing_array && *pfc != '$') {

test/brpc_redis_unittest.cpp

@@ -580,7 +580,7 @@ TEST_F(RedisTest, command_parser) {
         ASSERT_EQ(command, GetCompleteCommand(command_out));
     }
     {
-        // simulate parsing from network
+        // simulate parsing from network following RESP
         int t = 100;
         std::string raw_string("*3\r\n$3\r\nset\r\n$3\r\nabc\r\n$3\r\ndef\r\n");
         int size = raw_string.size();
@@ -600,6 +600,27 @@ TEST_F(RedisTest, command_parser) {
             ASSERT_EQ(GetCompleteCommand(command_out), "set abc def");
         }
     }
+    {
+        // simulate parsing from network under inline protocol
+        int t = 100;
+        std::string raw_string("set abc def\r\n");
+        int size = raw_string.size();
+        while (t--) {
+            for (int i = 0; i < size; ++i) {
+                buf.push_back(raw_string[i]);
+                if (i == size - 1) {
+                    ASSERT_EQ(brpc::PARSE_OK, parser.Consume(buf, &command_out, &arena));
+                } else {
+                    if (butil::fast_rand_less_than(2) == 0) {
+                        ASSERT_EQ(brpc::PARSE_ERROR_NOT_ENOUGH_DATA,
+                                parser.Consume(buf, &command_out, &arena));
+                    }
+                }
+            }
+            ASSERT_TRUE(buf.empty());
+            ASSERT_EQ(GetCompleteCommand(command_out), "set abc def");
+        }
+    }
     {
         // there is a non-string message in command and parse should fail
         buf.append("*3\r\n$3");