Merge pull request #1261 from ctiller/registered_calls

Registered calls
pull/1303/head
Yang Gao 10 years ago
commit b12dc6b5bc
  1. 253
      Makefile
  2. 1
      include/grpc++/channel_interface.h
  3. 9
      include/grpc++/impl/internal_stub.h
  4. 10
      include/grpc++/impl/rpc_method.h
  5. 2
      include/grpc++/impl/rpc_service_method.h
  6. 12
      include/grpc/grpc.h
  7. 127
      src/compiler/cpp_generator.cc
  8. 87
      src/core/surface/channel.c
  9. 11
      src/cpp/client/channel.cc
  10. 1
      src/cpp/client/channel.h
  11. 10
      src/cpp/client/generic_stub.cc
  12. 1
      test/core/end2end/gen_build_json.py
  13. 222
      test/core/end2end/tests/registered_call.c
  14. 63
      tools/run_tests/tests.json

File diff suppressed because one or more lines are too long

@ -51,6 +51,7 @@ class ChannelInterface : public CallHook {
public:
virtual ~ChannelInterface() {}
virtual void *RegisterMethod(const char *method_name) = 0;
virtual Call CreateCall(const RpcMethod& method, ClientContext* context,
CompletionQueue* cq) = 0;
};

@ -42,17 +42,14 @@ namespace grpc {
class InternalStub {
public:
InternalStub() {}
InternalStub(const std::shared_ptr<ChannelInterface>& channel)
: channel_(channel) {}
virtual ~InternalStub() {}
void set_channel(const std::shared_ptr<ChannelInterface>& channel) {
channel_ = channel;
}
ChannelInterface* channel() { return channel_.get(); }
private:
std::shared_ptr<ChannelInterface> channel_;
const std::shared_ptr<ChannelInterface> channel_;
};
} // namespace grpc

@ -45,17 +45,17 @@ class RpcMethod {
BIDI_STREAMING
};
explicit RpcMethod(const char* name)
: name_(name), method_type_(NORMAL_RPC) {}
RpcMethod(const char* name, RpcType type) : name_(name), method_type_(type) {}
RpcMethod(const char* name, RpcType type, void* channel_tag)
: name_(name), method_type_(type), channel_tag_(channel_tag) {}
const char* name() const { return name_; }
RpcType method_type() const { return method_type_; }
void* channel_tag() const { return channel_tag_; }
private:
const char* name_;
const char* const name_;
const RpcType method_type_;
void* const channel_tag_;
};
} // namespace grpc

@ -167,7 +167,7 @@ class RpcServiceMethod : public RpcMethod {
MethodHandler* handler,
grpc::protobuf::Message* request_prototype,
grpc::protobuf::Message* response_prototype)
: RpcMethod(name, type),
: RpcMethod(name, type, nullptr),
handler_(handler),
request_prototype_(request_prototype),
response_prototype_(response_prototype) {}

@ -421,6 +421,15 @@ grpc_call *grpc_channel_create_call(grpc_channel *channel,
const char *method, const char *host,
gpr_timespec deadline);
/* Pre-register a method/host pair on a channel. */
void *grpc_channel_register_call(grpc_channel *channel, const char *method,
const char *host);
/* Create a call given a handle returned from grpc_channel_register_call */
grpc_call *grpc_channel_create_registered_call(
grpc_channel *channel, grpc_completion_queue *completion_queue,
void *registered_call_handle, gpr_timespec deadline);
/* Start a batch of operations defined in the array ops; when complete, post a
completion of type 'tag' to the completion queue bound to the call.
The order of ops specified in the batch has no significance.
@ -579,8 +588,7 @@ grpc_call_error grpc_server_request_call_old(grpc_server *server,
grpc_call_error grpc_server_request_call(
grpc_server *server, grpc_call **call, grpc_call_details *details,
grpc_metadata_array *request_metadata,
grpc_completion_queue *cq_bound_to_call,
void *tag_new);
grpc_completion_queue *cq_bound_to_call, void *tag_new);
/* Registers a method in the server.
Methods to this (host, method) pair will not be reported by

@ -154,6 +154,7 @@ grpc::string GetHeaderIncludes(const grpc::protobuf::FileDescriptor *file,
const Parameters &params) {
grpc::string temp =
"#include <grpc++/impl/internal_stub.h>\n"
"#include <grpc++/impl/rpc_method.h>\n"
"#include <grpc++/impl/service_type.h>\n"
"#include <grpc++/status.h>\n"
"\n"
@ -172,7 +173,9 @@ grpc::string GetHeaderIncludes(const grpc::protobuf::FileDescriptor *file,
temp.append("template <class OutMessage> class ClientWriter;\n");
temp.append("template <class InMessage> class ServerReader;\n");
temp.append("template <class OutMessage> class ClientAsyncWriter;\n");
temp.append("template <class OutMessage, class InMessage> class ServerAsyncReader;\n");
temp.append(
"template <class OutMessage, class InMessage> class "
"ServerAsyncReader;\n");
}
if (HasServerOnlyStreaming(file)) {
temp.append("template <class InMessage> class ClientReader;\n");
@ -246,9 +249,9 @@ void PrintHeaderClientMethod(grpc::protobuf::io::Printer *printer,
*vars,
"std::unique_ptr< ::grpc::ClientReader< $Response$>> $Method$("
"::grpc::ClientContext* context, const $Request$& request);\n");
printer->Print(
*vars,
"std::unique_ptr< ::grpc::ClientAsyncReader< $Response$>> Async$Method$("
printer->Print(*vars,
"std::unique_ptr< ::grpc::ClientAsyncReader< $Response$>> "
"Async$Method$("
"::grpc::ClientContext* context, const $Request$& request, "
"::grpc::CompletionQueue* cq, void* tag);\n");
} else if (BidiStreaming(method)) {
@ -264,8 +267,14 @@ void PrintHeaderClientMethod(grpc::protobuf::io::Printer *printer,
}
}
void PrintHeaderServerMethodSync(
grpc::protobuf::io::Printer *printer,
void PrintHeaderClientMethodData(grpc::protobuf::io::Printer *printer,
const grpc::protobuf::MethodDescriptor *method,
std::map<grpc::string, grpc::string> *vars) {
(*vars)["Method"] = method->name();
printer->Print(*vars, "const ::grpc::RpcMethod rpcmethod_$Method$_;\n");
}
void PrintHeaderServerMethodSync(grpc::protobuf::io::Printer *printer,
const grpc::protobuf::MethodDescriptor *method,
std::map<grpc::string, grpc::string> *vars) {
(*vars)["Method"] = method->name();
@ -351,10 +360,18 @@ void PrintHeaderService(grpc::protobuf::io::Printer *printer,
"class Stub GRPC_FINAL : public ::grpc::InternalStub {\n"
" public:\n");
printer->Indent();
printer->Print(
"Stub(const std::shared_ptr< ::grpc::ChannelInterface>& channel);\n");
for (int i = 0; i < service->method_count(); ++i) {
PrintHeaderClientMethod(printer, service->method(i), vars);
}
printer->Outdent();
printer->Print(" private:\n");
printer->Indent();
for (int i = 0; i < service->method_count(); ++i) {
PrintHeaderClientMethodData(printer, service->method(i), vars);
}
printer->Outdent();
printer->Print("};\n");
printer->Print(
"static std::unique_ptr<Stub> NewStub(const std::shared_ptr< "
@ -479,7 +496,6 @@ grpc::string GetSourceIncludes(const grpc::protobuf::FileDescriptor *file,
printer.Print(vars, "#include <grpc++/async_unary_call.h>\n");
printer.Print(vars, "#include <grpc++/channel_interface.h>\n");
printer.Print(vars, "#include <grpc++/impl/client_unary_call.h>\n");
printer.Print(vars, "#include <grpc++/impl/rpc_method.h>\n");
printer.Print(vars, "#include <grpc++/impl/rpc_service_method.h>\n");
printer.Print(vars, "#include <grpc++/impl/service_type.h>\n");
printer.Print(vars, "#include <grpc++/stream.h>\n");
@ -514,7 +530,7 @@ void PrintSourceClientMethod(grpc::protobuf::io::Printer *printer,
"const $Request$& request, $Response$* response) {\n");
printer->Print(*vars,
" return ::grpc::BlockingUnaryCall(channel(), "
"::grpc::RpcMethod($prefix$$Service$_method_names[$Idx$]), "
"rpcmethod_$Method$_, "
"context, request, response);\n"
"}\n\n");
printer->Print(
@ -528,7 +544,7 @@ void PrintSourceClientMethod(grpc::protobuf::io::Printer *printer,
"::grpc::ClientAsyncResponseReader< $Response$>>(new "
"::grpc::ClientAsyncResponseReader< $Response$>("
"channel(), cq, "
"::grpc::RpcMethod($prefix$$Service$_method_names[$Idx$]), "
"rpcmethod_$Method$_, "
"context, request, tag));\n"
"}\n\n");
} else if (ClientOnlyStreaming(method)) {
@ -540,8 +556,7 @@ void PrintSourceClientMethod(grpc::protobuf::io::Printer *printer,
" return std::unique_ptr< ::grpc::ClientWriter< "
"$Request$>>(new ::grpc::ClientWriter< $Request$>("
"channel(),"
"::grpc::RpcMethod($prefix$$Service$_method_names[$Idx$], "
"::grpc::RpcMethod::RpcType::CLIENT_STREAMING), "
"rpcmethod_$Method$_, "
"context, response));\n"
"}\n\n");
printer->Print(*vars,
@ -553,8 +568,7 @@ void PrintSourceClientMethod(grpc::protobuf::io::Printer *printer,
" return std::unique_ptr< ::grpc::ClientAsyncWriter< "
"$Request$>>(new ::grpc::ClientAsyncWriter< $Request$>("
"channel(), cq, "
"::grpc::RpcMethod($prefix$$Service$_method_names[$Idx$], "
"::grpc::RpcMethod::RpcType::CLIENT_STREAMING), "
"rpcmethod_$Method$_, "
"context, response, tag));\n"
"}\n\n");
} else if (ServerOnlyStreaming(method)) {
@ -567,8 +581,7 @@ void PrintSourceClientMethod(grpc::protobuf::io::Printer *printer,
" return std::unique_ptr< ::grpc::ClientReader< "
"$Response$>>(new ::grpc::ClientReader< $Response$>("
"channel(),"
"::grpc::RpcMethod($prefix$$Service$_method_names[$Idx$], "
"::grpc::RpcMethod::RpcType::SERVER_STREAMING), "
"rpcmethod_$Method$_, "
"context, request));\n"
"}\n\n");
printer->Print(*vars,
@ -580,8 +593,7 @@ void PrintSourceClientMethod(grpc::protobuf::io::Printer *printer,
" return std::unique_ptr< ::grpc::ClientAsyncReader< "
"$Response$>>(new ::grpc::ClientAsyncReader< $Response$>("
"channel(), cq, "
"::grpc::RpcMethod($prefix$$Service$_method_names[$Idx$], "
"::grpc::RpcMethod::RpcType::SERVER_STREAMING), "
"rpcmethod_$Method$_, "
"context, request, tag));\n"
"}\n\n");
} else if (BidiStreaming(method)) {
@ -594,11 +606,11 @@ void PrintSourceClientMethod(grpc::protobuf::io::Printer *printer,
"$Request$, $Response$>>(new ::grpc::ClientReaderWriter< "
"$Request$, $Response$>("
"channel(),"
"::grpc::RpcMethod($prefix$$Service$_method_names[$Idx$], "
"::grpc::RpcMethod::RpcType::BIDI_STREAMING), "
"rpcmethod_$Method$_, "
"context));\n"
"}\n\n");
printer->Print(*vars,
printer->Print(
*vars,
"std::unique_ptr< ::grpc::ClientAsyncReaderWriter< "
"$Request$, $Response$>> "
"$ns$$Service$::Stub::Async$Method$(::grpc::ClientContext* context, "
@ -608,8 +620,7 @@ void PrintSourceClientMethod(grpc::protobuf::io::Printer *printer,
"$Request$, $Response$>>(new "
"::grpc::ClientAsyncReaderWriter< $Request$, $Response$>("
"channel(), cq, "
"::grpc::RpcMethod($prefix$$Service$_method_names[$Idx$], "
"::grpc::RpcMethod::RpcType::BIDI_STREAMING), "
"rpcmethod_$Method$_, "
"context, tag));\n"
"}\n\n");
}
@ -681,9 +692,9 @@ void PrintSourceServerAsyncMethod(
"$Request$* request, "
"::grpc::ServerAsyncResponseWriter< $Response$>* response, "
"::grpc::CompletionQueue* cq, void* tag) {\n");
printer->Print(
*vars,
" AsynchronousService::RequestAsyncUnary($Idx$, context, request, response, cq, tag);\n");
printer->Print(*vars,
" AsynchronousService::RequestAsyncUnary($Idx$, context, "
"request, response, cq, tag);\n");
printer->Print("}\n\n");
} else if (ClientOnlyStreaming(method)) {
printer->Print(*vars,
@ -691,9 +702,9 @@ void PrintSourceServerAsyncMethod(
"::grpc::ServerContext* context, "
"::grpc::ServerAsyncReader< $Response$, $Request$>* reader, "
"::grpc::CompletionQueue* cq, void* tag) {\n");
printer->Print(
*vars,
" AsynchronousService::RequestClientStreaming($Idx$, context, reader, cq, tag);\n");
printer->Print(*vars,
" AsynchronousService::RequestClientStreaming($Idx$, "
"context, reader, cq, tag);\n");
printer->Print("}\n\n");
} else if (ServerOnlyStreaming(method)) {
printer->Print(*vars,
@ -702,9 +713,9 @@ void PrintSourceServerAsyncMethod(
"$Request$* request, "
"::grpc::ServerAsyncWriter< $Response$>* writer, "
"::grpc::CompletionQueue* cq, void* tag) {\n");
printer->Print(
*vars,
" AsynchronousService::RequestServerStreaming($Idx$, context, request, writer, cq, tag);\n");
printer->Print(*vars,
" AsynchronousService::RequestServerStreaming($Idx$, "
"context, request, writer, cq, tag);\n");
printer->Print("}\n\n");
} else if (BidiStreaming(method)) {
printer->Print(
@ -713,9 +724,9 @@ void PrintSourceServerAsyncMethod(
"::grpc::ServerContext* context, "
"::grpc::ServerAsyncReaderWriter< $Response$, $Request$>* stream, "
"::grpc::CompletionQueue* cq, void *tag) {\n");
printer->Print(
*vars,
" AsynchronousService::RequestBidiStreaming($Idx$, context, stream, cq, tag);\n");
printer->Print(*vars,
" AsynchronousService::RequestBidiStreaming($Idx$, "
"context, stream, cq, tag);\n");
printer->Print("}\n\n");
}
}
@ -725,7 +736,8 @@ void PrintSourceService(grpc::protobuf::io::Printer *printer,
std::map<grpc::string, grpc::string> *vars) {
(*vars)["Service"] = service->name();
printer->Print(*vars, "static const char* $prefix$$Service$_method_names[] = {\n");
printer->Print(*vars,
"static const char* $prefix$$Service$_method_names[] = {\n");
for (int i = 0; i < service->method_count(); ++i) {
(*vars)["Method"] = service->method(i)->name();
printer->Print(*vars, " \"/$Package$$Service$/$Method$\",\n");
@ -736,20 +748,50 @@ void PrintSourceService(grpc::protobuf::io::Printer *printer,
*vars,
"std::unique_ptr< $ns$$Service$::Stub> $ns$$Service$::NewStub("
"const std::shared_ptr< ::grpc::ChannelInterface>& channel) {\n"
" std::unique_ptr< $ns$$Service$::Stub> stub(new $ns$$Service$::Stub());\n"
" stub->set_channel(channel);\n"
" std::unique_ptr< $ns$$Service$::Stub> stub(new "
"$ns$$Service$::Stub(channel));\n"
" return stub;\n"
"}\n\n");
printer->Print(*vars,
"$ns$$Service$::Stub::Stub(const std::shared_ptr< "
"::grpc::ChannelInterface>& channel)\n");
printer->Indent();
printer->Print(": ::grpc::InternalStub(channel)");
for (int i = 0; i < service->method_count(); ++i) {
const grpc::protobuf::MethodDescriptor *method = service->method(i);
(*vars)["Method"] = method->name();
(*vars)["Idx"] = as_string(i);
if (NoStreaming(method)) {
(*vars)["StreamingType"] = "NORMAL_RPC";
} else if (ClientOnlyStreaming(method)) {
(*vars)["StreamingType"] = "CLIENT_STREAMING";
} else if (ServerOnlyStreaming(method)) {
(*vars)["StreamingType"] = "SERVER_STREAMING";
} else {
(*vars)["StreamingType"] = "BIDI_STREAMING";
}
printer->Print(
*vars,
", rpcmethod_$Method$_("
"$prefix$$Service$_method_names[$Idx$], "
"::grpc::RpcMethod::$StreamingType$, "
"channel->RegisterMethod($prefix$$Service$_method_names[$Idx$])"
")\n");
}
printer->Print("{}\n\n");
printer->Outdent();
for (int i = 0; i < service->method_count(); ++i) {
(*vars)["Idx"] = as_string(i);
PrintSourceClientMethod(printer, service->method(i), vars);
}
(*vars)["MethodCount"] = as_string(service->method_count());
printer->Print(
*vars,
"$ns$$Service$::AsyncService::AsyncService(::grpc::CompletionQueue* cq) : "
"::grpc::AsynchronousService(cq, $prefix$$Service$_method_names, $MethodCount$) "
printer->Print(*vars,
"$ns$$Service$::AsyncService::AsyncService(::grpc::"
"CompletionQueue* cq) : "
"::grpc::AsynchronousService(cq, "
"$prefix$$Service$_method_names, $MethodCount$) "
"{}\n\n");
printer->Print(*vars,
@ -783,7 +825,8 @@ void PrintSourceService(grpc::protobuf::io::Printer *printer,
"service_->AddMethod(new ::grpc::RpcServiceMethod(\n"
" $prefix$$Service$_method_names[$Idx$],\n"
" ::grpc::RpcMethod::NORMAL_RPC,\n"
" new ::grpc::RpcMethodHandler< $ns$$Service$::Service, $Request$, "
" new ::grpc::RpcMethodHandler< $ns$$Service$::Service, "
"$Request$, "
"$Response$>(\n"
" std::function< ::grpc::Status($ns$$Service$::Service*, "
"::grpc::ServerContext*, const $Request$*, $Response$*)>("

@ -43,6 +43,12 @@
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
typedef struct registered_call {
grpc_mdelem *path;
grpc_mdelem *authority;
struct registered_call *next;
} registered_call;
struct grpc_channel {
int is_client;
gpr_refcount refs;
@ -51,10 +57,14 @@ struct grpc_channel {
grpc_mdstr *grpc_message_string;
grpc_mdstr *path_string;
grpc_mdstr *authority_string;
gpr_mu registered_call_mu;
registered_call *registered_calls;
};
#define CHANNEL_STACK_FROM_CHANNEL(c) ((grpc_channel_stack *)((c)+1))
#define CHANNEL_FROM_CHANNEL_STACK(channel_stack) (((grpc_channel *)(channel_stack)) - 1)
#define CHANNEL_FROM_CHANNEL_STACK(channel_stack) \
(((grpc_channel *)(channel_stack)) - 1)
#define CHANNEL_FROM_TOP_ELEM(top_elem) \
CHANNEL_FROM_CHANNEL_STACK(grpc_channel_stack_from_top_element(top_elem))
@ -66,7 +76,8 @@ grpc_channel *grpc_channel_create_from_filters(
grpc_channel *channel = gpr_malloc(size);
GPR_ASSERT(grpc_is_initialized() && "call grpc_init()");
channel->is_client = is_client;
/* decremented by grpc_channel_destroy, and grpc_client_channel_closed if is_client */
/* decremented by grpc_channel_destroy, and grpc_client_channel_closed if
* is_client */
gpr_ref_init(&channel->refs, 1 + is_client);
channel->metadata_context = mdctx;
channel->grpc_status_string = grpc_mdstr_from_string(mdctx, "grpc-status");
@ -75,18 +86,17 @@ grpc_channel *grpc_channel_create_from_filters(
channel->authority_string = grpc_mdstr_from_string(mdctx, ":authority");
grpc_channel_stack_init(filters, num_filters, args, channel->metadata_context,
CHANNEL_STACK_FROM_CHANNEL(channel));
gpr_mu_init(&channel->registered_call_mu);
channel->registered_calls = NULL;
return channel;
}
static void do_nothing(void *ignored, grpc_op_error error) {}
grpc_call *grpc_channel_create_call(grpc_channel *channel,
grpc_completion_queue *cq,
const char *method, const char *host,
gpr_timespec absolute_deadline) {
static grpc_call *grpc_channel_create_call_internal(
grpc_channel *channel, grpc_completion_queue *cq, grpc_mdelem *path_mdelem,
grpc_mdelem *authority_mdelem, gpr_timespec deadline) {
grpc_call *call;
grpc_mdelem *path_mdelem;
grpc_mdelem *authority_mdelem;
grpc_call_op op;
if (!channel->is_client) {
@ -97,11 +107,6 @@ grpc_call *grpc_channel_create_call(grpc_channel *channel,
call = grpc_call_create(channel, cq, NULL);
/* Add :path and :authority headers. */
/* TODO(klempner): Consider optimizing this by stashing mdelems for common
values of method and host. */
path_mdelem = grpc_mdelem_from_metadata_strings(
channel->metadata_context, grpc_mdstr_ref(channel->path_string),
grpc_mdstr_from_string(channel->metadata_context, method));
op.type = GRPC_SEND_METADATA;
op.dir = GRPC_CALL_DOWN;
op.flags = 0;
@ -110,18 +115,14 @@ grpc_call *grpc_channel_create_call(grpc_channel *channel,
op.user_data = NULL;
grpc_call_execute_op(call, &op);
grpc_mdstr_ref(channel->authority_string);
authority_mdelem = grpc_mdelem_from_metadata_strings(
channel->metadata_context, channel->authority_string,
grpc_mdstr_from_string(channel->metadata_context, host));
op.data.metadata = authority_mdelem;
grpc_call_execute_op(call, &op);
if (0 != gpr_time_cmp(absolute_deadline, gpr_inf_future)) {
if (0 != gpr_time_cmp(deadline, gpr_inf_future)) {
op.type = GRPC_SEND_DEADLINE;
op.dir = GRPC_CALL_DOWN;
op.flags = 0;
op.data.deadline = absolute_deadline;
op.data.deadline = deadline;
op.done_cb = do_nothing;
op.user_data = NULL;
grpc_call_execute_op(call, &op);
@ -130,6 +131,21 @@ grpc_call *grpc_channel_create_call(grpc_channel *channel,
return call;
}
grpc_call *grpc_channel_create_call(grpc_channel *channel,
grpc_completion_queue *cq,
const char *method, const char *host,
gpr_timespec deadline) {
return grpc_channel_create_call_internal(
channel, cq,
grpc_mdelem_from_metadata_strings(
channel->metadata_context, grpc_mdstr_ref(channel->path_string),
grpc_mdstr_from_string(channel->metadata_context, method)),
grpc_mdelem_from_metadata_strings(
channel->metadata_context, grpc_mdstr_ref(channel->authority_string),
grpc_mdstr_from_string(channel->metadata_context, host)),
deadline);
}
grpc_call *grpc_channel_create_call_old(grpc_channel *channel,
const char *method, const char *host,
gpr_timespec absolute_deadline) {
@ -137,6 +153,31 @@ grpc_call *grpc_channel_create_call_old(grpc_channel *channel,
absolute_deadline);
}
void *grpc_channel_register_call(grpc_channel *channel, const char *method,
const char *host) {
registered_call *rc = gpr_malloc(sizeof(registered_call));
rc->path = grpc_mdelem_from_metadata_strings(
channel->metadata_context, grpc_mdstr_ref(channel->path_string),
grpc_mdstr_from_string(channel->metadata_context, method));
rc->authority = grpc_mdelem_from_metadata_strings(
channel->metadata_context, grpc_mdstr_ref(channel->authority_string),
grpc_mdstr_from_string(channel->metadata_context, host));
gpr_mu_lock(&channel->registered_call_mu);
rc->next = channel->registered_calls;
channel->registered_calls = rc;
gpr_mu_unlock(&channel->registered_call_mu);
return rc;
}
grpc_call *grpc_channel_create_registered_call(
grpc_channel *channel, grpc_completion_queue *completion_queue,
void *registered_call_handle, gpr_timespec deadline) {
registered_call *rc = registered_call_handle;
return grpc_channel_create_call_internal(
channel, completion_queue, grpc_mdelem_ref(rc->path),
grpc_mdelem_ref(rc->authority), deadline);
}
void grpc_channel_internal_ref(grpc_channel *channel) {
gpr_ref(&channel->refs);
}
@ -148,7 +189,15 @@ static void destroy_channel(void *p, int ok) {
grpc_mdstr_unref(channel->grpc_message_string);
grpc_mdstr_unref(channel->path_string);
grpc_mdstr_unref(channel->authority_string);
while (channel->registered_calls) {
registered_call *rc = channel->registered_calls;
channel->registered_calls = rc->next;
grpc_mdelem_unref(rc->path);
grpc_mdelem_unref(rc->authority);
gpr_free(rc);
}
grpc_mdctx_unref(channel->metadata_context);
gpr_mu_destroy(&channel->registered_call_mu);
gpr_free(channel);
}

@ -61,7 +61,12 @@ Channel::~Channel() { grpc_channel_destroy(c_channel_); }
Call Channel::CreateCall(const RpcMethod& method, ClientContext* context,
CompletionQueue* cq) {
auto c_call = grpc_channel_create_call(c_channel_, cq->cq(), method.name(),
auto c_call =
method.channel_tag()
? grpc_channel_create_registered_call(c_channel_, cq->cq(),
method.channel_tag(),
context->RawDeadline())
: grpc_channel_create_call(c_channel_, cq->cq(), method.name(),
context->authority().empty()
? target_.c_str()
: context->authority().c_str(),
@ -82,4 +87,8 @@ void Channel::PerformOpsOnCall(CallOpBuffer* buf, Call* call) {
GRPC_TIMER_MARK(PERFORM_OPS_END, call->call());
}
void* Channel::RegisterMethod(const char* method) {
return grpc_channel_register_call(c_channel_, method, target_.c_str());
}
} // namespace grpc

@ -54,6 +54,7 @@ class Channel GRPC_FINAL : public ChannelInterface {
Channel(const grpc::string& target, grpc_channel* c_channel);
~Channel() GRPC_OVERRIDE;
virtual void *RegisterMethod(const char *method) GRPC_OVERRIDE;
virtual Call CreateCall(const RpcMethod& method, ClientContext* context,
CompletionQueue* cq) GRPC_OVERRIDE;
virtual void PerformOpsOnCall(CallOpBuffer* ops, Call* call) GRPC_OVERRIDE;

@ -39,13 +39,13 @@ namespace grpc {
// begin a call to a named method
std::unique_ptr<GenericClientAsyncReaderWriter> GenericStub::Call(
ClientContext* context, const grpc::string& method,
CompletionQueue* cq, void* tag) {
ClientContext* context, const grpc::string& method, CompletionQueue* cq,
void* tag) {
return std::unique_ptr<GenericClientAsyncReaderWriter>(
new GenericClientAsyncReaderWriter(
channel_.get(), cq, RpcMethod(method.c_str()), context, tag));
channel_.get(), cq,
RpcMethod(method.c_str(), RpcMethod::BIDI_STREAMING, nullptr),
context, tag));
}
} // namespace grpc

@ -69,6 +69,7 @@ END2END_TESTS = [
'request_with_payload',
'simple_delayed_request',
'simple_request',
'registered_call',
'thread_stress',
'writes_done_hangs_with_pending_read',

@ -0,0 +1,222 @@
/*
*
* Copyright 2015, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are
* met:
*
* * Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
* * Redistributions in binary form must reproduce the above
* copyright notice, this list of conditions and the following disclaimer
* in the documentation and/or other materials provided with the
* distribution.
* * Neither the name of Google Inc. nor the names of its
* contributors may be used to endorse or promote products derived from
* this software without specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*
*/
#include "test/core/end2end/end2end_tests.h"
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include "src/core/support/string.h"
#include <grpc/byte_buffer.h>
#include <grpc/grpc.h>
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
#include <grpc/support/time.h>
#include <grpc/support/useful.h>
#include "test/core/end2end/cq_verifier.h"
enum { TIMEOUT = 200000 };
static void *tag(gpr_intptr t) { return (void *)t; }
static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
const char *test_name,
grpc_channel_args *client_args,
grpc_channel_args *server_args) {
grpc_end2end_test_fixture f;
gpr_log(GPR_INFO, "%s/%s", test_name, config.name);
f = config.create_fixture(client_args, server_args);
config.init_client(&f, client_args);
config.init_server(&f, server_args);
return f;
}
static gpr_timespec n_seconds_time(int n) {
return GRPC_TIMEOUT_SECONDS_TO_DEADLINE(n);
}
static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
static void drain_cq(grpc_completion_queue *cq) {
grpc_event *ev;
grpc_completion_type type;
do {
ev = grpc_completion_queue_next(cq, five_seconds_time());
GPR_ASSERT(ev);
type = ev->type;
grpc_event_finish(ev);
} while (type != GRPC_QUEUE_SHUTDOWN);
}
static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return;
grpc_server_shutdown(f->server);
grpc_server_destroy(f->server);
f->server = NULL;
}
static void shutdown_client(grpc_end2end_test_fixture *f) {
if (!f->client) return;
grpc_channel_destroy(f->client);
f->client = NULL;
}
static void end_test(grpc_end2end_test_fixture *f) {
shutdown_server(f);
shutdown_client(f);
grpc_completion_queue_shutdown(f->server_cq);
drain_cq(f->server_cq);
grpc_completion_queue_destroy(f->server_cq);
grpc_completion_queue_shutdown(f->client_cq);
drain_cq(f->client_cq);
grpc_completion_queue_destroy(f->client_cq);
}
static void simple_request_body(grpc_end2end_test_fixture f, void *rc) {
grpc_call *c;
grpc_call *s;
gpr_timespec deadline = five_seconds_time();
cq_verifier *v_client = cq_verifier_create(f.client_cq);
cq_verifier *v_server = cq_verifier_create(f.server_cq);
grpc_op ops[6];
grpc_op *op;
grpc_metadata_array initial_metadata_recv;
grpc_metadata_array trailing_metadata_recv;
grpc_metadata_array request_metadata_recv;
grpc_call_details call_details;
grpc_status_code status;
char *details = NULL;
size_t details_capacity = 0;
int was_cancelled = 2;
c = grpc_channel_create_registered_call(f.client, f.client_cq, rc, deadline);
GPR_ASSERT(c);
grpc_metadata_array_init(&initial_metadata_recv);
grpc_metadata_array_init(&trailing_metadata_recv);
grpc_metadata_array_init(&request_metadata_recv);
grpc_call_details_init(&call_details);
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
op->data.send_initial_metadata.count = 0;
op++;
op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
op++;
op->op = GRPC_OP_RECV_INITIAL_METADATA;
op->data.recv_initial_metadata = &initial_metadata_recv;
op++;
op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
op->data.recv_status_on_client.status = &status;
op->data.recv_status_on_client.status_details = &details;
op->data.recv_status_on_client.status_details_capacity = &details_capacity;
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1)));
GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s,
&call_details,
&request_metadata_recv,
f.server_cq, tag(101)));
cq_expect_completion(v_server, tag(101), GRPC_OP_OK);
cq_verify(v_server);
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
op->data.send_initial_metadata.count = 0;
op++;
op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
op->data.send_status_from_server.trailing_metadata_count = 0;
op->data.send_status_from_server.status = GRPC_STATUS_UNIMPLEMENTED;
op->data.send_status_from_server.status_details = "xyz";
op++;
op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
op->data.recv_close_on_server.cancelled = &was_cancelled;
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
cq_expect_completion(v_server, tag(102), GRPC_OP_OK);
cq_verify(v_server);
cq_expect_completion(v_client, tag(1), GRPC_OP_OK);
cq_verify(v_client);
GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
GPR_ASSERT(0 == strcmp(details, "xyz"));
GPR_ASSERT(0 == strcmp(call_details.method, "/foo"));
GPR_ASSERT(0 == strcmp(call_details.host, "foo.test.google.fr:1234"));
GPR_ASSERT(was_cancelled == 0);
gpr_free(details);
grpc_metadata_array_destroy(&initial_metadata_recv);
grpc_metadata_array_destroy(&trailing_metadata_recv);
grpc_metadata_array_destroy(&request_metadata_recv);
grpc_call_details_destroy(&call_details);
grpc_call_destroy(c);
grpc_call_destroy(s);
cq_verifier_destroy(v_client);
cq_verifier_destroy(v_server);
}
static void test_invoke_simple_request(grpc_end2end_test_config config) {
grpc_end2end_test_fixture f = begin_test(config, __FUNCTION__, NULL, NULL);
void *rc =
grpc_channel_register_call(f.client, "/foo", "foo.test.google.fr:1234");
simple_request_body(f, rc);
end_test(&f);
config.tear_down_data(&f);
}
static void test_invoke_10_simple_requests(grpc_end2end_test_config config) {
int i;
grpc_end2end_test_fixture f = begin_test(config, __FUNCTION__, NULL, NULL);
void *rc =
grpc_channel_register_call(f.client, "/foo", "foo.test.google.fr:1234");
for (i = 0; i < 10; i++) {
simple_request_body(f, rc);
gpr_log(GPR_INFO, "Passed simple request %d", i);
}
end_test(&f);
config.tear_down_data(&f);
}
void grpc_end2end_tests(grpc_end2end_test_config config) {
test_invoke_simple_request(config);
test_invoke_10_simple_requests(config);
}

@ -918,6 +918,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_fake_security_registered_call_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -1359,6 +1368,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_fullstack_registered_call_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -1800,6 +1818,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_fullstack_uds_registered_call_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -2241,6 +2268,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_simple_ssl_fullstack_registered_call_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -2682,6 +2718,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_simple_ssl_with_oauth2_fullstack_registered_call_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -3123,6 +3168,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_socket_pair_registered_call_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -3564,6 +3618,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_socket_pair_one_byte_at_a_time_registered_call_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",

Loading…
Cancel
Save