Fix many issues around impl. of grpc

example/grpc_c++/client.cpp

@@ -21,7 +21,7 @@
 #include <brpc/channel.h>
 #include "helloworld.pb.h"
 
-DEFINE_string(protocol, "grpc", "Protocol type. Defined in src/brpc/options.proto");
+DEFINE_string(protocol, "h2c", "Protocol type. Defined in src/brpc/options.proto");
 DEFINE_string(server, "0.0.0.0:50051", "IP Address of server");
 DEFINE_string(load_balancer, "", "The algorithm for load balancing");
 DEFINE_int32(timeout_ms, 100, "RPC timeout in milliseconds");
@@ -59,22 +59,21 @@ int main(int argc, char* argv[]) {
         helloworld::HelloReply response;
         brpc::Controller cntl;
 
-        request.set_name("grpc client example");
+        request.set_name("grpc_req_from_brpc");
+        cntl.http_request().set_content_type("application/grpc");
         if (FLAGS_gzip) {
             cntl.set_request_compress_type(brpc::COMPRESS_TYPE_GZIP);
         }
         // Because `done'(last parameter) is NULL, this function waits until
         // the response comes back or error occurs(including timedout).
         stub.SayHello(&cntl, &request, &response, NULL);
-        //cntl.http_request().uri() = FLAGS_server;
-        //channel.CallMethod(NULL, &cntl, &request, &response, NULL);
         if (!cntl.Failed()) {
             LOG(INFO) << "Received response from " << cntl.remote_side()
                 << " to " << cntl.local_side()
                 << ": " << response.message()
                 << " latency=" << cntl.latency_us() << "us";
         } else {
-            LOG(WARNING) << cntl.ErrorCode() << ": " << cntl.ErrorText();
+            LOG(WARNING) << cntl.ErrorText();
         }
         usleep(FLAGS_interval_ms * 1000L);
     }

example/grpc_c++/server.cpp

@@ -42,7 +42,6 @@ public:
         if (FLAGS_gzip) {
             cntl->set_response_compress_type(brpc::COMPRESS_TYPE_GZIP);
         }
-        LOG(INFO) << "req=" << req->name();
         res->set_message("Hello " + req->name());
         // If an error happens, use controller::set_grpc_error_code to set errors
         // e.g., cntl->set_grpc_error_code(brpc::GRPC_RESOURCEEXHAUSTED, "test grpc message");

src/brpc/details/http_message.cpp

