Commit eeb07057 authored by zhujiashun's avatar zhujiashun

* Do not expose grpc_status and grpc_message to user

* Delete trailer map
* Convert H2Error and grpc error to ErrorCode
parent 0a000b46
...@@ -22,7 +22,7 @@ ...@@ -22,7 +22,7 @@
#include "helloworld.pb.h" #include "helloworld.pb.h"
DEFINE_string(protocol, "grpc", "Protocol type. Defined in src/brpc/options.proto"); DEFINE_string(protocol, "grpc", "Protocol type. Defined in src/brpc/options.proto");
DEFINE_string(server, "0.0.0.0:8000", "IP Address of server"); DEFINE_string(server, "0.0.0.0:50051", "IP Address of server");
DEFINE_string(load_balancer, "", "The algorithm for load balancing"); DEFINE_string(load_balancer, "", "The algorithm for load balancing");
DEFINE_int32(timeout_ms, 100, "RPC timeout in milliseconds"); DEFINE_int32(timeout_ms, 100, "RPC timeout in milliseconds");
DEFINE_int32(max_retry, 3, "Max retries(not including the first RPC)"); DEFINE_int32(max_retry, 3, "Max retries(not including the first RPC)");
...@@ -52,7 +52,6 @@ int main(int argc, char* argv[]) { ...@@ -52,7 +52,6 @@ int main(int argc, char* argv[]) {
helloworld::Greeter_Stub stub(&channel); helloworld::Greeter_Stub stub(&channel);
// Send a request and wait for the response every 1 second. // Send a request and wait for the response every 1 second.
int log_id = 0;
while (!brpc::IsAskedToQuit()) { while (!brpc::IsAskedToQuit()) {
// We will receive response synchronously, safe to put variables // We will receive response synchronously, safe to put variables
// on stack. // on stack.
...@@ -61,7 +60,6 @@ int main(int argc, char* argv[]) { ...@@ -61,7 +60,6 @@ int main(int argc, char* argv[]) {
brpc::Controller cntl; brpc::Controller cntl;
request.set_name("grpc client example"); request.set_name("grpc client example");
cntl.set_log_id(log_id ++); // set by user
if (FLAGS_gzip) { if (FLAGS_gzip) {
cntl.set_request_compress_type(brpc::COMPRESS_TYPE_GZIP); cntl.set_request_compress_type(brpc::COMPRESS_TYPE_GZIP);
} }
...@@ -73,14 +71,10 @@ int main(int argc, char* argv[]) { ...@@ -73,14 +71,10 @@ int main(int argc, char* argv[]) {
if (!cntl.Failed()) { if (!cntl.Failed()) {
LOG(INFO) << "Received response from " << cntl.remote_side() LOG(INFO) << "Received response from " << cntl.remote_side()
<< " to " << cntl.local_side() << " to " << cntl.local_side()
<< ": " << response.message() << " (attached=" << ": " << response.message()
<< cntl.response_attachment() << ")"
<< " latency=" << cntl.latency_us() << "us"; << " latency=" << cntl.latency_us() << "us";
} else { } else {
LOG(WARNING) << cntl.ErrorCode() << ": " << cntl.ErrorText() LOG(WARNING) << cntl.ErrorCode() << ": " << cntl.ErrorText();
<< ", grpc-status=" << cntl.grpc_status()
<< ", grpc-message=" << cntl.grpc_message();
} }
usleep(FLAGS_interval_ms * 1000L); usleep(FLAGS_interval_ms * 1000L);
} }
......
syntax = "proto3"; syntax = "proto2";
package helloworld; package helloworld;
...@@ -12,10 +12,10 @@ service Greeter { ...@@ -12,10 +12,10 @@ service Greeter {
// The request message containing the user's name. // The request message containing the user's name.
message HelloRequest { message HelloRequest {
string name = 1; required string name = 1;
} }
// The response message containing the greetings // The response message containing the greetings
message HelloReply { message HelloReply {
string message = 1; required string message = 1;
} }
...@@ -22,24 +22,20 @@ ...@@ -22,24 +22,20 @@
#include <brpc/restful.h> #include <brpc/restful.h>
#include "helloworld.pb.h" #include "helloworld.pb.h"
DEFINE_int32(port, 8010, "TCP Port of this server"); DEFINE_int32(port, 50051, "TCP Port of this server");
DEFINE_int32(idle_timeout_s, -1, "Connection will be closed if there is no " DEFINE_int32(idle_timeout_s, -1, "Connection will be closed if there is no "
"read/write operations during the last `idle_timeout_s'"); "read/write operations during the last `idle_timeout_s'");
DEFINE_int32(logoff_ms, 2000, "Maximum duration of server's LOGOFF state " DEFINE_int32(logoff_ms, 2000, "Maximum duration of server's LOGOFF state "
"(waiting for client to close connection before server stops)"); "(waiting for client to close connection before server stops)");
DEFINE_bool(gzip, false, "compress body using gzip"); DEFINE_bool(gzip, false, "compress body using gzip");
// Service with static path.
using helloworld::HelloRequest;
using helloworld::HelloReply;
class GreeterImpl : public helloworld::Greeter { class GreeterImpl : public helloworld::Greeter {
public: public:
GreeterImpl() {}; GreeterImpl() {};
virtual ~GreeterImpl() {}; virtual ~GreeterImpl() {};
void SayHello(google::protobuf::RpcController* cntl_base, void SayHello(google::protobuf::RpcController* cntl_base,
const HelloRequest* req, const helloworld::HelloRequest* req,
HelloReply* res, helloworld::HelloReply* res,
google::protobuf::Closure* done) { google::protobuf::Closure* done) {
brpc::ClosureGuard done_guard(done); brpc::ClosureGuard done_guard(done);
brpc::Controller* cntl = static_cast<brpc::Controller*>(cntl_base); brpc::Controller* cntl = static_cast<brpc::Controller*>(cntl_base);
......
...@@ -254,8 +254,6 @@ void Controller::InternalReset(bool in_constructor) { ...@@ -254,8 +254,6 @@ void Controller::InternalReset(bool in_constructor) {
_response_stream = INVALID_STREAM_ID; _response_stream = INVALID_STREAM_ID;
_remote_stream_settings = NULL; _remote_stream_settings = NULL;
_thrift_method_name = ""; _thrift_method_name = "";
_grpc_status = GRPC_OK;
_grpc_message = "";
} }
Controller::Call::Call(Controller::Call* rhs) Controller::Call::Call(Controller::Call* rhs)
......
...@@ -477,13 +477,6 @@ public: ...@@ -477,13 +477,6 @@ public:
// Get sock option. .e.g get vip info through ttm kernel module hook, // Get sock option. .e.g get vip info through ttm kernel module hook,
int GetSockOption(int level, int optname, void* optval, socklen_t* optlen); int GetSockOption(int level, int optname, void* optval, socklen_t* optlen);
void set_grpc_error_code(GrpcStatus status, const std::string& msg) {
_grpc_status = status;
_grpc_message = msg;
}
GrpcStatus grpc_status() { return _grpc_status; }
std::string grpc_message() { return _grpc_message; }
private: private:
struct CompletionInfo { struct CompletionInfo {
CallId id; // call_id of the corresponding request CallId id; // call_id of the corresponding request
...@@ -724,10 +717,6 @@ private: ...@@ -724,10 +717,6 @@ private:
// Thrift method name, only used when thrift protocol enabled // Thrift method name, only used when thrift protocol enabled
std::string _thrift_method_name; std::string _thrift_method_name;
uint32_t _thrift_seq_id;
GrpcStatus _grpc_status;
std::string _grpc_message;
}; };
// Advises the RPC system that the caller desires that the RPC call be // Advises the RPC system that the caller desires that the RPC call be
......
...@@ -133,9 +133,6 @@ public: ...@@ -133,9 +133,6 @@ public:
_cntl->add_flag(Controller::FLAGS_REQUEST_WITH_AUTH); _cntl->add_flag(Controller::FLAGS_REQUEST_WITH_AUTH);
} }
GrpcStatus grpc_status() { return _cntl->_grpc_status; }
std::string grpc_message() { return _cntl->_grpc_message; }
private: private:
Controller* _cntl; Controller* _cntl;
}; };
......
...@@ -31,5 +31,4 @@ enum Errno { ...@@ -31,5 +31,4 @@ enum Errno {
ELIMIT = 2004; // Reached server's limit on resources ELIMIT = 2004; // Reached server's limit on resources
ECLOSE = 2005; // Close socket initiatively ECLOSE = 2005; // Close socket initiatively
EITP = 2006; // Failed Itp response EITP = 2006; // Failed Itp response
EGRPC = 2007; // Failed Grpc response
} }
...@@ -24,27 +24,7 @@ ...@@ -24,27 +24,7 @@
namespace brpc { namespace brpc {
// The mapping can be found in grpc-go internal/transport/http_util.go GrpcStatus ErrorCodeToGrpcStatus(int error_code) {
GrpcStatus HttpStatus2GrpcStatus(int http_status) {
switch(http_status) {
case HTTP_STATUS_BAD_REQUEST:
return GRPC_INTERNAL;
case HTTP_STATUS_UNAUTHORIZED:
return GRPC_UNAUTHENTICATED;
case HTTP_STATUS_FORBIDDEN:
return GRPC_PERMISSIONDENIED;
case HTTP_STATUS_NOT_FOUND:
return GRPC_UNIMPLEMENTED;
case HTTP_STATUS_BAD_GATEWAY:
case HTTP_STATUS_SERVICE_UNAVAILABLE:
case HTTP_STATUS_GATEWAY_TIMEOUT:
return GRPC_UNAVAILABLE;
default:
return GRPC_UNKNOWN;
}
}
GrpcStatus ErrorCode2GrpcStatus(int error_code) {
switch (error_code) { switch (error_code) {
case ENOSERVICE: case ENOSERVICE:
case ENOMETHOD: case ENOMETHOD:
...@@ -63,40 +43,13 @@ GrpcStatus ErrorCode2GrpcStatus(int error_code) { ...@@ -63,40 +43,13 @@ GrpcStatus ErrorCode2GrpcStatus(int error_code) {
case ETIMEDOUT: case ETIMEDOUT:
return GRPC_INTERNAL; return GRPC_INTERNAL;
default: default:
return GRPC_INTERNAL; return GRPC_OK;
} }
} }
// The mapping can be found in int GrpcStatusToErrorCode(GrpcStatus grpc_status) {
// https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#errors //TODO(zhujiashun)
GrpcStatus h2Error2GrpcStatus(H2Error h2_error) { return EINTERNAL;
switch(h2_error) {
case H2_NO_ERROR:
case H2_PROTOCOL_ERROR:
case H2_INTERNAL_ERROR:
return GRPC_INTERNAL;
case H2_FLOW_CONTROL_ERROR:
return GRPC_RESOURCEEXHAUSTED;
case H2_SETTINGS_TIMEOUT:
case H2_STREAM_CLOSED_ERROR:
case H2_FRAME_SIZE_ERROR:
return GRPC_INTERNAL;
case H2_REFUSED_STREAM:
return GRPC_UNAVAILABLE;
case H2_CANCEL:
return GRPC_CANCELED;
case H2_COMPRESSION_ERROR:
case H2_CONNECT_ERROR:
return GRPC_INTERNAL;
case H2_ENHANCE_YOUR_CALM:
return GRPC_RESOURCEEXHAUSTED;
case H2_INADEQUATE_SECURITY:
return GRPC_PERMISSIONDENIED;
case H2_HTTP_1_1_REQUIRED:
return GRPC_INTERNAL;
default:
return GRPC_INTERNAL;
}
} }
void percent_encode(const std::string& str, std::string* str_out) { void percent_encode(const std::string& str, std::string* str_out) {
......
...@@ -22,8 +22,6 @@ ...@@ -22,8 +22,6 @@
namespace brpc { namespace brpc {
typedef std::map<std::string, std::string> TrailerMessage;
enum GrpcStatus { enum GrpcStatus {
// OK is returned on success. // OK is returned on success.
GRPC_OK = 0, GRPC_OK = 0,
...@@ -144,11 +142,9 @@ enum GrpcStatus { ...@@ -144,11 +142,9 @@ enum GrpcStatus {
GRPC_MAX, GRPC_MAX,
}; };
GrpcStatus HttpStatus2GrpcStatus(int http_status); GrpcStatus ErrorCodeToGrpcStatus(int error_code);
GrpcStatus ErrorCode2GrpcStatus(int error_code);
GrpcStatus h2Error2GrpcStatus(H2Error h2_error); int GrpcStatusToErrorCode(GrpcStatus grpc_status);
void percent_encode(const std::string& str, std::string* str_out); void percent_encode(const std::string& str, std::string* str_out);
......
...@@ -24,10 +24,7 @@ namespace brpc { ...@@ -24,10 +24,7 @@ namespace brpc {
HttpHeader::HttpHeader() HttpHeader::HttpHeader()
: _status_code(HTTP_STATUS_OK) : _status_code(HTTP_STATUS_OK)
, _method(HTTP_METHOD_GET) , _method(HTTP_METHOD_GET)
, _version(1, 1) , _version(1, 1) {
, _h2_stream_id(0)
, _h2_error(H2_NO_ERROR)
, _has_h2_error(false) {
// NOTE: don't forget to clear the field in Clear() as well. // NOTE: don't forget to clear the field in Clear() as well.
} }
...@@ -51,9 +48,6 @@ void HttpHeader::Swap(HttpHeader &rhs) { ...@@ -51,9 +48,6 @@ void HttpHeader::Swap(HttpHeader &rhs) {
_content_type.swap(rhs._content_type); _content_type.swap(rhs._content_type);
_unresolved_path.swap(rhs._unresolved_path); _unresolved_path.swap(rhs._unresolved_path);
std::swap(_version, rhs._version); std::swap(_version, rhs._version);
std::swap(_h2_stream_id, rhs._h2_stream_id);
std::swap(_h2_error, rhs._h2_error);
std::swap(_has_h2_error, rhs._has_h2_error);
} }
void HttpHeader::Clear() { void HttpHeader::Clear() {
...@@ -64,9 +58,6 @@ void HttpHeader::Clear() { ...@@ -64,9 +58,6 @@ void HttpHeader::Clear() {
_content_type.clear(); _content_type.clear();
_unresolved_path.clear(); _unresolved_path.clear();
_version = std::make_pair(1, 1); _version = std::make_pair(1, 1);
_h2_stream_id = 0;
_h2_error = H2_NO_ERROR;
_has_h2_error = false;
} }
const char* HttpHeader::reason_phrase() const { const char* HttpHeader::reason_phrase() const {
......
...@@ -63,14 +63,6 @@ public: ...@@ -63,14 +63,6 @@ public:
// True if the message is from HTTP2. // True if the message is from HTTP2.
bool is_http2() const { return major_version() == 2; } bool is_http2() const { return major_version() == 2; }
// Id of the HTTP2 stream where the message is from.
// 0 when is_http2() is false.
int h2_stream_id() const { return _h2_stream_id; }
H2Error h2_error() const { return _h2_error; }
bool has_h2_error() { return _has_h2_error; }
// Get/set "Content-Type". Notice that you can't get "Content-Type" // Get/set "Content-Type". Notice that you can't get "Content-Type"
// via GetHeader(). // via GetHeader().
// possible values: "text/plain", "application/json" ... // possible values: "text/plain", "application/json" ...
...@@ -164,7 +156,6 @@ friend void policy::ProcessHttpRequest(InputMessageBase *msg); ...@@ -164,7 +156,6 @@ friend void policy::ProcessHttpRequest(InputMessageBase *msg);
std::pair<int, int> _version; std::pair<int, int> _version;
int _h2_stream_id; int _h2_stream_id;
H2Error _h2_error; H2Error _h2_error;
bool _has_h2_error;
}; };
const HttpHeader& DefaultHttpHeader(); const HttpHeader& DefaultHttpHeader();
......
...@@ -1629,16 +1629,19 @@ void H2UnsentRequest::Describe(butil::IOBuf* desc) const { ...@@ -1629,16 +1629,19 @@ void H2UnsentRequest::Describe(butil::IOBuf* desc) const {
} }
} }
H2UnsentResponse::H2UnsentResponse(Controller* c, int stream_id, const TrailerMessage& trailers) H2UnsentResponse::H2UnsentResponse(Controller* c, int stream_id, bool grpc)
: _size(0) : _size(0)
, _stream_id(stream_id) , _stream_id(stream_id)
, _http_response(c->release_http_response()) , _http_response(c->release_http_response())
, _trailers(trailers) { , _grpc(grpc) {
_data.swap(c->response_attachment()); _data.swap(c->response_attachment());
if (grpc) {
_grpc_status = ErrorCodeToGrpcStatus(c->ErrorCode());
percent_encode(c->ErrorText(), &_grpc_message);
}
} }
H2UnsentResponse* H2UnsentResponse::New(Controller* c, int stream_id, H2UnsentResponse* H2UnsentResponse::New(Controller* c, int stream_id, bool grpc) {
const TrailerMessage& trailers) {
const HttpHeader* const h = &c->http_response(); const HttpHeader* const h = &c->http_response();
const CommonStrings* const common = get_common_strings(); const CommonStrings* const common = get_common_strings();
const bool need_content_length = const bool need_content_length =
...@@ -1649,7 +1652,7 @@ H2UnsentResponse* H2UnsentResponse::New(Controller* c, int stream_id, ...@@ -1649,7 +1652,7 @@ H2UnsentResponse* H2UnsentResponse::New(Controller* c, int stream_id,
+ (size_t)need_content_type; + (size_t)need_content_type;
const size_t memsize = offsetof(H2UnsentResponse, _list) + const size_t memsize = offsetof(H2UnsentResponse, _list) +
sizeof(HPacker::Header) * maxsize; sizeof(HPacker::Header) * maxsize;
H2UnsentResponse* msg = new (malloc(memsize)) H2UnsentResponse(c, stream_id, trailers); H2UnsentResponse* msg = new (malloc(memsize)) H2UnsentResponse(c, stream_id, grpc);
// :status // :status
if (h->status_code() == 200) { if (h->status_code() == 200) {
msg->push(common->H2_STATUS, common->STATUS_200); msg->push(common->H2_STATUS, common->STATUS_200);
...@@ -1721,10 +1724,14 @@ H2UnsentResponse::AppendAndDestroySelf(butil::IOBuf* out, Socket* socket) { ...@@ -1721,10 +1724,14 @@ H2UnsentResponse::AppendAndDestroySelf(butil::IOBuf* out, Socket* socket) {
butil::IOBufAppender trailer_appender; butil::IOBufAppender trailer_appender;
butil::IOBuf trailer_frag; butil::IOBuf trailer_frag;
for (TrailerMessage::iterator it = _trailers.begin(); if (_grpc) {
it != _trailers.end(); ++it) { HPacker::Header status_header("grpc-status",
HPacker::Header header(it->first, it->second); butil::string_printf("%d", _grpc_status));
hpacker.Encode(&trailer_appender, header, options); hpacker.Encode(&trailer_appender, status_header, options);
if (!_grpc_message.empty()) {
HPacker::Header msg_header("grpc-message", _grpc_message);
hpacker.Encode(&trailer_appender, msg_header, options);
}
} }
trailer_appender.move_to(trailer_frag); trailer_appender.move_to(trailer_frag);
......
...@@ -194,7 +194,7 @@ private: ...@@ -194,7 +194,7 @@ private:
class H2UnsentResponse : public SocketMessage { class H2UnsentResponse : public SocketMessage {
public: public:
static H2UnsentResponse* New(Controller* c, int stream_id, const TrailerMessage& trailers); static H2UnsentResponse* New(Controller* c, int stream_id, bool grpc);
void Destroy(); void Destroy();
void Describe(butil::IOBuf*) const; void Describe(butil::IOBuf*) const;
// @SocketMessage // @SocketMessage
...@@ -208,7 +208,7 @@ private: ...@@ -208,7 +208,7 @@ private:
void push(const std::string& name, const std::string& value) void push(const std::string& name, const std::string& value)
{ new (&_list[_size++]) HPacker::Header(name, value); } { new (&_list[_size++]) HPacker::Header(name, value); }
H2UnsentResponse(Controller* c, int stream_id, const TrailerMessage& trailers); H2UnsentResponse(Controller* c, int stream_id, bool grpc);
~H2UnsentResponse() {} ~H2UnsentResponse() {}
H2UnsentResponse(const H2UnsentResponse&); H2UnsentResponse(const H2UnsentResponse&);
void operator=(const H2UnsentResponse&); void operator=(const H2UnsentResponse&);
...@@ -218,7 +218,9 @@ private: ...@@ -218,7 +218,9 @@ private:
uint32_t _stream_id; uint32_t _stream_id;
std::unique_ptr<HttpHeader> _http_response; std::unique_ptr<HttpHeader> _http_response;
butil::IOBuf _data; butil::IOBuf _data;
TrailerMessage _trailers; bool _grpc;
GrpcStatus _grpc_status;
std::string _grpc_message;
HPacker::Header _list[0]; HPacker::Header _list[0];
}; };
......
...@@ -356,6 +356,21 @@ void ProcessHttpResponse(InputMessageBase* msg) { ...@@ -356,6 +356,21 @@ void ProcessHttpResponse(InputMessageBase* msg) {
} }
break; break;
} }
if (grpc_protocol) {
const std::string* grpc_status =
res_header->GetHeader(common->GRPC_STATUS);
const std::string* grpc_message =
res_header->GetHeader(common->GRPC_MESSAGE);
if (grpc_status) {
GrpcStatus status = (GrpcStatus)strtol(grpc_status->data(), NULL, 10);
if (status != GRPC_OK) {
const std::string err =
grpc_message ? grpc_message->data() : "Unknown grpc error";
cntl->SetFailed(GrpcStatusToErrorCode(status), "%s", err.c_str());
break;
}
}
}
if (cntl->response() == NULL || if (cntl->response() == NULL ||
cntl->response()->GetDescriptor()->field_count() == 0) { cntl->response()->GetDescriptor()->field_count() == 0) {
// a http call, content is the "real response". // a http call, content is the "real response".
...@@ -410,48 +425,6 @@ void ProcessHttpResponse(InputMessageBase* msg) { ...@@ -410,48 +425,6 @@ void ProcessHttpResponse(InputMessageBase* msg) {
} }
} while (0); } while (0);
do {
if (!grpc_protocol) {
break;
}
// begin to handle grpc case
// Receive RST_Stream and h2_error is set correspondingly
if (res_header->has_h2_error()) {
cntl->set_grpc_error_code(h2Error2GrpcStatus(res_header->h2_error()), "");
cntl->SetFailed(EGRPC, "");
break;
}
const std::string* grpc_status = res_header->GetHeader(common->GRPC_STATUS);
const std::string* grpc_message = res_header->GetHeader(common->GRPC_MESSAGE);
if (grpc_status) {
GrpcStatus status = (GrpcStatus)strtol(grpc_status->data(), NULL, 10);
if (status != GRPC_OK) {
cntl->set_grpc_error_code(status, grpc_message? *grpc_message: "");
cntl->SetFailed(EGRPC, grpc_message? grpc_message->c_str(): "");
}
break;
}
// grpc-status is absent in http header, just convert error code
// to grpc status
if (!cntl->Failed()) {
break;
}
if (cntl->ErrorCode() == EHTTP) {
cntl->set_grpc_error_code(
HttpStatus2GrpcStatus(res_header->status_code()),
cntl->ErrorText());
// use empty string since gprc-status and grpc-message is absent
cntl->SetFailed(EGRPC, "");
break;
}
cntl->set_grpc_error_code(ErrorCode2GrpcStatus(cntl->ErrorCode()),
cntl->ErrorText());
// use empty string since gprc-status and grpc-message is absent
cntl->SetFailed(EGRPC, "");
} while (0);
// Unlocks correlation_id inside. Revert controller's // Unlocks correlation_id inside. Revert controller's
// error code if it version check of `cid' fails // error code if it version check of `cid' fails
imsg_guard.reset(); imsg_guard.reset();
...@@ -798,6 +771,7 @@ HttpResponseSender::~HttpResponseSender() { ...@@ -798,6 +771,7 @@ HttpResponseSender::~HttpResponseSender() {
ParseContentType(req_header->content_type()) == HTTP_CONTENT_GRPC; ParseContentType(req_header->content_type()) == HTTP_CONTENT_GRPC;
if (cntl->Failed()) { if (cntl->Failed()) {
// TODO(zhujiashun): really need this?
if (!grpc_protocol) { if (!grpc_protocol) {
// Set status-code with default value(converted from error code) // Set status-code with default value(converted from error code)
// if user did not set it. // if user did not set it.
...@@ -853,36 +827,16 @@ HttpResponseSender::~HttpResponseSender() { ...@@ -853,36 +827,16 @@ HttpResponseSender::~HttpResponseSender() {
} }
} }
TrailerMessage trailers;
if (grpc_protocol) { if (grpc_protocol) {
// status code is always 200 according to grpc protocol // status code is always 200 according to grpc protocol
res_header->set_status_code(HTTP_STATUS_OK); res_header->set_status_code(HTTP_STATUS_OK);
res_header->set_content_type(common->CONTENT_TYPE_GRPC); res_header->set_content_type(common->CONTENT_TYPE_GRPC);
GrpcStatus status = accessor.grpc_status();
std::string message = accessor.grpc_message();
if (status == GRPC_OK && cntl->Failed()) {
// In this case, request may not be handled by user-defined method or
// users use cntl.SetFailed to set grpc error instead of using
// cntl.set_grpc_error_code
status = ErrorCode2GrpcStatus(cntl->ErrorCode());
message = cntl->ErrorText();
}
std::string message_encoded;
percent_encode(message, &message_encoded);
trailers[common->GRPC_STATUS] = butil::string_printf("%d", status);
if (!message_encoded.empty()) {
trailers[common->GRPC_MESSAGE] = message_encoded;
}
// always tell client gzip support // always tell client gzip support
// TODO(zhujiashun): add zlib and snappy? // TODO(zhujiashun): add zlib and snappy?
res_header->SetHeader(common->GRPC_ACCEPT_ENCODING, res_header->SetHeader(common->GRPC_ACCEPT_ENCODING,
common->IDENTITY + "," + common->GZIP); common->IDENTITY + "," + common->GZIP);
if (!cntl->Failed()) {
if (status == GRPC_OK) { // Encode Length-Prefixed-Message
butil::IOBuf tmp_buf; butil::IOBuf tmp_buf;
// Compressed-Flag as 0 / 1, encoded as 1 byte unsigned integer // Compressed-Flag as 0 / 1, encoded as 1 byte unsigned integer
if (grpc_compressed) { if (grpc_compressed) {
...@@ -907,7 +861,7 @@ HttpResponseSender::~HttpResponseSender() { ...@@ -907,7 +861,7 @@ HttpResponseSender::~HttpResponseSender() {
wopt.ignore_eovercrowded = true; wopt.ignore_eovercrowded = true;
if (req_header->is_http2()) { if (req_header->is_http2()) {
SocketMessagePtr<H2UnsentResponse> h2_response( SocketMessagePtr<H2UnsentResponse> h2_response(
H2UnsentResponse::New(cntl, _h2_stream_id, trailers)); H2UnsentResponse::New(cntl, _h2_stream_id, grpc_protocol));
if (h2_response == NULL) { if (h2_response == NULL) {
LOG(ERROR) << "Fail to make http2 response"; LOG(ERROR) << "Fail to make http2 response";
errno = EINVAL; errno = EINVAL;
...@@ -1280,7 +1234,7 @@ void ProcessHttpRequest(InputMessageBase *msg) { ...@@ -1280,7 +1234,7 @@ void ProcessHttpRequest(InputMessageBase *msg) {
req_body.cut1(&grpc_compressed); req_body.cut1(&grpc_compressed);
req_body.cutn(buf, 4); req_body.cutn(buf, 4);
int message_length = ReadBigEndian4Bytes(buf); int message_length = ReadBigEndian4Bytes(buf);
CHECK(message_length == req_body.length()); CHECK((size_t)message_length == req_body.length());
} }
butil::EndPoint user_addr; butil::EndPoint user_addr;
......
...@@ -41,7 +41,7 @@ namespace { ...@@ -41,7 +41,7 @@ namespace {
const std::string g_server_addr = "127.0.0.1:8011"; const std::string g_server_addr = "127.0.0.1:8011";
const std::string g_prefix = "Hello, "; const std::string g_prefix = "Hello, ";
const std::string g_req = "wyt"; const std::string g_req = "wyt";
const int64_t g_timeout_ms = 3000; const int64_t g_timeout_ms = 1000;
const std::string g_protocol = "grpc"; const std::string g_protocol = "grpc";
class MyGrpcService : public ::test::GrpcService { class MyGrpcService : public ::test::GrpcService {
...@@ -61,9 +61,9 @@ public: ...@@ -61,9 +61,9 @@ public:
res->set_message(g_prefix + req->message()); res->set_message(g_prefix + req->message());
if (req->has_error_code()) { if (req->has_error_code()) {
const int error_code = req->error_code(); const std::string err_msg =
cntl->set_grpc_error_code((brpc::GrpcStatus)error_code, butil::string_printf("%s%d", g_prefix.c_str(), req->error_code());
butil::string_printf("%s%d", g_prefix.c_str(), error_code)); cntl->SetFailed(err_msg.c_str());
return; return;
} }
} }
...@@ -73,7 +73,8 @@ public: ...@@ -73,7 +73,8 @@ public:
::test::GrpcResponse* res, ::test::GrpcResponse* res,
::google::protobuf::Closure* done) { ::google::protobuf::Closure* done) {
brpc::ClosureGuard done_guard(done); brpc::ClosureGuard done_guard(done);
bthread_usleep(6 * 1000000L); bthread_usleep(2000000 /*2s*/);
res->set_message(g_prefix + req->message());
return; return;
} }
}; };
...@@ -108,7 +109,7 @@ protected: ...@@ -108,7 +109,7 @@ protected:
stub.Method(&cntl, &req, &res, NULL); stub.Method(&cntl, &req, &res, NULL);
EXPECT_FALSE(cntl.Failed()) << cntl.ErrorCode() << ": " << cntl.ErrorText(); EXPECT_FALSE(cntl.Failed()) << cntl.ErrorCode() << ": " << cntl.ErrorText();
EXPECT_EQ(res.message(), g_prefix + g_req); EXPECT_EQ(res.message(), g_prefix + g_req);
EXPECT_EQ(brpc::GRPC_OK, cntl.grpc_status()); //EXPECT_EQ(brpc::GRPC_OK, cntl.grpc_status());
} }
brpc::Server _server; brpc::Server _server;
...@@ -161,9 +162,9 @@ TEST_F(GrpcTest, return_error) { ...@@ -161,9 +162,9 @@ TEST_F(GrpcTest, return_error) {
test::GrpcService_Stub stub(&_channel); test::GrpcService_Stub stub(&_channel);
stub.Method(&cntl, &req, &res, NULL); stub.Method(&cntl, &req, &res, NULL);
EXPECT_TRUE(cntl.Failed()); EXPECT_TRUE(cntl.Failed());
EXPECT_EQ(cntl.ErrorCode(), brpc::EGRPC); // FIXME(zhujiashun): message body is empty so the Errorcode may be ERESPONSE
EXPECT_EQ((int)cntl.grpc_status(), i); // EXPECT_EQ(cntl.ErrorCode(), brpc::EINTERNAL);
EXPECT_EQ(cntl.grpc_message(), butil::string_printf("%s%d", g_prefix.c_str(), i)); // EXPECT_EQ(cntl.ErrorText(), butil::string_printf("%s%d", g_prefix.c_str(), i));
} }
} }
...@@ -192,9 +193,9 @@ TEST_F(GrpcTest, MethodNotExist) { ...@@ -192,9 +193,9 @@ TEST_F(GrpcTest, MethodNotExist) {
test::GrpcService_Stub stub(&_channel); test::GrpcService_Stub stub(&_channel);
stub.MethodNotExist(&cntl, &req, &res, NULL); stub.MethodNotExist(&cntl, &req, &res, NULL);
EXPECT_TRUE(cntl.Failed()); EXPECT_TRUE(cntl.Failed());
EXPECT_EQ(cntl.ErrorCode(), brpc::EGRPC); // FIXME(zhujiashun): message body is empty so the Errorcode may be ERESPONSE
EXPECT_EQ((int)cntl.grpc_status(), brpc::GRPC_INTERNAL); //EXPECT_EQ(cntl.ErrorCode(), brpc::EINTERNAL);
ASSERT_TRUE(butil::StringPiece(cntl.grpc_message()).ends_with("Method MethodNotExist() not implemented.")); //ASSERT_TRUE(butil::StringPiece(cntl.ErrorText()).ends_with("Method MethodNotExist() not implemented."));
} }
} // namespace } // namespace
...@@ -216,8 +216,7 @@ protected: ...@@ -216,8 +216,7 @@ protected:
butil::IOBufAsZeroCopyOutputStream wrapper(&cntl.response_attachment()); butil::IOBufAsZeroCopyOutputStream wrapper(&cntl.response_attachment());
EXPECT_TRUE(res.SerializeToZeroCopyStream(&wrapper)); EXPECT_TRUE(res.SerializeToZeroCopyStream(&wrapper));
} }
const brpc::TrailerMessage dummy; brpc::policy::H2UnsentResponse* h2_res = brpc::policy::H2UnsentResponse::New(&cntl, h2_stream_id, false /*is grpc*/);
brpc::policy::H2UnsentResponse* h2_res = brpc::policy::H2UnsentResponse::New(&cntl, h2_stream_id, dummy);
butil::Status st = h2_res->AppendAndDestroySelf(out, _h2_client_sock.get()); butil::Status st = h2_res->AppendAndDestroySelf(out, _h2_client_sock.get());
ASSERT_TRUE(st.ok()); ASSERT_TRUE(st.ok());
} }
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment