Merge github.com:grpc/grpc into propagate
diff --git a/include/grpc++/server_context.h b/include/grpc++/server_context.h
index cf2732b..23273f4 100644
--- a/include/grpc++/server_context.h
+++ b/include/grpc++/server_context.h
@@ -125,6 +125,14 @@
const struct census_context* census_context() const;
+ // Async only. Has to be called before the rpc starts.
+ // Returns the tag in completion queue when the rpc finishes.
+ // IsCancelled() can then be called to check whether the rpc was cancelled.
+ void AsyncNotifyWhenDone(void* tag) {
+ has_notify_when_done_tag_ = true;
+ async_notify_when_done_tag_ = tag;
+ }
+
private:
friend class ::grpc::testing::InteropContextInspector;
friend class ::grpc::Server;
@@ -165,6 +173,8 @@
void set_call(grpc_call* call);
CompletionOp* completion_op_;
+ bool has_notify_when_done_tag_;
+ void* async_notify_when_done_tag_;
gpr_timespec deadline_;
grpc_call* call_;
diff --git a/include/grpc/grpc.h b/include/grpc/grpc.h
index b928f12..407f36a 100644
--- a/include/grpc/grpc.h
+++ b/include/grpc/grpc.h
@@ -589,7 +589,7 @@
/** Add a HTTP2 over plaintext over tcp listener.
Returns bound port number on success, 0 on failure.
REQUIRES: server not started */
-int grpc_server_add_http2_port(grpc_server *server, const char *addr);
+int grpc_server_add_insecure_http2_port(grpc_server *server, const char *addr);
/** Start a server - tells all listeners to start listening */
void grpc_server_start(grpc_server *server);
diff --git a/src/core/surface/server_chttp2.c b/src/core/surface/server_chttp2.c
index 78c5346..4ab845b 100644
--- a/src/core/surface/server_chttp2.c
+++ b/src/core/surface/server_chttp2.c
@@ -80,7 +80,7 @@
grpc_tcp_server_destroy(tcp, grpc_server_listener_destroy_done, server);
}
-int grpc_server_add_http2_port(grpc_server *server, const char *addr) {
+int grpc_server_add_insecure_http2_port(grpc_server *server, const char *addr) {
grpc_resolved_addresses *resolved = NULL;
grpc_tcp_server *tcp = NULL;
size_t i;
diff --git a/src/cpp/server/insecure_server_credentials.cc b/src/cpp/server/insecure_server_credentials.cc
index aca3568..800cd36 100644
--- a/src/cpp/server/insecure_server_credentials.cc
+++ b/src/cpp/server/insecure_server_credentials.cc
@@ -41,7 +41,7 @@
public:
int AddPortToServer(const grpc::string& addr,
grpc_server* server) GRPC_OVERRIDE {
- return grpc_server_add_http2_port(server, addr.c_str());
+ return grpc_server_add_insecure_http2_port(server, addr.c_str());
}
};
} // namespace
diff --git a/src/cpp/server/server_context.cc b/src/cpp/server/server_context.cc
index cf19556..0437339 100644
--- a/src/cpp/server/server_context.cc
+++ b/src/cpp/server/server_context.cc
@@ -50,16 +50,23 @@
class ServerContext::CompletionOp GRPC_FINAL : public CallOpSetInterface {
public:
// initial refs: one in the server context, one in the cq
- CompletionOp() : refs_(2), finalized_(false), cancelled_(0) {}
+ CompletionOp() : has_tag_(false), tag_(nullptr), refs_(2), finalized_(false), cancelled_(0) {}
void FillOps(grpc_op* ops, size_t* nops) GRPC_OVERRIDE;
bool FinalizeResult(void** tag, bool* status) GRPC_OVERRIDE;
bool CheckCancelled(CompletionQueue* cq);
+ void set_tag(void* tag) {
+ has_tag_ = true;
+ tag_ = tag;
+ }
+
void Unref();
private:
+ bool has_tag_;
+ void* tag_;
grpc::mutex mu_;
int refs_;
bool finalized_;
@@ -90,18 +97,25 @@
bool ServerContext::CompletionOp::FinalizeResult(void** tag, bool* status) {
grpc::unique_lock<grpc::mutex> lock(mu_);
finalized_ = true;
+ bool ret = false;
+ if (has_tag_) {
+ *tag = tag_;
+ ret = true;
+ }
if (!*status) cancelled_ = 1;
if (--refs_ == 0) {
lock.unlock();
delete this;
}
- return false;
+ return ret;
}
// ServerContext body
ServerContext::ServerContext()
: completion_op_(nullptr),
+ has_notify_when_done_tag_(false),
+ async_notify_when_done_tag_(nullptr),
call_(nullptr),
cq_(nullptr),
sent_initial_metadata_(false) {}
@@ -109,6 +123,8 @@
ServerContext::ServerContext(gpr_timespec deadline, grpc_metadata* metadata,
size_t metadata_count)
: completion_op_(nullptr),
+ has_notify_when_done_tag_(false),
+ async_notify_when_done_tag_(nullptr),
deadline_(deadline),
call_(nullptr),
cq_(nullptr),
@@ -133,6 +149,9 @@
void ServerContext::BeginCompletionOp(Call* call) {
GPR_ASSERT(!completion_op_);
completion_op_ = new CompletionOp();
+ if (has_notify_when_done_tag_) {
+ completion_op_->set_tag(async_notify_when_done_tag_);
+ }
call->PerformOps(completion_op_);
}
diff --git a/src/csharp/ext/grpc_csharp_ext.c b/src/csharp/ext/grpc_csharp_ext.c
index 65f3303..c12b636 100644
--- a/src/csharp/ext/grpc_csharp_ext.c
+++ b/src/csharp/ext/grpc_csharp_ext.c
@@ -732,7 +732,7 @@
GPR_EXPORT gpr_int32 GPR_CALLTYPE
grpcsharp_server_add_insecure_http2_port(grpc_server *server, const char *addr) {
- return grpc_server_add_http2_port(server, addr);
+ return grpc_server_add_insecure_http2_port(server, addr);
}
GPR_EXPORT void GPR_CALLTYPE grpcsharp_server_start(grpc_server *server) {
diff --git a/src/node/ext/server.cc b/src/node/ext/server.cc
index 04fabc8..1dc179d 100644
--- a/src/node/ext/server.cc
+++ b/src/node/ext/server.cc
@@ -265,8 +265,8 @@
grpc_server_credentials *creds = creds_object->GetWrappedServerCredentials();
int port;
if (creds == NULL) {
- port = grpc_server_add_http2_port(server->wrapped_server,
- *NanUtf8String(args[0]));
+ port = grpc_server_add_insecure_http2_port(server->wrapped_server,
+ *NanUtf8String(args[0]));
} else {
port = grpc_server_add_secure_http2_port(server->wrapped_server,
*NanUtf8String(args[0]),
diff --git a/src/php/ext/grpc/server.c b/src/php/ext/grpc/server.c
index 8b8d5b2..d58aa88 100644
--- a/src/php/ext/grpc/server.c
+++ b/src/php/ext/grpc/server.c
@@ -182,7 +182,7 @@
"add_http2_port expects a string", 1 TSRMLS_CC);
return;
}
- RETURN_LONG(grpc_server_add_http2_port(server->wrapped, addr));
+ RETURN_LONG(grpc_server_add_insecure_http2_port(server->wrapped, addr));
}
PHP_METHOD(Server, addSecureHttp2Port) {
diff --git a/src/python/grpcio/grpc/_adapter/_c/types/server.c b/src/python/grpcio/grpc/_adapter/_c/types/server.c
index 2a00f34..c2190ea 100644
--- a/src/python/grpcio/grpc/_adapter/_c/types/server.c
+++ b/src/python/grpcio/grpc/_adapter/_c/types/server.c
@@ -155,7 +155,7 @@
port = grpc_server_add_secure_http2_port(
self->c_serv, addr, creds->c_creds);
} else {
- port = grpc_server_add_http2_port(self->c_serv, addr);
+ port = grpc_server_add_insecure_http2_port(self->c_serv, addr);
}
return PyInt_FromLong(port);
diff --git a/src/ruby/ext/grpc/rb_server.c b/src/ruby/ext/grpc/rb_server.c
index 375a651..79a4ae8 100644
--- a/src/ruby/ext/grpc/rb_server.c
+++ b/src/ruby/ext/grpc/rb_server.c
@@ -357,7 +357,8 @@
rb_raise(rb_eRuntimeError, "destroyed!");
return Qnil;
} else if (rb_creds == Qnil) {
- recvd_port = grpc_server_add_http2_port(s->wrapped, StringValueCStr(port));
+ recvd_port =
+ grpc_server_add_insecure_http2_port(s->wrapped, StringValueCStr(port));
if (recvd_port == 0) {
rb_raise(rb_eRuntimeError,
"could not add port %s to server, not sure why",
diff --git a/test/core/end2end/dualstack_socket_test.c b/test/core/end2end/dualstack_socket_test.c
index 3c227b7..48a88be 100644
--- a/test/core/end2end/dualstack_socket_test.c
+++ b/test/core/end2end/dualstack_socket_test.c
@@ -96,8 +96,8 @@
cq = grpc_completion_queue_create();
server = grpc_server_create(NULL);
grpc_server_register_completion_queue(server, cq);
- GPR_ASSERT((got_port = grpc_server_add_http2_port(server, server_hostport)) >
- 0);
+ GPR_ASSERT((got_port = grpc_server_add_insecure_http2_port(
+ server, server_hostport)) > 0);
if (port == 0) {
port = got_port;
} else {
diff --git a/test/core/end2end/fixtures/chttp2_fullstack.c b/test/core/end2end/fixtures/chttp2_fullstack.c
index 6647b94..53a6f0d 100644
--- a/test/core/end2end/fixtures/chttp2_fullstack.c
+++ b/test/core/end2end/fixtures/chttp2_fullstack.c
@@ -84,7 +84,7 @@
}
f->server = grpc_server_create(server_args);
grpc_server_register_completion_queue(f->server, f->cq);
- GPR_ASSERT(grpc_server_add_http2_port(f->server, ffd->localaddr));
+ GPR_ASSERT(grpc_server_add_insecure_http2_port(f->server, ffd->localaddr));
grpc_server_start(f->server);
}
diff --git a/test/core/end2end/fixtures/chttp2_fullstack_compression.c b/test/core/end2end/fixtures/chttp2_fullstack_compression.c
index 7eb247b..a75075d 100644
--- a/test/core/end2end/fixtures/chttp2_fullstack_compression.c
+++ b/test/core/end2end/fixtures/chttp2_fullstack_compression.c
@@ -99,7 +99,7 @@
}
f->server = grpc_server_create(ffd->server_args_compression);
grpc_server_register_completion_queue(f->server, f->cq);
- GPR_ASSERT(grpc_server_add_http2_port(f->server, ffd->localaddr));
+ GPR_ASSERT(grpc_server_add_insecure_http2_port(f->server, ffd->localaddr));
grpc_server_start(f->server);
}
diff --git a/test/core/end2end/fixtures/chttp2_fullstack_uds_posix.c b/test/core/end2end/fixtures/chttp2_fullstack_uds_posix.c
index 89ad7b8..20afdb8 100644
--- a/test/core/end2end/fixtures/chttp2_fullstack_uds_posix.c
+++ b/test/core/end2end/fixtures/chttp2_fullstack_uds_posix.c
@@ -89,7 +89,7 @@
}
f->server = grpc_server_create(server_args);
grpc_server_register_completion_queue(f->server, f->cq);
- GPR_ASSERT(grpc_server_add_http2_port(f->server, ffd->localaddr));
+ GPR_ASSERT(grpc_server_add_insecure_http2_port(f->server, ffd->localaddr));
grpc_server_start(f->server);
}
diff --git a/test/core/end2end/fixtures/chttp2_fullstack_uds_posix_with_poll.c b/test/core/end2end/fixtures/chttp2_fullstack_uds_posix_with_poll.c
index a2ab25d..8491ea6 100644
--- a/test/core/end2end/fixtures/chttp2_fullstack_uds_posix_with_poll.c
+++ b/test/core/end2end/fixtures/chttp2_fullstack_uds_posix_with_poll.c
@@ -89,7 +89,7 @@
}
f->server = grpc_server_create(server_args);
grpc_server_register_completion_queue(f->server, f->cq);
- GPR_ASSERT(grpc_server_add_http2_port(f->server, ffd->localaddr));
+ GPR_ASSERT(grpc_server_add_insecure_http2_port(f->server, ffd->localaddr));
grpc_server_start(f->server);
}
diff --git a/test/core/end2end/fixtures/chttp2_fullstack_with_poll.c b/test/core/end2end/fixtures/chttp2_fullstack_with_poll.c
index 93786d0..2a4835a 100644
--- a/test/core/end2end/fixtures/chttp2_fullstack_with_poll.c
+++ b/test/core/end2end/fixtures/chttp2_fullstack_with_poll.c
@@ -83,7 +83,7 @@
}
f->server = grpc_server_create(server_args);
grpc_server_register_completion_queue(f->server, f->cq);
- GPR_ASSERT(grpc_server_add_http2_port(f->server, ffd->localaddr));
+ GPR_ASSERT(grpc_server_add_insecure_http2_port(f->server, ffd->localaddr));
grpc_server_start(f->server);
}
diff --git a/test/core/end2end/multiple_server_queues_test.c b/test/core/end2end/multiple_server_queues_test.c
index 208d42e..7772d14 100644
--- a/test/core/end2end/multiple_server_queues_test.c
+++ b/test/core/end2end/multiple_server_queues_test.c
@@ -45,7 +45,7 @@
cq2 = grpc_completion_queue_create();
server = grpc_server_create(NULL);
grpc_server_register_completion_queue(server, cq1);
- grpc_server_add_http2_port(server, "[::]:0");
+ grpc_server_add_insecure_http2_port(server, "[::]:0");
grpc_server_register_completion_queue(server, cq2);
grpc_server_start(server);
grpc_server_shutdown_and_notify(server, cq2, NULL);
diff --git a/test/core/fling/server.c b/test/core/fling/server.c
index 8f34904..f445c68 100644
--- a/test/core/fling/server.c
+++ b/test/core/fling/server.c
@@ -216,7 +216,7 @@
grpc_server_credentials_release(ssl_creds);
} else {
server = grpc_server_create(NULL);
- GPR_ASSERT(grpc_server_add_http2_port(server, addr));
+ GPR_ASSERT(grpc_server_add_insecure_http2_port(server, addr));
}
grpc_server_register_completion_queue(server, cq);
grpc_server_start(server);
diff --git a/test/cpp/end2end/async_end2end_test.cc b/test/cpp/end2end/async_end2end_test.cc
index b95bdf6..9b53bdc 100644
--- a/test/cpp/end2end/async_end2end_test.cc
+++ b/test/cpp/end2end/async_end2end_test.cc
@@ -592,6 +592,80 @@
EXPECT_EQ(meta6.second, server_trailing_metadata.find(meta6.first)->second);
EXPECT_GE(server_trailing_metadata.size(), static_cast<size_t>(2));
}
+
+// Server uses AsyncNotifyWhenDone API to check for cancellation
+TEST_F(AsyncEnd2endTest, ServerCheckCancellation) {
+ ResetStub();
+
+ EchoRequest send_request;
+ EchoRequest recv_request;
+ EchoResponse send_response;
+ EchoResponse recv_response;
+ Status recv_status;
+
+ ClientContext cli_ctx;
+ ServerContext srv_ctx;
+ grpc::ServerAsyncResponseWriter<EchoResponse> response_writer(&srv_ctx);
+
+ send_request.set_message("Hello");
+ std::unique_ptr<ClientAsyncResponseReader<EchoResponse> > response_reader(
+ stub_->AsyncEcho(&cli_ctx, send_request, cq_.get()));
+
+ srv_ctx.AsyncNotifyWhenDone(tag(5));
+ service_.RequestEcho(&srv_ctx, &recv_request, &response_writer, cq_.get(),
+ cq_.get(), tag(2));
+
+ Verifier().Expect(2, true).Verify(cq_.get());
+ EXPECT_EQ(send_request.message(), recv_request.message());
+
+ cli_ctx.TryCancel();
+ Verifier().Expect(5, true).Verify(cq_.get());
+ EXPECT_TRUE(srv_ctx.IsCancelled());
+
+ response_reader->Finish(&recv_response, &recv_status, tag(4));
+ Verifier().Expect(4, false).Verify(cq_.get());
+
+ EXPECT_EQ(StatusCode::CANCELLED, recv_status.error_code());
+}
+
+// Server uses AsyncNotifyWhenDone API to check for normal finish
+TEST_F(AsyncEnd2endTest, ServerCheckDone) {
+ ResetStub();
+
+ EchoRequest send_request;
+ EchoRequest recv_request;
+ EchoResponse send_response;
+ EchoResponse recv_response;
+ Status recv_status;
+
+ ClientContext cli_ctx;
+ ServerContext srv_ctx;
+ grpc::ServerAsyncResponseWriter<EchoResponse> response_writer(&srv_ctx);
+
+ send_request.set_message("Hello");
+ std::unique_ptr<ClientAsyncResponseReader<EchoResponse> > response_reader(
+ stub_->AsyncEcho(&cli_ctx, send_request, cq_.get()));
+
+ srv_ctx.AsyncNotifyWhenDone(tag(5));
+ service_.RequestEcho(&srv_ctx, &recv_request, &response_writer, cq_.get(),
+ cq_.get(), tag(2));
+
+ Verifier().Expect(2, true).Verify(cq_.get());
+ EXPECT_EQ(send_request.message(), recv_request.message());
+
+ send_response.set_message(recv_request.message());
+ response_writer.Finish(send_response, Status::OK, tag(3));
+ Verifier().Expect(3, true).Verify(cq_.get());
+ Verifier().Expect(5, true).Verify(cq_.get());
+ EXPECT_FALSE(srv_ctx.IsCancelled());
+
+ response_reader->Finish(&recv_response, &recv_status, tag(4));
+ Verifier().Expect(4, true).Verify(cq_.get());
+
+ EXPECT_EQ(send_response.message(), recv_response.message());
+ EXPECT_TRUE(recv_status.ok());
+}
+
} // namespace
} // namespace testing
} // namespace grpc