@@ -225,12 +225,14 @@ int HttpMessage::OnBody(const char *at, const size_t length) {
             delete _vmsgbuilder;
             _vmsgbuilder = NULL;
         } else {
-            if (_body_length < (size_t)FLAGS_http_verbose_max_body_length) {
+            if (_vbodylen < (size_t)FLAGS_http_verbose_max_body_length) {
                 int plen = std::min(length, (size_t)FLAGS_http_verbose_max_body_length
-                                    - _body_length);
-                _vmsgbuilder->write(at, plen);
+                                    - _vbodylen);
+                std::string str = butil::ToPrintableString(
+                    at, plen, std::numeric_limits<size_t>::max());
+                _vmsgbuilder->write(str.data(), str.size());
             }
-            _body_length += length;
+            _vbodylen += length;
         }
     }
     if (_stage != HTTP_ON_BODY) {
@@ -280,8 +282,8 @@ int HttpMessage::OnBody(const char *at, const size_t length) {
 
 int HttpMessage::OnMessageComplete() {
     if (_vmsgbuilder) {
-        if (_body_length > (size_t)FLAGS_http_verbose_max_body_length) {
-            *_vmsgbuilder << "\n<skipped " << _body_length
+        if (_vbodylen > (size_t)FLAGS_http_verbose_max_body_length) {
+            *_vmsgbuilder << "\n<skipped " << _vbodylen
                 - (size_t)FLAGS_http_verbose_max_body_length << " bytes>";
         }
         std::cerr << _vmsgbuilder->buf() << std::endl;
@@ -396,7 +398,7 @@ HttpMessage::HttpMessage(bool read_body_progressively)
     , _body_reader(NULL)
     , _cur_value(NULL)
     , _vmsgbuilder(NULL)
-    , _body_length(0) {
+    , _vbodylen(0) {
     http_parser_init(&_parser, HTTP_BOTH);
     _parser.data = this;
 }
@@ -534,10 +536,10 @@ std::ostream& operator<<(std::ostream& os, const http_parser& parser) {
 //                | "CONNECT"                ; Section 9.9
 //                | extension-method
 // extension-method = token
-void SerializeHttpRequest(butil::IOBuf* request,
-                          HttpHeader* h,
-                          const butil::EndPoint& remote_side,
-                          const butil::IOBuf* content) {
+void MakeRawHttpRequest(butil::IOBuf* request,
+                        HttpHeader* h,
+                        const butil::EndPoint& remote_side,
+                        const butil::IOBuf* content) {
     butil::IOBufBuilder os;
     os << HttpMethod2Str(h->method()) << ' ';
     const URI& uri = h->uri();
@@ -611,9 +613,9 @@ void SerializeHttpRequest(butil::IOBuf* request,
 //                CRLF
 //                [ message-body ]          ; Section 7.2
 // Status-Line = HTTP-Version SP Status-Code SP Reason-Phrase CRLF
-void SerializeHttpResponse(butil::IOBuf* response,
-                           HttpHeader* h,
-                           butil::IOBuf* content) {
+void MakeRawHttpResponse(butil::IOBuf* response,
+                         HttpHeader* h,
+                         butil::IOBuf* content) {
     butil::IOBufBuilder os;
     os << "HTTP/" << h->major_version() << '.'
        << h->minor_version() << ' ' << h->status_code()

src/brpc/details/http_message.h

@@ -113,7 +113,7 @@ private:
 protected:
     // Only valid when -http_verbose is on
     butil::IOBufBuilder* _vmsgbuilder;
-    size_t _body_length;
+    size_t _vbodylen;
 };
 
 std::ostream& operator<<(std::ostream& os, const http_parser& parser);
@@ -122,17 +122,17 @@ std::ostream& operator<<(std::ostream& os, const http_parser& parser);
 // header: may be modified in some cases
 // remote_side: used when "Host" is absent
 // content: could be NULL.
-void SerializeHttpRequest(butil::IOBuf* request,
-                          HttpHeader* header,
-                          const butil::EndPoint& remote_side,
-                          const butil::IOBuf* content);
+void MakeRawHttpRequest(butil::IOBuf* request,
+                        HttpHeader* header,
+                        const butil::EndPoint& remote_side,
+                        const butil::IOBuf* content);
 
 // Serialize a http response.
 // header: may be modified in some cases
 // content: cleared after usage. could be NULL. 
-void SerializeHttpResponse(butil::IOBuf* response,
-                           HttpHeader* header,
-                           butil::IOBuf* content);
+void MakeRawHttpResponse(butil::IOBuf* response,
+                         HttpHeader* header,
+                         butil::IOBuf* content);
 
 } // namespace brpc
 

src/brpc/global.cpp

@@ -504,18 +504,6 @@ static void GlobalInitializeOrDieImpl() {
     }
 #endif
 
-    // grpc protocol is based on http2
-    Protocol grpc_protocol = { ParseH2Message,
-                               SerializeHttpRequest, PackH2Request,
-                               ProcessHttpRequest, ProcessHttpResponse,
-                               VerifyHttpRequest, ParseHttpServerAddress,
-                               GetHttpMethodName,
-                               CONNECTION_TYPE_SINGLE,
-                               "grpc" };
-    if (RegisterProtocol(PROTOCOL_GRPC, grpc_protocol) != 0) {
-        exit(1);
-    }
-
     // Only valid at client side
     Protocol ubrpc_compack_protocol = {
         ParseNsheadMessage,

src/brpc/grpc.cpp

@@ -24,6 +24,30 @@
 
 namespace brpc {
 
+const char* GrpcStatusToString(GrpcStatus s) {
+    switch (s) {
+        case GRPC_OK: return "GRPC_OK";
+        case GRPC_CANCELED: return "GRPC_CANCELED";
+        case GRPC_UNKNOWN: return "GRPC_UNKNOWN";
+        case GRPC_INVALIDARGUMENT: return "GRPC_INVALIDARGUMENT";
+        case GRPC_DEADLINEEXCEEDED: return "GRPC_DEADLINEEXCEEDED";
+        case GRPC_NOTFOUND: return "GRPC_NOTFOUND";
+        case GRPC_ALREADYEXISTS: return "GRPC_ALREADYEXISTS";
+        case GRPC_PERMISSIONDENIED: return "GRPC_PERMISSIONDENIED";
+        case GRPC_RESOURCEEXHAUSTED: return "GRPC_RESOURCEEXHAUSTED";
+        case GRPC_FAILEDPRECONDITION: return "GRPC_FAILEDPRECONDITION";
+        case GPRC_ABORTED: return "GPRC_ABORTED";
+        case GRPC_OUTOFRANGE: return "GRPC_OUTOFRANGE";
+        case GRPC_UNIMPLEMENTED: return "GRPC_UNIMPLEMENTED";
+        case GRPC_INTERNAL: return "GRPC_INTERNAL";
+        case GRPC_UNAVAILABLE: return "GRPC_UNAVAILABLE";
+        case GRPC_DATALOSS: return "GRPC_DATALOSS";
+        case GRPC_UNAUTHENTICATED: return "GRPC_UNAUTHENTICATED";
+        case GRPC_MAX: return "GRPC_MAX";
+    }
+    return "Unknown-GrpcStatus";
+}
+
 GrpcStatus ErrorCodeToGrpcStatus(int error_code) {
     switch (error_code) {
     case 0:

src/brpc/grpc.h

@@ -142,8 +142,11 @@ enum GrpcStatus {
     GRPC_MAX,
 }; 
 
-GrpcStatus ErrorCodeToGrpcStatus(int error_code);
+// Get description of the error.
+const char* GrpcStatusToString(GrpcStatus);
 
+// Convert between error code and grpc status with similar semantics
+GrpcStatus ErrorCodeToGrpcStatus(int error_code);
 int GrpcStatusToErrorCode(GrpcStatus grpc_status);
 
 void PercentEncode(const std::string& str, std::string* str_out);

src/brpc/http_header.h

@@ -154,8 +154,6 @@ friend void policy::ProcessHttpRequest(InputMessageBase *msg);
     std::string _content_type;
     std::string _unresolved_path;
     std::pair<int, int> _version;
-    int _h2_stream_id;
-    H2Error _h2_error;
 };
 
 const HttpHeader& DefaultHttpHeader();

src/brpc/options.proto

@@ -47,7 +47,6 @@ enum ProtocolType {
     PROTOCOL_CDS_AGENT = 24;           // Client side only
     PROTOCOL_ESP = 25;                 // Client side only
     PROTOCOL_HTTP2 = 26;
-    PROTOCOL_GRPC = 27;
 }
 
 enum CompressType {

src/brpc/policy/http2_rpc_protocol.cpp

@@ -1592,8 +1592,7 @@ size_t H2UnsentRequest::EstimatedByteSize() {
     return sz;
 }
 
-void H2UnsentRequest::Describe(butil::IOBuf* desc) const {
-    butil::IOBufBuilder os;
+void H2UnsentRequest::Print(std::ostream& os) const {
     os << "[ H2 REQUEST @" << butil::my_ip() << " ]\n";
     for (size_t i = 0; i < _size; ++i) {
         os << "> " << _list[i].name << " = " << _list[i].value << '\n';
@@ -1613,33 +1612,23 @@ void H2UnsentRequest::Describe(butil::IOBuf* desc) const {
     if (!body->empty()) {
         os << "> \n";
     }
-    os.move_to(*desc);
-    if (body->size() > (size_t)FLAGS_http_verbose_max_body_length) {
-        size_t nskipped = body->size() - (size_t)FLAGS_http_verbose_max_body_length;
-        body->append_to(desc, FLAGS_http_verbose_max_body_length);
-        if (nskipped) {
-            char str[48];
-            snprintf(str, sizeof(str), "\n<skipped %" PRIu64 " bytes>", nskipped);
-            desc->append(str);
-        }
-    } else {
-        desc->append(*body);
-    }
+    os << butil::BinaryPrinter(*body, FLAGS_http_verbose_max_body_length);
+
 }
 
-H2UnsentResponse::H2UnsentResponse(Controller* c, int stream_id, bool grpc)
+H2UnsentResponse::H2UnsentResponse(Controller* c, int stream_id, bool is_grpc)
     : _size(0)
     , _stream_id(stream_id)
     , _http_response(c->release_http_response())
-    , _grpc(grpc) {
+    , _is_grpc(is_grpc) {
     _data.swap(c->response_attachment());
-    if (grpc) {
+    if (is_grpc) {
         _grpc_status = ErrorCodeToGrpcStatus(c->ErrorCode());
         PercentEncode(c->ErrorText(), &_grpc_message);
     }
 }
 
-H2UnsentResponse* H2UnsentResponse::New(Controller* c, int stream_id, bool grpc) {
+H2UnsentResponse* H2UnsentResponse::New(Controller* c, int stream_id, bool is_grpc) {
     const HttpHeader* const h = &c->http_response();
     const CommonStrings* const common = get_common_strings();
     const bool need_content_length =
@@ -1650,7 +1639,7 @@ H2UnsentResponse* H2UnsentResponse::New(Controller* c, int stream_id, bool grpc)
         + (size_t)need_content_type;
     const size_t memsize = offsetof(H2UnsentResponse, _list) +
         sizeof(HPacker::Header) * maxsize;
-    H2UnsentResponse* msg = new (malloc(memsize)) H2UnsentResponse(c, stream_id, grpc);
+    H2UnsentResponse* msg = new (malloc(memsize)) H2UnsentResponse(c, stream_id, is_grpc);
     // :status
     if (h->status_code() == 200) {
         msg->push(common->H2_STATUS, common->STATUS_200);
@@ -1720,18 +1709,17 @@ H2UnsentResponse::AppendAndDestroySelf(butil::IOBuf* out, Socket* socket) {
     butil::IOBuf frag;
     appender.move_to(frag);
 
-    butil::IOBufAppender trailer_appender;
     butil::IOBuf trailer_frag;
-    if (_grpc) {
+    if (_is_grpc) {
         HPacker::Header status_header("grpc-status",
                                       butil::string_printf("%d", _grpc_status));
-        hpacker.Encode(&trailer_appender, status_header, options);
+        hpacker.Encode(&appender, status_header, options);
         if (!_grpc_message.empty()) {
             HPacker::Header msg_header("grpc-message", _grpc_message);
-            hpacker.Encode(&trailer_appender, msg_header, options);
+            hpacker.Encode(&appender, msg_header, options);
         }
+        appender.move_to(trailer_frag);
     }
-    trailer_appender.move_to(trailer_frag);
 
     PackH2Message(out, frag, trailer_frag, _data, _stream_id, ctx);
     return butil::Status::OK();
@@ -1752,8 +1740,7 @@ size_t H2UnsentResponse::EstimatedByteSize() {
     return sz;
 }
 
-void H2UnsentResponse::Describe(butil::IOBuf* desc) const {
-    butil::IOBufBuilder os;
+void H2UnsentResponse::Print(std::ostream& os) const {
     os << "[ H2 RESPONSE @" << butil::my_ip() << " ]\n";
     for (size_t i = 0; i < _size; ++i) {
         os << "> " << _list[i].name << " = " << _list[i].value << '\n';
@@ -1767,18 +1754,7 @@ void H2UnsentResponse::Describe(butil::IOBuf* desc) const {
     if (!_data.empty()) {
         os << "> \n";
     }
-    os.move_to(*desc);
-    if (_data.size() > (size_t)FLAGS_http_verbose_max_body_length) {
-        size_t nskipped = _data.size() - (size_t)FLAGS_http_verbose_max_body_length;
-        _data.append_to(desc, FLAGS_http_verbose_max_body_length);
-        if (nskipped) {
-            char str[48];
-            snprintf(str, sizeof(str), "\n<skipped %" PRIu64 " bytes>", nskipped);
-            desc->append(str);
-        }
-    } else {
-        desc->append(_data);
-    }
+    os << butil::BinaryPrinter(_data, FLAGS_http_verbose_max_body_length);
 }
 
 void PackH2Request(butil::IOBuf*,
@@ -1806,9 +1782,7 @@ void PackH2Request(butil::IOBuf*,
     *user_message = h2_req;
     
     if (FLAGS_http_verbose) {
-        butil::IOBuf desc;
-        h2_req->Describe(&desc);
-        std::cerr << desc << std::endl;
+        std::cerr << *h2_req << std::endl;
     }
 }
 

src/brpc/policy/http2_rpc_protocol.h

@@ -135,7 +135,7 @@ friend void PackH2Request(butil::IOBuf*, SocketMessage**,
                           Controller*, const butil::IOBuf&, const Authenticator*);
 public:
     static H2UnsentRequest* New(Controller* c);
-    void Describe(butil::IOBuf*) const;
+    void Print(std::ostream&) const;
 
     int AddRefManually()
     { return _nref.fetch_add(1, butil::memory_order_relaxed); }
@@ -194,9 +194,9 @@ private:
 
 class H2UnsentResponse : public SocketMessage {
 public:
-    static H2UnsentResponse* New(Controller* c, int stream_id, bool grpc);
+    static H2UnsentResponse* New(Controller* c, int stream_id, bool is_grpc);
     void Destroy();
-    void Describe(butil::IOBuf*) const;
+    void Print(std::ostream&) const;
     // @SocketMessage
     butil::Status AppendAndDestroySelf(butil::IOBuf* out, Socket*) override;
     size_t EstimatedByteSize() override;
@@ -208,7 +208,7 @@ private:
     void push(const std::string& name, const std::string& value)
     { new (&_list[_size++]) HPacker::Header(name, value); }
 
-    H2UnsentResponse(Controller* c, int stream_id, bool grpc);
+    H2UnsentResponse(Controller* c, int stream_id, bool is_grpc);
     ~H2UnsentResponse() {}
     H2UnsentResponse(const H2UnsentResponse&);
     void operator=(const H2UnsentResponse&);
@@ -218,7 +218,7 @@ private:
     uint32_t _stream_id;
     std::unique_ptr<HttpHeader> _http_response;
     butil::IOBuf _data;
-    bool _grpc;
+    bool _is_grpc;
     GrpcStatus _grpc_status;
     std::string _grpc_message;
     HPacker::Header _list[0];
@@ -409,6 +409,15 @@ inline bool H2Context::RunOutStreams() const {
     return (_last_client_stream_id > 0x7FFFFFFF);
 }
 
+inline std::ostream& operator<<(std::ostream& os, const H2UnsentRequest& req) {
+    req.Print(os);
+    return os;
+}
+inline std::ostream& operator<<(std::ostream& os, const H2UnsentResponse& res) {
+    res.Print(os);
+    return os;
+}
+
 }  // namespace policy
 } // namespace brpc
 

src/brpc/policy/http_rpc_protocol.h

@@ -41,7 +41,6 @@ struct CommonStrings {
     std::string ACCEPT_ENCODING;
     std::string CONTENT_ENCODING;
     std::string CONTENT_LENGTH;
-    std::string IDENTITY;
     std::string GZIP;
     std::string CONNECTION;
     std::string KEEP_ALIVE;
@@ -69,6 +68,7 @@ struct CommonStrings {
     std::string TRAILERS;
     std::string GRPC_ENCODING;
     std::string GRPC_ACCEPT_ENCODING;
+    std::string GRPC_ACCEPT_ENCODING_VALUE;
     std::string GRPC_STATUS;
     std::string GRPC_MESSAGE;
 
@@ -77,8 +77,8 @@ struct CommonStrings {
 
 // Used in UT.
 class HttpContext : public ReadableProgressiveAttachment
-                       , public InputMessageBase
-                       , public HttpMessage {
+                  , public InputMessageBase
+                  , public HttpMessage {
 public:
     HttpContext(bool read_body_progressively = false)
         : InputMessageBase()
@@ -138,13 +138,14 @@ enum HttpContentType {
     HTTP_CONTENT_OTHERS = 0,
     HTTP_CONTENT_JSON = 1,
     HTTP_CONTENT_PROTO = 2,
-    HTTP_CONTENT_GRPC = 3
 };
 
-HttpContentType ParseContentType(butil::StringPiece content_type);
+// Parse from the textual content type. One type may have more than one literals.
+// Returns a numerical type. *is_grpc_ct is set to true if the content-type is
+// set by gRPC.
+HttpContentType ParseContentType(butil::StringPiece content_type, bool* is_grpc_ct);
 
-}  // namespace policy
+} // namespace policy
 } // namespace brpc
 
-
 #endif // BRPC_POLICY_HTTP_RPC_PROTOCOL_H