aboutsummaryrefslogtreecommitdiffhomepage
path: root/src
diff options
context:
space:
mode:
authorGravatar David Garcia Quintas <dgq@google.com>2015-07-23 04:26:24 -0700
committerGravatar David Garcia Quintas <dgq@google.com>2015-07-23 04:26:24 -0700
commitbaa2aa644226b00ad9cb493660356f4473acd212 (patch)
tree13ffebed603924ac83e0c77df5133b855f826de0 /src
parent8ec09f6530938c6126a6579ce85ee07dbf71d785 (diff)
parentd435caa13eb3710a4d3d17825b0efe3a06336ab7 (diff)
Merge branch 'master' of github.com:grpc/grpc into compression-accept-encoding
Diffstat (limited to 'src')
-rw-r--r--src/compiler/csharp_generator.cc4
-rw-r--r--src/compiler/objective_c_generator.cc3
-rw-r--r--src/compiler/objective_c_plugin.cc5
-rw-r--r--src/core/channel/client_channel.c28
-rw-r--r--src/core/channel/compress_filter.c2
-rw-r--r--src/core/channel/http_client_filter.c74
-rw-r--r--src/core/client_config/subchannel.c4
-rw-r--r--src/core/iomgr/alarm.c8
-rw-r--r--src/core/iomgr/iomgr.c10
-rw-r--r--src/core/iomgr/pollset_posix.c4
-rw-r--r--src/core/iomgr/pollset_windows.c2
-rw-r--r--src/core/iomgr/tcp_client_posix.c8
-rw-r--r--src/core/iomgr/tcp_client_windows.c2
-rw-r--r--src/core/support/sync_posix.c3
-rw-r--r--src/core/support/sync_win32.c4
-rw-r--r--src/core/support/time.c27
-rw-r--r--src/core/support/time_posix.c2
-rw-r--r--src/core/support/time_win32.c2
-rw-r--r--src/core/surface/call.c6
-rw-r--r--src/core/surface/call.h7
-rw-r--r--src/core/surface/call_log_batch.c8
-rw-r--r--src/core/surface/completion_queue.c4
-rw-r--r--src/core/surface/server.c2
-rw-r--r--src/core/transport/chttp2/parsing.c2
-rw-r--r--src/core/transport/chttp2/stream_encoder.c9
-rw-r--r--src/core/transport/transport_op_string.c2
-rw-r--r--src/cpp/client/channel_arguments.cc40
-rw-r--r--src/cpp/client/client_context.cc2
-rw-r--r--src/cpp/client/create_channel.cc9
-rw-r--r--src/csharp/Grpc.Auth/GoogleCredential.cs16
-rw-r--r--src/csharp/Grpc.Core.Tests/ClientServerTest.cs133
-rw-r--r--src/csharp/Grpc.Core.Tests/Internal/MetadataArraySafeHandleTest.cs24
-rw-r--r--src/csharp/Grpc.Core.Tests/TimespecTest.cs13
-rw-r--r--src/csharp/Grpc.Core/AsyncClientStreamingCall.cs18
-rw-r--r--src/csharp/Grpc.Core/AsyncDuplexStreamingCall.cs24
-rw-r--r--src/csharp/Grpc.Core/AsyncServerStreamingCall.cs24
-rw-r--r--src/csharp/Grpc.Core/AsyncUnaryCall.cs106
-rw-r--r--src/csharp/Grpc.Core/Calls.cs10
-rw-r--r--src/csharp/Grpc.Core/Channel.cs25
-rw-r--r--src/csharp/Grpc.Core/ChannelOptions.cs30
-rw-r--r--src/csharp/Grpc.Core/Grpc.Core.csproj9
-rw-r--r--src/csharp/Grpc.Core/Internal/AsyncCall.cs43
-rw-r--r--src/csharp/Grpc.Core/Internal/AsyncCallServer.cs7
-rw-r--r--src/csharp/Grpc.Core/Internal/BatchContextSafeHandle.cs146
-rw-r--r--src/csharp/Grpc.Core/Internal/CallSafeHandle.cs6
-rw-r--r--src/csharp/Grpc.Core/Internal/MetadataArraySafeHandle.cs46
-rw-r--r--src/csharp/Grpc.Core/Internal/ServerCallHandler.cs79
-rw-r--r--src/csharp/Grpc.Core/Internal/ServerResponseStream.cs4
-rw-r--r--src/csharp/Grpc.Core/Internal/Timespec.cs14
-rw-r--r--src/csharp/Grpc.Core/Metadata.cs5
-rw-r--r--src/csharp/Grpc.Core/Server.cs28
-rw-r--r--src/csharp/Grpc.Core/ServerCallContext.cs90
-rw-r--r--src/csharp/Grpc.Core/ServerMethods.cs8
-rw-r--r--src/csharp/Grpc.Core/Version.cs2
-rw-r--r--src/csharp/Grpc.Core/VersionInfo.cs13
-rw-r--r--src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs2
-rw-r--r--src/csharp/Grpc.Examples/MathExamples.cs7
-rw-r--r--src/csharp/Grpc.Examples/MathGrpc.cs12
-rw-r--r--src/csharp/Grpc.Examples/MathServiceImpl.cs8
-rw-r--r--src/csharp/Grpc.HealthCheck.Tests/HealthClientServerTest.cs4
-rw-r--r--src/csharp/Grpc.HealthCheck.Tests/HealthServiceImplTest.cs2
-rw-r--r--src/csharp/Grpc.HealthCheck/HealthGrpc.cs6
-rw-r--r--src/csharp/Grpc.HealthCheck/HealthServiceImpl.cs2
-rw-r--r--src/csharp/Grpc.HealthCheck/Settings.StyleCop10
-rw-r--r--src/csharp/Grpc.IntegrationTesting.Client/Grpc.IntegrationTesting.Client.csproj6
-rw-r--r--src/csharp/Grpc.IntegrationTesting.Server/Grpc.IntegrationTesting.Server.csproj6
-rw-r--r--src/csharp/Grpc.IntegrationTesting/Grpc.IntegrationTesting.csproj32
-rw-r--r--src/csharp/Grpc.IntegrationTesting/InteropClient.cs57
-rw-r--r--src/csharp/Grpc.IntegrationTesting/TestGrpc.cs20
-rw-r--r--src/csharp/Grpc.IntegrationTesting/TestServiceImpl.cs12
-rw-r--r--src/csharp/ext/grpc_csharp_ext.c64
-rw-r--r--src/node/README.md4
-rw-r--r--src/node/examples/math_server.js18
-rw-r--r--src/node/examples/route_guide_server.js16
-rw-r--r--src/node/examples/stock_server.js15
-rw-r--r--src/node/index.js6
-rw-r--r--src/node/interop/interop_client.js49
-rw-r--r--src/node/interop/interop_server.js20
-rw-r--r--src/node/src/client.js8
-rw-r--r--src/node/src/server.js255
-rw-r--r--src/node/test/health_test.js9
-rw-r--r--src/node/test/interop_sanity_test.js2
-rw-r--r--src/node/test/math_client_test.js2
-rw-r--r--src/node/test/surface_test.js296
-rw-r--r--src/objective-c/GRPCClient/GRPCCall.m32
-rw-r--r--src/objective-c/GRPCClient/private/GRPCCompletionQueue.m3
-rw-r--r--src/objective-c/GRPCClient/private/GRPCWrappedCall.m7
-rw-r--r--src/objective-c/RxLibrary/GRXConcurrentWriteable.h (renamed from src/objective-c/GRPCClient/private/GRPCDelegateWrapper.h)52
-rw-r--r--src/objective-c/RxLibrary/GRXConcurrentWriteable.m (renamed from src/objective-c/GRPCClient/private/GRPCDelegateWrapper.m)50
-rw-r--r--src/objective-c/RxLibrary/GRXImmediateWriter.m19
-rw-r--r--src/objective-c/RxLibrary/private/GRXNSFastEnumerator.m1
-rw-r--r--src/objective-c/generated_libraries/RemoteTestClient/RemoteTest.podspec8
-rw-r--r--src/objective-c/generated_libraries/RouteGuideClient/RouteGuide.podspec8
-rw-r--r--src/objective-c/tests/Tests.xcodeproj/project.pbxproj2
-rw-r--r--src/objective-c/tests/Tests.xcodeproj/xcshareddata/xcschemes/AllTests.xcscheme104
-rwxr-xr-xsrc/objective-c/tests/run_tests.sh49
-rw-r--r--src/php/ext/grpc/server.c1
-rwxr-xr-xsrc/php/tests/unit_tests/EndToEndTest.php2
-rwxr-xr-xsrc/php/tests/unit_tests/SecureEndToEndTest.php1
-rw-r--r--src/python/src/grpc/_adapter/_c/utility.c2
-rw-r--r--src/python/src/grpc/_adapter/_low_test.py5
-rw-r--r--src/python/src/grpc/_links/_transmission_test.py9
-rw-r--r--src/python/src/grpc/framework/interfaces/links/test_cases.py11
-rw-r--r--src/ruby/spec/generic/rpc_server_spec.rb18
104 files changed, 1873 insertions, 667 deletions
diff --git a/src/compiler/csharp_generator.cc b/src/compiler/csharp_generator.cc
index 1910e9bd2d..64371047e0 100644
--- a/src/compiler/csharp_generator.cc
+++ b/src/compiler/csharp_generator.cc
@@ -149,7 +149,7 @@ std::string GetMethodRequestParamMaybe(const MethodDescriptor *method) {
std::string GetMethodReturnTypeClient(const MethodDescriptor *method) {
switch (GetMethodType(method)) {
case METHODTYPE_NO_STREAMING:
- return "Task<" + GetClassName(method->output_type()) + ">";
+ return "AsyncUnaryCall<" + GetClassName(method->output_type()) + ">";
case METHODTYPE_CLIENT_STREAMING:
return "AsyncClientStreamingCall<" + GetClassName(method->input_type())
+ ", " + GetClassName(method->output_type()) + ">";
@@ -298,7 +298,7 @@ void GenerateServerInterface(Printer* out, const ServiceDescriptor *service) {
out->Indent();
for (int i = 0; i < service->method_count(); i++) {
const MethodDescriptor *method = service->method(i);
- out->Print("$returntype$ $methodname$(ServerCallContext context, $request$$response_stream_maybe$);\n",
+ out->Print("$returntype$ $methodname$($request$$response_stream_maybe$, ServerCallContext context);\n",
"methodname", method->name(), "returntype",
GetMethodReturnTypeServer(method), "request",
GetMethodRequestParamServer(method), "response_stream_maybe",
diff --git a/src/compiler/objective_c_generator.cc b/src/compiler/objective_c_generator.cc
index 711d0d5870..69b3805bb1 100644
--- a/src/compiler/objective_c_generator.cc
+++ b/src/compiler/objective_c_generator.cc
@@ -186,9 +186,6 @@ string GetHeader(const ServiceDescriptor *service) {
grpc::protobuf::io::StringOutputStream output_stream(&output);
Printer printer(&output_stream, '$');
- printer.Print("@protocol GRXWriteable;\n");
- printer.Print("@protocol GRXWriter;\n\n");
-
map<string, string> vars = {{"service_class", ServiceClassName(service)}};
printer.Print(vars, "@protocol $service_class$ <NSObject>\n\n");
diff --git a/src/compiler/objective_c_plugin.cc b/src/compiler/objective_c_plugin.cc
index 2b5ab758fc..10f06ad4df 100644
--- a/src/compiler/objective_c_plugin.cc
+++ b/src/compiler/objective_c_plugin.cc
@@ -63,7 +63,9 @@ class ObjectiveCGrpcGenerator : public grpc::protobuf::compiler::CodeGenerator {
// Generate .pbrpc.h
string imports = string("#import \"") + file_name + ".pbobjc.h\"\n\n"
- "#import <ProtoRPC/ProtoService.h>\n";
+ "#import <ProtoRPC/ProtoService.h>\n"
+ "#import <RxLibrary/GRXWriteable.h>\n"
+ "#import <RxLibrary/GRXWriter.h>\n";
// TODO(jcanizales): Instead forward-declare the input and output types
// and import the files in the .pbrpc.m
@@ -89,7 +91,6 @@ class ObjectiveCGrpcGenerator : public grpc::protobuf::compiler::CodeGenerator {
string imports = string("#import \"") + file_name + ".pbrpc.h\"\n\n"
"#import <ProtoRPC/ProtoRPC.h>\n"
- "#import <RxLibrary/GRXWriteable.h>\n"
"#import <RxLibrary/GRXWriter+Immediate.h>\n";
string definitions;
diff --git a/src/core/channel/client_channel.c b/src/core/channel/client_channel.c
index 10e01ebbb4..c1aa580b2d 100644
--- a/src/core/channel/client_channel.c
+++ b/src/core/channel/client_channel.c
@@ -236,21 +236,6 @@ static void picked_target(void *arg, int iomgr_success) {
}
}
-static void pick_target(grpc_lb_policy *lb_policy, call_data *calld) {
- grpc_metadata_batch *initial_metadata;
- grpc_transport_stream_op *op = &calld->waiting_op;
-
- GPR_ASSERT(op->bind_pollset);
- GPR_ASSERT(op->send_ops);
- GPR_ASSERT(op->send_ops->nops >= 1);
- GPR_ASSERT(op->send_ops->ops[0].type == GRPC_OP_METADATA);
- initial_metadata = &op->send_ops->ops[0].data.metadata;
-
- grpc_iomgr_closure_init(&calld->async_setup_task, picked_target, calld);
- grpc_lb_policy_pick(lb_policy, op->bind_pollset, initial_metadata,
- &calld->picked_channel, &calld->async_setup_task);
-}
-
static grpc_iomgr_closure *merge_into_waiting_op(
grpc_call_element *elem, grpc_transport_stream_op *new_op) {
call_data *calld = elem->call_data;
@@ -358,12 +343,23 @@ static void perform_transport_stream_op(grpc_call_element *elem,
gpr_mu_lock(&chand->mu_config);
lb_policy = chand->lb_policy;
if (lb_policy) {
+ grpc_transport_stream_op *op = &calld->waiting_op;
+ grpc_pollset *bind_pollset = op->bind_pollset;
+ grpc_metadata_batch *initial_metadata = &op->send_ops->ops[0].data.metadata;
GRPC_LB_POLICY_REF(lb_policy, "pick");
gpr_mu_unlock(&chand->mu_config);
calld->state = CALL_WAITING_FOR_PICK;
+
+ GPR_ASSERT(op->bind_pollset);
+ GPR_ASSERT(op->send_ops);
+ GPR_ASSERT(op->send_ops->nops >= 1);
+ GPR_ASSERT(
+ op->send_ops->ops[0].type == GRPC_OP_METADATA);
gpr_mu_unlock(&calld->mu_state);
- pick_target(lb_policy, calld);
+ grpc_iomgr_closure_init(&calld->async_setup_task, picked_target, calld);
+ grpc_lb_policy_pick(lb_policy, bind_pollset, initial_metadata,
+ &calld->picked_channel, &calld->async_setup_task);
GRPC_LB_POLICY_UNREF(lb_policy, "pick");
} else if (chand->resolver != NULL) {
diff --git a/src/core/channel/compress_filter.c b/src/core/channel/compress_filter.c
index 9ff679df18..f656484fac 100644
--- a/src/core/channel/compress_filter.c
+++ b/src/core/channel/compress_filter.c
@@ -208,7 +208,7 @@ static void process_send_ops(grpc_call_element *elem,
calld->has_compression_algorithm = 1; /* GPR_TRUE */
}
/* hint compression algorithm */
- grpc_metadata_batch_add_head(
+ grpc_metadata_batch_add_tail(
&(sop->data.metadata), &calld->compression_algorithm_storage,
GRPC_MDELEM_REF(channeld->mdelem_compression_algorithms
[calld->compression_algorithm]));
diff --git a/src/core/channel/http_client_filter.c b/src/core/channel/http_client_filter.c
index 63e4912397..6e93103a6a 100644
--- a/src/core/channel/http_client_filter.c
+++ b/src/core/channel/http_client_filter.c
@@ -32,13 +32,17 @@
#include "src/core/channel/http_client_filter.h"
#include <string.h>
+#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include "src/core/support/string.h"
typedef struct call_data {
grpc_linked_mdelem method;
grpc_linked_mdelem scheme;
grpc_linked_mdelem te_trailers;
grpc_linked_mdelem content_type;
+ grpc_linked_mdelem user_agent;
int sent_initial_metadata;
int got_initial_metadata;
@@ -58,6 +62,8 @@ typedef struct channel_data {
grpc_mdelem *scheme;
grpc_mdelem *content_type;
grpc_mdelem *status;
+ /** complete user agent mdelem */
+ grpc_mdelem *user_agent;
} channel_data;
/* used to silence 'variable not used' warnings */
@@ -92,6 +98,18 @@ static void hc_on_recv(void *user_data, int success) {
calld->on_done_recv->cb(calld->on_done_recv->cb_arg, success);
}
+static grpc_mdelem *client_strip_filter(void *user_data, grpc_mdelem *md) {
+ grpc_call_element *elem = user_data;
+ channel_data *channeld = elem->channel_data;
+ /* eat the things we'd like to set ourselves */
+ if (md->key == channeld->method->key) return NULL;
+ if (md->key == channeld->scheme->key) return NULL;
+ if (md->key == channeld->te_trailers->key) return NULL;
+ if (md->key == channeld->content_type->key) return NULL;
+ if (md->key == channeld->user_agent->key) return NULL;
+ return md;
+}
+
static void hc_mutate_op(grpc_call_element *elem,
grpc_transport_stream_op *op) {
/* grab pointers to our data from the call element */
@@ -105,6 +123,7 @@ static void hc_mutate_op(grpc_call_element *elem,
grpc_stream_op *op = &ops[i];
if (op->type != GRPC_OP_METADATA) continue;
calld->sent_initial_metadata = 1;
+ grpc_metadata_batch_filter(&op->data.metadata, client_strip_filter, elem);
/* Send : prefixed headers, which have to be before any application
layer headers. */
grpc_metadata_batch_add_head(&op->data.metadata, &calld->method,
@@ -115,6 +134,8 @@ static void hc_mutate_op(grpc_call_element *elem,
GRPC_MDELEM_REF(channeld->te_trailers));
grpc_metadata_batch_add_tail(&op->data.metadata, &calld->content_type,
GRPC_MDELEM_REF(channeld->content_type));
+ grpc_metadata_batch_add_tail(&op->data.metadata, &calld->user_agent,
+ GRPC_MDELEM_REF(channeld->user_agent));
break;
}
}
@@ -169,6 +190,55 @@ static const char *scheme_from_args(const grpc_channel_args *args) {
return "http";
}
+static grpc_mdstr *user_agent_from_args(grpc_mdctx *mdctx,
+ const grpc_channel_args *args) {
+ gpr_strvec v;
+ size_t i;
+ int is_first = 1;
+ char *tmp;
+ grpc_mdstr *result;
+
+ gpr_strvec_init(&v);
+
+ for (i = 0; args && i < args->num_args; i++) {
+ if (0 == strcmp(args->args[i].key, GRPC_ARG_PRIMARY_USER_AGENT_STRING)) {
+ if (args->args[i].type != GRPC_ARG_STRING) {
+ gpr_log(GPR_ERROR, "Channel argument '%s' should be a string",
+ GRPC_ARG_PRIMARY_USER_AGENT_STRING);
+ } else {
+ if (!is_first) gpr_strvec_add(&v, gpr_strdup(" "));
+ is_first = 0;
+ gpr_strvec_add(&v, gpr_strdup(args->args[i].value.string));
+ }
+ }
+ }
+
+ gpr_asprintf(&tmp, "%sgrpc-c/%s (%s)", is_first ? "" : " ",
+ grpc_version_string(), GPR_PLATFORM_STRING);
+ is_first = 0;
+ gpr_strvec_add(&v, tmp);
+
+ for (i = 0; args && i < args->num_args; i++) {
+ if (0 == strcmp(args->args[i].key, GRPC_ARG_SECONDARY_USER_AGENT_STRING)) {
+ if (args->args[i].type != GRPC_ARG_STRING) {
+ gpr_log(GPR_ERROR, "Channel argument '%s' should be a string",
+ GRPC_ARG_SECONDARY_USER_AGENT_STRING);
+ } else {
+ if (!is_first) gpr_strvec_add(&v, gpr_strdup(" "));
+ is_first = 0;
+ gpr_strvec_add(&v, gpr_strdup(args->args[i].value.string));
+ }
+ }
+ }
+
+ tmp = gpr_strvec_flatten(&v, NULL);
+ gpr_strvec_destroy(&v);
+ result = grpc_mdstr_from_string(mdctx, tmp);
+ gpr_free(tmp);
+
+ return result;
+}
+
/* Constructor for channel_data */
static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master,
const grpc_channel_args *args, grpc_mdctx *mdctx,
@@ -189,6 +259,9 @@ static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master,
channeld->content_type =
grpc_mdelem_from_strings(mdctx, "content-type", "application/grpc");
channeld->status = grpc_mdelem_from_strings(mdctx, ":status", "200");
+ channeld->user_agent = grpc_mdelem_from_metadata_strings(
+ mdctx, grpc_mdstr_from_string(mdctx, "user-agent"),
+ user_agent_from_args(mdctx, args));
}
/* Destructor for channel data */
@@ -201,6 +274,7 @@ static void destroy_channel_elem(grpc_channel_element *elem) {
GRPC_MDELEM_UNREF(channeld->scheme);
GRPC_MDELEM_UNREF(channeld->content_type);
GRPC_MDELEM_UNREF(channeld->status);
+ GRPC_MDELEM_UNREF(channeld->user_agent);
}
const grpc_channel_filter grpc_http_client_filter = {
diff --git a/src/core/client_config/subchannel.c b/src/core/client_config/subchannel.c
index 35f172683a..487f5afb35 100644
--- a/src/core/client_config/subchannel.c
+++ b/src/core/client_config/subchannel.c
@@ -300,7 +300,7 @@ static void continue_connect(grpc_subchannel *c) {
}
static void start_connect(grpc_subchannel *c) {
- gpr_timespec now = gpr_now(GPR_CLOCK_REALTIME);
+ gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC);
c->next_attempt = now;
c->backoff_delta = gpr_time_from_seconds(1, GPR_TIMESPAN);
@@ -585,7 +585,7 @@ static void subchannel_connected(void *arg, int iomgr_success) {
c->have_alarm = 1;
c->next_attempt = gpr_time_add(c->next_attempt, c->backoff_delta);
c->backoff_delta = gpr_time_add(c->backoff_delta, c->backoff_delta);
- grpc_alarm_init(&c->alarm, c->next_attempt, on_alarm, c, gpr_now(GPR_CLOCK_REALTIME));
+ grpc_alarm_init(&c->alarm, c->next_attempt, on_alarm, c, gpr_now(GPR_CLOCK_MONOTONIC));
gpr_mu_unlock(&c->mu);
}
}
diff --git a/src/core/iomgr/alarm.c b/src/core/iomgr/alarm.c
index 5b9a37e5dd..931f746f75 100644
--- a/src/core/iomgr/alarm.c
+++ b/src/core/iomgr/alarm.c
@@ -36,6 +36,7 @@
#include "src/core/iomgr/alarm_heap.h"
#include "src/core/iomgr/alarm_internal.h"
#include "src/core/iomgr/time_averaged_stats.h"
+#include <grpc/support/log.h>
#include <grpc/support/sync.h>
#include <grpc/support/useful.h>
@@ -67,6 +68,7 @@ typedef struct {
static gpr_mu g_mu;
/* Allow only one run_some_expired_alarms at once */
static gpr_mu g_checker_mu;
+static gpr_clock_type g_clock_type;
static shard_type g_shards[NUM_SHARDS];
/* Protected by g_mu */
static shard_type *g_shard_queue[NUM_SHARDS];
@@ -85,6 +87,7 @@ void grpc_alarm_list_init(gpr_timespec now) {
gpr_mu_init(&g_mu);
gpr_mu_init(&g_checker_mu);
+ g_clock_type = now.clock_type;
for (i = 0; i < NUM_SHARDS; i++) {
shard_type *shard = &g_shards[i];
@@ -102,7 +105,7 @@ void grpc_alarm_list_init(gpr_timespec now) {
void grpc_alarm_list_shutdown(void) {
int i;
- while (run_some_expired_alarms(NULL, gpr_inf_future(GPR_CLOCK_REALTIME), NULL,
+ while (run_some_expired_alarms(NULL, gpr_inf_future(g_clock_type), NULL,
0))
;
for (i = 0; i < NUM_SHARDS; i++) {
@@ -175,6 +178,8 @@ void grpc_alarm_init(grpc_alarm *alarm, gpr_timespec deadline,
gpr_timespec now) {
int is_first_alarm = 0;
shard_type *shard = &g_shards[shard_idx(alarm)];
+ GPR_ASSERT(deadline.clock_type == g_clock_type);
+ GPR_ASSERT(now.clock_type == g_clock_type);
alarm->cb = alarm_cb;
alarm->cb_arg = alarm_cb_arg;
alarm->deadline = deadline;
@@ -355,6 +360,7 @@ static int run_some_expired_alarms(gpr_mu *drop_mu, gpr_timespec now,
}
int grpc_alarm_check(gpr_mu *drop_mu, gpr_timespec now, gpr_timespec *next) {
+ GPR_ASSERT(now.clock_type == g_clock_type);
return run_some_expired_alarms(drop_mu, now, next, 1);
}
diff --git a/src/core/iomgr/iomgr.c b/src/core/iomgr/iomgr.c
index 0244f689b1..a18c176b30 100644
--- a/src/core/iomgr/iomgr.c
+++ b/src/core/iomgr/iomgr.c
@@ -57,9 +57,9 @@ static grpc_iomgr_object g_root_object;
static void background_callback_executor(void *ignored) {
gpr_mu_lock(&g_mu);
while (!g_shutdown) {
- gpr_timespec deadline = gpr_inf_future(GPR_CLOCK_REALTIME);
+ gpr_timespec deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC);
gpr_timespec short_deadline = gpr_time_add(
- gpr_now(GPR_CLOCK_REALTIME), gpr_time_from_millis(100, GPR_TIMESPAN));
+ gpr_now(GPR_CLOCK_MONOTONIC), gpr_time_from_millis(100, GPR_TIMESPAN));
if (g_cbs_head) {
grpc_iomgr_closure *closure = g_cbs_head;
g_cbs_head = closure->next;
@@ -67,7 +67,7 @@ static void background_callback_executor(void *ignored) {
gpr_mu_unlock(&g_mu);
closure->cb(closure->cb_arg, closure->success);
gpr_mu_lock(&g_mu);
- } else if (grpc_alarm_check(&g_mu, gpr_now(GPR_CLOCK_REALTIME),
+ } else if (grpc_alarm_check(&g_mu, gpr_now(GPR_CLOCK_MONOTONIC),
&deadline)) {
} else {
gpr_mu_unlock(&g_mu);
@@ -90,7 +90,7 @@ void grpc_iomgr_init(void) {
gpr_thd_id id;
gpr_mu_init(&g_mu);
gpr_cv_init(&g_rcv);
- grpc_alarm_list_init(gpr_now(GPR_CLOCK_REALTIME));
+ grpc_alarm_list_init(gpr_now(GPR_CLOCK_MONOTONIC));
g_root_object.next = g_root_object.prev = &g_root_object;
g_root_object.name = "root";
grpc_iomgr_platform_init();
@@ -145,7 +145,7 @@ void grpc_iomgr_shutdown(void) {
} while (g_cbs_head);
continue;
}
- if (grpc_alarm_check(&g_mu, gpr_inf_future(GPR_CLOCK_REALTIME), NULL)) {
+ if (grpc_alarm_check(&g_mu, gpr_inf_future(GPR_CLOCK_MONOTONIC), NULL)) {
gpr_log(GPR_DEBUG, "got late alarm");
continue;
}
diff --git a/src/core/iomgr/pollset_posix.c b/src/core/iomgr/pollset_posix.c
index efb301d81c..c8646af615 100644
--- a/src/core/iomgr/pollset_posix.c
+++ b/src/core/iomgr/pollset_posix.c
@@ -136,7 +136,7 @@ static void finish_shutdown(grpc_pollset *pollset) {
int grpc_pollset_work(grpc_pollset *pollset, gpr_timespec deadline) {
/* pollset->mu already held */
- gpr_timespec now = gpr_now(GPR_CLOCK_REALTIME);
+ gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC);
if (gpr_time_cmp(now, deadline) > 0) {
return 0;
}
@@ -205,7 +205,7 @@ int grpc_poll_deadline_to_millis_timeout(gpr_timespec deadline,
gpr_timespec now) {
gpr_timespec timeout;
static const int max_spin_polling_us = 10;
- if (gpr_time_cmp(deadline, gpr_inf_future(GPR_CLOCK_REALTIME)) == 0) {
+ if (gpr_time_cmp(deadline, gpr_inf_future(deadline.clock_type)) == 0) {
return -1;
}
if (gpr_time_cmp(deadline, gpr_time_add(now, gpr_time_from_micros(
diff --git a/src/core/iomgr/pollset_windows.c b/src/core/iomgr/pollset_windows.c
index 24226cc980..a9c4739c7c 100644
--- a/src/core/iomgr/pollset_windows.c
+++ b/src/core/iomgr/pollset_windows.c
@@ -70,7 +70,7 @@ void grpc_pollset_destroy(grpc_pollset *pollset) {
int grpc_pollset_work(grpc_pollset *pollset, gpr_timespec deadline) {
gpr_timespec now;
- now = gpr_now(GPR_CLOCK_REALTIME);
+ now = gpr_now(GPR_CLOCK_MONOTONIC);
if (gpr_time_cmp(now, deadline) > 0) {
return 0 /* GPR_FALSE */;
}
diff --git a/src/core/iomgr/tcp_client_posix.c b/src/core/iomgr/tcp_client_posix.c
index dc0489e64f..41d8b169e0 100644
--- a/src/core/iomgr/tcp_client_posix.c
+++ b/src/core/iomgr/tcp_client_posix.c
@@ -114,6 +114,8 @@ static void on_writable(void *acp, int success) {
void (*cb)(void *arg, grpc_endpoint *tcp) = ac->cb;
void *cb_arg = ac->cb_arg;
+ grpc_alarm_cancel(&ac->alarm);
+
gpr_mu_lock(&ac->mu);
if (success) {
do {
@@ -178,8 +180,6 @@ finish:
if (done) {
gpr_mu_destroy(&ac->mu);
gpr_free(ac);
- } else {
- grpc_alarm_cancel(&ac->alarm);
}
cb(cb_arg, ep);
}
@@ -253,8 +253,8 @@ void grpc_tcp_client_connect(void (*cb)(void *arg, grpc_endpoint *ep),
ac->write_closure.cb_arg = ac;
gpr_mu_lock(&ac->mu);
- grpc_alarm_init(&ac->alarm, deadline, on_alarm, ac,
- gpr_now(GPR_CLOCK_REALTIME));
+ grpc_alarm_init(&ac->alarm, gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC),
+ on_alarm, ac, gpr_now(GPR_CLOCK_MONOTONIC));
grpc_fd_notify_on_write(ac->fd, &ac->write_closure);
gpr_mu_unlock(&ac->mu);
diff --git a/src/core/iomgr/tcp_client_windows.c b/src/core/iomgr/tcp_client_windows.c
index 16741452b9..39fd43130b 100644
--- a/src/core/iomgr/tcp_client_windows.c
+++ b/src/core/iomgr/tcp_client_windows.c
@@ -216,7 +216,7 @@ void grpc_tcp_client_connect(void (*cb)(void *arg, grpc_endpoint *tcp),
ac->aborted = 0;
grpc_alarm_init(&ac->alarm, deadline, on_alarm, ac,
- gpr_now(GPR_CLOCK_REALTIME));
+ gpr_now(GPR_CLOCK_MONOTONIC));
socket->write_info.outstanding = 1;
grpc_socket_notify_on_write(socket, on_connect, ac);
return;
diff --git a/src/core/support/sync_posix.c b/src/core/support/sync_posix.c
index 41af8ceb0a..61572b9a8e 100644
--- a/src/core/support/sync_posix.c
+++ b/src/core/support/sync_posix.c
@@ -63,10 +63,11 @@ void gpr_cv_destroy(gpr_cv *cv) { GPR_ASSERT(pthread_cond_destroy(cv) == 0); }
int gpr_cv_wait(gpr_cv *cv, gpr_mu *mu, gpr_timespec abs_deadline) {
int err = 0;
- if (gpr_time_cmp(abs_deadline, gpr_inf_future(GPR_CLOCK_REALTIME)) == 0) {
+ if (gpr_time_cmp(abs_deadline, gpr_inf_future(abs_deadline.clock_type)) == 0) {
err = pthread_cond_wait(cv, mu);
} else {
struct timespec abs_deadline_ts;
+ abs_deadline = gpr_convert_clock_type(abs_deadline, GPR_CLOCK_REALTIME);
abs_deadline_ts.tv_sec = abs_deadline.tv_sec;
abs_deadline_ts.tv_nsec = abs_deadline.tv_nsec;
err = pthread_cond_timedwait(cv, mu, &abs_deadline_ts);
diff --git a/src/core/support/sync_win32.c b/src/core/support/sync_win32.c
index 63196d10d3..54f84a46ac 100644
--- a/src/core/support/sync_win32.c
+++ b/src/core/support/sync_win32.c
@@ -83,10 +83,10 @@ int gpr_cv_wait(gpr_cv *cv, gpr_mu *mu, gpr_timespec abs_deadline) {
int timeout = 0;
DWORD timeout_max_ms;
mu->locked = 0;
- if (gpr_time_cmp(abs_deadline, gpr_inf_future(GPR_CLOCK_REALTIME)) == 0) {
+ if (gpr_time_cmp(abs_deadline, gpr_inf_future(abs_deadline.clock_type)) == 0) {
SleepConditionVariableCS(cv, &mu->cs, INFINITE);
} else {
- gpr_timespec now = gpr_now(GPR_CLOCK_REALTIME);
+ gpr_timespec now = gpr_now(abs_deadline.clock_type);
gpr_int64 now_ms = now.tv_sec * 1000 + now.tv_nsec / 1000000;
gpr_int64 deadline_ms =
abs_deadline.tv_sec * 1000 + abs_deadline.tv_nsec / 1000000;
diff --git a/src/core/support/time.c b/src/core/support/time.c
index 570f195bd1..b523ae01cc 100644
--- a/src/core/support/time.c
+++ b/src/core/support/time.c
@@ -290,3 +290,30 @@ gpr_int32 gpr_time_to_millis(gpr_timespec t) {
double gpr_timespec_to_micros(gpr_timespec t) {
return (double)t.tv_sec * GPR_US_PER_SEC + t.tv_nsec * 1e-3;
}
+
+gpr_timespec gpr_convert_clock_type(gpr_timespec t, gpr_clock_type clock_type) {
+ if (t.clock_type == clock_type) {
+ return t;
+ }
+
+ if (t.tv_nsec == 0) {
+ if (t.tv_sec == TYPE_MAX(time_t)) {
+ t.clock_type = clock_type;
+ return t;
+ }
+ if (t.tv_sec == TYPE_MIN(time_t)) {
+ t.clock_type = clock_type;
+ return t;
+ }
+ }
+
+ if (clock_type == GPR_TIMESPAN) {
+ return gpr_time_sub(t, gpr_now(t.clock_type));
+ }
+
+ if (t.clock_type == GPR_TIMESPAN) {
+ return gpr_time_add(gpr_now(clock_type), t);
+ }
+
+ return gpr_time_add(gpr_now(clock_type), gpr_time_sub(t, gpr_now(t.clock_type)));
+}
diff --git a/src/core/support/time_posix.c b/src/core/support/time_posix.c
index 258b2e640e..841485c4b4 100644
--- a/src/core/support/time_posix.c
+++ b/src/core/support/time_posix.c
@@ -120,7 +120,7 @@ void gpr_sleep_until(gpr_timespec until) {
for (;;) {
/* We could simplify by using clock_nanosleep instead, but it might be
* slightly less portable. */
- now = gpr_now(GPR_CLOCK_REALTIME);
+ now = gpr_now(until.clock_type);
if (gpr_time_cmp(until, now) <= 0) {
return;
}
diff --git a/src/core/support/time_win32.c b/src/core/support/time_win32.c
index 238cd07ebc..7f64c80e27 100644
--- a/src/core/support/time_win32.c
+++ b/src/core/support/time_win32.c
@@ -80,7 +80,7 @@ void gpr_sleep_until(gpr_timespec until) {
for (;;) {
/* We could simplify by using clock_nanosleep instead, but it might be
* slightly less portable. */
- now = gpr_now(GPR_CLOCK_REALTIME);
+ now = gpr_now(until.clock_type);
if (gpr_time_cmp(until, now) <= 0) {
return;
}
diff --git a/src/core/surface/call.c b/src/core/surface/call.c
index 54f10261e3..0cdb9e22a5 100644
--- a/src/core/surface/call.c
+++ b/src/core/surface/call.c
@@ -352,7 +352,7 @@ grpc_call *grpc_call_create(grpc_channel *channel, grpc_completion_queue *cq,
}
grpc_call_stack_init(channel_stack, server_transport_data, initial_op_ptr,
CALL_STACK_FROM_CALL(call));
- if (gpr_time_cmp(send_deadline, gpr_inf_future(GPR_CLOCK_REALTIME)) != 0) {
+ if (gpr_time_cmp(send_deadline, gpr_inf_future(send_deadline.clock_type)) != 0) {
set_deadline_alarm(call, send_deadline);
}
return call;
@@ -1309,8 +1309,8 @@ static void set_deadline_alarm(grpc_call *call, gpr_timespec deadline) {
}
GRPC_CALL_INTERNAL_REF(call, "alarm");
call->have_alarm = 1;
- grpc_alarm_init(&call->alarm, deadline, call_alarm, call,
- gpr_now(GPR_CLOCK_REALTIME));
+ grpc_alarm_init(&call->alarm, gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC), call_alarm, call,
+ gpr_now(GPR_CLOCK_MONOTONIC));
}
/* we offset status by a small amount when storing it into transport metadata
diff --git a/src/core/surface/call.h b/src/core/surface/call.h
index 5736e97b59..49fb4de47a 100644
--- a/src/core/surface/call.h
+++ b/src/core/surface/call.h
@@ -134,6 +134,10 @@ void grpc_server_log_request_call(char *file, int line,
grpc_completion_queue *cq_for_notification,
void *tag);
+void grpc_server_log_shutdown(char *file, int line, gpr_log_severity severity,
+ grpc_server *server, grpc_completion_queue *cq,
+ void *tag);
+
/* Set a context pointer.
No thread safety guarantees are made wrt this value. */
void grpc_call_context_set(grpc_call *call, grpc_context_index elem,
@@ -151,6 +155,9 @@ void *grpc_call_context_get(grpc_call *call, grpc_context_index elem);
grpc_server_log_request_call(sev, server, call, details, initial_metadata, \
cq_bound_to_call, cq_for_notifications, tag)
+#define GRPC_SERVER_LOG_SHUTDOWN(sev, server, cq, tag) \
+ if (grpc_trace_batch) grpc_server_log_shutdown(sev, server, cq, tag)
+
gpr_uint8 grpc_call_is_client(grpc_call *call);
/** Returns a bitset for the encodings (compression algorithms) supported by \a
diff --git a/src/core/surface/call_log_batch.c b/src/core/surface/call_log_batch.c
index 997046d954..7bf8cafc24 100644
--- a/src/core/surface/call_log_batch.c
+++ b/src/core/surface/call_log_batch.c
@@ -136,3 +136,11 @@ void grpc_server_log_request_call(char *file, int line,
"tag=%p)", server, call, details, initial_metadata,
cq_bound_to_call, cq_for_notification, tag);
}
+
+void grpc_server_log_shutdown(char *file, int line, gpr_log_severity severity,
+ grpc_server *server, grpc_completion_queue *cq,
+ void *tag) {
+ gpr_log(file, line, severity,
+ "grpc_server_shutdown_and_notify(server=%p, cq=%p, tag=%p)", server,
+ cq, tag);
+}
diff --git a/src/core/surface/completion_queue.c b/src/core/surface/completion_queue.c
index 8484418247..3f60b0b0ba 100644
--- a/src/core/surface/completion_queue.c
+++ b/src/core/surface/completion_queue.c
@@ -148,6 +148,8 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc,
gpr_timespec deadline) {
grpc_event ret;
+ deadline = gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC);
+
GRPC_CQ_INTERNAL_REF(cc, "next");
gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
for (;;) {
@@ -188,6 +190,8 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag,
grpc_cq_completion *c;
grpc_cq_completion *prev;
+ deadline = gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC);
+
GRPC_CQ_INTERNAL_REF(cc, "pluck");
gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
for (;;) {
diff --git a/src/core/surface/server.c b/src/core/surface/server.c
index f2d6b11bc7..439452aea2 100644
--- a/src/core/surface/server.c
+++ b/src/core/surface/server.c
@@ -980,6 +980,8 @@ void grpc_server_shutdown_and_notify(grpc_server *server,
channel_broadcaster broadcaster;
request_killer reqkill;
+ GRPC_SERVER_LOG_SHUTDOWN(GPR_INFO, server, cq, tag);
+
/* lock, and gather up some stuff to do */
gpr_mu_lock(&server->mu_global);
grpc_cq_begin_op(cq);
diff --git a/src/core/transport/chttp2/parsing.c b/src/core/transport/chttp2/parsing.c
index aa32f2e44a..904b9afce7 100644
--- a/src/core/transport/chttp2/parsing.c
+++ b/src/core/transport/chttp2/parsing.c
@@ -588,7 +588,7 @@ static void on_header(void *tp, grpc_mdelem *md) {
GPR_ASSERT(stream_parsing);
GRPC_CHTTP2_IF_TRACING(gpr_log(
- GPR_INFO, "HTTP:%d:HDR: %s: %s", stream_parsing->id,
+ GPR_INFO, "HTTP:%d:HDR:%s: %s: %s", stream_parsing->id,
transport_parsing->is_client ? "CLI" : "SVR",
grpc_mdstr_as_c_string(md->key), grpc_mdstr_as_c_string(md->value)));
diff --git a/src/core/transport/chttp2/stream_encoder.c b/src/core/transport/chttp2/stream_encoder.c
index d30059abf8..65b31a5afd 100644
--- a/src/core/transport/chttp2/stream_encoder.c
+++ b/src/core/transport/chttp2/stream_encoder.c
@@ -438,7 +438,7 @@ static void deadline_enc(grpc_chttp2_hpack_compressor *c, gpr_timespec deadline,
char timeout_str[GRPC_CHTTP2_TIMEOUT_ENCODE_MIN_BUFSIZE];
grpc_mdelem *mdelem;
grpc_chttp2_encode_timeout(
- gpr_time_sub(deadline, gpr_now(GPR_CLOCK_REALTIME)), timeout_str);
+ gpr_time_sub(deadline, gpr_now(deadline.clock_type)), timeout_str);
mdelem = grpc_mdelem_from_metadata_strings(
c->mdctx, GRPC_MDSTR_REF(c->timeout_key_str),
grpc_mdstr_from_string(c->mdctx, timeout_str));
@@ -560,6 +560,7 @@ void grpc_chttp2_encode(grpc_stream_op *ops, size_t ops_count, int eof,
grpc_mdctx *mdctx = compressor->mdctx;
grpc_linked_mdelem *l;
int need_unref = 0;
+ gpr_timespec deadline;
GPR_ASSERT(stream_id != 0);
@@ -589,9 +590,9 @@ void grpc_chttp2_encode(grpc_stream_op *ops, size_t ops_count, int eof,
l->md = hpack_enc(compressor, l->md, &st);
need_unref |= l->md != NULL;
}
- if (gpr_time_cmp(op->data.metadata.deadline,
- gpr_inf_future(GPR_CLOCK_REALTIME)) != 0) {
- deadline_enc(compressor, op->data.metadata.deadline, &st);
+ deadline = op->data.metadata.deadline;
+ if (gpr_time_cmp(deadline, gpr_inf_future(deadline.clock_type)) != 0) {
+ deadline_enc(compressor, deadline, &st);
}
curop++;
break;
diff --git a/src/core/transport/transport_op_string.c b/src/core/transport/transport_op_string.c
index 9d127c5472..10d796fc15 100644
--- a/src/core/transport/transport_op_string.c
+++ b/src/core/transport/transport_op_string.c
@@ -61,7 +61,7 @@ static void put_metadata_list(gpr_strvec *b, grpc_metadata_batch md) {
if (m != md.list.head) gpr_strvec_add(b, gpr_strdup(", "));
put_metadata(b, m->md);
}
- if (gpr_time_cmp(md.deadline, gpr_inf_future(GPR_CLOCK_REALTIME)) != 0) {
+ if (gpr_time_cmp(md.deadline, gpr_inf_future(md.deadline.clock_type)) != 0) {
char *tmp;
gpr_asprintf(&tmp, " deadline=%d.%09d", md.deadline.tv_sec,
md.deadline.tv_nsec);
diff --git a/src/cpp/client/channel_arguments.cc b/src/cpp/client/channel_arguments.cc
index 92ac5ea6fd..da6602e7af 100644
--- a/src/cpp/client/channel_arguments.cc
+++ b/src/cpp/client/channel_arguments.cc
@@ -33,11 +33,49 @@
#include <grpc++/channel_arguments.h>
+#include <grpc/support/log.h>
+
#include "src/core/channel/channel_args.h"
namespace grpc {
-void ChannelArguments::_Experimental_SetCompressionAlgorithm(
+ChannelArguments::ChannelArguments(const ChannelArguments& other)
+ : strings_(other.strings_) {
+ args_.reserve(other.args_.size());
+ auto list_it_dst = strings_.begin();
+ auto list_it_src = other.strings_.begin();
+ for (auto a = other.args_.begin(); a != other.args_.end(); ++a) {
+ grpc_arg ap;
+ ap.type = a->type;
+ GPR_ASSERT(list_it_src->c_str() == a->key);
+ ap.key = const_cast<char*>(list_it_dst->c_str());
+ ++list_it_src;
+ ++list_it_dst;
+ switch (a->type) {
+ case GRPC_ARG_INTEGER:
+ ap.value.integer = a->value.integer;
+ break;
+ case GRPC_ARG_STRING:
+ GPR_ASSERT(list_it_src->c_str() == a->value.string);
+ ap.value.string = const_cast<char*>(list_it_dst->c_str());
+ ++list_it_src;
+ ++list_it_dst;
+ break;
+ case GRPC_ARG_POINTER:
+ ap.value.pointer = a->value.pointer;
+ ap.value.pointer.p = a->value.pointer.copy(ap.value.pointer.p);
+ break;
+ }
+ args_.push_back(ap);
+ }
+}
+
+void ChannelArguments::Swap(ChannelArguments& other) {
+ args_.swap(other.args_);
+ strings_.swap(other.strings_);
+}
+
+void ChannelArguments::SetCompressionAlgorithm(
grpc_compression_algorithm algorithm) {
SetInt(GRPC_COMPRESSION_ALGORITHM_ARG, algorithm);
}
diff --git a/src/cpp/client/client_context.cc b/src/cpp/client/client_context.cc
index 69216d2030..14ab772e50 100644
--- a/src/cpp/client/client_context.cc
+++ b/src/cpp/client/client_context.cc
@@ -79,7 +79,7 @@ void ClientContext::set_call(grpc_call* call,
}
}
-void ClientContext::_experimental_set_compression_algorithm(
+void ClientContext::set_compression_algorithm(
grpc_compression_algorithm algorithm) {
char* algorithm_name = NULL;
if (!grpc_compression_algorithm_name(algorithm, &algorithm_name)) {
diff --git a/src/cpp/client/create_channel.cc b/src/cpp/client/create_channel.cc
index 510af2bb00..38eeda0dc0 100644
--- a/src/cpp/client/create_channel.cc
+++ b/src/cpp/client/create_channel.cc
@@ -32,9 +32,11 @@
*/
#include <memory>
+#include <sstream>
#include "src/cpp/client/channel.h"
#include <grpc++/channel_interface.h>
+#include <grpc++/channel_arguments.h>
#include <grpc++/create_channel.h>
namespace grpc {
@@ -43,7 +45,12 @@ class ChannelArguments;
std::shared_ptr<ChannelInterface> CreateChannel(
const grpc::string& target, const std::shared_ptr<Credentials>& creds,
const ChannelArguments& args) {
- return creds ? creds->CreateChannel(target, args)
+ ChannelArguments cp_args = args;
+ std::ostringstream user_agent_prefix;
+ user_agent_prefix << "grpc-c++/" << grpc_version_string();
+ cp_args.SetString(GRPC_ARG_PRIMARY_USER_AGENT_STRING,
+ user_agent_prefix.str());
+ return creds ? creds->CreateChannel(target, cp_args)
: std::shared_ptr<ChannelInterface>(
new Channel(target, grpc_lame_client_channel_create()));
}
diff --git a/src/csharp/Grpc.Auth/GoogleCredential.cs b/src/csharp/Grpc.Auth/GoogleCredential.cs
index 8d5e543a21..7edf19ed67 100644
--- a/src/csharp/Grpc.Auth/GoogleCredential.cs
+++ b/src/csharp/Grpc.Auth/GoogleCredential.cs
@@ -35,8 +35,11 @@ using System;
using System.Collections.Generic;
using System.IO;
using System.Security.Cryptography;
+using System.Threading;
+using System.Threading.Tasks;
using Google.Apis.Auth.OAuth2;
+using Google.Apis.Auth.OAuth2.Responses;
using Newtonsoft.Json.Linq;
using Org.BouncyCastle.Crypto.Parameters;
using Org.BouncyCastle.Security;
@@ -100,6 +103,19 @@ namespace Grpc.Auth
return new GoogleCredential(serviceCredential);
}
+ public Task<bool> RequestAccessTokenAsync(CancellationToken taskCancellationToken)
+ {
+ return credential.RequestAccessTokenAsync(taskCancellationToken);
+ }
+
+ public TokenResponse Token
+ {
+ get
+ {
+ return credential.Token;
+ }
+ }
+
internal ServiceCredential InternalCredential
{
get
diff --git a/src/csharp/Grpc.Core.Tests/ClientServerTest.cs b/src/csharp/Grpc.Core.Tests/ClientServerTest.cs
index e797dd82f2..8ba2c8a9a2 100644
--- a/src/csharp/Grpc.Core.Tests/ClientServerTest.cs
+++ b/src/csharp/Grpc.Core.Tests/ClientServerTest.cs
@@ -33,6 +33,7 @@
using System;
using System.Diagnostics;
+using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Grpc.Core;
@@ -99,17 +100,17 @@ namespace Grpc.Core.Tests
[Test]
public void UnaryCall()
{
- var call = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty);
- Assert.AreEqual("ABC", Calls.BlockingUnaryCall(call, "ABC", CancellationToken.None));
+ var internalCall = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty);
+ Assert.AreEqual("ABC", Calls.BlockingUnaryCall(internalCall, "ABC", CancellationToken.None));
}
[Test]
public void UnaryCall_ServerHandlerThrows()
{
- var call = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty);
+ var internalCall = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty);
try
{
- Calls.BlockingUnaryCall(call, "THROW", CancellationToken.None);
+ Calls.BlockingUnaryCall(internalCall, "THROW", CancellationToken.None);
Assert.Fail();
}
catch (RpcException e)
@@ -119,10 +120,40 @@ namespace Grpc.Core.Tests
}
[Test]
+ public void UnaryCall_ServerHandlerThrowsRpcException()
+ {
+ var internalCall = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty);
+ try
+ {
+ Calls.BlockingUnaryCall(internalCall, "THROW_UNAUTHENTICATED", CancellationToken.None);
+ Assert.Fail();
+ }
+ catch (RpcException e)
+ {
+ Assert.AreEqual(StatusCode.Unauthenticated, e.Status.StatusCode);
+ }
+ }
+
+ [Test]
+ public void UnaryCall_ServerHandlerSetsStatus()
+ {
+ var internalCall = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty);
+ try
+ {
+ Calls.BlockingUnaryCall(internalCall, "SET_UNAUTHENTICATED", CancellationToken.None);
+ Assert.Fail();
+ }
+ catch (RpcException e)
+ {
+ Assert.AreEqual(StatusCode.Unauthenticated, e.Status.StatusCode);
+ }
+ }
+
+ [Test]
public void AsyncUnaryCall()
{
- var call = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty);
- var result = Calls.AsyncUnaryCall(call, "ABC", CancellationToken.None).Result;
+ var internalCall = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty);
+ var result = Calls.AsyncUnaryCall(internalCall, "ABC", CancellationToken.None).ResponseAsync.Result;
Assert.AreEqual("ABC", result);
}
@@ -131,10 +162,10 @@ namespace Grpc.Core.Tests
{
Task.Run(async () =>
{
- var call = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty);
+ var internalCall = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty);
try
{
- await Calls.AsyncUnaryCall(call, "THROW", CancellationToken.None);
+ await Calls.AsyncUnaryCall(internalCall, "THROW", CancellationToken.None);
Assert.Fail();
}
catch (RpcException e)
@@ -149,11 +180,11 @@ namespace Grpc.Core.Tests
{
Task.Run(async () =>
{
- var call = new Call<string, string>(ServiceName, ConcatAndEchoMethod, channel, Metadata.Empty);
- var callResult = Calls.AsyncClientStreamingCall(call, CancellationToken.None);
+ var internalCall = new Call<string, string>(ServiceName, ConcatAndEchoMethod, channel, Metadata.Empty);
+ var call = Calls.AsyncClientStreamingCall(internalCall, CancellationToken.None);
- await callResult.RequestStream.WriteAll(new string[] { "A", "B", "C" });
- Assert.AreEqual("ABC", await callResult.Result);
+ await call.RequestStream.WriteAll(new string[] { "A", "B", "C" });
+ Assert.AreEqual("ABC", await call.ResponseAsync);
}).Wait();
}
@@ -162,10 +193,10 @@ namespace Grpc.Core.Tests
{
Task.Run(async () =>
{
- var call = new Call<string, string>(ServiceName, ConcatAndEchoMethod, channel, Metadata.Empty);
+ var internalCall = new Call<string, string>(ServiceName, ConcatAndEchoMethod, channel, Metadata.Empty);
var cts = new CancellationTokenSource();
- var callResult = Calls.AsyncClientStreamingCall(call, cts.Token);
+ var call = Calls.AsyncClientStreamingCall(internalCall, cts.Token);
// TODO(jtattermusch): we need this to ensure call has been initiated once we cancel it.
await Task.Delay(1000);
@@ -173,7 +204,7 @@ namespace Grpc.Core.Tests
try
{
- await callResult.Result;
+ await call.ResponseAsync;
}
catch (RpcException e)
{
@@ -183,29 +214,53 @@ namespace Grpc.Core.Tests
}
[Test]
+ public void AsyncUnaryCall_EchoMetadata()
+ {
+ var headers = new Metadata
+ {
+ new Metadata.Entry("asciiHeader", "abcdefg"),
+ new Metadata.Entry("binaryHeader-bin", new byte[] { 1, 2, 3, 0, 0xff }),
+ };
+ var internalCall = new Call<string, string>(ServiceName, EchoMethod, channel, headers);
+ var call = Calls.AsyncUnaryCall(internalCall, "ABC", CancellationToken.None);
+
+ Assert.AreEqual("ABC", call.ResponseAsync.Result);
+
+ Assert.AreEqual(StatusCode.OK, call.GetStatus().StatusCode);
+
+ var trailers = call.GetTrailers();
+ Assert.AreEqual(2, trailers.Count);
+ Assert.AreEqual(headers[0].Key, trailers[0].Key);
+ Assert.AreEqual(headers[0].Value, trailers[0].Value);
+
+ Assert.AreEqual(headers[1].Key, trailers[1].Key);
+ CollectionAssert.AreEqual(headers[1].ValueBytes, trailers[1].ValueBytes);
+ }
+
+ [Test]
public void UnaryCall_DisposedChannel()
{
channel.Dispose();
- var call = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty);
- Assert.Throws(typeof(ObjectDisposedException), () => Calls.BlockingUnaryCall(call, "ABC", CancellationToken.None));
+ var internalCall = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty);
+ Assert.Throws(typeof(ObjectDisposedException), () => Calls.BlockingUnaryCall(internalCall, "ABC", CancellationToken.None));
}
[Test]
public void UnaryCallPerformance()
{
- var call = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty);
+ var internalCall = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty);
BenchmarkUtil.RunBenchmark(100, 100,
- () => { Calls.BlockingUnaryCall(call, "ABC", default(CancellationToken)); });
+ () => { Calls.BlockingUnaryCall(internalCall, "ABC", default(CancellationToken)); });
}
[Test]
public void UnknownMethodHandler()
{
- var call = new Call<string, string>(ServiceName, NonexistentMethod, channel, Metadata.Empty);
+ var internalCall = new Call<string, string>(ServiceName, NonexistentMethod, channel, Metadata.Empty);
try
{
- Calls.BlockingUnaryCall(call, "ABC", default(CancellationToken));
+ Calls.BlockingUnaryCall(internalCall, "ABC", default(CancellationToken));
Assert.Fail();
}
catch (RpcException e)
@@ -214,16 +269,48 @@ namespace Grpc.Core.Tests
}
}
- private static async Task<string> EchoHandler(ServerCallContext context, string request)
+ [Test]
+ public void UserAgentStringPresent()
{
+ var internalCall = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty);
+ string userAgent = Calls.BlockingUnaryCall(internalCall, "RETURN-USER-AGENT", CancellationToken.None);
+ Assert.IsTrue(userAgent.StartsWith("grpc-csharp/"));
+ }
+
+ private static async Task<string> EchoHandler(string request, ServerCallContext context)
+ {
+ foreach (Metadata.Entry metadataEntry in context.RequestHeaders)
+ {
+ if (metadataEntry.Key != "user-agent")
+ {
+ context.ResponseTrailers.Add(metadataEntry);
+ }
+ }
+
+ if (request == "RETURN-USER-AGENT")
+ {
+ return context.RequestHeaders.Where(entry => entry.Key == "user-agent").Single().Value;
+ }
+
if (request == "THROW")
{
throw new Exception("This was thrown on purpose by a test");
}
+
+ if (request == "THROW_UNAUTHENTICATED")
+ {
+ throw new RpcException(new Status(StatusCode.Unauthenticated, ""));
+ }
+
+ if (request == "SET_UNAUTHENTICATED")
+ {
+ context.Status = new Status(StatusCode.Unauthenticated, "");
+ }
+
return request;
}
- private static async Task<string> ConcatAndEchoHandler(ServerCallContext context, IAsyncStreamReader<string> requestStream)
+ private static async Task<string> ConcatAndEchoHandler(IAsyncStreamReader<string> requestStream, ServerCallContext context)
{
string result = "";
await requestStream.ForEach(async (request) =>
diff --git a/src/csharp/Grpc.Core.Tests/Internal/MetadataArraySafeHandleTest.cs b/src/csharp/Grpc.Core.Tests/Internal/MetadataArraySafeHandleTest.cs
index 320423b245..46469113c5 100644
--- a/src/csharp/Grpc.Core.Tests/Internal/MetadataArraySafeHandleTest.cs
+++ b/src/csharp/Grpc.Core.Tests/Internal/MetadataArraySafeHandleTest.cs
@@ -51,12 +51,34 @@ namespace Grpc.Core.Internal.Tests
[Test]
public void CreateAndDestroy()
{
- var metadata = new Metadata {
+ var metadata = new Metadata
+ {
new Metadata.Entry("host", "somehost"),
new Metadata.Entry("header2", "header value"),
};
var nativeMetadata = MetadataArraySafeHandle.Create(metadata);
nativeMetadata.Dispose();
}
+
+ [Test]
+ public void ReadMetadataFromPtrUnsafe()
+ {
+ var metadata = new Metadata
+ {
+ new Metadata.Entry("host", "somehost"),
+ new Metadata.Entry("header2", "header value"),
+ };
+ var nativeMetadata = MetadataArraySafeHandle.Create(metadata);
+
+ var copy = MetadataArraySafeHandle.ReadMetadataFromPtrUnsafe(nativeMetadata.Handle);
+ Assert.AreEqual(2, copy.Count);
+
+ Assert.AreEqual("host", copy[0].Key);
+ Assert.AreEqual("somehost", copy[0].Value);
+ Assert.AreEqual("header2", copy[1].Key);
+ Assert.AreEqual("header value", copy[1].Value);
+
+ nativeMetadata.Dispose();
+ }
}
}
diff --git a/src/csharp/Grpc.Core.Tests/TimespecTest.cs b/src/csharp/Grpc.Core.Tests/TimespecTest.cs
index 5831121add..a34b407a01 100644
--- a/src/csharp/Grpc.Core.Tests/TimespecTest.cs
+++ b/src/csharp/Grpc.Core.Tests/TimespecTest.cs
@@ -59,6 +59,19 @@ namespace Grpc.Core.Internal.Tests
}
[Test]
+ public void ToDateTime()
+ {
+ Assert.AreEqual(new DateTime(1970, 1, 1, 0, 0, 0, DateTimeKind.Utc),
+ new Timespec(IntPtr.Zero, 0).ToDateTime());
+
+ Assert.AreEqual(new DateTime(1970, 1, 1, 0, 0, 10, DateTimeKind.Utc).AddTicks(50),
+ new Timespec(new IntPtr(10), 5000).ToDateTime());
+
+ Assert.AreEqual(new DateTime(2015, 7, 21, 4, 21, 48, DateTimeKind.Utc),
+ new Timespec(new IntPtr(1437452508), 0).ToDateTime());
+ }
+
+ [Test]
public void Add()
{
var t = new Timespec { tv_sec = new IntPtr(12345), tv_nsec = 123456789 };
diff --git a/src/csharp/Grpc.Core/AsyncClientStreamingCall.cs b/src/csharp/Grpc.Core/AsyncClientStreamingCall.cs
index d66b0d4974..bf020cd627 100644
--- a/src/csharp/Grpc.Core/AsyncClientStreamingCall.cs
+++ b/src/csharp/Grpc.Core/AsyncClientStreamingCall.cs
@@ -43,24 +43,28 @@ namespace Grpc.Core
public sealed class AsyncClientStreamingCall<TRequest, TResponse> : IDisposable
{
readonly IClientStreamWriter<TRequest> requestStream;
- readonly Task<TResponse> result;
+ readonly Task<TResponse> responseAsync;
+ readonly Func<Status> getStatusFunc;
+ readonly Func<Metadata> getTrailersFunc;
readonly Action disposeAction;
- public AsyncClientStreamingCall(IClientStreamWriter<TRequest> requestStream, Task<TResponse> result, Action disposeAction)
+ public AsyncClientStreamingCall(IClientStreamWriter<TRequest> requestStream, Task<TResponse> responseAsync, Func<Status> getStatusFunc, Func<Metadata> getTrailersFunc, Action disposeAction)
{
this.requestStream = requestStream;
- this.result = result;
+ this.responseAsync = responseAsync;
+ this.getStatusFunc = getStatusFunc;
+ this.getTrailersFunc = getTrailersFunc;
this.disposeAction = disposeAction;
}
/// <summary>
/// Asynchronous call result.
/// </summary>
- public Task<TResponse> Result
+ public Task<TResponse> ResponseAsync
{
get
{
- return this.result;
+ return this.responseAsync;
}
}
@@ -81,11 +85,11 @@ namespace Grpc.Core
/// <returns></returns>
public TaskAwaiter<TResponse> GetAwaiter()
{
- return result.GetAwaiter();
+ return responseAsync.GetAwaiter();
}
/// <summary>
- /// Provides means to provide after the call.
+ /// Provides means to cleanup after the call.
/// If the call has already finished normally (request stream has been completed and call result has been received), doesn't do anything.
/// Otherwise, requests cancellation of the call which should terminate all pending async operations associated with the call.
/// As a result, all resources being used by the call should be released eventually.
diff --git a/src/csharp/Grpc.Core/AsyncDuplexStreamingCall.cs b/src/csharp/Grpc.Core/AsyncDuplexStreamingCall.cs
index 4c0d5936ac..0979de606f 100644
--- a/src/csharp/Grpc.Core/AsyncDuplexStreamingCall.cs
+++ b/src/csharp/Grpc.Core/AsyncDuplexStreamingCall.cs
@@ -44,12 +44,16 @@ namespace Grpc.Core
{
readonly IClientStreamWriter<TRequest> requestStream;
readonly IAsyncStreamReader<TResponse> responseStream;
+ readonly Func<Status> getStatusFunc;
+ readonly Func<Metadata> getTrailersFunc;
readonly Action disposeAction;
- public AsyncDuplexStreamingCall(IClientStreamWriter<TRequest> requestStream, IAsyncStreamReader<TResponse> responseStream, Action disposeAction)
+ public AsyncDuplexStreamingCall(IClientStreamWriter<TRequest> requestStream, IAsyncStreamReader<TResponse> responseStream, Func<Status> getStatusFunc, Func<Metadata> getTrailersFunc, Action disposeAction)
{
this.requestStream = requestStream;
this.responseStream = responseStream;
+ this.getStatusFunc = getStatusFunc;
+ this.getTrailersFunc = getTrailersFunc;
this.disposeAction = disposeAction;
}
@@ -76,6 +80,24 @@ namespace Grpc.Core
}
/// <summary>
+ /// Gets the call status if the call has already finished.
+ /// Throws InvalidOperationException otherwise.
+ /// </summary>
+ public Status GetStatus()
+ {
+ return getStatusFunc();
+ }
+
+ /// <summary>
+ /// Gets the call trailing metadata if the call has already finished.
+ /// Throws InvalidOperationException otherwise.
+ /// </summary>
+ public Metadata GetTrailers()
+ {
+ return getTrailersFunc();
+ }
+
+ /// <summary>
/// Provides means to cleanup after the call.
/// If the call has already finished normally (request stream has been completed and response stream has been fully read), doesn't do anything.
/// Otherwise, requests cancellation of the call which should terminate all pending async operations associated with the call.
diff --git a/src/csharp/Grpc.Core/AsyncServerStreamingCall.cs b/src/csharp/Grpc.Core/AsyncServerStreamingCall.cs
index 7a479b9a23..380efcdb0e 100644
--- a/src/csharp/Grpc.Core/AsyncServerStreamingCall.cs
+++ b/src/csharp/Grpc.Core/AsyncServerStreamingCall.cs
@@ -43,11 +43,15 @@ namespace Grpc.Core
public sealed class AsyncServerStreamingCall<TResponse> : IDisposable
{
readonly IAsyncStreamReader<TResponse> responseStream;
+ readonly Func<Status> getStatusFunc;
+ readonly Func<Metadata> getTrailersFunc;
readonly Action disposeAction;
- public AsyncServerStreamingCall(IAsyncStreamReader<TResponse> responseStream, Action disposeAction)
+ public AsyncServerStreamingCall(IAsyncStreamReader<TResponse> responseStream, Func<Status> getStatusFunc, Func<Metadata> getTrailersFunc, Action disposeAction)
{
this.responseStream = responseStream;
+ this.getStatusFunc = getStatusFunc;
+ this.getTrailersFunc = getTrailersFunc;
this.disposeAction = disposeAction;
}
@@ -63,6 +67,24 @@ namespace Grpc.Core
}
/// <summary>
+ /// Gets the call status if the call has already finished.
+ /// Throws InvalidOperationException otherwise.
+ /// </summary>
+ public Status GetStatus()
+ {
+ return getStatusFunc();
+ }
+
+ /// <summary>
+ /// Gets the call trailing metadata if the call has already finished.
+ /// Throws InvalidOperationException otherwise.
+ /// </summary>
+ public Metadata GetTrailers()
+ {
+ return getTrailersFunc();
+ }
+
+ /// <summary>
/// Provides means to cleanup after the call.
/// If the call has already finished normally (response stream has been fully read), doesn't do anything.
/// Otherwise, requests cancellation of the call which should terminate all pending async operations associated with the call.
diff --git a/src/csharp/Grpc.Core/AsyncUnaryCall.cs b/src/csharp/Grpc.Core/AsyncUnaryCall.cs
new file mode 100644
index 0000000000..224e343916
--- /dev/null
+++ b/src/csharp/Grpc.Core/AsyncUnaryCall.cs
@@ -0,0 +1,106 @@
+#region Copyright notice and license
+
+// 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.
+
+#endregion
+
+using System;
+using System.Runtime.CompilerServices;
+using System.Threading.Tasks;
+
+namespace Grpc.Core
+{
+ /// <summary>
+ /// Return type for single request - single response call.
+ /// </summary>
+ public sealed class AsyncUnaryCall<TResponse> : IDisposable
+ {
+ readonly Task<TResponse> responseAsync;
+ readonly Func<Status> getStatusFunc;
+ readonly Func<Metadata> getTrailersFunc;
+ readonly Action disposeAction;
+
+ public AsyncUnaryCall(Task<TResponse> responseAsync, Func<Status> getStatusFunc, Func<Metadata> getTrailersFunc, Action disposeAction)
+ {
+ this.responseAsync = responseAsync;
+ this.getStatusFunc = getStatusFunc;
+ this.getTrailersFunc = getTrailersFunc;
+ this.disposeAction = disposeAction;
+ }
+
+ /// <summary>
+ /// Asynchronous call result.
+ /// </summary>
+ public Task<TResponse> ResponseAsync
+ {
+ get
+ {
+ return this.responseAsync;
+ }
+ }
+
+ /// <summary>
+ /// Allows awaiting this object directly.
+ /// </summary>
+ public TaskAwaiter<TResponse> GetAwaiter()
+ {
+ return responseAsync.GetAwaiter();
+ }
+
+ /// <summary>
+ /// Gets the call status if the call has already finished.
+ /// Throws InvalidOperationException otherwise.
+ /// </summary>
+ public Status GetStatus()
+ {
+ return getStatusFunc();
+ }
+
+ /// <summary>
+ /// Gets the call trailing metadata if the call has already finished.
+ /// Throws InvalidOperationException otherwise.
+ /// </summary>
+ public Metadata GetTrailers()
+ {
+ return getTrailersFunc();
+ }
+
+ /// <summary>
+ /// Provides means to cleanup after the call.
+ /// If the call has already finished normally (request stream has been completed and call result has been received), doesn't do anything.
+ /// Otherwise, requests cancellation of the call which should terminate all pending async operations associated with the call.
+ /// As a result, all resources being used by the call should be released eventually.
+ /// </summary>
+ public void Dispose()
+ {
+ disposeAction.Invoke();
+ }
+ }
+}
diff --git a/src/csharp/Grpc.Core/Calls.cs b/src/csharp/Grpc.Core/Calls.cs
index 9e95182c72..359fe53741 100644
--- a/src/csharp/Grpc.Core/Calls.cs
+++ b/src/csharp/Grpc.Core/Calls.cs
@@ -53,7 +53,7 @@ namespace Grpc.Core
return asyncCall.UnaryCall(call.Channel, call.Name, req, call.Headers);
}
- public static async Task<TResponse> AsyncUnaryCall<TRequest, TResponse>(Call<TRequest, TResponse> call, TRequest req, CancellationToken token)
+ public static AsyncUnaryCall<TResponse> AsyncUnaryCall<TRequest, TResponse>(Call<TRequest, TResponse> call, TRequest req, CancellationToken token)
where TRequest : class
where TResponse : class
{
@@ -61,7 +61,7 @@ namespace Grpc.Core
asyncCall.Initialize(call.Channel, call.Channel.CompletionQueue, call.Name);
var asyncResult = asyncCall.UnaryCallAsync(req, call.Headers);
RegisterCancellationCallback(asyncCall, token);
- return await asyncResult;
+ return new AsyncUnaryCall<TResponse>(asyncResult, asyncCall.GetStatus, asyncCall.GetTrailers, asyncCall.Cancel);
}
public static AsyncServerStreamingCall<TResponse> AsyncServerStreamingCall<TRequest, TResponse>(Call<TRequest, TResponse> call, TRequest req, CancellationToken token)
@@ -73,7 +73,7 @@ namespace Grpc.Core
asyncCall.StartServerStreamingCall(req, call.Headers);
RegisterCancellationCallback(asyncCall, token);
var responseStream = new ClientResponseStream<TRequest, TResponse>(asyncCall);
- return new AsyncServerStreamingCall<TResponse>(responseStream, asyncCall.Cancel);
+ return new AsyncServerStreamingCall<TResponse>(responseStream, asyncCall.GetStatus, asyncCall.GetTrailers, asyncCall.Cancel);
}
public static AsyncClientStreamingCall<TRequest, TResponse> AsyncClientStreamingCall<TRequest, TResponse>(Call<TRequest, TResponse> call, CancellationToken token)
@@ -85,7 +85,7 @@ namespace Grpc.Core
var resultTask = asyncCall.ClientStreamingCallAsync(call.Headers);
RegisterCancellationCallback(asyncCall, token);
var requestStream = new ClientRequestStream<TRequest, TResponse>(asyncCall);
- return new AsyncClientStreamingCall<TRequest, TResponse>(requestStream, resultTask, asyncCall.Cancel);
+ return new AsyncClientStreamingCall<TRequest, TResponse>(requestStream, resultTask, asyncCall.GetStatus, asyncCall.GetTrailers, asyncCall.Cancel);
}
public static AsyncDuplexStreamingCall<TRequest, TResponse> AsyncDuplexStreamingCall<TRequest, TResponse>(Call<TRequest, TResponse> call, CancellationToken token)
@@ -98,7 +98,7 @@ namespace Grpc.Core
RegisterCancellationCallback(asyncCall, token);
var requestStream = new ClientRequestStream<TRequest, TResponse>(asyncCall);
var responseStream = new ClientResponseStream<TRequest, TResponse>(asyncCall);
- return new AsyncDuplexStreamingCall<TRequest, TResponse>(requestStream, responseStream, asyncCall.Cancel);
+ return new AsyncDuplexStreamingCall<TRequest, TResponse>(requestStream, responseStream, asyncCall.GetStatus, asyncCall.GetTrailers, asyncCall.Cancel);
}
private static void RegisterCancellationCallback<TRequest, TResponse>(AsyncCall<TRequest, TResponse> asyncCall, CancellationToken token)
diff --git a/src/csharp/Grpc.Core/Channel.cs b/src/csharp/Grpc.Core/Channel.cs
index 5baf260003..e5c6abd2cb 100644
--- a/src/csharp/Grpc.Core/Channel.cs
+++ b/src/csharp/Grpc.Core/Channel.cs
@@ -28,11 +28,14 @@
// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#endregion
+
using System;
using System.Collections.Generic;
+using System.Linq;
using System.Runtime.InteropServices;
using System.Threading;
using System.Threading.Tasks;
+
using Grpc.Core.Internal;
namespace Grpc.Core
@@ -44,6 +47,7 @@ namespace Grpc.Core
{
readonly GrpcEnvironment environment;
readonly ChannelSafeHandle handle;
+ readonly List<ChannelOption> options;
readonly string target;
bool disposed;
@@ -57,7 +61,10 @@ namespace Grpc.Core
public Channel(string host, Credentials credentials = null, IEnumerable<ChannelOption> options = null)
{
this.environment = GrpcEnvironment.GetInstance();
- using (ChannelArgsSafeHandle nativeChannelArgs = ChannelOptions.CreateChannelArgs(options))
+ this.options = options != null ? new List<ChannelOption>(options) : new List<ChannelOption>();
+
+ EnsureUserAgentChannelOption(this.options);
+ using (ChannelArgsSafeHandle nativeChannelArgs = ChannelOptions.CreateChannelArgs(this.options))
{
if (credentials != null)
{
@@ -71,7 +78,7 @@ namespace Grpc.Core
this.handle = ChannelSafeHandle.Create(host, nativeChannelArgs);
}
}
- this.target = GetOverridenTarget(host, options);
+ this.target = GetOverridenTarget(host, this.options);
}
/// <summary>
@@ -141,6 +148,20 @@ namespace Grpc.Core
}
}
+ private static void EnsureUserAgentChannelOption(List<ChannelOption> options)
+ {
+ if (!options.Any((option) => option.Name == ChannelOptions.PrimaryUserAgentString))
+ {
+ options.Add(new ChannelOption(ChannelOptions.PrimaryUserAgentString, GetUserAgentString()));
+ }
+ }
+
+ private static string GetUserAgentString()
+ {
+ // TODO(jtattermusch): it would be useful to also provide .NET/mono version.
+ return string.Format("grpc-csharp/{0}", VersionInfo.CurrentVersion);
+ }
+
/// <summary>
/// Look for SslTargetNameOverride option and return its value instead of originalTarget
/// if found.
diff --git a/src/csharp/Grpc.Core/ChannelOptions.cs b/src/csharp/Grpc.Core/ChannelOptions.cs
index bc23bb59b1..9fe03d2805 100644
--- a/src/csharp/Grpc.Core/ChannelOptions.cs
+++ b/src/csharp/Grpc.Core/ChannelOptions.cs
@@ -115,41 +115,49 @@ namespace Grpc.Core
}
}
+ /// <summary>
+ /// Defines names of supported channel options.
+ /// </summary>
public static class ChannelOptions
{
- // Override SSL target check. Only to be used for testing.
+ /// <summary>Override SSL target check. Only to be used for testing.</summary>
public const string SslTargetNameOverride = "grpc.ssl_target_name_override";
- // Enable census for tracing and stats collection
+ /// <summary>Enable census for tracing and stats collection</summary>
public const string Census = "grpc.census";
- // Maximum number of concurrent incoming streams to allow on a http2 connection
+ /// <summary>Maximum number of concurrent incoming streams to allow on a http2 connection</summary>
public const string MaxConcurrentStreams = "grpc.max_concurrent_streams";
- // Maximum message length that the channel can receive
+ /// <summary>Maximum message length that the channel can receive</summary>
public const string MaxMessageLength = "grpc.max_message_length";
- // Initial sequence number for http2 transports
+ /// <summary>Initial sequence number for http2 transports</summary>
public const string Http2InitialSequenceNumber = "grpc.http2.initial_sequence_number";
+ /// <summary>Primary user agent: goes at the start of the user-agent metadata</summary>
+ public const string PrimaryUserAgentString = "grpc.primary_user_agent";
+
+ /// <summary> Secondary user agent: goes at the end of the user-agent metadata</summary>
+ public const string SecondaryUserAgentString = "grpc.secondary_user_agent";
+
/// <summary>
/// Creates native object for a collection of channel options.
/// </summary>
/// <returns>The native channel arguments.</returns>
- internal static ChannelArgsSafeHandle CreateChannelArgs(IEnumerable<ChannelOption> options)
+ internal static ChannelArgsSafeHandle CreateChannelArgs(List<ChannelOption> options)
{
- if (options == null)
+ if (options == null || options.Count == 0)
{
return ChannelArgsSafeHandle.CreateNull();
}
- var optionList = new List<ChannelOption>(options); // It's better to do defensive copy
ChannelArgsSafeHandle nativeArgs = null;
try
{
- nativeArgs = ChannelArgsSafeHandle.Create(optionList.Count);
- for (int i = 0; i < optionList.Count; i++)
+ nativeArgs = ChannelArgsSafeHandle.Create(options.Count);
+ for (int i = 0; i < options.Count; i++)
{
- var option = optionList[i];
+ var option = options[i];
if (option.Type == ChannelOption.OptionType.Integer)
{
nativeArgs.SetInteger(i, option.Name, option.IntValue);
diff --git a/src/csharp/Grpc.Core/Grpc.Core.csproj b/src/csharp/Grpc.Core/Grpc.Core.csproj
index a227fe5477..fd68b91851 100644
--- a/src/csharp/Grpc.Core/Grpc.Core.csproj
+++ b/src/csharp/Grpc.Core/Grpc.Core.csproj
@@ -33,13 +33,12 @@
</PropertyGroup>
<ItemGroup>
<Reference Include="System" />
- <Reference Include="System.Collections.Immutable, Version=1.1.36.0, Culture=neutral, PublicKeyToken=b03f5f7f11d50a3a, processorArchitecture=MSIL">
- <SpecificVersion>False</SpecificVersion>
- <HintPath>..\packages\System.Collections.Immutable.1.1.36\lib\portable-net45+win8+wp8+wpa81\System.Collections.Immutable.dll</HintPath>
- </Reference>
<Reference Include="System.Interactive.Async">
<HintPath>..\packages\Ix-Async.1.2.3\lib\net45\System.Interactive.Async.dll</HintPath>
</Reference>
+ <Reference Include="System.Collections.Immutable">
+ <HintPath>..\packages\System.Collections.Immutable.1.1.36\lib\portable-net45+win8+wp8+wpa81\System.Collections.Immutable.dll</HintPath>
+ </Reference>
</ItemGroup>
<ItemGroup>
<Compile Include="AsyncDuplexStreamingCall.cs" />
@@ -102,6 +101,8 @@
<Compile Include="Internal\CompletionRegistry.cs" />
<Compile Include="Internal\BatchContextSafeHandle.cs" />
<Compile Include="ChannelOptions.cs" />
+ <Compile Include="AsyncUnaryCall.cs" />
+ <Compile Include="VersionInfo.cs" />
</ItemGroup>
<ItemGroup>
<None Include="Grpc.Core.nuspec" />
diff --git a/src/csharp/Grpc.Core/Internal/AsyncCall.cs b/src/csharp/Grpc.Core/Internal/AsyncCall.cs
index 24b75d1668..f983dbb759 100644
--- a/src/csharp/Grpc.Core/Internal/AsyncCall.cs
+++ b/src/csharp/Grpc.Core/Internal/AsyncCall.cs
@@ -52,8 +52,8 @@ namespace Grpc.Core.Internal
// Completion of a pending unary response if not null.
TaskCompletionSource<TResponse> unaryResponseTcs;
- // Set after status is received. Only used for streaming response calls.
- Status? finishedStatus;
+ // Set after status is received. Used for both unary and streaming response calls.
+ ClientSideStatus? finishedStatus;
bool readObserverCompleted; // True if readObserver has already been completed.
@@ -249,6 +249,32 @@ namespace Grpc.Core.Internal
}
/// <summary>
+ /// Gets the resulting status if the call has already finished.
+ /// Throws InvalidOperationException otherwise.
+ /// </summary>
+ public Status GetStatus()
+ {
+ lock (myLock)
+ {
+ Preconditions.CheckState(finishedStatus.HasValue, "Status can only be accessed once the call has finished.");
+ return finishedStatus.Value.Status;
+ }
+ }
+
+ /// <summary>
+ /// Gets the trailing metadata if the call has already finished.
+ /// Throws InvalidOperationException otherwise.
+ /// </summary>
+ public Metadata GetTrailers()
+ {
+ lock (myLock)
+ {
+ Preconditions.CheckState(finishedStatus.HasValue, "Trailers can only be accessed once the call has finished.");
+ return finishedStatus.Value.Trailers;
+ }
+ }
+
+ /// <summary>
/// On client-side, we only fire readCompletionDelegate once all messages have been read
/// and status has been received.
/// </summary>
@@ -265,7 +291,7 @@ namespace Grpc.Core.Internal
if (shouldComplete)
{
- var status = finishedStatus.Value;
+ var status = finishedStatus.Value.Status;
if (status.StatusCode != StatusCode.OK)
{
FireCompletion(completionDelegate, default(TResponse), new RpcException(status));
@@ -288,9 +314,13 @@ namespace Grpc.Core.Internal
/// </summary>
private void HandleUnaryResponse(bool success, BatchContextSafeHandle ctx)
{
+ var fullStatus = ctx.GetReceivedStatusOnClient();
+
lock (myLock)
{
finished = true;
+ finishedStatus = fullStatus;
+
halfclosed = true;
ReleaseResourcesIfPossible();
@@ -302,7 +332,8 @@ namespace Grpc.Core.Internal
return;
}
- var status = ctx.GetReceivedStatus();
+ var status = fullStatus.Status;
+
if (status.StatusCode != StatusCode.OK)
{
unaryResponseTcs.SetException(new RpcException(status));
@@ -321,13 +352,13 @@ namespace Grpc.Core.Internal
/// </summary>
private void HandleFinished(bool success, BatchContextSafeHandle ctx)
{
- var status = ctx.GetReceivedStatus();
+ var fullStatus = ctx.GetReceivedStatusOnClient();
AsyncCompletionDelegate<TResponse> origReadCompletionDelegate = null;
lock (myLock)
{
finished = true;
- finishedStatus = status;
+ finishedStatus = fullStatus;
origReadCompletionDelegate = readCompletionDelegate;
diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs
index 309067ea9d..f809f4a84c 100644
--- a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs
+++ b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs
@@ -101,14 +101,17 @@ namespace Grpc.Core.Internal
/// Only one pending send action is allowed at any given time.
/// completionDelegate is called when the operation finishes.
/// </summary>
- public void StartSendStatusFromServer(Status status, AsyncCompletionDelegate<object> completionDelegate)
+ public void StartSendStatusFromServer(Status status, Metadata trailers, AsyncCompletionDelegate<object> completionDelegate)
{
lock (myLock)
{
Preconditions.CheckNotNull(completionDelegate, "Completion delegate cannot be null");
CheckSendingAllowed();
- call.StartSendStatusFromServer(status, HandleHalfclosed);
+ using (var metadataArray = MetadataArraySafeHandle.Create(trailers))
+ {
+ call.StartSendStatusFromServer(status, HandleHalfclosed, metadataArray);
+ }
halfcloseRequested = true;
readingDone = true;
sendCompletionDelegate = completionDelegate;
diff --git a/src/csharp/Grpc.Core/Internal/BatchContextSafeHandle.cs b/src/csharp/Grpc.Core/Internal/BatchContextSafeHandle.cs
index 861cbbe4c6..6a2add54db 100644
--- a/src/csharp/Grpc.Core/Internal/BatchContextSafeHandle.cs
+++ b/src/csharp/Grpc.Core/Internal/BatchContextSafeHandle.cs
@@ -38,7 +38,6 @@ using Grpc.Core;
namespace Grpc.Core.Internal
{
/// <summary>
- /// Not owned version of
/// grpcsharp_batch_context
/// </summary>
internal class BatchContextSafeHandle : SafeHandleZeroIsInvalid
@@ -47,6 +46,9 @@ namespace Grpc.Core.Internal
static extern BatchContextSafeHandle grpcsharp_batch_context_create();
[DllImport("grpc_csharp_ext.dll")]
+ static extern IntPtr grpcsharp_batch_context_recv_initial_metadata(BatchContextSafeHandle ctx);
+
+ [DllImport("grpc_csharp_ext.dll")]
static extern IntPtr grpcsharp_batch_context_recv_message_length(BatchContextSafeHandle ctx);
[DllImport("grpc_csharp_ext.dll")]
@@ -59,12 +61,24 @@ namespace Grpc.Core.Internal
static extern IntPtr grpcsharp_batch_context_recv_status_on_client_details(BatchContextSafeHandle ctx); // returns const char*
[DllImport("grpc_csharp_ext.dll")]
+ static extern IntPtr grpcsharp_batch_context_recv_status_on_client_trailing_metadata(BatchContextSafeHandle ctx);
+
+ [DllImport("grpc_csharp_ext.dll")]
static extern CallSafeHandle grpcsharp_batch_context_server_rpc_new_call(BatchContextSafeHandle ctx);
[DllImport("grpc_csharp_ext.dll")]
static extern IntPtr grpcsharp_batch_context_server_rpc_new_method(BatchContextSafeHandle ctx); // returns const char*
[DllImport("grpc_csharp_ext.dll")]
+ static extern IntPtr grpcsharp_batch_context_server_rpc_new_host(BatchContextSafeHandle ctx); // returns const char*
+
+ [DllImport("grpc_csharp_ext.dll")]
+ static extern Timespec grpcsharp_batch_context_server_rpc_new_deadline(BatchContextSafeHandle ctx);
+
+ [DllImport("grpc_csharp_ext.dll")]
+ static extern IntPtr grpcsharp_batch_context_server_rpc_new_request_metadata(BatchContextSafeHandle ctx);
+
+ [DllImport("grpc_csharp_ext.dll")]
static extern int grpcsharp_batch_context_recv_close_on_server_cancelled(BatchContextSafeHandle ctx);
[DllImport("grpc_csharp_ext.dll")]
@@ -87,13 +101,26 @@ namespace Grpc.Core.Internal
}
}
- public Status GetReceivedStatus()
+ // Gets data of recv_initial_metadata completion.
+ public Metadata GetReceivedInitialMetadata()
+ {
+ IntPtr metadataArrayPtr = grpcsharp_batch_context_recv_initial_metadata(this);
+ return MetadataArraySafeHandle.ReadMetadataFromPtrUnsafe(metadataArrayPtr);
+ }
+
+ // Gets data of recv_status_on_client completion.
+ public ClientSideStatus GetReceivedStatusOnClient()
{
- // TODO: can the native method return string directly?
string details = Marshal.PtrToStringAnsi(grpcsharp_batch_context_recv_status_on_client_details(this));
- return new Status(grpcsharp_batch_context_recv_status_on_client_status(this), details);
+ var status = new Status(grpcsharp_batch_context_recv_status_on_client_status(this), details);
+
+ IntPtr metadataArrayPtr = grpcsharp_batch_context_recv_status_on_client_trailing_metadata(this);
+ var metadata = MetadataArraySafeHandle.ReadMetadataFromPtrUnsafe(metadataArrayPtr);
+
+ return new ClientSideStatus(status, metadata);
}
+ // Gets data of recv_message completion.
public byte[] GetReceivedMessage()
{
IntPtr len = grpcsharp_batch_context_recv_message_length(this);
@@ -106,16 +133,22 @@ namespace Grpc.Core.Internal
return data;
}
- public CallSafeHandle GetServerRpcNewCall()
+ // Gets data of server_rpc_new completion.
+ public ServerRpcNew GetServerRpcNew()
{
- return grpcsharp_batch_context_server_rpc_new_call(this);
- }
+ var call = grpcsharp_batch_context_server_rpc_new_call(this);
- public string GetServerRpcNewMethod()
- {
- return Marshal.PtrToStringAnsi(grpcsharp_batch_context_server_rpc_new_method(this));
+ var method = Marshal.PtrToStringAnsi(grpcsharp_batch_context_server_rpc_new_method(this));
+ var host = Marshal.PtrToStringAnsi(grpcsharp_batch_context_server_rpc_new_host(this));
+ var deadline = grpcsharp_batch_context_server_rpc_new_deadline(this);
+
+ IntPtr metadataArrayPtr = grpcsharp_batch_context_server_rpc_new_request_metadata(this);
+ var metadata = MetadataArraySafeHandle.ReadMetadataFromPtrUnsafe(metadataArrayPtr);
+
+ return new ServerRpcNew(call, method, host, deadline, metadata);
}
+ // Gets data of receive_close_on_server completion.
public bool GetReceivedCloseOnServerCancelled()
{
return grpcsharp_batch_context_recv_close_on_server_cancelled(this) != 0;
@@ -127,4 +160,97 @@ namespace Grpc.Core.Internal
return true;
}
}
+
+ /// <summary>
+ /// Status + metadata received on client side when call finishes.
+ /// (when receive_status_on_client operation finishes).
+ /// </summary>
+ internal struct ClientSideStatus
+ {
+ readonly Status status;
+ readonly Metadata trailers;
+
+ public ClientSideStatus(Status status, Metadata trailers)
+ {
+ this.status = status;
+ this.trailers = trailers;
+ }
+
+ public Status Status
+ {
+ get
+ {
+ return this.status;
+ }
+ }
+
+ public Metadata Trailers
+ {
+ get
+ {
+ return this.trailers;
+ }
+ }
+ }
+
+ /// <summary>
+ /// Details of a newly received RPC.
+ /// </summary>
+ internal struct ServerRpcNew
+ {
+ readonly CallSafeHandle call;
+ readonly string method;
+ readonly string host;
+ readonly Timespec deadline;
+ readonly Metadata requestMetadata;
+
+ public ServerRpcNew(CallSafeHandle call, string method, string host, Timespec deadline, Metadata requestMetadata)
+ {
+ this.call = call;
+ this.method = method;
+ this.host = host;
+ this.deadline = deadline;
+ this.requestMetadata = requestMetadata;
+ }
+
+ public CallSafeHandle Call
+ {
+ get
+ {
+ return this.call;
+ }
+ }
+
+ public string Method
+ {
+ get
+ {
+ return this.method;
+ }
+ }
+
+ public string Host
+ {
+ get
+ {
+ return this.host;
+ }
+ }
+
+ public Timespec Deadline
+ {
+ get
+ {
+ return this.deadline;
+ }
+ }
+
+ public Metadata RequestMetadata
+ {
+ get
+ {
+ return this.requestMetadata;
+ }
+ }
+ }
} \ No newline at end of file
diff --git a/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs b/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs
index 3b246ac01b..19dbb83f24 100644
--- a/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs
+++ b/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs
@@ -81,7 +81,7 @@ namespace Grpc.Core.Internal
[DllImport("grpc_csharp_ext.dll")]
static extern GRPCCallError grpcsharp_call_send_status_from_server(CallSafeHandle call,
- BatchContextSafeHandle ctx, StatusCode statusCode, string statusMessage);
+ BatchContextSafeHandle ctx, StatusCode statusCode, string statusMessage, MetadataArraySafeHandle metadataArray);
[DllImport("grpc_csharp_ext.dll")]
static extern GRPCCallError grpcsharp_call_recv_message(CallSafeHandle call,
@@ -159,11 +159,11 @@ namespace Grpc.Core.Internal
grpcsharp_call_send_close_from_client(this, ctx).CheckOk();
}
- public void StartSendStatusFromServer(Status status, BatchCompletionDelegate callback)
+ public void StartSendStatusFromServer(Status status, BatchCompletionDelegate callback, MetadataArraySafeHandle metadataArray)
{
var ctx = BatchContextSafeHandle.Create();
completionRegistry.RegisterBatchCompletion(ctx, callback);
- grpcsharp_call_send_status_from_server(this, ctx, status.StatusCode, status.Detail).CheckOk();
+ grpcsharp_call_send_status_from_server(this, ctx, status.StatusCode, status.Detail, metadataArray).CheckOk();
}
public void StartReceiveMessage(BatchCompletionDelegate callback)
diff --git a/src/csharp/Grpc.Core/Internal/MetadataArraySafeHandle.cs b/src/csharp/Grpc.Core/Internal/MetadataArraySafeHandle.cs
index 80aa7f5603..427c16fac6 100644
--- a/src/csharp/Grpc.Core/Internal/MetadataArraySafeHandle.cs
+++ b/src/csharp/Grpc.Core/Internal/MetadataArraySafeHandle.cs
@@ -46,12 +46,24 @@ namespace Grpc.Core.Internal
static extern void grpcsharp_metadata_array_add(MetadataArraySafeHandle array, string key, byte[] value, UIntPtr valueLength);
[DllImport("grpc_csharp_ext.dll")]
+ static extern UIntPtr grpcsharp_metadata_array_count(IntPtr metadataArray);
+
+ [DllImport("grpc_csharp_ext.dll")]
+ static extern IntPtr grpcsharp_metadata_array_get_key(IntPtr metadataArray, UIntPtr index);
+
+ [DllImport("grpc_csharp_ext.dll")]
+ static extern IntPtr grpcsharp_metadata_array_get_value(IntPtr metadataArray, UIntPtr index);
+
+ [DllImport("grpc_csharp_ext.dll")]
+ static extern UIntPtr grpcsharp_metadata_array_get_value_length(IntPtr metadataArray, UIntPtr index);
+
+ [DllImport("grpc_csharp_ext.dll")]
static extern void grpcsharp_metadata_array_destroy_full(IntPtr array);
private MetadataArraySafeHandle()
{
}
-
+
public static MetadataArraySafeHandle Create(Metadata metadata)
{
// TODO(jtattermusch): we might wanna check that the metadata is readonly
@@ -63,6 +75,38 @@ namespace Grpc.Core.Internal
return metadataArray;
}
+ /// <summary>
+ /// Reads metadata from pointer to grpc_metadata_array
+ /// </summary>
+ public static Metadata ReadMetadataFromPtrUnsafe(IntPtr metadataArray)
+ {
+ if (metadataArray == IntPtr.Zero)
+ {
+ return null;
+ }
+
+ ulong count = grpcsharp_metadata_array_count(metadataArray).ToUInt64();
+
+ var metadata = new Metadata();
+ for (ulong i = 0; i < count; i++)
+ {
+ var index = new UIntPtr(i);
+ string key = Marshal.PtrToStringAnsi(grpcsharp_metadata_array_get_key(metadataArray, index));
+ var bytes = new byte[grpcsharp_metadata_array_get_value_length(metadataArray, index).ToUInt64()];
+ Marshal.Copy(grpcsharp_metadata_array_get_value(metadataArray, index), bytes, 0, bytes.Length);
+ metadata.Add(new Metadata.Entry(key, bytes));
+ }
+ return metadata;
+ }
+
+ internal IntPtr Handle
+ {
+ get
+ {
+ return handle;
+ }
+ }
+
protected override bool ReleaseHandle()
{
grpcsharp_metadata_array_destroy_full(handle);
diff --git a/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs b/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs
index 594e46b159..3680b1e791 100644
--- a/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs
+++ b/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs
@@ -34,6 +34,7 @@
using System;
using System.Collections.Generic;
using System.Linq;
+using System.Threading;
using System.Threading.Tasks;
using Grpc.Core.Internal;
using Grpc.Core.Utils;
@@ -42,7 +43,7 @@ namespace Grpc.Core.Internal
{
internal interface IServerCallHandler
{
- Task HandleCall(string methodName, CallSafeHandle call, GrpcEnvironment environment);
+ Task HandleCall(ServerRpcNew newRpc, GrpcEnvironment environment);
}
internal class UnaryServerCallHandler<TRequest, TResponse> : IServerCallHandler
@@ -58,27 +59,28 @@ namespace Grpc.Core.Internal
this.handler = handler;
}
- public async Task HandleCall(string methodName, CallSafeHandle call, GrpcEnvironment environment)
+ public async Task HandleCall(ServerRpcNew newRpc, GrpcEnvironment environment)
{
var asyncCall = new AsyncCallServer<TRequest, TResponse>(
method.ResponseMarshaller.Serializer,
method.RequestMarshaller.Deserializer,
environment);
- asyncCall.Initialize(call);
+ asyncCall.Initialize(newRpc.Call);
var finishedTask = asyncCall.ServerSideCallAsync();
var requestStream = new ServerRequestStream<TRequest, TResponse>(asyncCall);
var responseStream = new ServerResponseStream<TRequest, TResponse>(asyncCall);
- Status status = Status.DefaultSuccess;
+ Status status;
+ var context = HandlerUtils.NewContext(newRpc);
try
{
Preconditions.CheckArgument(await requestStream.MoveNext());
var request = requestStream.Current;
// TODO(jtattermusch): we need to read the full stream so that native callhandle gets deallocated.
Preconditions.CheckArgument(!await requestStream.MoveNext());
- var context = new ServerCallContext(); // TODO(jtattermusch): initialize the context
- var result = await handler(context, request);
+ var result = await handler(request, context);
+ status = context.Status;
await responseStream.WriteAsync(result);
}
catch (Exception e)
@@ -88,7 +90,7 @@ namespace Grpc.Core.Internal
}
try
{
- await responseStream.WriteStatusAsync(status);
+ await responseStream.WriteStatusAsync(status, context.ResponseTrailers);
}
catch (OperationCanceledException)
{
@@ -111,28 +113,28 @@ namespace Grpc.Core.Internal
this.handler = handler;
}
- public async Task HandleCall(string methodName, CallSafeHandle call, GrpcEnvironment environment)
+ public async Task HandleCall(ServerRpcNew newRpc, GrpcEnvironment environment)
{
var asyncCall = new AsyncCallServer<TRequest, TResponse>(
method.ResponseMarshaller.Serializer,
method.RequestMarshaller.Deserializer,
environment);
- asyncCall.Initialize(call);
+ asyncCall.Initialize(newRpc.Call);
var finishedTask = asyncCall.ServerSideCallAsync();
var requestStream = new ServerRequestStream<TRequest, TResponse>(asyncCall);
var responseStream = new ServerResponseStream<TRequest, TResponse>(asyncCall);
- Status status = Status.DefaultSuccess;
+ Status status;
+ var context = HandlerUtils.NewContext(newRpc);
try
{
Preconditions.CheckArgument(await requestStream.MoveNext());
var request = requestStream.Current;
// TODO(jtattermusch): we need to read the full stream so that native callhandle gets deallocated.
Preconditions.CheckArgument(!await requestStream.MoveNext());
-
- var context = new ServerCallContext(); // TODO(jtattermusch): initialize the context
- await handler(context, request, responseStream);
+ await handler(request, responseStream, context);
+ status = context.Status;
}
catch (Exception e)
{
@@ -142,7 +144,7 @@ namespace Grpc.Core.Internal
try
{
- await responseStream.WriteStatusAsync(status);
+ await responseStream.WriteStatusAsync(status, context.ResponseTrailers);
}
catch (OperationCanceledException)
{
@@ -165,23 +167,24 @@ namespace Grpc.Core.Internal
this.handler = handler;
}
- public async Task HandleCall(string methodName, CallSafeHandle call, GrpcEnvironment environment)
+ public async Task HandleCall(ServerRpcNew newRpc, GrpcEnvironment environment)
{
var asyncCall = new AsyncCallServer<TRequest, TResponse>(
method.ResponseMarshaller.Serializer,
method.RequestMarshaller.Deserializer,
environment);
- asyncCall.Initialize(call);
+ asyncCall.Initialize(newRpc.Call);
var finishedTask = asyncCall.ServerSideCallAsync();
var requestStream = new ServerRequestStream<TRequest, TResponse>(asyncCall);
var responseStream = new ServerResponseStream<TRequest, TResponse>(asyncCall);
- var context = new ServerCallContext(); // TODO(jtattermusch): initialize the context
- Status status = Status.DefaultSuccess;
+ Status status;
+ var context = HandlerUtils.NewContext(newRpc);
try
{
- var result = await handler(context, requestStream);
+ var result = await handler(requestStream, context);
+ status = context.Status;
try
{
await responseStream.WriteAsync(result);
@@ -199,7 +202,7 @@ namespace Grpc.Core.Internal
try
{
- await responseStream.WriteStatusAsync(status);
+ await responseStream.WriteStatusAsync(status, context.ResponseTrailers);
}
catch (OperationCanceledException)
{
@@ -222,23 +225,24 @@ namespace Grpc.Core.Internal
this.handler = handler;
}
- public async Task HandleCall(string methodName, CallSafeHandle call, GrpcEnvironment environment)
+ public async Task HandleCall(ServerRpcNew newRpc, GrpcEnvironment environment)
{
var asyncCall = new AsyncCallServer<TRequest, TResponse>(
method.ResponseMarshaller.Serializer,
method.RequestMarshaller.Deserializer,
environment);
- asyncCall.Initialize(call);
+ asyncCall.Initialize(newRpc.Call);
var finishedTask = asyncCall.ServerSideCallAsync();
var requestStream = new ServerRequestStream<TRequest, TResponse>(asyncCall);
var responseStream = new ServerResponseStream<TRequest, TResponse>(asyncCall);
- var context = new ServerCallContext(); // TODO(jtattermusch): initialize the context
- Status status = Status.DefaultSuccess;
+ Status status;
+ var context = HandlerUtils.NewContext(newRpc);
try
{
- await handler(context, requestStream, responseStream);
+ await handler(requestStream, responseStream, context);
+ status = context.Status;
}
catch (Exception e)
{
@@ -247,7 +251,7 @@ namespace Grpc.Core.Internal
}
try
{
- await responseStream.WriteStatusAsync(status);
+ await responseStream.WriteStatusAsync(status, context.ResponseTrailers);
}
catch (OperationCanceledException)
{
@@ -259,18 +263,19 @@ namespace Grpc.Core.Internal
internal class NoSuchMethodCallHandler : IServerCallHandler
{
- public async Task HandleCall(string methodName, CallSafeHandle call, GrpcEnvironment environment)
+ public static readonly NoSuchMethodCallHandler Instance = new NoSuchMethodCallHandler();
+
+ public async Task HandleCall(ServerRpcNew newRpc, GrpcEnvironment environment)
{
// We don't care about the payload type here.
var asyncCall = new AsyncCallServer<byte[], byte[]>(
(payload) => payload, (payload) => payload, environment);
- asyncCall.Initialize(call);
+ asyncCall.Initialize(newRpc.Call);
var finishedTask = asyncCall.ServerSideCallAsync();
- var requestStream = new ServerRequestStream<byte[], byte[]>(asyncCall);
var responseStream = new ServerResponseStream<byte[], byte[]>(asyncCall);
- await responseStream.WriteStatusAsync(new Status(StatusCode.Unimplemented, "No such method."));
+ await responseStream.WriteStatusAsync(new Status(StatusCode.Unimplemented, "No such method."), Metadata.Empty);
await finishedTask;
}
}
@@ -279,8 +284,22 @@ namespace Grpc.Core.Internal
{
public static Status StatusFromException(Exception e)
{
+ var rpcException = e as RpcException;
+ if (rpcException != null)
+ {
+ // use the status thrown by handler.
+ return rpcException.Status;
+ }
+
// TODO(jtattermusch): what is the right status code here?
return new Status(StatusCode.Unknown, "Exception was thrown by handler.");
}
+
+ public static ServerCallContext NewContext(ServerRpcNew newRpc)
+ {
+ return new ServerCallContext(
+ newRpc.Method, newRpc.Host, newRpc.Deadline.ToDateTime(),
+ newRpc.RequestMetadata, CancellationToken.None);
+ }
}
}
diff --git a/src/csharp/Grpc.Core/Internal/ServerResponseStream.cs b/src/csharp/Grpc.Core/Internal/ServerResponseStream.cs
index a2d77dd5b7..756dcee87f 100644
--- a/src/csharp/Grpc.Core/Internal/ServerResponseStream.cs
+++ b/src/csharp/Grpc.Core/Internal/ServerResponseStream.cs
@@ -56,10 +56,10 @@ namespace Grpc.Core.Internal
return taskSource.Task;
}
- public Task WriteStatusAsync(Status status)
+ public Task WriteStatusAsync(Status status, Metadata trailers)
{
var taskSource = new AsyncCompletionTaskSource<object>();
- call.StartSendStatusFromServer(status, taskSource.CompletionDelegate);
+ call.StartSendStatusFromServer(status, trailers, taskSource.CompletionDelegate);
return taskSource.Task;
}
}
diff --git a/src/csharp/Grpc.Core/Internal/Timespec.cs b/src/csharp/Grpc.Core/Internal/Timespec.cs
index de783f5a4b..da2819f14d 100644
--- a/src/csharp/Grpc.Core/Internal/Timespec.cs
+++ b/src/csharp/Grpc.Core/Internal/Timespec.cs
@@ -43,6 +43,8 @@ namespace Grpc.Core.Internal
const int NanosPerSecond = 1000 * 1000 * 1000;
const int NanosPerTick = 100;
+ static readonly DateTime UnixEpoch = new DateTime(1970, 1, 1, 0, 0, 0, 0, DateTimeKind.Utc);
+
[DllImport("grpc_csharp_ext.dll")]
static extern Timespec gprsharp_now();
@@ -52,6 +54,13 @@ namespace Grpc.Core.Internal
[DllImport("grpc_csharp_ext.dll")]
static extern int gprsharp_sizeof_timespec();
+ public Timespec(IntPtr tv_sec, int tv_nsec)
+ {
+ this.tv_sec = tv_sec;
+ this.tv_nsec = tv_nsec;
+ this.clock_type = GPRClockType.Realtime;
+ }
+
// NOTE: on linux 64bit sizeof(gpr_timespec) = 16, on windows 32bit sizeof(gpr_timespec) = 8
// so IntPtr seems to have the right size to work on both.
public System.IntPtr tv_sec;
@@ -76,6 +85,11 @@ namespace Grpc.Core.Internal
return gprsharp_now();
}
}
+
+ public DateTime ToDateTime()
+ {
+ return UnixEpoch.AddTicks(tv_sec.ToInt64() * (NanosPerSecond / NanosPerTick) + tv_nsec / NanosPerTick);
+ }
internal static int NativeSize
{
diff --git a/src/csharp/Grpc.Core/Metadata.cs b/src/csharp/Grpc.Core/Metadata.cs
index 4552d39d88..2f308cbb11 100644
--- a/src/csharp/Grpc.Core/Metadata.cs
+++ b/src/csharp/Grpc.Core/Metadata.cs
@@ -220,6 +220,11 @@ namespace Grpc.Core
return value;
}
}
+
+ public override string ToString()
+ {
+ return string.Format("[Entry: key={0}, value={1}]", Key, Value);
+ }
}
}
}
diff --git a/src/csharp/Grpc.Core/Server.cs b/src/csharp/Grpc.Core/Server.cs
index cbf77196cf..fd30735359 100644
--- a/src/csharp/Grpc.Core/Server.cs
+++ b/src/csharp/Grpc.Core/Server.cs
@@ -53,6 +53,7 @@ namespace Grpc.Core
public const int PickUnusedPort = 0;
readonly GrpcEnvironment environment;
+ readonly List<ChannelOption> options;
readonly ServerSafeHandle handle;
readonly object myLock = new object();
@@ -69,7 +70,8 @@ namespace Grpc.Core
public Server(IEnumerable<ChannelOption> options = null)
{
this.environment = GrpcEnvironment.GetInstance();
- using (var channelArgs = ChannelOptions.CreateChannelArgs(options))
+ this.options = options != null ? new List<ChannelOption>(options) : new List<ChannelOption>();
+ using (var channelArgs = ChannelOptions.CreateChannelArgs(this.options))
{
this.handle = ServerSafeHandle.NewServer(environment.CompletionQueue, channelArgs);
}
@@ -218,16 +220,16 @@ namespace Grpc.Core
/// <summary>
/// Selects corresponding handler for given call and handles the call.
/// </summary>
- private async Task InvokeCallHandler(CallSafeHandle call, string method)
+ private async Task HandleCallAsync(ServerRpcNew newRpc)
{
try
{
IServerCallHandler callHandler;
- if (!callHandlers.TryGetValue(method, out callHandler))
+ if (!callHandlers.TryGetValue(newRpc.Method, out callHandler))
{
- callHandler = new NoSuchMethodCallHandler();
+ callHandler = NoSuchMethodCallHandler.Instance;
}
- await callHandler.HandleCall(method, call, environment);
+ await callHandler.HandleCall(newRpc, environment);
}
catch (Exception e)
{
@@ -240,15 +242,15 @@ namespace Grpc.Core
/// </summary>
private void HandleNewServerRpc(bool success, BatchContextSafeHandle ctx)
{
- // TODO: handle error
-
- CallSafeHandle call = ctx.GetServerRpcNewCall();
- string method = ctx.GetServerRpcNewMethod();
-
- // after server shutdown, the callback returns with null call
- if (!call.IsInvalid)
+ if (success)
{
- Task.Run(async () => await InvokeCallHandler(call, method));
+ ServerRpcNew newRpc = ctx.GetServerRpcNew();
+
+ // after server shutdown, the callback returns with null call
+ if (!newRpc.Call.IsInvalid)
+ {
+ Task.Run(async () => await HandleCallAsync(newRpc));
+ }
}
AllowOneRpc();
diff --git a/src/csharp/Grpc.Core/ServerCallContext.cs b/src/csharp/Grpc.Core/ServerCallContext.cs
index bc9a499c51..17a2eefd07 100644
--- a/src/csharp/Grpc.Core/ServerCallContext.cs
+++ b/src/csharp/Grpc.Core/ServerCallContext.cs
@@ -33,6 +33,7 @@
using System;
using System.Runtime.CompilerServices;
+using System.Threading;
using System.Threading.Tasks;
namespace Grpc.Core
@@ -42,14 +43,93 @@ namespace Grpc.Core
/// </summary>
public sealed class ServerCallContext
{
- // TODO(jtattermusch): add cancellationToken
+ // TODO(jtattermusch): expose method to send initial metadata back to client
- // TODO(jtattermusch): add deadline info
+ private readonly string method;
+ private readonly string host;
+ private readonly DateTime deadline;
+ private readonly Metadata requestHeaders;
+ private readonly CancellationToken cancellationToken;
+ private readonly Metadata responseTrailers = new Metadata();
- // TODO(jtattermusch): expose initial metadata sent by client for reading
+ private Status status = Status.DefaultSuccess;
- // TODO(jtattermusch): expose method to send initial metadata back to client
+ public ServerCallContext(string method, string host, DateTime deadline, Metadata requestHeaders, CancellationToken cancellationToken)
+ {
+ this.method = method;
+ this.host = host;
+ this.deadline = deadline;
+ this.requestHeaders = requestHeaders;
+ this.cancellationToken = cancellationToken;
+ }
+
+ /// <summary> Name of method called in this RPC. </summary>
+ public string Method
+ {
+ get
+ {
+ return this.method;
+ }
+ }
+
+ /// <summary> Name of host called in this RPC. </summary>
+ public string Host
+ {
+ get
+ {
+ return this.host;
+ }
+ }
+
+ /// <summary> Deadline for this RPC. </summary>
+ public DateTime Deadline
+ {
+ get
+ {
+ return this.deadline;
+ }
+ }
+
+ /// <summary> Initial metadata sent by client. </summary>
+ public Metadata RequestHeaders
+ {
+ get
+ {
+ return this.requestHeaders;
+ }
+ }
+
+ // TODO(jtattermusch): support signalling cancellation.
+ /// <summary> Cancellation token signals when call is cancelled. </summary>
+ public CancellationToken CancellationToken
+ {
+ get
+ {
+ return this.cancellationToken;
+ }
+ }
+
+ /// <summary> Trailers to send back to client after RPC finishes.</summary>
+ public Metadata ResponseTrailers
+ {
+ get
+ {
+ return this.responseTrailers;
+ }
+ }
+
+ /// <summary> Status to send back to client after RPC finishes.</summary>
+ public Status Status
+ {
+ get
+ {
+ return this.status;
+ }
- // TODO(jtattermusch): allow setting status and trailing metadata to send after handler completes.
+ set
+ {
+ status = value;
+ }
+ }
}
}
diff --git a/src/csharp/Grpc.Core/ServerMethods.cs b/src/csharp/Grpc.Core/ServerMethods.cs
index 377b78eb30..d457770203 100644
--- a/src/csharp/Grpc.Core/ServerMethods.cs
+++ b/src/csharp/Grpc.Core/ServerMethods.cs
@@ -42,28 +42,28 @@ namespace Grpc.Core
/// <summary>
/// Server-side handler for unary call.
/// </summary>
- public delegate Task<TResponse> UnaryServerMethod<TRequest, TResponse>(ServerCallContext context, TRequest request)
+ public delegate Task<TResponse> UnaryServerMethod<TRequest, TResponse>(TRequest request, ServerCallContext context)
where TRequest : class
where TResponse : class;
/// <summary>
/// Server-side handler for client streaming call.
/// </summary>
- public delegate Task<TResponse> ClientStreamingServerMethod<TRequest, TResponse>(ServerCallContext context, IAsyncStreamReader<TRequest> requestStream)
+ public delegate Task<TResponse> ClientStreamingServerMethod<TRequest, TResponse>(IAsyncStreamReader<TRequest> requestStream, ServerCallContext context)
where TRequest : class
where TResponse : class;
/// <summary>
/// Server-side handler for server streaming call.
/// </summary>
- public delegate Task ServerStreamingServerMethod<TRequest, TResponse>(ServerCallContext context, TRequest request, IServerStreamWriter<TResponse> responseStream)
+ public delegate Task ServerStreamingServerMethod<TRequest, TResponse>(TRequest request, IServerStreamWriter<TResponse> responseStream, ServerCallContext context)
where TRequest : class
where TResponse : class;
/// <summary>
/// Server-side handler for bidi streaming call.
/// </summary>
- public delegate Task DuplexStreamingServerMethod<TRequest, TResponse>(ServerCallContext context, IAsyncStreamReader<TRequest> requestStream, IServerStreamWriter<TResponse> responseStream)
+ public delegate Task DuplexStreamingServerMethod<TRequest, TResponse>(IAsyncStreamReader<TRequest> requestStream, IServerStreamWriter<TResponse> responseStream, ServerCallContext context)
where TRequest : class
where TResponse : class;
}
diff --git a/src/csharp/Grpc.Core/Version.cs b/src/csharp/Grpc.Core/Version.cs
index f1db1f6157..b5cb652945 100644
--- a/src/csharp/Grpc.Core/Version.cs
+++ b/src/csharp/Grpc.Core/Version.cs
@@ -2,4 +2,4 @@ using System.Reflection;
using System.Runtime.CompilerServices;
// The current version of gRPC C#.
-[assembly: AssemblyVersion("0.6.0.*")]
+[assembly: AssemblyVersion(Grpc.Core.VersionInfo.CurrentVersion + ".*")]
diff --git a/src/csharp/Grpc.Core/VersionInfo.cs b/src/csharp/Grpc.Core/VersionInfo.cs
new file mode 100644
index 0000000000..656a3d47bb
--- /dev/null
+++ b/src/csharp/Grpc.Core/VersionInfo.cs
@@ -0,0 +1,13 @@
+using System.Reflection;
+using System.Runtime.CompilerServices;
+
+namespace Grpc.Core
+{
+ public static class VersionInfo
+ {
+ /// <summary>
+ /// Current version of gRPC
+ /// </summary>
+ public const string CurrentVersion = "0.6.0";
+ }
+}
diff --git a/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs b/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs
index e7c4b33120..7a957c5b6f 100644
--- a/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs
+++ b/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs
@@ -144,7 +144,7 @@ namespace math.Tests
n => Num.CreateBuilder().SetNum_(n).Build());
await call.RequestStream.WriteAll(numbers);
- var result = await call.Result;
+ var result = await call.ResponseAsync;
Assert.AreEqual(60, result.Num_);
}
}).Wait();
diff --git a/src/csharp/Grpc.Examples/MathExamples.cs b/src/csharp/Grpc.Examples/MathExamples.cs
index 7deb651689..06d81a4d83 100644
--- a/src/csharp/Grpc.Examples/MathExamples.cs
+++ b/src/csharp/Grpc.Examples/MathExamples.cs
@@ -46,8 +46,7 @@ namespace math
public static async Task DivAsyncExample(Math.IMathClient client)
{
- Task<DivReply> resultTask = client.DivAsync(new DivArgs.Builder { Dividend = 4, Divisor = 5 }.Build());
- DivReply result = await resultTask;
+ DivReply result = await client.DivAsync(new DivArgs.Builder { Dividend = 4, Divisor = 5 }.Build());
Console.WriteLine("DivAsync Result: " + result);
}
@@ -72,7 +71,7 @@ namespace math
using (var call = client.Sum())
{
await call.RequestStream.WriteAll(numbers);
- Console.WriteLine("Sum Result: " + await call.Result);
+ Console.WriteLine("Sum Result: " + await call.ResponseAsync);
}
}
@@ -104,7 +103,7 @@ namespace math
using (var sumCall = client.Sum())
{
await sumCall.RequestStream.WriteAll(numbers);
- sum = await sumCall.Result;
+ sum = await sumCall.ResponseAsync;
}
DivReply result = await client.DivAsync(new DivArgs.Builder { Dividend = sum.Num_, Divisor = numbers.Count }.Build());
diff --git a/src/csharp/Grpc.Examples/MathGrpc.cs b/src/csharp/Grpc.Examples/MathGrpc.cs
index 1805972ce3..ef787cf1d8 100644
--- a/src/csharp/Grpc.Examples/MathGrpc.cs
+++ b/src/csharp/Grpc.Examples/MathGrpc.cs
@@ -45,7 +45,7 @@ namespace math {
public interface IMathClient
{
global::math.DivReply Div(global::math.DivArgs request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
- Task<global::math.DivReply> DivAsync(global::math.DivArgs request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
+ AsyncUnaryCall<global::math.DivReply> DivAsync(global::math.DivArgs request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
AsyncDuplexStreamingCall<global::math.DivArgs, global::math.DivReply> DivMany(Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
AsyncServerStreamingCall<global::math.Num> Fib(global::math.FibArgs request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
AsyncClientStreamingCall<global::math.Num, global::math.Num> Sum(Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
@@ -54,10 +54,10 @@ namespace math {
// server-side interface
public interface IMath
{
- Task<global::math.DivReply> Div(ServerCallContext context, global::math.DivArgs request);
- Task DivMany(ServerCallContext context, IAsyncStreamReader<global::math.DivArgs> requestStream, IServerStreamWriter<global::math.DivReply> responseStream);
- Task Fib(ServerCallContext context, global::math.FibArgs request, IServerStreamWriter<global::math.Num> responseStream);
- Task<global::math.Num> Sum(ServerCallContext context, IAsyncStreamReader<global::math.Num> requestStream);
+ Task<global::math.DivReply> Div(global::math.DivArgs request, ServerCallContext context);
+ Task DivMany(IAsyncStreamReader<global::math.DivArgs> requestStream, IServerStreamWriter<global::math.DivReply> responseStream, ServerCallContext context);
+ Task Fib(global::math.FibArgs request, IServerStreamWriter<global::math.Num> responseStream, ServerCallContext context);
+ Task<global::math.Num> Sum(IAsyncStreamReader<global::math.Num> requestStream, ServerCallContext context);
}
// client stub
@@ -71,7 +71,7 @@ namespace math {
var call = CreateCall(__ServiceName, __Method_Div, headers);
return Calls.BlockingUnaryCall(call, request, cancellationToken);
}
- public Task<global::math.DivReply> DivAsync(global::math.DivArgs request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken))
+ public AsyncUnaryCall<global::math.DivReply> DivAsync(global::math.DivArgs request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken))
{
var call = CreateCall(__ServiceName, __Method_Div, headers);
return Calls.AsyncUnaryCall(call, request, cancellationToken);
diff --git a/src/csharp/Grpc.Examples/MathServiceImpl.cs b/src/csharp/Grpc.Examples/MathServiceImpl.cs
index e247ac9d73..3dd0f53a0d 100644
--- a/src/csharp/Grpc.Examples/MathServiceImpl.cs
+++ b/src/csharp/Grpc.Examples/MathServiceImpl.cs
@@ -45,12 +45,12 @@ namespace math
/// </summary>
public class MathServiceImpl : Math.IMath
{
- public Task<DivReply> Div(ServerCallContext context, DivArgs request)
+ public Task<DivReply> Div(DivArgs request, ServerCallContext context)
{
return Task.FromResult(DivInternal(request));
}
- public async Task Fib(ServerCallContext context, FibArgs request, IServerStreamWriter<Num> responseStream)
+ public async Task Fib(FibArgs request, IServerStreamWriter<Num> responseStream, ServerCallContext context)
{
if (request.Limit <= 0)
{
@@ -67,7 +67,7 @@ namespace math
}
}
- public async Task<Num> Sum(ServerCallContext context, IAsyncStreamReader<Num> requestStream)
+ public async Task<Num> Sum(IAsyncStreamReader<Num> requestStream, ServerCallContext context)
{
long sum = 0;
await requestStream.ForEach(async num =>
@@ -77,7 +77,7 @@ namespace math
return Num.CreateBuilder().SetNum_(sum).Build();
}
- public async Task DivMany(ServerCallContext context, IAsyncStreamReader<DivArgs> requestStream, IServerStreamWriter<DivReply> responseStream)
+ public async Task DivMany(IAsyncStreamReader<DivArgs> requestStream, IServerStreamWriter<DivReply> responseStream, ServerCallContext context)
{
await requestStream.ForEach(async divArgs =>
{
diff --git a/src/csharp/Grpc.HealthCheck.Tests/HealthClientServerTest.cs b/src/csharp/Grpc.HealthCheck.Tests/HealthClientServerTest.cs
index 73ff0e74b5..bc14a0a62f 100644
--- a/src/csharp/Grpc.HealthCheck.Tests/HealthClientServerTest.cs
+++ b/src/csharp/Grpc.HealthCheck.Tests/HealthClientServerTest.cs
@@ -87,9 +87,7 @@ namespace Grpc.HealthCheck.Tests
[Test]
public void ServiceDoesntExist()
{
- // TODO(jtattermusch): currently, this returns wrong status code, because we don't enable sending arbitrary status code from
- // server handlers yet.
- Assert.Throws(typeof(RpcException), () => client.Check(HealthCheckRequest.CreateBuilder().SetHost("").SetService("nonexistent.service").Build()));
+ Assert.Throws(Is.TypeOf(typeof(RpcException)).And.Property("Status").Property("StatusCode").EqualTo(StatusCode.NotFound), () => client.Check(HealthCheckRequest.CreateBuilder().SetHost("").SetService("nonexistent.service").Build()));
}
// TODO(jtattermusch): add test with timeout once timeouts are supported
diff --git a/src/csharp/Grpc.HealthCheck.Tests/HealthServiceImplTest.cs b/src/csharp/Grpc.HealthCheck.Tests/HealthServiceImplTest.cs
index 9b7c4f2140..7184415655 100644
--- a/src/csharp/Grpc.HealthCheck.Tests/HealthServiceImplTest.cs
+++ b/src/csharp/Grpc.HealthCheck.Tests/HealthServiceImplTest.cs
@@ -101,7 +101,7 @@ namespace Grpc.HealthCheck.Tests
private static HealthCheckResponse.Types.ServingStatus GetStatusHelper(HealthServiceImpl impl, string host, string service)
{
- return impl.Check(null, HealthCheckRequest.CreateBuilder().SetHost(host).SetService(service).Build()).Result.Status;
+ return impl.Check(HealthCheckRequest.CreateBuilder().SetHost(host).SetService(service).Build(), null).Result.Status;
}
}
}
diff --git a/src/csharp/Grpc.HealthCheck/HealthGrpc.cs b/src/csharp/Grpc.HealthCheck/HealthGrpc.cs
index 3aebdcb557..217127eca7 100644
--- a/src/csharp/Grpc.HealthCheck/HealthGrpc.cs
+++ b/src/csharp/Grpc.HealthCheck/HealthGrpc.cs
@@ -25,13 +25,13 @@ namespace Grpc.Health.V1Alpha {
public interface IHealthClient
{
global::Grpc.Health.V1Alpha.HealthCheckResponse Check(global::Grpc.Health.V1Alpha.HealthCheckRequest request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
- Task<global::Grpc.Health.V1Alpha.HealthCheckResponse> CheckAsync(global::Grpc.Health.V1Alpha.HealthCheckRequest request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
+ AsyncUnaryCall<global::Grpc.Health.V1Alpha.HealthCheckResponse> CheckAsync(global::Grpc.Health.V1Alpha.HealthCheckRequest request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
}
// server-side interface
public interface IHealth
{
- Task<global::Grpc.Health.V1Alpha.HealthCheckResponse> Check(ServerCallContext context, global::Grpc.Health.V1Alpha.HealthCheckRequest request);
+ Task<global::Grpc.Health.V1Alpha.HealthCheckResponse> Check(global::Grpc.Health.V1Alpha.HealthCheckRequest request, ServerCallContext context);
}
// client stub
@@ -45,7 +45,7 @@ namespace Grpc.Health.V1Alpha {
var call = CreateCall(__ServiceName, __Method_Check, headers);
return Calls.BlockingUnaryCall(call, request, cancellationToken);
}
- public Task<global::Grpc.Health.V1Alpha.HealthCheckResponse> CheckAsync(global::Grpc.Health.V1Alpha.HealthCheckRequest request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken))
+ public AsyncUnaryCall<global::Grpc.Health.V1Alpha.HealthCheckResponse> CheckAsync(global::Grpc.Health.V1Alpha.HealthCheckRequest request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken))
{
var call = CreateCall(__ServiceName, __Method_Check, headers);
return Calls.AsyncUnaryCall(call, request, cancellationToken);
diff --git a/src/csharp/Grpc.HealthCheck/HealthServiceImpl.cs b/src/csharp/Grpc.HealthCheck/HealthServiceImpl.cs
index db3a2a0942..3c3b9c35f1 100644
--- a/src/csharp/Grpc.HealthCheck/HealthServiceImpl.cs
+++ b/src/csharp/Grpc.HealthCheck/HealthServiceImpl.cs
@@ -95,7 +95,7 @@ namespace Grpc.HealthCheck
}
}
- public Task<HealthCheckResponse> Check(ServerCallContext context, HealthCheckRequest request)
+ public Task<HealthCheckResponse> Check(HealthCheckRequest request, ServerCallContext context)
{
lock (myLock)
{
diff --git a/src/csharp/Grpc.HealthCheck/Settings.StyleCop b/src/csharp/Grpc.HealthCheck/Settings.StyleCop
new file mode 100644
index 0000000000..2942add962
--- /dev/null
+++ b/src/csharp/Grpc.HealthCheck/Settings.StyleCop
@@ -0,0 +1,10 @@
+<StyleCopSettings Version="105">
+ <SourceFileList>
+ <SourceFile>Health.cs</SourceFile>
+ <Settings>
+ <GlobalSettings>
+ <BooleanProperty Name="RulesEnabledByDefault">False</BooleanProperty>
+ </GlobalSettings>
+ </Settings>
+ </SourceFileList>
+</StyleCopSettings>
diff --git a/src/csharp/Grpc.IntegrationTesting.Client/Grpc.IntegrationTesting.Client.csproj b/src/csharp/Grpc.IntegrationTesting.Client/Grpc.IntegrationTesting.Client.csproj
index 328acb5b47..dc1d0a44c0 100644
--- a/src/csharp/Grpc.IntegrationTesting.Client/Grpc.IntegrationTesting.Client.csproj
+++ b/src/csharp/Grpc.IntegrationTesting.Client/Grpc.IntegrationTesting.Client.csproj
@@ -3,8 +3,6 @@
<PropertyGroup>
<Configuration Condition=" '$(Configuration)' == '' ">Debug</Configuration>
<Platform Condition=" '$(Platform)' == '' ">x86</Platform>
- <ProductVersion>10.0.0</ProductVersion>
- <SchemaVersion>2.0</SchemaVersion>
<ProjectGuid>{3D166931-BA2D-416E-95A3-D36E8F6E90B9}</ProjectGuid>
<OutputType>Exe</OutputType>
<RootNamespace>Grpc.IntegrationTesting.Client</RootNamespace>
@@ -48,6 +46,10 @@
<Project>{C61154BA-DD4A-4838-8420-0162A28925E0}</Project>
<Name>Grpc.IntegrationTesting</Name>
</ProjectReference>
+ <ProjectReference Include="..\Grpc.Core\Grpc.Core.csproj">
+ <Project>{CCC4440E-49F7-4790-B0AF-FEABB0837AE7}</Project>
+ <Name>Grpc.Core</Name>
+ </ProjectReference>
</ItemGroup>
<ItemGroup>
<None Include="app.config" />
diff --git a/src/csharp/Grpc.IntegrationTesting.Server/Grpc.IntegrationTesting.Server.csproj b/src/csharp/Grpc.IntegrationTesting.Server/Grpc.IntegrationTesting.Server.csproj
index ae184c1dc7..f03c8f3ce3 100644
--- a/src/csharp/Grpc.IntegrationTesting.Server/Grpc.IntegrationTesting.Server.csproj
+++ b/src/csharp/Grpc.IntegrationTesting.Server/Grpc.IntegrationTesting.Server.csproj
@@ -3,8 +3,6 @@
<PropertyGroup>
<Configuration Condition=" '$(Configuration)' == '' ">Debug</Configuration>
<Platform Condition=" '$(Platform)' == '' ">x86</Platform>
- <ProductVersion>10.0.0</ProductVersion>
- <SchemaVersion>2.0</SchemaVersion>
<ProjectGuid>{A654F3B8-E859-4E6A-B30D-227527DBEF0D}</ProjectGuid>
<OutputType>Exe</OutputType>
<RootNamespace>Grpc.IntegrationTesting.Server</RootNamespace>
@@ -48,6 +46,10 @@
<Project>{C61154BA-DD4A-4838-8420-0162A28925E0}</Project>
<Name>Grpc.IntegrationTesting</Name>
</ProjectReference>
+ <ProjectReference Include="..\Grpc.Core\Grpc.Core.csproj">
+ <Project>{CCC4440E-49F7-4790-B0AF-FEABB0837AE7}</Project>
+ <Name>Grpc.Core</Name>
+ </ProjectReference>
</ItemGroup>
<ItemGroup>
<None Include="app.config" />
diff --git a/src/csharp/Grpc.IntegrationTesting/Grpc.IntegrationTesting.csproj b/src/csharp/Grpc.IntegrationTesting/Grpc.IntegrationTesting.csproj
index af4a75a034..d3c69ab9eb 100644
--- a/src/csharp/Grpc.IntegrationTesting/Grpc.IntegrationTesting.csproj
+++ b/src/csharp/Grpc.IntegrationTesting/Grpc.IntegrationTesting.csproj
@@ -32,21 +32,34 @@
<PlatformTarget>x86</PlatformTarget>
</PropertyGroup>
<ItemGroup>
- <Reference Include="Google.Apis.Auth.PlatformServices">
+ <Reference Include="Google.Apis.Auth, Version=1.9.1.12395, Culture=neutral, processorArchitecture=MSIL">
+ <SpecificVersion>False</SpecificVersion>
+ <HintPath>..\packages\Google.Apis.Auth.1.9.1\lib\net40\Google.Apis.Auth.dll</HintPath>
+ </Reference>
+ <Reference Include="Google.Apis.Auth.PlatformServices, Version=1.9.1.12399, Culture=neutral, processorArchitecture=MSIL">
+ <SpecificVersion>False</SpecificVersion>
<HintPath>..\packages\Google.Apis.Auth.1.9.1\lib\net40\Google.Apis.Auth.PlatformServices.dll</HintPath>
</Reference>
- <Reference Include="Google.Apis.Core">
+ <Reference Include="Google.Apis.Core, Version=1.9.1.12394, Culture=neutral, processorArchitecture=MSIL">
+ <SpecificVersion>False</SpecificVersion>
<HintPath>..\packages\Google.Apis.Core.1.9.1\lib\portable-net40+sl50+win+wpa81+wp80\Google.Apis.Core.dll</HintPath>
</Reference>
- <Reference Include="Microsoft.Threading.Tasks">
+ <Reference Include="Microsoft.Threading.Tasks, Version=1.0.12.0, Culture=neutral, PublicKeyToken=b03f5f7f11d50a3a, processorArchitecture=MSIL">
+ <SpecificVersion>False</SpecificVersion>
<HintPath>..\packages\Microsoft.Bcl.Async.1.0.168\lib\net40\Microsoft.Threading.Tasks.dll</HintPath>
</Reference>
- <Reference Include="Microsoft.Threading.Tasks.Extensions">
+ <Reference Include="Microsoft.Threading.Tasks.Extensions, Version=1.0.12.0, Culture=neutral, PublicKeyToken=b03f5f7f11d50a3a, processorArchitecture=MSIL">
+ <SpecificVersion>False</SpecificVersion>
<HintPath>..\packages\Microsoft.Bcl.Async.1.0.168\lib\net40\Microsoft.Threading.Tasks.Extensions.dll</HintPath>
</Reference>
- <Reference Include="Microsoft.Threading.Tasks.Extensions.Desktop">
+ <Reference Include="Microsoft.Threading.Tasks.Extensions.Desktop, Version=1.0.168.0, Culture=neutral, PublicKeyToken=b03f5f7f11d50a3a, processorArchitecture=MSIL">
+ <SpecificVersion>False</SpecificVersion>
<HintPath>..\packages\Microsoft.Bcl.Async.1.0.168\lib\net40\Microsoft.Threading.Tasks.Extensions.Desktop.dll</HintPath>
</Reference>
+ <Reference Include="Newtonsoft.Json, Version=6.0.0.0, Culture=neutral, PublicKeyToken=30ad4fe6b2a6aeed, processorArchitecture=MSIL">
+ <SpecificVersion>False</SpecificVersion>
+ <HintPath>..\packages\Newtonsoft.Json.6.0.6\lib\net45\Newtonsoft.Json.dll</HintPath>
+ </Reference>
<Reference Include="nunit.framework">
<HintPath>..\packages\NUnit.2.6.4\lib\nunit.framework.dll</HintPath>
</Reference>
@@ -63,16 +76,15 @@
</Reference>
<Reference Include="System.Net" />
<Reference Include="System.Net.Http" />
- <Reference Include="System.Net.Http.Extensions">
+ <Reference Include="System.Net.Http.Extensions, Version=2.2.28.0, Culture=neutral, PublicKeyToken=b03f5f7f11d50a3a, processorArchitecture=MSIL">
+ <SpecificVersion>False</SpecificVersion>
<HintPath>..\packages\Microsoft.Net.Http.2.2.28\lib\net45\System.Net.Http.Extensions.dll</HintPath>
</Reference>
- <Reference Include="System.Net.Http.Primitives">
+ <Reference Include="System.Net.Http.Primitives, Version=4.2.28.0, Culture=neutral, PublicKeyToken=b03f5f7f11d50a3a, processorArchitecture=MSIL">
+ <SpecificVersion>False</SpecificVersion>
<HintPath>..\packages\Microsoft.Net.Http.2.2.28\lib\net45\System.Net.Http.Primitives.dll</HintPath>
</Reference>
<Reference Include="System.Net.Http.WebRequest" />
- <Reference Include="Newtonsoft.Json">
- <HintPath>..\packages\Newtonsoft.Json.6.0.6\lib\net45\Newtonsoft.Json.dll</HintPath>
- </Reference>
</ItemGroup>
<ItemGroup>
<Compile Include="..\Grpc.Core\Version.cs">
diff --git a/src/csharp/Grpc.IntegrationTesting/InteropClient.cs b/src/csharp/Grpc.IntegrationTesting/InteropClient.cs
index 05e732dbd4..ce255f9423 100644
--- a/src/csharp/Grpc.IntegrationTesting/InteropClient.cs
+++ b/src/csharp/Grpc.IntegrationTesting/InteropClient.cs
@@ -135,7 +135,7 @@ namespace Grpc.IntegrationTesting
GrpcEnvironment.Shutdown();
}
- private void RunTestCase(string testCase, TestService.ITestServiceClient client)
+ private void RunTestCase(string testCase, TestService.TestServiceClient client)
{
switch (testCase)
{
@@ -163,6 +163,12 @@ namespace Grpc.IntegrationTesting
case "compute_engine_creds":
RunComputeEngineCreds(client);
break;
+ case "oauth2_auth_token":
+ RunOAuth2AuthToken(client);
+ break;
+ case "per_rpc_creds":
+ RunPerRpcCreds(client);
+ break;
case "cancel_after_begin":
RunCancelAfterBegin(client);
break;
@@ -213,7 +219,7 @@ namespace Grpc.IntegrationTesting
{
await call.RequestStream.WriteAll(bodySizes);
- var response = await call.Result;
+ var response = await call.ResponseAsync;
Assert.AreEqual(74922, response.AggregatedPayloadSize);
}
Console.WriteLine("Passed!");
@@ -355,6 +361,51 @@ namespace Grpc.IntegrationTesting
Console.WriteLine("Passed!");
}
+ public static void RunOAuth2AuthToken(TestService.TestServiceClient client)
+ {
+ Console.WriteLine("running oauth2_auth_token");
+ var credential = GoogleCredential.GetApplicationDefault().CreateScoped(new[] { AuthScope });
+ Assert.IsTrue(credential.RequestAccessTokenAsync(CancellationToken.None).Result);
+ string oauth2Token = credential.Token.AccessToken;
+
+ // Intercept calls with an OAuth2 token obtained out-of-band.
+ client.HeaderInterceptor = new MetadataInterceptorDelegate((metadata) =>
+ {
+ metadata.Add(new Metadata.Entry("Authorization", "Bearer " + oauth2Token));
+ });
+
+ var request = SimpleRequest.CreateBuilder()
+ .SetFillUsername(true)
+ .SetFillOauthScope(true)
+ .Build();
+
+ var response = client.UnaryCall(request);
+
+ Assert.AreEqual(AuthScopeResponse, response.OauthScope);
+ Assert.AreEqual(ServiceAccountUser, response.Username);
+ Console.WriteLine("Passed!");
+ }
+
+ public static void RunPerRpcCreds(TestService.TestServiceClient client)
+ {
+ Console.WriteLine("running per_rpc_creds");
+
+ var credential = GoogleCredential.GetApplicationDefault().CreateScoped(new[] { AuthScope });
+ Assert.IsTrue(credential.RequestAccessTokenAsync(CancellationToken.None).Result);
+ string oauth2Token = credential.Token.AccessToken;
+
+ var request = SimpleRequest.CreateBuilder()
+ .SetFillUsername(true)
+ .SetFillOauthScope(true)
+ .Build();
+
+ var response = client.UnaryCall(request, headers: new Metadata { new Metadata.Entry("Authorization", "Bearer " + oauth2Token) });
+
+ Assert.AreEqual(AuthScopeResponse, response.OauthScope);
+ Assert.AreEqual(ServiceAccountUser, response.Username);
+ Console.WriteLine("Passed!");
+ }
+
public static void RunCancelAfterBegin(TestService.ITestServiceClient client)
{
Task.Run(async () =>
@@ -370,7 +421,7 @@ namespace Grpc.IntegrationTesting
try
{
- var response = await call.Result;
+ var response = await call.ResponseAsync;
Assert.Fail();
}
catch (RpcException e)
diff --git a/src/csharp/Grpc.IntegrationTesting/TestGrpc.cs b/src/csharp/Grpc.IntegrationTesting/TestGrpc.cs
index 96d9b23717..de2fa07441 100644
--- a/src/csharp/Grpc.IntegrationTesting/TestGrpc.cs
+++ b/src/csharp/Grpc.IntegrationTesting/TestGrpc.cs
@@ -60,9 +60,9 @@ namespace grpc.testing {
public interface ITestServiceClient
{
global::grpc.testing.Empty EmptyCall(global::grpc.testing.Empty request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
- Task<global::grpc.testing.Empty> EmptyCallAsync(global::grpc.testing.Empty request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
+ AsyncUnaryCall<global::grpc.testing.Empty> EmptyCallAsync(global::grpc.testing.Empty request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
global::grpc.testing.SimpleResponse UnaryCall(global::grpc.testing.SimpleRequest request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
- Task<global::grpc.testing.SimpleResponse> UnaryCallAsync(global::grpc.testing.SimpleRequest request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
+ AsyncUnaryCall<global::grpc.testing.SimpleResponse> UnaryCallAsync(global::grpc.testing.SimpleRequest request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
AsyncServerStreamingCall<global::grpc.testing.StreamingOutputCallResponse> StreamingOutputCall(global::grpc.testing.StreamingOutputCallRequest request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
AsyncClientStreamingCall<global::grpc.testing.StreamingInputCallRequest, global::grpc.testing.StreamingInputCallResponse> StreamingInputCall(Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
AsyncDuplexStreamingCall<global::grpc.testing.StreamingOutputCallRequest, global::grpc.testing.StreamingOutputCallResponse> FullDuplexCall(Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken));
@@ -72,12 +72,12 @@ namespace grpc.testing {
// server-side interface
public interface ITestService
{
- Task<global::grpc.testing.Empty> EmptyCall(ServerCallContext context, global::grpc.testing.Empty request);
- Task<global::grpc.testing.SimpleResponse> UnaryCall(ServerCallContext context, global::grpc.testing.SimpleRequest request);
- Task StreamingOutputCall(ServerCallContext context, global::grpc.testing.StreamingOutputCallRequest request, IServerStreamWriter<global::grpc.testing.StreamingOutputCallResponse> responseStream);
- Task<global::grpc.testing.StreamingInputCallResponse> StreamingInputCall(ServerCallContext context, IAsyncStreamReader<global::grpc.testing.StreamingInputCallRequest> requestStream);
- Task FullDuplexCall(ServerCallContext context, IAsyncStreamReader<global::grpc.testing.StreamingOutputCallRequest> requestStream, IServerStreamWriter<global::grpc.testing.StreamingOutputCallResponse> responseStream);
- Task HalfDuplexCall(ServerCallContext context, IAsyncStreamReader<global::grpc.testing.StreamingOutputCallRequest> requestStream, IServerStreamWriter<global::grpc.testing.StreamingOutputCallResponse> responseStream);
+ Task<global::grpc.testing.Empty> EmptyCall(global::grpc.testing.Empty request, ServerCallContext context);
+ Task<global::grpc.testing.SimpleResponse> UnaryCall(global::grpc.testing.SimpleRequest request, ServerCallContext context);
+ Task StreamingOutputCall(global::grpc.testing.StreamingOutputCallRequest request, IServerStreamWriter<global::grpc.testing.StreamingOutputCallResponse> responseStream, ServerCallContext context);
+ Task<global::grpc.testing.StreamingInputCallResponse> StreamingInputCall(IAsyncStreamReader<global::grpc.testing.StreamingInputCallRequest> requestStream, ServerCallContext context);
+ Task FullDuplexCall(IAsyncStreamReader<global::grpc.testing.StreamingOutputCallRequest> requestStream, IServerStreamWriter<global::grpc.testing.StreamingOutputCallResponse> responseStream, ServerCallContext context);
+ Task HalfDuplexCall(IAsyncStreamReader<global::grpc.testing.StreamingOutputCallRequest> requestStream, IServerStreamWriter<global::grpc.testing.StreamingOutputCallResponse> responseStream, ServerCallContext context);
}
// client stub
@@ -91,7 +91,7 @@ namespace grpc.testing {
var call = CreateCall(__ServiceName, __Method_EmptyCall, headers);
return Calls.BlockingUnaryCall(call, request, cancellationToken);
}
- public Task<global::grpc.testing.Empty> EmptyCallAsync(global::grpc.testing.Empty request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken))
+ public AsyncUnaryCall<global::grpc.testing.Empty> EmptyCallAsync(global::grpc.testing.Empty request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken))
{
var call = CreateCall(__ServiceName, __Method_EmptyCall, headers);
return Calls.AsyncUnaryCall(call, request, cancellationToken);
@@ -101,7 +101,7 @@ namespace grpc.testing {
var call = CreateCall(__ServiceName, __Method_UnaryCall, headers);
return Calls.BlockingUnaryCall(call, request, cancellationToken);
}
- public Task<global::grpc.testing.SimpleResponse> UnaryCallAsync(global::grpc.testing.SimpleRequest request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken))
+ public AsyncUnaryCall<global::grpc.testing.SimpleResponse> UnaryCallAsync(global::grpc.testing.SimpleRequest request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken))
{
var call = CreateCall(__ServiceName, __Method_UnaryCall, headers);
return Calls.AsyncUnaryCall(call, request, cancellationToken);
diff --git a/src/csharp/Grpc.IntegrationTesting/TestServiceImpl.cs b/src/csharp/Grpc.IntegrationTesting/TestServiceImpl.cs
index 6bd997d1f4..ccf9fe6ced 100644
--- a/src/csharp/Grpc.IntegrationTesting/TestServiceImpl.cs
+++ b/src/csharp/Grpc.IntegrationTesting/TestServiceImpl.cs
@@ -46,19 +46,19 @@ namespace grpc.testing
/// </summary>
public class TestServiceImpl : TestService.ITestService
{
- public Task<Empty> EmptyCall(ServerCallContext context, Empty request)
+ public Task<Empty> EmptyCall(Empty request, ServerCallContext context)
{
return Task.FromResult(Empty.DefaultInstance);
}
- public Task<SimpleResponse> UnaryCall(ServerCallContext context, SimpleRequest request)
+ public Task<SimpleResponse> UnaryCall(SimpleRequest request, ServerCallContext context)
{
var response = SimpleResponse.CreateBuilder()
.SetPayload(CreateZerosPayload(request.ResponseSize)).Build();
return Task.FromResult(response);
}
- public async Task StreamingOutputCall(ServerCallContext context, StreamingOutputCallRequest request, IServerStreamWriter<StreamingOutputCallResponse> responseStream)
+ public async Task StreamingOutputCall(StreamingOutputCallRequest request, IServerStreamWriter<StreamingOutputCallResponse> responseStream, ServerCallContext context)
{
foreach (var responseParam in request.ResponseParametersList)
{
@@ -68,7 +68,7 @@ namespace grpc.testing
}
}
- public async Task<StreamingInputCallResponse> StreamingInputCall(ServerCallContext context, IAsyncStreamReader<StreamingInputCallRequest> requestStream)
+ public async Task<StreamingInputCallResponse> StreamingInputCall(IAsyncStreamReader<StreamingInputCallRequest> requestStream, ServerCallContext context)
{
int sum = 0;
await requestStream.ForEach(async request =>
@@ -78,7 +78,7 @@ namespace grpc.testing
return StreamingInputCallResponse.CreateBuilder().SetAggregatedPayloadSize(sum).Build();
}
- public async Task FullDuplexCall(ServerCallContext context, IAsyncStreamReader<StreamingOutputCallRequest> requestStream, IServerStreamWriter<StreamingOutputCallResponse> responseStream)
+ public async Task FullDuplexCall(IAsyncStreamReader<StreamingOutputCallRequest> requestStream, IServerStreamWriter<StreamingOutputCallResponse> responseStream, ServerCallContext context)
{
await requestStream.ForEach(async request =>
{
@@ -91,7 +91,7 @@ namespace grpc.testing
});
}
- public async Task HalfDuplexCall(ServerCallContext context, IAsyncStreamReader<StreamingOutputCallRequest> requestStream, IServerStreamWriter<StreamingOutputCallResponse> responseStream)
+ public async Task HalfDuplexCall(IAsyncStreamReader<StreamingOutputCallRequest> requestStream, IServerStreamWriter<StreamingOutputCallResponse> responseStream, ServerCallContext context)
{
throw new NotImplementedException();
}
diff --git a/src/csharp/ext/grpc_csharp_ext.c b/src/csharp/ext/grpc_csharp_ext.c
index 7dd1959a5f..682521446f 100644
--- a/src/csharp/ext/grpc_csharp_ext.c
+++ b/src/csharp/ext/grpc_csharp_ext.c
@@ -167,6 +167,29 @@ grpcsharp_metadata_array_add(grpc_metadata_array *array, const char *key,
array->count++;
}
+GPR_EXPORT gpr_intptr GPR_CALLTYPE
+grpcsharp_metadata_array_count(grpc_metadata_array *array) {
+ return (gpr_intptr) array->count;
+}
+
+GPR_EXPORT const char *GPR_CALLTYPE
+grpcsharp_metadata_array_get_key(grpc_metadata_array *array, size_t index) {
+ GPR_ASSERT(index < array->count);
+ return array->metadata[index].key;
+}
+
+GPR_EXPORT const char *GPR_CALLTYPE
+grpcsharp_metadata_array_get_value(grpc_metadata_array *array, size_t index) {
+ GPR_ASSERT(index < array->count);
+ return array->metadata[index].value;
+}
+
+GPR_EXPORT gpr_intptr GPR_CALLTYPE
+grpcsharp_metadata_array_get_value_length(grpc_metadata_array *array, size_t index) {
+ GPR_ASSERT(index < array->count);
+ return (gpr_intptr) array->metadata[index].value_length;
+}
+
/* Move contents of metadata array */
void grpcsharp_metadata_array_move(grpc_metadata_array *dest,
grpc_metadata_array *src) {
@@ -218,6 +241,12 @@ GPR_EXPORT void GPR_CALLTYPE grpcsharp_batch_context_destroy(grpcsharp_batch_con
gpr_free(ctx);
}
+GPR_EXPORT const grpc_metadata_array *GPR_CALLTYPE
+grpcsharp_batch_context_recv_initial_metadata(
+ const grpcsharp_batch_context *ctx) {
+ return &(ctx->recv_initial_metadata);
+}
+
GPR_EXPORT gpr_intptr GPR_CALLTYPE grpcsharp_batch_context_recv_message_length(
const grpcsharp_batch_context *ctx) {
if (!ctx->recv_message) {
@@ -260,6 +289,12 @@ grpcsharp_batch_context_recv_status_on_client_details(
return ctx->recv_status_on_client.status_details;
}
+GPR_EXPORT const grpc_metadata_array *GPR_CALLTYPE
+grpcsharp_batch_context_recv_status_on_client_trailing_metadata(
+ const grpcsharp_batch_context *ctx) {
+ return &(ctx->recv_status_on_client.trailing_metadata);
+}
+
GPR_EXPORT grpc_call *GPR_CALLTYPE grpcsharp_batch_context_server_rpc_new_call(
const grpcsharp_batch_context *ctx) {
return ctx->server_rpc_new.call;
@@ -271,6 +306,24 @@ grpcsharp_batch_context_server_rpc_new_method(
return ctx->server_rpc_new.call_details.method;
}
+GPR_EXPORT const char *GPR_CALLTYPE
+grpcsharp_batch_context_server_rpc_new_host(
+ const grpcsharp_batch_context *ctx) {
+ return ctx->server_rpc_new.call_details.host;
+}
+
+GPR_EXPORT gpr_timespec GPR_CALLTYPE
+grpcsharp_batch_context_server_rpc_new_deadline(
+ const grpcsharp_batch_context *ctx) {
+ return ctx->server_rpc_new.call_details.deadline;
+}
+
+GPR_EXPORT const grpc_metadata_array *GPR_CALLTYPE
+grpcsharp_batch_context_server_rpc_new_request_metadata(
+ const grpcsharp_batch_context *ctx) {
+ return &(ctx->server_rpc_new.request_metadata);
+}
+
GPR_EXPORT gpr_int32 GPR_CALLTYPE
grpcsharp_batch_context_recv_close_on_server_cancelled(
const grpcsharp_batch_context *ctx) {
@@ -589,15 +642,20 @@ GPR_EXPORT grpc_call_error GPR_CALLTYPE
grpcsharp_call_send_status_from_server(grpc_call *call,
grpcsharp_batch_context *ctx,
grpc_status_code status_code,
- const char *status_details) {
+ const char *status_details,
+ grpc_metadata_array *trailing_metadata) {
/* TODO: don't use magic number */
grpc_op ops[1];
ops[0].op = GRPC_OP_SEND_STATUS_FROM_SERVER;
ops[0].data.send_status_from_server.status = status_code;
ops[0].data.send_status_from_server.status_details =
gpr_strdup(status_details);
- ops[0].data.send_status_from_server.trailing_metadata = NULL;
- ops[0].data.send_status_from_server.trailing_metadata_count = 0;
+ grpcsharp_metadata_array_move(&(ctx->send_status_from_server.trailing_metadata),
+ trailing_metadata);
+ ops[0].data.send_status_from_server.trailing_metadata_count =
+ ctx->send_status_from_server.trailing_metadata.count;
+ ops[0].data.send_status_from_server.trailing_metadata =
+ ctx->send_status_from_server.trailing_metadata.metadata;
ops[0].flags = 0;
return grpc_call_start_batch(call, ops, sizeof(ops) / sizeof(ops[0]), ctx);
diff --git a/src/node/README.md b/src/node/README.md
index 2f4c49096d..78781dab14 100644
--- a/src/node/README.md
+++ b/src/node/README.md
@@ -54,10 +54,10 @@ function loadObject(reflectionObject)
Returns the same structure that `load` returns, but takes a reflection object from `ProtoBuf.js` instead of a file name.
```javascript
-function buildServer(serviceArray)
+function Server([serverOpions])
```
-Takes an array of service objects and returns a constructor for a server that handles requests to all of those services.
+Constructs a server to which service/implementation pairs can be added.
```javascript
diff --git a/src/node/examples/math_server.js b/src/node/examples/math_server.js
index 0a86e7eaff..b1f8a6323f 100644
--- a/src/node/examples/math_server.js
+++ b/src/node/examples/math_server.js
@@ -36,8 +36,6 @@
var grpc = require('..');
var math = grpc.load(__dirname + '/math.proto').math;
-var Server = grpc.buildServer([math.Math.service]);
-
/**
* Server function for division. Provides the /Math/DivMany and /Math/Div
* functions (Div is just DivMany with only one stream element). For each
@@ -108,19 +106,17 @@ function mathDivMany(stream) {
stream.end();
});
}
-
-var server = new Server({
- 'math.Math' : {
- div: mathDiv,
- fib: mathFib,
- sum: mathSum,
- divMany: mathDivMany
- }
+var server = new grpc.Server();
+server.addProtoService(math.Math.service, {
+ div: mathDiv,
+ fib: mathFib,
+ sum: mathSum,
+ divMany: mathDivMany
});
if (require.main === module) {
server.bind('0.0.0.0:50051');
- server.listen();
+ server.start();
}
/**
diff --git a/src/node/examples/route_guide_server.js b/src/node/examples/route_guide_server.js
index c777eab7bc..70044a322c 100644
--- a/src/node/examples/route_guide_server.js
+++ b/src/node/examples/route_guide_server.js
@@ -40,8 +40,6 @@ var _ = require('lodash');
var grpc = require('..');
var examples = grpc.load(__dirname + '/route_guide.proto').examples;
-var Server = grpc.buildServer([examples.RouteGuide.service]);
-
var COORD_FACTOR = 1e7;
/**
@@ -228,14 +226,14 @@ function routeChat(call) {
* @return {Server} The new server object
*/
function getServer() {
- return new Server({
- 'examples.RouteGuide' : {
- getFeature: getFeature,
- listFeatures: listFeatures,
- recordRoute: recordRoute,
- routeChat: routeChat
- }
+ var server = new grpc.Server();
+ server.addProtoService(examples.RouteGuide.service, {
+ getFeature: getFeature,
+ listFeatures: listFeatures,
+ recordRoute: recordRoute,
+ routeChat: routeChat
});
+ return server;
}
if (require.main === module) {
diff --git a/src/node/examples/stock_server.js b/src/node/examples/stock_server.js
index caaf9f99ba..f2eb6ad4ab 100644
--- a/src/node/examples/stock_server.js
+++ b/src/node/examples/stock_server.js
@@ -37,8 +37,6 @@ var _ = require('lodash');
var grpc = require('..');
var examples = grpc.load(__dirname + '/stock.proto').examples;
-var StockServer = grpc.buildServer([examples.Stock.service]);
-
function getLastTradePrice(call, callback) {
callback(null, {symbol: call.request.symbol, price: 88});
}
@@ -73,13 +71,12 @@ function getLastTradePriceMultiple(call) {
});
}
-var stockServer = new StockServer({
- 'examples.Stock' : {
- getLastTradePrice: getLastTradePrice,
- getLastTradePriceMultiple: getLastTradePriceMultiple,
- watchFutureTrades: watchFutureTrades,
- getHighestTradePrice: getHighestTradePrice
- }
+var stockServer = new grpc.Server();
+stockServer.addProtoService(examples.Stock.service, {
+ getLastTradePrice: getLastTradePrice,
+ getLastTradePriceMultiple: getLastTradePriceMultiple,
+ watchFutureTrades: watchFutureTrades,
+ getHighestTradePrice: getHighestTradePrice
});
if (require.main === module) {
diff --git a/src/node/index.js b/src/node/index.js
index b6a4e2d0ee..d81e780443 100644
--- a/src/node/index.js
+++ b/src/node/index.js
@@ -133,9 +133,9 @@ exports.loadObject = loadObject;
exports.load = load;
/**
- * See docs for server.makeServerConstructor
+ * See docs for Server
*/
-exports.buildServer = server.makeProtobufServerConstructor;
+exports.Server = server.Server;
/**
* Status name to code number mapping
@@ -159,5 +159,3 @@ exports.ServerCredentials = grpc.ServerCredentials;
exports.getGoogleAuthDelegate = getGoogleAuthDelegate;
exports.makeGenericClientConstructor = client.makeClientConstructor;
-
-exports.makeGenericServerConstructor = server.makeServerConstructor;
diff --git a/src/node/interop/interop_client.js b/src/node/interop/interop_client.js
index b61b0b63c0..e810e68e45 100644
--- a/src/node/interop/interop_client.js
+++ b/src/node/interop/interop_client.js
@@ -318,6 +318,51 @@ function authTest(expected_user, scope, client, done) {
});
}
+function oauth2Test(expected_user, scope, per_rpc, client, done) {
+ (new GoogleAuth()).getApplicationDefault(function(err, credential) {
+ assert.ifError(err);
+ var arg = {
+ fill_username: true,
+ fill_oauth_scope: true
+ };
+ credential = credential.createScoped(scope);
+ credential.getAccessToken(function(err, token) {
+ assert.ifError(err);
+ var updateMetadata = function(authURI, metadata, callback) {
+ metadata = _.clone(metadata);
+ if (metadata.Authorization) {
+ metadata.Authorization = _.clone(metadata.Authorization);
+ } else {
+ metadata.Authorization = [];
+ }
+ metadata.Authorization.push('Bearer ' + token);
+ callback(null, metadata);
+ };
+ var makeTestCall = function(error, client_metadata) {
+ assert.ifError(error);
+ var call = client.unaryCall(arg, function(err, resp) {
+ assert.ifError(err);
+ assert.strictEqual(resp.username, expected_user);
+ assert.strictEqual(resp.oauth_scope, AUTH_SCOPE_RESPONSE);
+ });
+ call.on('status', function(status) {
+ assert.strictEqual(status.code, grpc.status.OK);
+ if (done) {
+ done();
+ }
+ });
+ };
+ if (per_rpc) {
+ updateMetadata('', {}, makeTestCall);
+ } else {
+ client.updateMetadata = updateMetadata;
+ makeTestCall(null, {});
+ }
+
+ });
+ });
+}
+
/**
* Map from test case names to test functions
*/
@@ -333,7 +378,9 @@ var test_cases = {
timeout_on_sleeping_server: timeoutOnSleepingServer,
compute_engine_creds: _.partial(authTest, COMPUTE_ENGINE_USER, null),
service_account_creds: _.partial(authTest, AUTH_USER, AUTH_SCOPE),
- jwt_token_creds: _.partial(authTest, AUTH_USER, null)
+ jwt_token_creds: _.partial(authTest, AUTH_USER, null),
+ oauth2_auth_token: _.partial(oauth2Test, AUTH_USER, AUTH_SCOPE, false),
+ per_rpc_creds: _.partial(oauth2Test, AUTH_USER, AUTH_SCOPE, true)
};
/**
diff --git a/src/node/interop/interop_server.js b/src/node/interop/interop_server.js
index 0baa78a094..505c6bb537 100644
--- a/src/node/interop/interop_server.js
+++ b/src/node/interop/interop_server.js
@@ -38,7 +38,6 @@ var path = require('path');
var _ = require('lodash');
var grpc = require('..');
var testProto = grpc.load(__dirname + '/test.proto').grpc.testing;
-var Server = grpc.buildServer([testProto.TestService.service]);
/**
* Create a buffer filled with size zeroes
@@ -173,16 +172,15 @@ function getServer(port, tls) {
key_data,
pem_data);
}
- var server = new Server({
- 'grpc.testing.TestService' : {
- emptyCall: handleEmpty,
- unaryCall: handleUnary,
- streamingOutputCall: handleStreamingOutput,
- streamingInputCall: handleStreamingInput,
- fullDuplexCall: handleFullDuplex,
- halfDuplexCall: handleHalfDuplex
- }
- }, null, options);
+ var server = new grpc.Server(options);
+ server.addProtoService(testProto.TestService.service, {
+ emptyCall: handleEmpty,
+ unaryCall: handleUnary,
+ streamingOutputCall: handleStreamingOutput,
+ streamingInputCall: handleStreamingInput,
+ fullDuplexCall: handleFullDuplex,
+ halfDuplexCall: handleHalfDuplex
+ });
var port_num = server.bind('0.0.0.0:' + port, server_creds);
return {server: server, port: port_num};
}
diff --git a/src/node/src/client.js b/src/node/src/client.js
index b7bad949d4..da6327b432 100644
--- a/src/node/src/client.js
+++ b/src/node/src/client.js
@@ -47,6 +47,7 @@ var Readable = stream.Readable;
var Writable = stream.Writable;
var Duplex = stream.Duplex;
var util = require('util');
+var version = require('../package.json').version;
util.inherits(ClientWritableStream, Writable);
@@ -517,9 +518,12 @@ function makeClientConstructor(methods, serviceName) {
callback(null, metadata);
};
}
-
- this.server_address = address.replace(/\/$/, '');
+ if (!options) {
+ options = {};
+ }
+ options['grpc.primary_user_agent'] = 'grpc-node/' + version;
this.channel = new grpc.Channel(address, options);
+ this.server_address = address.replace(/\/$/, '');
this.auth_uri = this.server_address + '/' + serviceName;
this.updateMetadata = updateMetadata;
}
diff --git a/src/node/src/server.js b/src/node/src/server.js
index 00be400e61..0a3a0031bd 100644
--- a/src/node/src/server.js
+++ b/src/node/src/server.js
@@ -72,6 +72,9 @@ function handleError(call, error) {
status.metadata = error.metadata;
}
var error_batch = {};
+ if (!call.metadataSent) {
+ error_batch[grpc.opType.SEND_INITIAL_METADATA] = {};
+ }
error_batch[grpc.opType.SEND_STATUS_FROM_SERVER] = status;
call.startBatch(error_batch, function(){});
}
@@ -115,6 +118,10 @@ function sendUnaryResponse(call, value, serialize, metadata) {
if (metadata) {
status.metadata = metadata;
}
+ if (!call.metadataSent) {
+ end_batch[grpc.opType.SEND_INITIAL_METADATA] = {};
+ call.metadataSent = true;
+ }
end_batch[grpc.opType.SEND_MESSAGE] = serialize(value);
end_batch[grpc.opType.SEND_STATUS_FROM_SERVER] = status;
call.startBatch(end_batch, function (){});
@@ -136,6 +143,10 @@ function setUpWritable(stream, serialize) {
stream.serialize = common.wrapIgnoreNull(serialize);
function sendStatus() {
var batch = {};
+ if (!stream.call.metadataSent) {
+ stream.call.metadataSent = true;
+ batch[grpc.opType.SEND_INITIAL_METADATA] = {};
+ }
batch[grpc.opType.SEND_STATUS_FROM_SERVER] = stream.status;
stream.call.startBatch(batch, function(){});
}
@@ -239,6 +250,10 @@ function ServerWritableStream(call, serialize) {
function _write(chunk, encoding, callback) {
/* jshint validthis: true */
var batch = {};
+ if (!this.call.metadataSent) {
+ batch[grpc.opType.SEND_INITIAL_METADATA] = {};
+ this.call.metadataSent = true;
+ }
batch[grpc.opType.SEND_MESSAGE] = this.serialize(chunk);
this.call.startBatch(batch, function(err, value) {
if (err) {
@@ -251,6 +266,23 @@ function _write(chunk, encoding, callback) {
ServerWritableStream.prototype._write = _write;
+function sendMetadata(responseMetadata) {
+ /* jshint validthis: true */
+ if (!this.call.metadataSent) {
+ this.call.metadataSent = true;
+ var batch = [];
+ batch[grpc.opType.SEND_INITIAL_METADATA] = responseMetadata;
+ this.call.startBatch(batch, function(err) {
+ if (err) {
+ this.emit('error', err);
+ return;
+ }
+ });
+ }
+}
+
+ServerWritableStream.prototype.sendMetadata = sendMetadata;
+
util.inherits(ServerReadableStream, Readable);
/**
@@ -339,6 +371,7 @@ function ServerDuplexStream(call, serialize, deserialize) {
ServerDuplexStream.prototype._read = _read;
ServerDuplexStream.prototype._write = _write;
+ServerDuplexStream.prototype.sendMetadata = sendMetadata;
/**
* Fully handle a unary call
@@ -348,12 +381,20 @@ ServerDuplexStream.prototype._write = _write;
*/
function handleUnary(call, handler, metadata) {
var emitter = new EventEmitter();
+ emitter.sendMetadata = function(responseMetadata) {
+ if (!call.metadataSent) {
+ call.metadataSent = true;
+ var batch = {};
+ batch[grpc.opType.SEND_INITIAL_METADATA] = responseMetadata;
+ call.startBatch(batch, function() {});
+ }
+ };
emitter.on('error', function(error) {
handleError(call, error);
});
+ emitter.metadata = metadata;
waitForCancel(call, emitter);
var batch = {};
- batch[grpc.opType.SEND_INITIAL_METADATA] = metadata;
batch[grpc.opType.RECV_MESSAGE] = true;
call.startBatch(batch, function(err, result) {
if (err) {
@@ -392,8 +433,8 @@ function handleUnary(call, handler, metadata) {
function handleServerStreaming(call, handler, metadata) {
var stream = new ServerWritableStream(call, handler.serialize);
waitForCancel(call, stream);
+ stream.metadata = metadata;
var batch = {};
- batch[grpc.opType.SEND_INITIAL_METADATA] = metadata;
batch[grpc.opType.RECV_MESSAGE] = true;
call.startBatch(batch, function(err, result) {
if (err) {
@@ -419,13 +460,19 @@ function handleServerStreaming(call, handler, metadata) {
*/
function handleClientStreaming(call, handler, metadata) {
var stream = new ServerReadableStream(call, handler.deserialize);
+ stream.sendMetadata = function(responseMetadata) {
+ if (!call.metadataSent) {
+ call.metadataSent = true;
+ var batch = {};
+ batch[grpc.opType.SEND_INITIAL_METADATA] = responseMetadata;
+ call.startBatch(batch, function() {});
+ }
+ };
stream.on('error', function(error) {
handleError(call, error);
});
waitForCancel(call, stream);
- var metadata_batch = {};
- metadata_batch[grpc.opType.SEND_INITIAL_METADATA] = metadata;
- call.startBatch(metadata_batch, function() {});
+ stream.metadata = metadata;
handler.func(stream, function(err, value, trailer) {
stream.terminate();
if (err) {
@@ -449,9 +496,7 @@ function handleBidiStreaming(call, handler, metadata) {
var stream = new ServerDuplexStream(call, handler.serialize,
handler.deserialize);
waitForCancel(call, stream);
- var metadata_batch = {};
- metadata_batch[grpc.opType.SEND_INITIAL_METADATA] = metadata;
- call.startBatch(metadata_batch, function() {});
+ stream.metadata = metadata;
handler.func(stream);
}
@@ -466,29 +511,28 @@ var streamHandlers = {
* Constructs a server object that stores request handlers and delegates
* incoming requests to those handlers
* @constructor
- * @param {function(string, Object<string, Array<Buffer>>):
- Object<string, Array<Buffer|string>>=} getMetadata Callback that gets
- * metatada for a given method
* @param {Object=} options Options that should be passed to the internal server
* implementation
*/
-function Server(getMetadata, options) {
+function Server(options) {
this.handlers = {};
var handlers = this.handlers;
var server = new grpc.Server(options);
this._server = server;
+ this.started = false;
/**
* Start the server and begin handling requests
* @this Server
*/
- this.listen = function() {
+ this.start = function() {
+ if (this.started) {
+ throw new Error('Server is already running');
+ }
+ this.started = true;
console.log('Server starting');
_.each(handlers, function(handler, handler_name) {
console.log('Serving', handler_name);
});
- if (this.started) {
- throw 'Server is already running';
- }
server.start();
/**
* Handles the SERVER_RPC_NEW event. If there is a handler associated with
@@ -523,11 +567,7 @@ function Server(getMetadata, options) {
call.startBatch(batch, function() {});
return;
}
- var response_metadata = {};
- if (getMetadata) {
- response_metadata = getMetadata(method, metadata);
- }
- streamHandlers[handler.type](call, handler, response_metadata);
+ streamHandlers[handler.type](call, handler, metadata);
}
server.requestCall(handleNewCall);
};
@@ -565,6 +605,47 @@ Server.prototype.register = function(name, handler, serialize, deserialize,
return true;
};
+Server.prototype.addService = function(service, implementation) {
+ if (this.started) {
+ throw new Error('Can\'t add a service to a started server.');
+ }
+ var self = this;
+ _.each(service, function(attrs, name) {
+ var method_type;
+ if (attrs.requestStream) {
+ if (attrs.responseStream) {
+ method_type = 'bidi';
+ } else {
+ method_type = 'client_stream';
+ }
+ } else {
+ if (attrs.responseStream) {
+ method_type = 'server_stream';
+ } else {
+ method_type = 'unary';
+ }
+ }
+ if (implementation[name] === undefined) {
+ throw new Error('Method handler for ' + attrs.path +
+ ' not provided.');
+ }
+ var serialize = attrs.responseSerialize;
+ var deserialize = attrs.requestDeserialize;
+ var register_success = self.register(attrs.path,
+ _.bind(implementation[name],
+ implementation),
+ serialize, deserialize, method_type);
+ if (!register_success) {
+ throw new Error('Method handler for ' + attrs.path +
+ ' already provided.');
+ }
+ });
+};
+
+Server.prototype.addProtoService = function(service, implementation) {
+ this.addService(common.getProtobufServiceAttrs(service), implementation);
+};
+
/**
* Binds the server to the given port, with SSL enabled if creds is given
* @param {string} port The port that the server should bind on, in the format
@@ -573,6 +654,9 @@ Server.prototype.register = function(name, handler, serialize, deserialize,
* nothing for an insecure port
*/
Server.prototype.bind = function(port, creds) {
+ if (this.started) {
+ throw new Error('Can\'t bind an already running server to an address');
+ }
if (creds) {
return this._server.addSecureHttp2Port(port, creds);
} else {
@@ -581,131 +665,6 @@ Server.prototype.bind = function(port, creds) {
};
/**
- * Create a constructor for servers with services defined by service_attr_map.
- * That is an object that maps (namespaced) service names to objects that in
- * turn map method names to objects with the following keys:
- * path: The path on the server for accessing the method. For example, for
- * protocol buffers, we use "/service_name/method_name"
- * requestStream: bool indicating whether the client sends a stream
- * resonseStream: bool indicating whether the server sends a stream
- * requestDeserialize: function to deserialize request objects
- * responseSerialize: function to serialize response objects
- * @param {Object} service_attr_map An object mapping service names to method
- * attribute map objects
- * @return {function(Object, function, Object=)} New server constructor
- */
-function makeServerConstructor(service_attr_map) {
- /**
- * Create a server with the given handlers for all of the methods.
- * @constructor
- * @param {Object} service_handlers Map from service names to map from method
- * names to handlers
- * @param {function(string, Object<string, Array<Buffer>>):
- Object<string, Array<Buffer|string>>=} getMetadata Callback that
- * gets metatada for a given method
- * @param {Object=} options Options to pass to the underlying server
- */
- function SurfaceServer(service_handlers, getMetadata, options) {
- var server = new Server(getMetadata, options);
- this.inner_server = server;
- _.each(service_attr_map, function(service_attrs, service_name) {
- if (service_handlers[service_name] === undefined) {
- throw new Error('Handlers for service ' +
- service_name + ' not provided.');
- }
- _.each(service_attrs, function(attrs, name) {
- var method_type;
- if (attrs.requestStream) {
- if (attrs.responseStream) {
- method_type = 'bidi';
- } else {
- method_type = 'client_stream';
- }
- } else {
- if (attrs.responseStream) {
- method_type = 'server_stream';
- } else {
- method_type = 'unary';
- }
- }
- if (service_handlers[service_name][name] === undefined) {
- throw new Error('Method handler for ' + attrs.path +
- ' not provided.');
- }
- var serialize = attrs.responseSerialize;
- var deserialize = attrs.requestDeserialize;
- server.register(attrs.path, _.bind(service_handlers[service_name][name],
- service_handlers[service_name]),
- serialize, deserialize, method_type);
- });
- }, this);
- }
-
- /**
- * Binds the server to the given port, with SSL enabled if creds is supplied
- * @param {string} port The port that the server should bind on, in the format
- * "address:port"
- * @param {boolean=} creds Credentials to use for SSL
- * @return {SurfaceServer} this
- */
- SurfaceServer.prototype.bind = function(port, creds) {
- return this.inner_server.bind(port, creds);
- };
-
- /**
- * Starts the server listening on any bound ports
- * @return {SurfaceServer} this
- */
- SurfaceServer.prototype.listen = function() {
- this.inner_server.listen();
- return this;
- };
-
- /**
- * Shuts the server down; tells it to stop listening for new requests and to
- * kill old requests.
- */
- SurfaceServer.prototype.shutdown = function() {
- this.inner_server.shutdown();
- };
-
- return SurfaceServer;
-}
-
-/**
- * Create a constructor for servers that serve the given services.
- * @param {Array<ProtoBuf.Reflect.Service>} services The services that the
- * servers will serve
- * @return {function(Object, function, Object=)} New server constructor
- */
-function makeProtobufServerConstructor(services) {
- var qual_names = [];
- var service_attr_map = {};
- _.each(services, function(service) {
- var service_name = common.fullyQualifiedName(service);
- _.each(service.children, function(method) {
- var name = common.fullyQualifiedName(method);
- if (_.indexOf(qual_names, name) !== -1) {
- throw new Error('Method ' + name + ' exposed by more than one service');
- }
- qual_names.push(name);
- });
- var method_attrs = common.getProtobufServiceAttrs(service);
- if (!service_attr_map.hasOwnProperty(service_name)) {
- service_attr_map[service_name] = {};
- }
- service_attr_map[service_name] = _.extend(service_attr_map[service_name],
- method_attrs);
- });
- return makeServerConstructor(service_attr_map);
-}
-
-/**
- * See documentation for makeServerConstructor
- */
-exports.makeServerConstructor = makeServerConstructor;
-
-/**
- * See documentation for makeProtobufServerConstructor
+ * See documentation for Server
*/
-exports.makeProtobufServerConstructor = makeProtobufServerConstructor;
+exports.Server = Server;
diff --git a/src/node/test/health_test.js b/src/node/test/health_test.js
index 4d1a5082e0..bb700cc46c 100644
--- a/src/node/test/health_test.js
+++ b/src/node/test/health_test.js
@@ -49,14 +49,13 @@ describe('Health Checking', function() {
'grpc.test.TestService': 'SERVING'
}
};
- var HealthServer = grpc.buildServer([health.service]);
- var healthServer = new HealthServer({
- 'grpc.health.v1alpha.Health': new health.Implementation(statusMap)
- });
+ var healthServer = new grpc.Server();
+ healthServer.addProtoService(health.service,
+ new health.Implementation(statusMap));
var healthClient;
before(function() {
var port_num = healthServer.bind('0.0.0.0:0');
- healthServer.listen();
+ healthServer.start();
healthClient = new health.Client('localhost:' + port_num);
});
after(function() {
diff --git a/src/node/test/interop_sanity_test.js b/src/node/test/interop_sanity_test.js
index fcd8eb6403..0a5eb29c0c 100644
--- a/src/node/test/interop_sanity_test.js
+++ b/src/node/test/interop_sanity_test.js
@@ -46,7 +46,7 @@ describe('Interop tests', function() {
before(function(done) {
var server_obj = interop_server.getServer(0, true);
server = server_obj.server;
- server.listen();
+ server.start();
port = 'localhost:' + server_obj.port;
done();
});
diff --git a/src/node/test/math_client_test.js b/src/node/test/math_client_test.js
index 3461922e66..f2751857ff 100644
--- a/src/node/test/math_client_test.js
+++ b/src/node/test/math_client_test.js
@@ -52,7 +52,7 @@ var server = require('../examples/math_server.js');
describe('Math client', function() {
before(function(done) {
var port_num = server.bind('0.0.0.0:0');
- server.listen();
+ server.start();
math_client = new math.Math('localhost:' + port_num);
done();
});
diff --git a/src/node/test/surface_test.js b/src/node/test/surface_test.js
index 125957277f..3cb68f8cd8 100644
--- a/src/node/test/surface_test.js
+++ b/src/node/test/surface_test.js
@@ -69,34 +69,45 @@ describe('File loader', function() {
});
});
});
-describe('Surface server constructor', function() {
- it('Should fail with conflicting method names', function() {
- assert.throws(function() {
- grpc.buildServer([mathService, mathService]);
- });
+describe('Server.prototype.addProtoService', function() {
+ var server;
+ var dummyImpls = {
+ 'div': function() {},
+ 'divMany': function() {},
+ 'fib': function() {},
+ 'sum': function() {}
+ };
+ beforeEach(function() {
+ server = new grpc.Server();
+ });
+ afterEach(function() {
+ server.shutdown();
});
it('Should succeed with a single service', function() {
assert.doesNotThrow(function() {
- grpc.buildServer([mathService]);
+ server.addProtoService(mathService, dummyImpls);
+ });
+ });
+ it('Should fail with conflicting method names', function() {
+ server.addProtoService(mathService, dummyImpls);
+ assert.throws(function() {
+ server.addProtoService(mathService, dummyImpls);
});
});
it('Should fail with missing handlers', function() {
- var Server = grpc.buildServer([mathService]);
assert.throws(function() {
- new Server({
- 'math.Math': {
- 'div': function() {},
- 'divMany': function() {},
- 'fib': function() {}
- }
+ server.addProtoService(mathService, {
+ 'div': function() {},
+ 'divMany': function() {},
+ 'fib': function() {}
});
}, /math.Math.Sum/);
});
- it('Should fail with no handlers for the service', function() {
- var Server = grpc.buildServer([mathService]);
+ it('Should fail if the server has been started', function() {
+ server.start();
assert.throws(function() {
- new Server({});
- }, /math.Math/);
+ server.addProtoService(mathService, dummyImpls);
+ });
});
});
describe('Echo service', function() {
@@ -105,18 +116,16 @@ describe('Echo service', function() {
before(function() {
var test_proto = ProtoBuf.loadProtoFile(__dirname + '/echo_service.proto');
var echo_service = test_proto.lookup('EchoService');
- var Server = grpc.buildServer([echo_service]);
- server = new Server({
- 'EchoService': {
- echo: function(call, callback) {
- callback(null, call.request);
- }
+ server = new grpc.Server();
+ server.addProtoService(echo_service, {
+ echo: function(call, callback) {
+ callback(null, call.request);
}
});
var port = server.bind('localhost:0');
var Client = surface_client.makeProtobufClientConstructor(echo_service);
client = new Client('localhost:' + port);
- server.listen();
+ server.start();
});
after(function() {
server.shutdown();
@@ -151,18 +160,14 @@ describe('Generic client and server', function() {
var client;
var server;
before(function() {
- var Server = grpc.makeGenericServerConstructor({
- string: string_service_attrs
- });
- server = new Server({
- string: {
- capitalize: function(call, callback) {
- callback(null, _.capitalize(call.request));
- }
+ server = new grpc.Server();
+ server.addService(string_service_attrs, {
+ capitalize: function(call, callback) {
+ callback(null, _.capitalize(call.request));
}
});
var port = server.bind('localhost:0');
- server.listen();
+ server.start();
var Client = grpc.makeGenericClientConstructor(string_service_attrs);
client = new Client('localhost:' + port);
});
@@ -178,6 +183,92 @@ describe('Generic client and server', function() {
});
});
});
+describe('Echo metadata', function() {
+ var client;
+ var server;
+ before(function() {
+ var test_proto = ProtoBuf.loadProtoFile(__dirname + '/test_service.proto');
+ var test_service = test_proto.lookup('TestService');
+ server = new grpc.Server();
+ server.addProtoService(test_service, {
+ unary: function(call, cb) {
+ call.sendMetadata(call.metadata);
+ cb(null, {});
+ },
+ clientStream: function(stream, cb){
+ stream.on('data', function(data) {});
+ stream.on('end', function() {
+ stream.sendMetadata(stream.metadata);
+ cb(null, {});
+ });
+ },
+ serverStream: function(stream) {
+ stream.sendMetadata(stream.metadata);
+ stream.end();
+ },
+ bidiStream: function(stream) {
+ stream.on('data', function(data) {});
+ stream.on('end', function() {
+ stream.sendMetadata(stream.metadata);
+ stream.end();
+ });
+ }
+ });
+ var port = server.bind('localhost:0');
+ var Client = surface_client.makeProtobufClientConstructor(test_service);
+ client = new Client('localhost:' + port);
+ server.start();
+ });
+ after(function() {
+ server.shutdown();
+ });
+ it('with unary call', function(done) {
+ var call = client.unary({}, function(err, data) {
+ assert.ifError(err);
+ }, {key: ['value']});
+ call.on('metadata', function(metadata) {
+ assert.deepEqual(metadata.key, ['value']);
+ done();
+ });
+ });
+ it('with client stream call', function(done) {
+ var call = client.clientStream(function(err, data) {
+ assert.ifError(err);
+ }, {key: ['value']});
+ call.on('metadata', function(metadata) {
+ assert.deepEqual(metadata.key, ['value']);
+ done();
+ });
+ call.end();
+ });
+ it('with server stream call', function(done) {
+ var call = client.serverStream({}, {key: ['value']});
+ call.on('data', function() {});
+ call.on('metadata', function(metadata) {
+ assert.deepEqual(metadata.key, ['value']);
+ done();
+ });
+ });
+ it('with bidi stream call', function(done) {
+ var call = client.bidiStream({key: ['value']});
+ call.on('data', function() {});
+ call.on('metadata', function(metadata) {
+ assert.deepEqual(metadata.key, ['value']);
+ done();
+ });
+ call.end();
+ });
+ it('shows the correct user-agent string', function(done) {
+ var version = require('../package.json').version;
+ var call = client.unary({}, function(err, data) {
+ assert.ifError(err);
+ }, {key: ['value']});
+ call.on('metadata', function(metadata) {
+ assert(_.startsWith(metadata['user-agent'], 'grpc-node/' + version));
+ done();
+ });
+ });
+});
describe('Other conditions', function() {
var client;
var server;
@@ -185,72 +276,70 @@ describe('Other conditions', function() {
before(function() {
var test_proto = ProtoBuf.loadProtoFile(__dirname + '/test_service.proto');
var test_service = test_proto.lookup('TestService');
- var Server = grpc.buildServer([test_service]);
- server = new Server({
- TestService: {
- unary: function(call, cb) {
- var req = call.request;
- if (req.error) {
- cb(new Error('Requested error'), null, {metadata: ['yes']});
+ server = new grpc.Server();
+ server.addProtoService(test_service, {
+ unary: function(call, cb) {
+ var req = call.request;
+ if (req.error) {
+ cb(new Error('Requested error'), null, {trailer_present: ['yes']});
+ } else {
+ cb(null, {count: 1}, {trailer_present: ['yes']});
+ }
+ },
+ clientStream: function(stream, cb){
+ var count = 0;
+ var errored;
+ stream.on('data', function(data) {
+ if (data.error) {
+ errored = true;
+ cb(new Error('Requested error'), null, {trailer_present: ['yes']});
} else {
- cb(null, {count: 1}, {metadata: ['yes']});
+ count += 1;
}
- },
- clientStream: function(stream, cb){
- var count = 0;
- var errored;
- stream.on('data', function(data) {
- if (data.error) {
- errored = true;
- cb(new Error('Requested error'), null, {metadata: ['yes']});
- } else {
- count += 1;
- }
- });
- stream.on('end', function() {
- if (!errored) {
- cb(null, {count: count}, {metadata: ['yes']});
- }
- });
- },
- serverStream: function(stream) {
- var req = stream.request;
- if (req.error) {
+ });
+ stream.on('end', function() {
+ if (!errored) {
+ cb(null, {count: count}, {trailer_present: ['yes']});
+ }
+ });
+ },
+ serverStream: function(stream) {
+ var req = stream.request;
+ if (req.error) {
+ var err = new Error('Requested error');
+ err.metadata = {trailer_present: ['yes']};
+ stream.emit('error', err);
+ } else {
+ for (var i = 0; i < 5; i++) {
+ stream.write({count: i});
+ }
+ stream.end({trailer_present: ['yes']});
+ }
+ },
+ bidiStream: function(stream) {
+ var count = 0;
+ stream.on('data', function(data) {
+ if (data.error) {
var err = new Error('Requested error');
- err.metadata = {metadata: ['yes']};
+ err.metadata = {
+ trailer_present: ['yes'],
+ count: ['' + count]
+ };
stream.emit('error', err);
} else {
- for (var i = 0; i < 5; i++) {
- stream.write({count: i});
- }
- stream.end({metadata: ['yes']});
+ stream.write({count: count});
+ count += 1;
}
- },
- bidiStream: function(stream) {
- var count = 0;
- stream.on('data', function(data) {
- if (data.error) {
- var err = new Error('Requested error');
- err.metadata = {
- metadata: ['yes'],
- count: ['' + count]
- };
- stream.emit('error', err);
- } else {
- stream.write({count: count});
- count += 1;
- }
- });
- stream.on('end', function() {
- stream.end({metadata: ['yes']});
- });
- }
+ });
+ stream.on('end', function() {
+ stream.end({trailer_present: ['yes']});
+ });
}
});
port = server.bind('localhost:0');
var Client = surface_client.makeProtobufClientConstructor(test_service);
client = new Client('localhost:' + port);
- server.listen();
+ server.start();
});
after(function() {
server.shutdown();
@@ -340,7 +429,7 @@ describe('Other conditions', function() {
assert.ifError(err);
});
call.on('status', function(status) {
- assert.deepEqual(status.metadata.metadata, ['yes']);
+ assert.deepEqual(status.metadata.trailer_present, ['yes']);
done();
});
});
@@ -349,7 +438,7 @@ describe('Other conditions', function() {
assert(err);
});
call.on('status', function(status) {
- assert.deepEqual(status.metadata.metadata, ['yes']);
+ assert.deepEqual(status.metadata.trailer_present, ['yes']);
done();
});
});
@@ -361,7 +450,7 @@ describe('Other conditions', function() {
call.write({error: false});
call.end();
call.on('status', function(status) {
- assert.deepEqual(status.metadata.metadata, ['yes']);
+ assert.deepEqual(status.metadata.trailer_present, ['yes']);
done();
});
});
@@ -373,7 +462,7 @@ describe('Other conditions', function() {
call.write({error: true});
call.end();
call.on('status', function(status) {
- assert.deepEqual(status.metadata.metadata, ['yes']);
+ assert.deepEqual(status.metadata.trailer_present, ['yes']);
done();
});
});
@@ -382,7 +471,7 @@ describe('Other conditions', function() {
call.on('data', function(){});
call.on('status', function(status) {
assert.strictEqual(status.code, grpc.status.OK);
- assert.deepEqual(status.metadata.metadata, ['yes']);
+ assert.deepEqual(status.metadata.trailer_present, ['yes']);
done();
});
});
@@ -390,7 +479,7 @@ describe('Other conditions', function() {
var call = client.serverStream({error: true});
call.on('data', function(){});
call.on('error', function(error) {
- assert.deepEqual(error.metadata.metadata, ['yes']);
+ assert.deepEqual(error.metadata.trailer_present, ['yes']);
done();
});
});
@@ -402,7 +491,7 @@ describe('Other conditions', function() {
call.on('data', function(){});
call.on('status', function(status) {
assert.strictEqual(status.code, grpc.status.OK);
- assert.deepEqual(status.metadata.metadata, ['yes']);
+ assert.deepEqual(status.metadata.trailer_present, ['yes']);
done();
});
});
@@ -413,7 +502,7 @@ describe('Other conditions', function() {
call.end();
call.on('data', function(){});
call.on('error', function(error) {
- assert.deepEqual(error.metadata.metadata, ['yes']);
+ assert.deepEqual(error.metadata.trailer_present, ['yes']);
done();
});
});
@@ -465,18 +554,17 @@ describe('Cancelling surface client', function() {
var client;
var server;
before(function() {
- var Server = grpc.buildServer([mathService]);
- server = new Server({
- 'math.Math': {
- 'div': function(stream) {},
- 'divMany': function(stream) {},
- 'fib': function(stream) {},
- 'sum': function(stream) {}
- }
+ server = new grpc.Server();
+ server.addProtoService(mathService, {
+ 'div': function(stream) {},
+ 'divMany': function(stream) {},
+ 'fib': function(stream) {},
+ 'sum': function(stream) {}
});
var port = server.bind('localhost:0');
var Client = surface_client.makeProtobufClientConstructor(mathService);
client = new Client('localhost:' + port);
+ server.start();
});
after(function() {
server.shutdown();
diff --git a/src/objective-c/GRPCClient/GRPCCall.m b/src/objective-c/GRPCClient/GRPCCall.m
index 53e5abe177..9435bf2b35 100644
--- a/src/objective-c/GRPCClient/GRPCCall.m
+++ b/src/objective-c/GRPCClient/GRPCCall.m
@@ -35,10 +35,10 @@
#include <grpc/grpc.h>
#include <grpc/support/time.h>
+#import <RxLibrary/GRXConcurrentWriteable.h>
#import "private/GRPCChannel.h"
#import "private/GRPCCompletionQueue.h"
-#import "private/GRPCDelegateWrapper.h"
#import "private/GRPCWrappedCall.h"
#import "private/NSData+GRPC.h"
#import "private/NSDictionary+GRPC.h"
@@ -78,9 +78,13 @@ NSString * const kGRPCStatusMetadataKey = @"io.grpc.StatusMetadataKey";
// do. Particularly, in the face of errors, there's no ordering guarantee at
// all. This wrapper over our actual writeable ensures thread-safety and
// correct ordering.
- GRPCDelegateWrapper *_responseWriteable;
+ GRXConcurrentWriteable *_responseWriteable;
GRXWriter *_requestWriter;
+ // To create a retain cycle when a call is started, up until it finishes. See
+ // |startWithWriteable:| and |finishWithError:|.
+ GRPCCall *_self;
+
NSMutableDictionary *_requestMetadata;
NSMutableDictionary *_responseMetadata;
}
@@ -143,8 +147,13 @@ NSString * const kGRPCStatusMetadataKey = @"io.grpc.StatusMetadataKey";
#pragma mark Finish
- (void)finishWithError:(NSError *)errorOrNil {
+ // If the call isn't retained anywhere else, it can be deallocated now.
+ _self = nil;
+
+ // If there were still request messages coming, stop them.
_requestWriter.state = GRXWriterStateFinished;
_requestWriter = nil;
+
if (errorOrNil) {
[_responseWriteable cancelWithError:errorOrNil];
} else {
@@ -191,7 +200,7 @@ NSString * const kGRPCStatusMetadataKey = @"io.grpc.StatusMetadataKey";
return;
}
__weak GRPCCall *weakSelf = self;
- __weak GRPCDelegateWrapper *weakWriteable = _responseWriteable;
+ __weak GRXConcurrentWriteable *weakWriteable = _responseWriteable;
dispatch_async(_callQueue, ^{
[weakSelf startReadWithHandler:^(grpc_byte_buffer *message) {
@@ -216,7 +225,7 @@ NSString * const kGRPCStatusMetadataKey = @"io.grpc.StatusMetadataKey";
[weakSelf cancelCall];
return;
}
- [weakWriteable enqueueMessage:data completionHandler:^{
+ [weakWriteable enqueueValue:data completionHandler:^{
[weakSelf startNextRead];
}];
}];
@@ -276,6 +285,7 @@ NSString * const kGRPCStatusMetadataKey = @"io.grpc.StatusMetadataKey";
}
- (void)writesFinishedWithError:(NSError *)errorOrNil {
+ _requestWriter = nil;
if (errorOrNil) {
[self cancel];
} else {
@@ -335,12 +345,14 @@ NSString * const kGRPCStatusMetadataKey = @"io.grpc.StatusMetadataKey";
#pragma mark GRXWriter implementation
- (void)startWithWriteable:(id<GRXWriteable>)writeable {
- // The following produces a retain cycle self:_responseWriteable:self, which is only
- // broken when writesFinishedWithError: is sent to the wrapped writeable.
- // Care is taken not to retain self strongly in any of the blocks used in
- // the implementation of GRPCCall, so that the life of the instance is
- // determined by this retain cycle.
- _responseWriteable = [[GRPCDelegateWrapper alloc] initWithWriteable:writeable writer:self];
+ // Create a retain cycle so that this instance lives until the RPC finishes (or is cancelled).
+ // This makes RPCs in which the call isn't externally retained possible (as long as it is started
+ // before being autoreleased).
+ // Care is taken not to retain self strongly in any of the blocks used in this implementation, so
+ // that the life of the instance is determined by this retain cycle.
+ _self = self;
+
+ _responseWriteable = [[GRXConcurrentWriteable alloc] initWithWriteable:writeable];
[self sendHeaders:_requestMetadata];
[self invokeCall];
}
diff --git a/src/objective-c/GRPCClient/private/GRPCCompletionQueue.m b/src/objective-c/GRPCClient/private/GRPCCompletionQueue.m
index 40aade4f9a..12535c9616 100644
--- a/src/objective-c/GRPCClient/private/GRPCCompletionQueue.m
+++ b/src/objective-c/GRPCClient/private/GRPCCompletionQueue.m
@@ -65,7 +65,8 @@
dispatch_async(gDefaultConcurrentQueue, ^{
while (YES) {
// The following call blocks until an event is available.
- grpc_event event = grpc_completion_queue_next(unmanagedQueue, gpr_inf_future);
+ grpc_event event = grpc_completion_queue_next(unmanagedQueue,
+ gpr_inf_future(GPR_CLOCK_REALTIME));
GRPCQueueCompletionHandler handler;
switch (event.type) {
case GRPC_OP_COMPLETE:
diff --git a/src/objective-c/GRPCClient/private/GRPCWrappedCall.m b/src/objective-c/GRPCClient/private/GRPCWrappedCall.m
index 45f10f5d63..1db63df77f 100644
--- a/src/objective-c/GRPCClient/private/GRPCWrappedCall.m
+++ b/src/objective-c/GRPCClient/private/GRPCWrappedCall.m
@@ -246,8 +246,11 @@
if (!_queue) {
return nil;
}
- _call = grpc_channel_create_call(channel.unmanagedChannel, _queue.unmanagedQueue,
- path.UTF8String, host.UTF8String, gpr_inf_future);
+ _call = grpc_channel_create_call(channel.unmanagedChannel,
+ _queue.unmanagedQueue,
+ path.UTF8String,
+ host.UTF8String,
+ gpr_inf_future(GPR_CLOCK_REALTIME));
if (_call == NULL) {
return nil;
}
diff --git a/src/objective-c/GRPCClient/private/GRPCDelegateWrapper.h b/src/objective-c/RxLibrary/GRXConcurrentWriteable.h
index 9a30a2f966..1080001905 100644
--- a/src/objective-c/GRPCClient/private/GRPCDelegateWrapper.h
+++ b/src/objective-c/RxLibrary/GRXConcurrentWriteable.h
@@ -33,49 +33,39 @@
#import <Foundation/Foundation.h>
-#import <RxLibrary/GRXWriter.h>
+#import "GRXWriter.h"
+#import "GRXWriteable.h"
-@protocol GRXWriteable;
-
-// This is a thread-safe wrapper over a GRXWriteable instance. It lets one
-// enqueue calls to a GRXWriteable instance for the main thread, guaranteeing
-// that writesFinishedWithError: is the last message sent to it (no matter what
-// messages are sent to the wrapper, in what order, nor from which thread). It
-// also guarantees that, if cancelWithError: is called from the main thread
-// (e.g. by the app cancelling the writes), no further messages are sent to the
-// writeable except writesFinishedWithError:.
+// This is a thread-safe wrapper over a GRXWriteable instance. It lets one enqueue calls to a
+// GRXWriteable instance for the main thread, guaranteeing that writesFinishedWithError: is the last
+// message sent to it (no matter what messages are sent to the wrapper, in what order, nor from
+// which thread). It also guarantees that, if cancelWithError: is called from the main thread (e.g.
+// by the app cancelling the writes), no further messages are sent to the writeable except
+// writesFinishedWithError:.
//
-// TODO(jcanizales): Let the user specify another queue for the writeable
-// callbacks.
-// TODO(jcanizales): Rename to GRXWriteableWrapper and move to the Rx library.
-@interface GRPCDelegateWrapper : NSObject
+// TODO(jcanizales): Let the user specify another queue for the writeable callbacks.
+@interface GRXConcurrentWriteable : NSObject
// The GRXWriteable passed is the wrapped writeable.
-// Both the GRXWriter instance and the GRXWriteable instance are retained until
-// writesFinishedWithError: is sent to the writeable, and released after that.
-// This is used to create a retain cycle that keeps both objects alive until the
-// writing is explicitly finished.
-- (instancetype)initWithWriteable:(id<GRXWriteable>)writeable writer:(GRXWriter *)writer
- NS_DESIGNATED_INITIALIZER;
+// The GRXWriteable instance is retained until writesFinishedWithError: is sent to it, and released
+// after that.
+- (instancetype)initWithWriteable:(id<GRXWriteable>)writeable NS_DESIGNATED_INITIALIZER;
// Enqueues writeValue: to be sent to the writeable in the main thread.
// The passed handler is invoked from the main thread after writeValue: returns.
-- (void)enqueueMessage:(NSData *)message completionHandler:(void (^)())handler;
+- (void)enqueueValue:(id)value completionHandler:(void (^)())handler;
-// Enqueues writesFinishedWithError:nil to be sent to the writeable in the main
-// thread. After that message is sent to the writeable, all other methods of
-// this object are effectively noops.
+// Enqueues writesFinishedWithError:nil to be sent to the writeable in the main thread. After that
+// message is sent to the writeable, all other methods of this object are effectively noops.
- (void)enqueueSuccessfulCompletion;
-// If the writeable has not yet received a writesFinishedWithError: message, this
-// will enqueue one to be sent to it in the main thread, and cancel all other
-// pending messages to the writeable enqueued by this object (both past and
-// future).
+// If the writeable has not yet received a writesFinishedWithError: message, this will enqueue one
+// to be sent to it in the main thread, and cancel all other pending messages to the writeable
+// enqueued by this object (both past and future).
// The error argument cannot be nil.
- (void)cancelWithError:(NSError *)error;
-// Cancels all pending messages to the writeable enqueued by this object (both
-// past and future). Because the writeable won't receive writesFinishedWithError:,
-// this also releases the writeable and the writer.
+// Cancels all pending messages to the writeable enqueued by this object (both past and future).
+// Because the writeable won't receive writesFinishedWithError:, this also releases the writeable.
- (void)cancelSilently;
@end
diff --git a/src/objective-c/GRPCClient/private/GRPCDelegateWrapper.m b/src/objective-c/RxLibrary/GRXConcurrentWriteable.m
index 294cfb7e23..08bd079aea 100644
--- a/src/objective-c/GRPCClient/private/GRPCDelegateWrapper.m
+++ b/src/objective-c/RxLibrary/GRXConcurrentWriteable.m
@@ -31,45 +31,42 @@
*
*/
-#import "GRPCDelegateWrapper.h"
+#import "GRXConcurrentWriteable.h"
#import <RxLibrary/GRXWriteable.h>
-@interface GRPCDelegateWrapper ()
-// These are atomic so that cancellation can nillify them from any thread.
+@interface GRXConcurrentWriteable ()
+// This is atomic so that cancellation can nillify it from any thread.
@property(atomic, strong) id<GRXWriteable> writeable;
-@property(atomic, strong) GRXWriter *writer;
@end
-@implementation GRPCDelegateWrapper {
+@implementation GRXConcurrentWriteable {
dispatch_queue_t _writeableQueue;
// This ensures that writesFinishedWithError: is only sent once to the writeable.
dispatch_once_t _alreadyFinished;
}
- (instancetype)init {
- return [self initWithWriteable:nil writer:nil];
+ return [self initWithWriteable:nil];
}
// Designated initializer
-- (instancetype)initWithWriteable:(id<GRXWriteable>)writeable writer:(GRXWriter *)writer {
+- (instancetype)initWithWriteable:(id<GRXWriteable>)writeable {
if (self = [super init]) {
_writeableQueue = dispatch_get_main_queue();
_writeable = writeable;
- _writer = writer;
}
return self;
}
-- (void)enqueueMessage:(NSData *)message completionHandler:(void (^)())handler {
+- (void)enqueueValue:(id)value completionHandler:(void (^)())handler {
dispatch_async(_writeableQueue, ^{
- // We're racing a possible cancellation performed by another thread. To turn
- // all already-enqueued messages into noops, cancellation nillifies the
- // writeable property. If we get it before it's nil, we won
- // the race.
+ // We're racing a possible cancellation performed by another thread. To turn all already-
+ // enqueued messages into noops, cancellation nillifies the writeable property. If we get it
+ // before it's nil, we won the race.
id<GRXWriteable> writeable = self.writeable;
if (writeable) {
- [writeable writeValue:message];
+ [writeable writeValue:value];
handler();
}
});
@@ -78,13 +75,11 @@
- (void)enqueueSuccessfulCompletion {
dispatch_async(_writeableQueue, ^{
dispatch_once(&_alreadyFinished, ^{
- // Cancellation is now impossible. None of the other three blocks can run
- // concurrently with this one.
+ // Cancellation is now impossible. None of the other three blocks can run concurrently with
+ // this one.
[self.writeable writesFinishedWithError:nil];
- // Break the retain cycle with writer, and skip any possible message to the
- // wrapped writeable enqueued after this one.
+ // Skip any possible message to the wrapped writeable enqueued after this one.
self.writeable = nil;
- self.writer = nil;
});
});
}
@@ -92,29 +87,24 @@
- (void)cancelWithError:(NSError *)error {
NSAssert(error, @"For a successful completion, use enqueueSuccessfulCompletion.");
dispatch_once(&_alreadyFinished, ^{
- // Skip any of the still-enqueued messages to the wrapped writeable. We use
- // the atomic setter to nillify writer and writeable because we might be
- // running concurrently with the blocks in _writeableQueue, and assignment
- // with ARC isn't atomic.
+ // Skip any of the still-enqueued messages to the wrapped writeable. We use the atomic setter to
+ // nillify writeable because we might be running concurrently with the blocks in
+ // _writeableQueue, and assignment with ARC isn't atomic.
id<GRXWriteable> writeable = self.writeable;
self.writeable = nil;
dispatch_async(_writeableQueue, ^{
[writeable writesFinishedWithError:error];
- // Break the retain cycle with writer.
- self.writer = nil;
});
});
}
- (void)cancelSilently {
dispatch_once(&_alreadyFinished, ^{
- // Skip any of the still-enqueued messages to the wrapped writeable. We use
- // the atomic setter to nillify writer and writeable because we might be
- // running concurrently with the blocks in _writeableQueue, and assignment
- // with ARC isn't atomic.
+ // Skip any of the still-enqueued messages to the wrapped writeable. We use the atomic setter to
+ // nillify writeable because we might be running concurrently with the blocks in
+ // _writeableQueue, and assignment with ARC isn't atomic.
self.writeable = nil;
- self.writer = nil;
});
}
@end
diff --git a/src/objective-c/RxLibrary/GRXImmediateWriter.m b/src/objective-c/RxLibrary/GRXImmediateWriter.m
index b6d2b2cac0..3edae788ab 100644
--- a/src/objective-c/RxLibrary/GRXImmediateWriter.m
+++ b/src/objective-c/RxLibrary/GRXImmediateWriter.m
@@ -76,28 +76,15 @@
}
+ (GRXWriter *)writerWithValue:(id)value {
- if (value) {
- return [self writerWithEnumerator:[NSEnumerator grx_enumeratorWithSingleValue:value]];
- } else {
- return [self emptyWriter];
- }
+ return [self writerWithEnumerator:[NSEnumerator grx_enumeratorWithSingleValue:value]];
}
+ (GRXWriter *)writerWithError:(NSError *)error {
- if (error) {
- return [self writerWithEnumerator:nil error:error];
- } else {
- return [self emptyWriter];
- }
+ return [self writerWithEnumerator:nil error:error];
}
+ (GRXWriter *)emptyWriter {
- static GRXImmediateWriter *emptyWriter;
- static dispatch_once_t onceToken;
- dispatch_once(&onceToken, ^{
- emptyWriter = [self writerWithEnumerator:nil error:nil];
- });
- return emptyWriter;
+ return [self writerWithEnumerator:nil error:nil];
}
#pragma mark Conformance with GRXWriter
diff --git a/src/objective-c/RxLibrary/private/GRXNSFastEnumerator.m b/src/objective-c/RxLibrary/private/GRXNSFastEnumerator.m
index 2050fa98ec..0387c9954e 100644
--- a/src/objective-c/RxLibrary/private/GRXNSFastEnumerator.m
+++ b/src/objective-c/RxLibrary/private/GRXNSFastEnumerator.m
@@ -59,7 +59,6 @@
// Designated initializer.
- (instancetype)initWithContainer:(id<NSFastEnumeration>)container {
- NSAssert(container, @"container can't be nil");
if ((self = [super init])) {
_container = container;
}
diff --git a/src/objective-c/generated_libraries/RemoteTestClient/RemoteTest.podspec b/src/objective-c/generated_libraries/RemoteTestClient/RemoteTest.podspec
index dd0dab352d..7cc9a040fe 100644
--- a/src/objective-c/generated_libraries/RemoteTestClient/RemoteTest.podspec
+++ b/src/objective-c/generated_libraries/RemoteTestClient/RemoteTest.podspec
@@ -7,7 +7,13 @@ Pod::Spec.new do |s|
s.osx.deployment_target = "10.8"
# Run protoc with the Objective-C and gRPC plugins to generate protocol messages and gRPC clients.
- s.prepare_command = "protoc --objc_out=. --objcgrpc_out=. *.proto"
+ s.prepare_command = <<-CMD
+ cd ../../../..
+ # TODO(jcanizales): Make only Objective-C plugin.
+ make plugins
+ cd -
+ protoc --plugin=protoc-gen-grpc=../../../../bins/opt/grpc_objective_c_plugin --objc_out=. --grpc_out=. *.proto
+ CMD
s.subspec "Messages" do |ms|
ms.source_files = "*.pbobjc.{h,m}"
diff --git a/src/objective-c/generated_libraries/RouteGuideClient/RouteGuide.podspec b/src/objective-c/generated_libraries/RouteGuideClient/RouteGuide.podspec
index e26e62f5bb..0e8dacd1c4 100644
--- a/src/objective-c/generated_libraries/RouteGuideClient/RouteGuide.podspec
+++ b/src/objective-c/generated_libraries/RouteGuideClient/RouteGuide.podspec
@@ -7,7 +7,13 @@ Pod::Spec.new do |s|
s.osx.deployment_target = "10.8"
# Run protoc with the Objective-C and gRPC plugins to generate protocol messages and gRPC clients.
- s.prepare_command = "protoc --objc_out=. --objcgrpc_out=. *.proto"
+ s.prepare_command = <<-CMD
+ cd ../../../..
+ # TODO(jcanizales): Make only Objective-C plugin.
+ make plugins
+ cd -
+ protoc --plugin=protoc-gen-grpc=../../../../bins/opt/grpc_objective_c_plugin --objc_out=. --grpc_out=. *.proto
+ CMD
s.subspec "Messages" do |ms|
ms.source_files = "*.pbobjc.{h,m}"
diff --git a/src/objective-c/tests/Tests.xcodeproj/project.pbxproj b/src/objective-c/tests/Tests.xcodeproj/project.pbxproj
index 34be705db2..f13fb8288b 100644
--- a/src/objective-c/tests/Tests.xcodeproj/project.pbxproj
+++ b/src/objective-c/tests/Tests.xcodeproj/project.pbxproj
@@ -391,7 +391,6 @@
635697DC1B14FC11007A7283 /* Debug */ = {
isa = XCBuildConfiguration;
buildSettings = {
- OTHER_LDFLAGS = "-ObjC";
PRODUCT_NAME = "$(TARGET_NAME)";
SKIP_INSTALL = YES;
};
@@ -400,7 +399,6 @@
635697DD1B14FC11007A7283 /* Release */ = {
isa = XCBuildConfiguration;
buildSettings = {
- OTHER_LDFLAGS = "-ObjC";
PRODUCT_NAME = "$(TARGET_NAME)";
SKIP_INSTALL = YES;
};
diff --git a/src/objective-c/tests/Tests.xcodeproj/xcshareddata/xcschemes/AllTests.xcscheme b/src/objective-c/tests/Tests.xcodeproj/xcshareddata/xcschemes/AllTests.xcscheme
new file mode 100644
index 0000000000..3a6e2c3591
--- /dev/null
+++ b/src/objective-c/tests/Tests.xcodeproj/xcshareddata/xcschemes/AllTests.xcscheme
@@ -0,0 +1,104 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<Scheme
+ LastUpgradeVersion = "0630"
+ version = "1.3">
+ <BuildAction
+ parallelizeBuildables = "YES"
+ buildImplicitDependencies = "YES">
+ <BuildActionEntries>
+ <BuildActionEntry
+ buildForTesting = "YES"
+ buildForRunning = "YES"
+ buildForProfiling = "NO"
+ buildForArchiving = "NO"
+ buildForAnalyzing = "YES">
+ <BuildableReference
+ BuildableIdentifier = "primary"
+ BlueprintIdentifier = "63423F431B150A5F006CF63C"
+ BuildableName = "AllTests.xctest"
+ BlueprintName = "AllTests"
+ ReferencedContainer = "container:Tests.xcodeproj">
+ </BuildableReference>
+ </BuildActionEntry>
+ </BuildActionEntries>
+ </BuildAction>
+ <TestAction
+ selectedDebuggerIdentifier = "Xcode.DebuggerFoundation.Debugger.LLDB"
+ selectedLauncherIdentifier = "Xcode.DebuggerFoundation.Launcher.LLDB"
+ shouldUseLaunchSchemeArgsEnv = "YES"
+ buildConfiguration = "Debug">
+ <Testables>
+ <TestableReference
+ skipped = "NO">
+ <BuildableReference
+ BuildableIdentifier = "primary"
+ BlueprintIdentifier = "63423F431B150A5F006CF63C"
+ BuildableName = "AllTests.xctest"
+ BlueprintName = "AllTests"
+ ReferencedContainer = "container:Tests.xcodeproj">
+ </BuildableReference>
+ <SkippedTests>
+ <Test
+ Identifier = "LocalClearTextTests">
+ </Test>
+ <Test
+ Identifier = "LocalClearTextTests/testConnectionToLocalServer">
+ </Test>
+ </SkippedTests>
+ </TestableReference>
+ </Testables>
+ <MacroExpansion>
+ <BuildableReference
+ BuildableIdentifier = "primary"
+ BlueprintIdentifier = "63423F431B150A5F006CF63C"
+ BuildableName = "AllTests.xctest"
+ BlueprintName = "AllTests"
+ ReferencedContainer = "container:Tests.xcodeproj">
+ </BuildableReference>
+ </MacroExpansion>
+ </TestAction>
+ <LaunchAction
+ selectedDebuggerIdentifier = "Xcode.DebuggerFoundation.Debugger.LLDB"
+ selectedLauncherIdentifier = "Xcode.DebuggerFoundation.Launcher.LLDB"
+ launchStyle = "0"
+ useCustomWorkingDirectory = "NO"
+ buildConfiguration = "Debug"
+ ignoresPersistentStateOnLaunch = "NO"
+ debugDocumentVersioning = "YES"
+ allowLocationSimulation = "YES">
+ <MacroExpansion>
+ <BuildableReference
+ BuildableIdentifier = "primary"
+ BlueprintIdentifier = "63423F431B150A5F006CF63C"
+ BuildableName = "AllTests.xctest"
+ BlueprintName = "AllTests"
+ ReferencedContainer = "container:Tests.xcodeproj">
+ </BuildableReference>
+ </MacroExpansion>
+ <AdditionalOptions>
+ </AdditionalOptions>
+ </LaunchAction>
+ <ProfileAction
+ shouldUseLaunchSchemeArgsEnv = "YES"
+ savedToolIdentifier = ""
+ useCustomWorkingDirectory = "NO"
+ buildConfiguration = "Release"
+ debugDocumentVersioning = "YES">
+ <MacroExpansion>
+ <BuildableReference
+ BuildableIdentifier = "primary"
+ BlueprintIdentifier = "63423F431B150A5F006CF63C"
+ BuildableName = "AllTests.xctest"
+ BlueprintName = "AllTests"
+ ReferencedContainer = "container:Tests.xcodeproj">
+ </BuildableReference>
+ </MacroExpansion>
+ </ProfileAction>
+ <AnalyzeAction
+ buildConfiguration = "Debug">
+ </AnalyzeAction>
+ <ArchiveAction
+ buildConfiguration = "Release"
+ revealArchiveInOrganizer = "YES">
+ </ArchiveAction>
+</Scheme>
diff --git a/src/objective-c/tests/run_tests.sh b/src/objective-c/tests/run_tests.sh
new file mode 100755
index 0000000000..37fced3a62
--- /dev/null
+++ b/src/objective-c/tests/run_tests.sh
@@ -0,0 +1,49 @@
+#!/bin/bash
+# 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.
+
+set -e
+
+cd $(dirname $0)
+
+# TODO(jcanizales): Remove when Cocoapods issue #3823 is resolved.
+export COCOAPODS_DISABLE_DETERMINISTIC_UUIDS=YES
+pod install
+
+# xcodebuild is very verbose. We filter its output and tell Bash to fail if any
+# element of the pipe fails.
+# TODO(jcanizales): Use xctool instead? Issue #2540.
+set -o pipefail
+XCODEBUILD_FILTER='(^===|^\*\*|\bfatal\b|\berror\b|\bwarning\b|\bfail)'
+xcodebuild \
+ -workspace Tests.xcworkspace \
+ -scheme AllTests \
+ -destination name="iPhone 6" \
+ test \
+ | egrep "$XCODEBUILD_FILTER" -
diff --git a/src/php/ext/grpc/server.c b/src/php/ext/grpc/server.c
index c319526e42..8b8d5b2f47 100644
--- a/src/php/ext/grpc/server.c
+++ b/src/php/ext/grpc/server.c
@@ -64,6 +64,7 @@ void free_wrapped_grpc_server(void *object TSRMLS_DC) {
wrapped_grpc_server *server = (wrapped_grpc_server *)object;
if (server->wrapped != NULL) {
grpc_server_shutdown_and_notify(server->wrapped, completion_queue, NULL);
+ grpc_server_cancel_all_calls(server->wrapped);
grpc_completion_queue_pluck(completion_queue, NULL,
gpr_inf_future(GPR_CLOCK_REALTIME));
grpc_server_destroy(server->wrapped);
diff --git a/src/php/tests/unit_tests/EndToEndTest.php b/src/php/tests/unit_tests/EndToEndTest.php
index 296873fa8f..2980dca4a7 100755
--- a/src/php/tests/unit_tests/EndToEndTest.php
+++ b/src/php/tests/unit_tests/EndToEndTest.php
@@ -61,7 +61,6 @@ class EndToEndTest extends PHPUnit_Framework_TestCase{
$event = $this->server->requestCall();
$this->assertSame('dummy_method', $event->method);
- $this->assertSame([], $event->metadata);
$server_call = $event->call;
$event = $server_call->startBatch([
@@ -83,7 +82,6 @@ class EndToEndTest extends PHPUnit_Framework_TestCase{
Grpc\OP_RECV_STATUS_ON_CLIENT => true
]);
- $this->assertSame([], $event->metadata);
$status = $event->status;
$this->assertSame([], $status->metadata);
$this->assertSame(Grpc\STATUS_OK, $status->code);
diff --git a/src/php/tests/unit_tests/SecureEndToEndTest.php b/src/php/tests/unit_tests/SecureEndToEndTest.php
index 0c18cd3e91..f91c006da5 100755
--- a/src/php/tests/unit_tests/SecureEndToEndTest.php
+++ b/src/php/tests/unit_tests/SecureEndToEndTest.php
@@ -73,7 +73,6 @@ class SecureEndToEndTest extends PHPUnit_Framework_TestCase{
$event = $this->server->requestCall();
$this->assertSame('dummy_method', $event->method);
- $this->assertSame([], $event->metadata);
$server_call = $event->call;
$event = $server_call->startBatch([
diff --git a/src/python/src/grpc/_adapter/_c/utility.c b/src/python/src/grpc/_adapter/_c/utility.c
index 000c8d0c38..d9f911a41a 100644
--- a/src/python/src/grpc/_adapter/_c/utility.c
+++ b/src/python/src/grpc/_adapter/_c/utility.c
@@ -489,10 +489,10 @@ PyObject *pygrpc_cast_metadata_array_to_pyseq(grpc_metadata_array metadata) {
void pygrpc_byte_buffer_to_bytes(
grpc_byte_buffer *buffer, char **result, size_t *result_size) {
grpc_byte_buffer_reader reader;
- grpc_byte_buffer_reader_init(&reader, buffer);
gpr_slice slice;
char *read_result = NULL;
size_t size = 0;
+ grpc_byte_buffer_reader_init(&reader, buffer);
while (grpc_byte_buffer_reader_next(&reader, &slice)) {
read_result = gpr_realloc(read_result, size + GPR_SLICE_LENGTH(slice));
memcpy(read_result + size, GPR_SLICE_START_PTR(slice),
diff --git a/src/python/src/grpc/_adapter/_low_test.py b/src/python/src/grpc/_adapter/_low_test.py
index 268e5fe765..a49cd007bf 100644
--- a/src/python/src/grpc/_adapter/_low_test.py
+++ b/src/python/src/grpc/_adapter/_low_test.py
@@ -129,7 +129,10 @@ class InsecureServerInsecureClient(unittest.TestCase):
self.assertIsInstance(request_event.call, _low.Call)
self.assertIs(server_request_tag, request_event.tag)
self.assertEquals(1, len(request_event.results))
- self.assertEquals(dict(client_initial_metadata), dict(request_event.results[0].initial_metadata))
+ got_initial_metadata = dict(request_event.results[0].initial_metadata)
+ self.assertEquals(
+ dict(client_initial_metadata),
+ dict((x, got_initial_metadata[x]) for x in zip(*client_initial_metadata)[0]))
self.assertEquals(METHOD, request_event.call_details.method)
self.assertEquals(HOST, request_event.call_details.host)
self.assertLess(abs(DEADLINE - request_event.call_details.deadline), DEADLINE_TOLERANCE)
diff --git a/src/python/src/grpc/_links/_transmission_test.py b/src/python/src/grpc/_links/_transmission_test.py
index c5ef1edb25..3eeec03f46 100644
--- a/src/python/src/grpc/_links/_transmission_test.py
+++ b/src/python/src/grpc/_links/_transmission_test.py
@@ -93,8 +93,13 @@ class TransmissionTest(test_cases.TransmissionTest, unittest.TestCase):
def create_service_completion(self):
return _intermediary_low.Code.OK, 'An exuberant test "details" message!'
- def assertMetadataEqual(self, original_metadata, transmitted_metadata):
- self.assertSequenceEqual(original_metadata, transmitted_metadata)
+ def assertMetadataTransmitted(self, original_metadata, transmitted_metadata):
+ # we need to filter out any additional metadata added in transmitted_metadata
+ # since implementations are allowed to add to what is sent (in any position)
+ keys, _ = zip(*original_metadata)
+ self.assertSequenceEqual(
+ original_metadata,
+ [x for x in transmitted_metadata if x[0] in keys])
class RoundTripTest(unittest.TestCase):
diff --git a/src/python/src/grpc/framework/interfaces/links/test_cases.py b/src/python/src/grpc/framework/interfaces/links/test_cases.py
index 3ac212ebdf..bf1f09d99d 100644
--- a/src/python/src/grpc/framework/interfaces/links/test_cases.py
+++ b/src/python/src/grpc/framework/interfaces/links/test_cases.py
@@ -161,8 +161,8 @@ class TransmissionTest(object):
raise NotImplementedError()
@abc.abstractmethod
- def assertMetadataEqual(self, original_metadata, transmitted_metadata):
- """Asserts that two metadata objects are equal.
+ def assertMetadataTransmitted(self, original_metadata, transmitted_metadata):
+ """Asserts that transmitted_metadata contains original_metadata.
Args:
original_metadata: A metadata object used in this test.
@@ -170,7 +170,8 @@ class TransmissionTest(object):
through the system under test.
Raises:
- AssertionError: if the two metadata objects are not equal.
+ AssertionError: if the transmitted_metadata object does not contain
+ original_metadata.
"""
raise NotImplementedError()
@@ -239,7 +240,7 @@ class TransmissionTest(object):
self.assertFalse(initial_metadata_seen)
self.assertFalse(seen_payloads)
self.assertFalse(terminal_metadata_seen)
- self.assertMetadataEqual(initial_metadata, ticket.initial_metadata)
+ self.assertMetadataTransmitted(initial_metadata, ticket.initial_metadata)
initial_metadata_seen = True
if ticket.payload is not None:
@@ -248,7 +249,7 @@ class TransmissionTest(object):
if ticket.terminal_metadata is not None:
self.assertFalse(terminal_metadata_seen)
- self.assertMetadataEqual(terminal_metadata, ticket.terminal_metadata)
+ self.assertMetadataTransmitted(terminal_metadata, ticket.terminal_metadata)
terminal_metadata_seen = True
self.assertSequenceEqual(payloads, seen_payloads)
diff --git a/src/ruby/spec/generic/rpc_server_spec.rb b/src/ruby/spec/generic/rpc_server_spec.rb
index f2403de77c..0326f6e894 100644
--- a/src/ruby/spec/generic/rpc_server_spec.rb
+++ b/src/ruby/spec/generic/rpc_server_spec.rb
@@ -35,6 +35,14 @@ def load_test_certs
files.map { |f| File.open(File.join(test_root, f)).read }
end
+def check_md(wanted_md, received_md)
+ wanted_md.zip(received_md).each do |w, r|
+ w.each do |key, value|
+ expect(r[key]).to eq(value)
+ end
+ end
+end
+
# A test message
class EchoMsg
def self.marshal(_o)
@@ -376,7 +384,7 @@ describe GRPC::RpcServer do
stub = EchoStub.new(@host, **client_opts)
expect(stub.an_rpc(req, k1: 'v1', k2: 'v2')).to be_a(EchoMsg)
wanted_md = [{ 'k1' => 'v1', 'k2' => 'v2' }]
- expect(service.received_md).to eq(wanted_md)
+ check_md(wanted_md, service.received_md)
@srv.stop
t.join
end
@@ -391,7 +399,7 @@ describe GRPC::RpcServer do
deadline = service.delay + 1.0 # wait for long enough
expect(stub.an_rpc(req, deadline, k1: 'v1', k2: 'v2')).to be_a(EchoMsg)
wanted_md = [{ 'k1' => 'v1', 'k2' => 'v2' }]
- expect(service.received_md).to eq(wanted_md)
+ check_md(wanted_md, service.received_md)
@srv.stop
t.join
end
@@ -443,7 +451,7 @@ describe GRPC::RpcServer do
expect(stub.an_rpc(req, k1: 'v1', k2: 'v2')).to be_a(EchoMsg)
wanted_md = [{ 'k1' => 'updated-v1', 'k2' => 'v2',
'jwt_aud_uri' => "https://#{@host}/EchoService" }]
- expect(service.received_md).to eq(wanted_md)
+ check_md(wanted_md, service.received_md)
@srv.stop
t.join
end
@@ -535,7 +543,9 @@ describe GRPC::RpcServer do
'method' => '/EchoService/an_rpc',
'connect_k1' => 'connect_v1'
}
- expect(op.metadata).to eq(wanted_md)
+ wanted_md.each do |key, value|
+ expect(op.metadata[key]).to eq(value)
+ end
@srv.stop
t.join
end