Support various payload of baidu-std: json, proto-json and proto-text

Author: chenBright

src/brpc/compress.cpp

@@ -17,14 +17,16 @@
 
 
 #include "butil/logging.h"
+#include "json2pb/json_to_pb.h"
 #include "brpc/compress.h"
 #include "brpc/protocol.h"
 
-
 namespace brpc {
 
 static const int MAX_HANDLER_SIZE = 1024;
-static CompressHandler s_handler_map[MAX_HANDLER_SIZE] = { { NULL, NULL, NULL } };
+static CompressHandler s_handler_map[MAX_HANDLER_SIZE] = {
+    { NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL }
+};
 
 int RegisterCompressHandler(CompressType type, 
                             CompressHandler handler) {
@@ -47,7 +49,7 @@ int RegisterCompressHandler(CompressType type,
 
 // Find CompressHandler by type.
 // Returns NULL if not found
-inline const CompressHandler* FindCompressHandler(CompressType type) {
+const CompressHandler* FindCompressHandler(CompressType type) {
     int index = type;
     if (index < 0 || index >= MAX_HANDLER_SIZE) {
         LOG(ERROR) << "CompressType=" << type << " is out of range";

src/brpc/compress.h

@@ -21,6 +21,8 @@
 
 #include <google/protobuf/message.h>              // Message
 #include "butil/iobuf.h"                           // butil::IOBuf
+#include "json2pb/pb_to_json.h"
+#include "json2pb/json_to_pb.h"
 #include "brpc/options.pb.h"                     // CompressType
 
 namespace brpc {
@@ -29,10 +31,20 @@ struct CompressHandler {
     // Compress serialized `msg' into `buf'.
     // Returns true on success, false otherwise
     bool (*Compress)(const google::protobuf::Message& msg, butil::IOBuf* buf);
+    bool (*Compress2Json)(const google::protobuf::Message& msg, butil::IOBuf* buf,
+                          const json2pb::Pb2JsonOptions& options);
+    bool (*Compress2ProtoJson)(const google::protobuf::Message& msg, butil::IOBuf* buf,
+                               const json2pb::Pb2ProtoJsonOptions& options);
+    bool (*Compress2ProtoText)(const google::protobuf::Message& msg, butil::IOBuf* buf);
 
     // Parse decompressed `data' as `msg'.
     // Returns true on success, false otherwise
     bool (*Decompress)(const butil::IOBuf& data, google::protobuf::Message* msg);
+    bool (*DecompressFromJson)(const butil::IOBuf& data, google::protobuf::Message* msg,
+                               const json2pb::Json2PbOptions& options);
+    bool (*DecompressFromProtoJson)(const butil::IOBuf& data, google::protobuf::Message* msg,
+                                    const json2pb::ProtoJson2PbOptions& options);
+    bool (*DecompressFromProtoText)(const butil::IOBuf& data, google::protobuf::Message* msg);
 
     // Name of the compression algorithm, must be string constant.
     const char* name;
@@ -42,6 +54,9 @@ struct CompressHandler {
 // Returns 0 on success, -1 otherwise
 int RegisterCompressHandler(CompressType type, CompressHandler handler);
 
+// Returns CompressHandler pointer of `type' if registered, NULL otherwise.
+const CompressHandler* FindCompressHandler(CompressType type);
+
 // Returns the `name' of the CompressType if registered
 const char* CompressTypeToCStr(CompressType type);
 

src/brpc/controller.cpp

@@ -290,6 +290,8 @@ void Controller::ResetPods() {
     _http_response = NULL;
     _request_user_fields = NULL;
     _response_user_fields = NULL;
+    _request_content_type = CONTENT_TYPE_PB;
+    _response_content_type = CONTENT_TYPE_PB;
     _request_streams.clear();
     _response_streams.clear();
     _remote_stream_settings = NULL;

src/brpc/controller.h

@@ -616,6 +616,20 @@ friend void policy::ProcessThriftRequest(InputMessageBase*);
 
     void CallAfterRpcResp(const google::protobuf::Message* req, const google::protobuf::Message* res);
 
+    void set_request_content_type(ContentType type) {
+        _request_content_type = type;
+    }
+    ContentType request_content_type() const {
+        return _request_content_type;
+    }
+
+    void set_response_content_type(ContentType type) {
+        _response_content_type = type;
+    }
+    ContentType response_content_type() const {
+        return _response_content_type;
+    }
+
 private:
     struct CompletionInfo {
         CallId id;           // call_id of the corresponding request
@@ -859,6 +873,11 @@ friend void policy::ProcessThriftRequest(InputMessageBase*);
     butil::IOBuf _request_attachment;
     butil::IOBuf _response_attachment;
 
+    // Only SerializedRequest supports `_request_content_type'.
+    ContentType _request_content_type;
+    // Only SerializedResponse supports `_response_content_type'.
+    ContentType _response_content_type;
+
     // Writable progressive attachment
     butil::intrusive_ptr<ProgressiveAttachment> _wpa;
     // Readable progressive attachment

src/brpc/global.cpp

@@ -388,25 +388,34 @@ static void GlobalInitializeOrDieImpl() {
     LoadBalancerExtension()->RegisterOrDie("_dynpart", &g_ext->dynpart_lb);
 
     // Compress Handlers
-    const CompressHandler gzip_compress =
-        { GzipCompress, GzipDecompress, "gzip" };
+    CompressHandler gzip_compress = {
+        GzipCompress, GzipCompress2Json, GzipCompress2ProtoJson,
+        GzipCompress2ProtoText, GzipDecompress, GzipDecompressFromJson,
+        GzipDecompressFromProtoJson, GzipDecompressFromProtoText, "gzip"
+    };
     if (RegisterCompressHandler(COMPRESS_TYPE_GZIP, gzip_compress) != 0) {
         exit(1);
     }
-    const CompressHandler zlib_compress =
-        { ZlibCompress, ZlibDecompress, "zlib" };
+    CompressHandler zlib_compress = {
+        ZlibCompress, ZlibCompress2Json, ZlibCompress2ProtoJson,
+        ZlibCompress2ProtoText, ZlibDecompress, ZlibDecompressFromJson,
+        ZlibDecompressFromProtoJson, ZlibDecompressFromProtoText, "zlib"
+    };
     if (RegisterCompressHandler(COMPRESS_TYPE_ZLIB, zlib_compress) != 0) {
         exit(1);
     }
-    const CompressHandler snappy_compress =
-        { SnappyCompress, SnappyDecompress, "snappy" };
+    CompressHandler snappy_compress = {
+        SnappyCompress, SnappyCompress2Json, SnappyCompress2ProtoJson,
+        SnappyCompress2ProtoText, SnappyDecompress, SnappyDecompressFromJson,
+        SnappyDecompressFromProtoJson, SnappyDecompressFromProtoText, "snappy"
+    };
     if (RegisterCompressHandler(COMPRESS_TYPE_SNAPPY, snappy_compress) != 0) {
         exit(1);
     }
 
     // Protocols
     Protocol baidu_protocol = { ParseRpcMessage,
-                                SerializeRequestDefault, PackRpcRequest,
+                                SerializeRpcRequest, PackRpcRequest,
                                 ProcessRpcRequest, ProcessRpcResponse,
                                 VerifyRpcRequest, NULL, NULL,
                                 CONNECTION_TYPE_ALL, "baidu_std" };

src/brpc/options.proto

@@ -74,6 +74,13 @@ enum CompressType {
     COMPRESS_TYPE_LZ4 = 4;
 }
 
+enum ContentType {
+    CONTENT_TYPE_PB = 0;
+    CONTENT_TYPE_JSON = 1;
+    CONTENT_TYPE_PROTO_JSON = 2;
+    CONTENT_TYPE_PROTO_TEXT = 3;
+}
+
 message ChunkInfo {
     required int64 stream_id = 1;
     required int64 chunk_id = 2;

src/brpc/policy/baidu_rpc_meta.proto

@@ -33,6 +33,7 @@ message RpcMeta {
     optional bytes authentication_data = 7;
     optional StreamSettings stream_settings = 8;
     map<string, string> user_fields = 9;
+    optional ContentType content_type = 10;
 }
 
 message RpcRequestMeta {