diff --git a/src/core/surface/channel.c b/src/core/surface/channel.c
index e38734c6a4921fb4e7e63a6d247262b13637e1c9..e764a3b9af0a82666137eecbf22a83eef2f66edb 100644
--- a/src/core/surface/channel.c
+++ b/src/core/surface/channel.c
@@ -39,6 +39,7 @@
#include "src/core/iomgr/iomgr.h"
#include "src/core/surface/call.h"
#include "src/core/surface/client.h"
+#include "src/core/surface/init.h"
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
@@ -63,6 +64,7 @@ grpc_channel *grpc_channel_create_from_filters(
size_t size =
sizeof(grpc_channel) + grpc_channel_stack_size(filters, num_filters);
grpc_channel *channel = gpr_malloc(size);
+ GPR_ASSERT(grpc_is_initialized() && "call grpc_init()");
channel->is_client = is_client;
/* decremented by grpc_channel_destroy, and grpc_client_channel_closed if is_client */
gpr_ref_init(&channel->refs, 1 + is_client);
diff --git a/src/core/surface/init.c b/src/core/surface/init.c
index 4db66fb66eb05b485a428f35113df0a775664d95..e48c4202e5831004453791d7004a0756879048de 100644
--- a/src/core/surface/init.c
+++ b/src/core/surface/init.c
@@ -40,17 +40,17 @@
#include "src/core/surface/surface_trace.h"
#include "src/core/transport/chttp2_transport.h"
-static gpr_once g_init = GPR_ONCE_INIT;
+static gpr_once g_basic_init = GPR_ONCE_INIT;
static gpr_mu g_init_mu;
static int g_initializations;
-static void do_init(void) {
+static void do_basic_init(void) {
gpr_mu_init(&g_init_mu);
g_initializations = 0;
}
void grpc_init(void) {
- gpr_once_init(&g_init, do_init);
+ gpr_once_init(&g_basic_init, do_basic_init);
gpr_mu_lock(&g_init_mu);
if (++g_initializations == 1) {
@@ -73,3 +73,13 @@ void grpc_shutdown(void) {
}
gpr_mu_unlock(&g_init_mu);
}
+
+int grpc_is_initialized(void) {
+ int r;
+ gpr_once_init(&g_basic_init, do_basic_init);
+ gpr_mu_lock(&g_init_mu);
+ r = g_initializations > 0;
+ gpr_mu_unlock(&g_init_mu);
+ return r;
+}
+
diff --git a/src/core/surface/init.h b/src/core/surface/init.h
index ab40bedf8754918fd965d52e5dfa4c2bcc1872c7..416874020d7c79b090b2e66487d443b15febd93a 100644
--- a/src/core/surface/init.h
+++ b/src/core/surface/init.h
@@ -35,5 +35,6 @@
#define GRPC_INTERNAL_CORE_SURFACE_INIT_H
void grpc_security_pre_init(void);
+int grpc_is_initialized(void);
#endif /* GRPC_INTERNAL_CORE_SURFACE_INIT_H */
diff --git a/src/core/surface/server.c b/src/core/surface/server.c
index c99a1b4cc9c9c72c3c144607890f6a143b69ba39..424734c54ce5615fcbe700bb22bd834803004d82 100644
--- a/src/core/surface/server.c
+++ b/src/core/surface/server.c
@@ -44,6 +44,7 @@
#include "src/core/surface/call.h"
#include "src/core/surface/channel.h"
#include "src/core/surface/completion_queue.h"
+#include "src/core/surface/init.h"
#include "src/core/transport/metadata.h"
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
@@ -612,6 +613,9 @@ grpc_server *grpc_server_create_from_filters(grpc_completion_queue *cq,
int census_enabled = grpc_channel_args_is_census_enabled(args);
grpc_server *server = gpr_malloc(sizeof(grpc_server));
+
+ GPR_ASSERT(grpc_is_initialized() && "call grpc_init()");
+
memset(server, 0, sizeof(grpc_server));
if (cq) addcq(server, cq);
diff --git a/src/csharp/.gitignore b/src/csharp/.gitignore
index 4f4cd1f7d1ad47282abab67f2c431147e8dd92b4..dbaf60de0cf6dfbeb27fa4948a42a70982b10ef6 100644
--- a/src/csharp/.gitignore
+++ b/src/csharp/.gitignore
@@ -1,4 +1,5 @@
*.userprefs
+StyleCop.Cache
test-results
packages
Grpc.v12.suo
diff --git a/src/csharp/Grpc.Core.Tests/PInvokeTest.cs b/src/csharp/Grpc.Core.Tests/PInvokeTest.cs
index 282d521ba3798004837e5c593d268c1103cd39f0..9db08d2f02f7ba5042a694cdd30ef065eedf1c9a 100644
--- a/src/csharp/Grpc.Core.Tests/PInvokeTest.cs
+++ b/src/csharp/Grpc.Core.Tests/PInvokeTest.cs
@@ -127,8 +127,6 @@ namespace Grpc.Core.Tests
[Test]
public void NopPInvokeBenchmark()
{
- CompletionCallbackDelegate handler = Handler;
-
BenchmarkUtil.RunBenchmark(
1000000, 100000000,
() => {
diff --git a/src/csharp/Grpc.Core/ChannelArgs.cs b/src/csharp/Grpc.Core/ChannelArgs.cs
index 653a5780a33340be868f96efe22b1a283c9129cf..298b6edf2066f78e031ea6ca990a10bd47961364 100644
--- a/src/csharp/Grpc.Core/ChannelArgs.cs
+++ b/src/csharp/Grpc.Core/ChannelArgs.cs
@@ -99,7 +99,7 @@ namespace Grpc.Core
}
return nativeArgs;
}
- catch (Exception e)
+ catch (Exception)
{
if (nativeArgs != null)
{
diff --git a/src/csharp/Grpc.Core/Grpc.Core.csproj b/src/csharp/Grpc.Core/Grpc.Core.csproj
index 93d5430591b5018bb7d52b1f414db3c013f7fec7..78b6cdde59bc11f7052a290d3d15c04e8141b310 100644
--- a/src/csharp/Grpc.Core/Grpc.Core.csproj
+++ b/src/csharp/Grpc.Core/Grpc.Core.csproj
@@ -51,7 +51,6 @@
<Compile Include="Internal\SafeHandleZeroIsInvalid.cs" />
<Compile Include="Internal\Timespec.cs" />
<Compile Include="Internal\GrpcThreadPool.cs" />
- <Compile Include="Internal\AsyncCall.cs" />
<Compile Include="Internal\ServerSafeHandle.cs" />
<Compile Include="Method.cs" />
<Compile Include="ServerCalls.cs" />
@@ -69,6 +68,12 @@
<Compile Include="Credentials.cs" />
<Compile Include="Internal\ChannelArgsSafeHandle.cs" />
<Compile Include="ChannelArgs.cs" />
+ <Compile Include="Internal\AsyncCompletion.cs" />
+ <Compile Include="Internal\AsyncCallBase.cs" />
+ <Compile Include="Internal\AsyncCallServer.cs" />
+ <Compile Include="OperationFailedException.cs" />
+ <Compile Include="Internal\AsyncCall.cs" />
+ <Compile Include="Utils\Preconditions.cs" />
</ItemGroup>
<Choose>
<!-- Under older versions of Monodevelop, Choose is not supported and is just
diff --git a/src/csharp/Grpc.Core/Internal/AsyncCall.cs b/src/csharp/Grpc.Core/Internal/AsyncCall.cs
index 6f37b059f75dcc22b1d85b0dbf85ef7d95d68f38..5ae036298b00c16f032dc6832f1ba7bf91b76d7f 100644
--- a/src/csharp/Grpc.Core/Internal/AsyncCall.cs
+++ b/src/csharp/Grpc.Core/Internal/AsyncCall.cs
@@ -43,84 +43,47 @@ using Grpc.Core.Utils;
namespace Grpc.Core.Internal
{
/// <summary>
- /// Handles native call lifecycle and provides convenience methods.
+ /// Handles client side native call lifecycle.
/// </summary>
- internal class AsyncCall<TWrite, TRead>
+ internal class AsyncCall<TRequest, TResponse> : AsyncCallBase<TRequest, TResponse>
{
- readonly Func<TWrite, byte[]> serializer;
- readonly Func<byte[], TRead> deserializer;
-
readonly CompletionCallbackDelegate unaryResponseHandler;
readonly CompletionCallbackDelegate finishedHandler;
- readonly CompletionCallbackDelegate writeFinishedHandler;
- readonly CompletionCallbackDelegate readFinishedHandler;
- readonly CompletionCallbackDelegate halfclosedHandler;
- readonly CompletionCallbackDelegate finishedServersideHandler;
-
- object myLock = new object();
- GCHandle gchandle;
- CallSafeHandle call;
- bool disposed;
-
- bool server;
-
- bool started;
- bool errorOccured;
- bool cancelRequested;
- bool readingDone;
- bool halfcloseRequested;
- bool halfclosed;
- bool finished;
-
- // Completion of a pending write if not null.
- TaskCompletionSource<object> writeTcs;
-
- // Completion of a pending read if not null.
- TaskCompletionSource<TRead> readTcs;
-
- // Completion of a pending halfclose if not null.
- TaskCompletionSource<object> halfcloseTcs;
// Completion of a pending unary response if not null.
- TaskCompletionSource<TRead> unaryResponseTcs;
+ TaskCompletionSource<TResponse> unaryResponseTcs;
- // Set after status is received on client. Only used for server streaming and duplex streaming calls.
+ // Set after status is received. Only used for streaming response calls.
Nullable<Status> finishedStatus;
- TaskCompletionSource<object> finishedServersideTcs = new TaskCompletionSource<object>();
- // For streaming, the reads will be delivered to this observer.
- IObserver<TRead> readObserver;
+ bool readObserverCompleted; // True if readObserver has already been completed.
- public AsyncCall(Func<TWrite, byte[]> serializer, Func<byte[], TRead> deserializer)
+ public AsyncCall(Func<TRequest, byte[]> serializer, Func<byte[], TResponse> deserializer) : base(serializer, deserializer)
{
- this.serializer = serializer;
- this.deserializer = deserializer;
- this.unaryResponseHandler = HandleUnaryResponse;
- this.finishedHandler = HandleFinished;
- this.writeFinishedHandler = HandleWriteFinished;
- this.readFinishedHandler = HandleReadFinished;
- this.halfclosedHandler = HandleHalfclosed;
- this.finishedServersideHandler = HandleFinishedServerside;
+ this.unaryResponseHandler = CreateBatchCompletionCallback(HandleUnaryResponse);
+ this.finishedHandler = CreateBatchCompletionCallback(HandleFinished);
}
public void Initialize(Channel channel, CompletionQueueSafeHandle cq, String methodName)
{
- InitializeInternal(CallSafeHandle.Create(channel.Handle, cq, methodName, channel.Target, Timespec.InfFuture), false);
+ var call = CallSafeHandle.Create(channel.Handle, cq, methodName, channel.Target, Timespec.InfFuture);
+ InitializeInternal(call);
}
- public void InitializeServer(CallSafeHandle call)
- {
- InitializeInternal(call, true);
- }
-
- public TRead UnaryCall(Channel channel, String methodName, TWrite msg)
+ // TODO: this method is not Async, so it shouldn't be in AsyncCall class, but
+ // it is reusing fair amount of code in this class, so we are leaving it here.
+ // TODO: for other calls, you need to call Initialize, this methods calls initialize
+ // on its own, so there's a usage inconsistency.
+ /// <summary>
+ /// Blocking unary request - unary response call.
+ /// </summary>
+ public TResponse UnaryCall(Channel channel, String methodName, TRequest msg)
{
using(CompletionQueueSafeHandle cq = CompletionQueueSafeHandle.Create())
{
- // TODO: handle serialization error...
- byte[] payload = serializer(msg);
+ byte[] payload = UnsafeSerialize(msg);
- unaryResponseTcs = new TaskCompletionSource<TRead>();
+ unaryResponseTcs = new TaskCompletionSource<TResponse>();
lock (myLock)
{
@@ -143,508 +106,200 @@ namespace Grpc.Core.Internal
}
}
- public Task<TRead> UnaryCallAsync(TWrite msg)
+ /// <summary>
+ /// Starts a unary request - unary response call.
+ /// </summary>
+ public Task<TResponse> UnaryCallAsync(TRequest msg)
{
lock (myLock)
{
+ Preconditions.CheckNotNull(call);
+
started = true;
halfcloseRequested = true;
readingDone = true;
- // TODO: handle serialization error...
- byte[] payload = serializer(msg);
+ byte[] payload = UnsafeSerialize(msg);
- unaryResponseTcs = new TaskCompletionSource<TRead>();
+ unaryResponseTcs = new TaskCompletionSource<TResponse>();
call.StartUnary(payload, unaryResponseHandler);
return unaryResponseTcs.Task;
}
}
- public Task<TRead> ClientStreamingCallAsync()
+ /// <summary>
+ /// Starts a streamed request - unary response call.
+ /// Use StartSendMessage and StartSendCloseFromClient to stream requests.
+ /// </summary>
+ public Task<TResponse> ClientStreamingCallAsync()
{
lock (myLock)
{
+ Preconditions.CheckNotNull(call);
+
started = true;
readingDone = true;
- unaryResponseTcs = new TaskCompletionSource<TRead>();
+ unaryResponseTcs = new TaskCompletionSource<TResponse>();
call.StartClientStreaming(unaryResponseHandler);
return unaryResponseTcs.Task;
}
}
- public void StartServerStreamingCall(TWrite msg, IObserver<TRead> readObserver)
+ /// <summary>
+ /// Starts a unary request - streamed response call.
+ /// </summary>
+ public void StartServerStreamingCall(TRequest msg, IObserver<TResponse> readObserver)
{
lock (myLock)
{
+ Preconditions.CheckNotNull(call);
+
started = true;
halfcloseRequested = true;
halfclosed = true; // halfclose not confirmed yet, but it will be once finishedHandler is called.
this.readObserver = readObserver;
- // TODO: handle serialization error...
- byte[] payload = serializer(msg);
+ byte[] payload = UnsafeSerialize(msg);
call.StartServerStreaming(payload, finishedHandler);
- ReceiveMessageAsync();
+ StartReceiveMessage();
}
}
- public void StartDuplexStreamingCall(IObserver<TRead> readObserver)
+ /// <summary>
+ /// Starts a streaming request - streaming response call.
+ /// Use StartSendMessage and StartSendCloseFromClient to stream requests.
+ /// </summary>
+ public void StartDuplexStreamingCall(IObserver<TResponse> readObserver)
{
lock (myLock)
{
+ Preconditions.CheckNotNull(call);
+
started = true;
this.readObserver = readObserver;
call.StartDuplexStreaming(finishedHandler);
- ReceiveMessageAsync();
+ StartReceiveMessage();
}
}
- public Task ServerSideUnaryRequestCallAsync()
- {
- lock (myLock)
- {
- started = true;
- call.StartServerSide(finishedServersideHandler);
- return finishedServersideTcs.Task;
- }
- }
-
- public Task ServerSideStreamingRequestCallAsync(IObserver<TRead> readObserver)
- {
- lock (myLock)
- {
- started = true;
- call.StartServerSide(finishedServersideHandler);
-
- if (this.readObserver != null)
- {
- throw new InvalidOperationException("Already registered an observer.");
- }
- this.readObserver = readObserver;
- ReceiveMessageAsync();
-
- return finishedServersideTcs.Task;
- }
- }
-
- public Task SendMessageAsync(TWrite msg)
+ /// <summary>
+ /// Sends a streaming request. Only one pending send action is allowed at any given time.
+ /// completionDelegate is called when the operation finishes.
+ /// </summary>
+ public void StartSendMessage(TRequest msg, AsyncCompletionDelegate completionDelegate)
{
- lock (myLock)
- {
- CheckNotDisposed();
- CheckStarted();
- CheckNoError();
-
- if (halfcloseRequested)
- {
- throw new InvalidOperationException("Already halfclosed.");
- }
-
- if (writeTcs != null)
- {
- throw new InvalidOperationException("Only one write can be pending at a time");
- }
-
- // TODO: wrap serialization...
- byte[] payload = serializer(msg);
-
- call.StartSendMessage(payload, writeFinishedHandler);
- writeTcs = new TaskCompletionSource<object>();
- return writeTcs.Task;
- }
+ StartSendMessageInternal(msg, completionDelegate);
}
- public Task SendCloseFromClientAsync()
+ /// <summary>
+ /// Sends halfclose, indicating client is done with streaming requests.
+ /// Only one pending send action is allowed at any given time.
+ /// completionDelegate is called when the operation finishes.
+ /// </summary>
+ public void StartSendCloseFromClient(AsyncCompletionDelegate completionDelegate)
{
lock (myLock)
{
- CheckNotDisposed();
- CheckStarted();
- CheckNoError();
-
- if (halfcloseRequested)
- {
- throw new InvalidOperationException("Already halfclosed.");
- }
+ Preconditions.CheckNotNull(completionDelegate, "Completion delegate cannot be null");
+ CheckSendingAllowed();
call.StartSendCloseFromClient(halfclosedHandler);
halfcloseRequested = true;
- halfcloseTcs = new TaskCompletionSource<object>();
- return halfcloseTcs.Task;
- }
- }
-
- public Task SendStatusFromServerAsync(Status status)
- {
- lock (myLock)
- {
- CheckNotDisposed();
- CheckStarted();
- CheckNoError();
-
- if (halfcloseRequested)
- {
- throw new InvalidOperationException("Already halfclosed.");
- }
-
- call.StartSendStatusFromServer(status, halfclosedHandler);
- halfcloseRequested = true;
- halfcloseTcs = new TaskCompletionSource<object>();
- return halfcloseTcs.Task;
+ sendCompletionDelegate = completionDelegate;
}
}
- public Task<TRead> ReceiveMessageAsync()
+ /// <summary>
+ /// On client-side, we only fire readObserver.OnCompleted once all messages have been read
+ /// and status has been received.
+ /// </summary>
+ protected override void CompleteReadObserver()
{
- lock (myLock)
+ if (readingDone && finishedStatus.HasValue)
{
- CheckNotDisposed();
- CheckStarted();
- CheckNoError();
-
- if (readingDone)
- {
- throw new InvalidOperationException("Already read the last message.");
- }
-
- if (readTcs != null)
+ bool shouldComplete;
+ lock (myLock)
{
- throw new InvalidOperationException("Only one read can be pending at a time");
+ shouldComplete = !readObserverCompleted;
+ readObserverCompleted = true;
}
- call.StartReceiveMessage(readFinishedHandler);
-
- readTcs = new TaskCompletionSource<TRead>();
- return readTcs.Task;
- }
- }
-
- public void Cancel()
- {
- lock (myLock)
- {
- CheckNotDisposed();
- CheckStarted();
- cancelRequested = true;
- }
- // grpc_call_cancel is threadsafe
- call.Cancel();
- }
-
- public void CancelWithStatus(Status status)
- {
- lock (myLock)
- {
- CheckNotDisposed();
- CheckStarted();
- cancelRequested = true;
- }
- // grpc_call_cancel_with_status is threadsafe
- call.CancelWithStatus(status);
- }
-
- private void InitializeInternal(CallSafeHandle call, bool server)
- {
- lock (myLock)
- {
- // Make sure this object and the delegated held by it will not be garbage collected
- // before we release this handle.
- gchandle = GCHandle.Alloc(this);
- this.call = call;
- this.server = server;
- }
- }
-
- private void CheckStarted()
- {
- if (!started)
- {
- throw new InvalidOperationException("Call not started");
- }
- }
-
- private void CheckNotDisposed()
- {
- if (disposed)
- {
- throw new InvalidOperationException("Call has already been disposed.");
- }
- }
-
- private void CheckNoError()
- {
- if (errorOccured)
- {
- throw new InvalidOperationException("Error occured when processing call.");
- }
- }
-
- private bool ReleaseResourcesIfPossible()
- {
- if (!disposed && call != null)
- {
- if (halfclosed && readingDone && finished)
+ if (shouldComplete)
{
- ReleaseResources();
- return true;
+ var status = finishedStatus.Value;
+ if (status.StatusCode != StatusCode.OK)
+ {
+ FireReadObserverOnError(new RpcException(status));
+ }
+ else
+ {
+ FireReadObserverOnCompleted();
+ }
}
}
- return false;
- }
-
- private void ReleaseResources()
- {
- if (call != null) {
- call.Dispose();
- }
- gchandle.Free();
- disposed = true;
- }
-
- private void CompleteStreamObserver(Status status)
- {
- if (status.StatusCode != StatusCode.OK)
- {
- // TODO: wrap to handle exceptions;
- readObserver.OnError(new RpcException(status));
- } else {
- // TODO: wrap to handle exceptions;
- readObserver.OnCompleted();
- }
}
/// <summary>
/// Handler for unary response completion.
/// </summary>
- private void HandleUnaryResponse(GRPCOpError error, IntPtr batchContextPtr)
+ private void HandleUnaryResponse(bool wasError, BatchContextSafeHandleNotOwned ctx)
{
- try
+ lock(myLock)
{
- TaskCompletionSource<TRead> tcs;
- lock(myLock)
- {
- finished = true;
- halfclosed = true;
- tcs = unaryResponseTcs;
-
- ReleaseResourcesIfPossible();
- }
-
- var ctx = new BatchContextSafeHandleNotOwned(batchContextPtr);
+ finished = true;
+ halfclosed = true;
- if (error != GRPCOpError.GRPC_OP_OK)
- {
- tcs.SetException(new RpcException(
- new Status(StatusCode.Internal, "Internal error occured.")
- ));
- return;
- }
-
- var status = ctx.GetReceivedStatus();
- if (status.StatusCode != StatusCode.OK)
- {
- tcs.SetException(new RpcException(status));
- return;
- }
-
- // TODO: handle deserialize error...
- var msg = deserializer(ctx.GetReceivedMessage());
- tcs.SetResult(msg);
- }
- catch(Exception e)
- {
- Console.WriteLine("Caught exception in a native handler: " + e);
+ ReleaseResourcesIfPossible();
}
- }
-
- private void HandleWriteFinished(GRPCOpError error, IntPtr batchContextPtr)
- {
- try
- {
- TaskCompletionSource<object> oldTcs = null;
- lock (myLock)
- {
- oldTcs = writeTcs;
- writeTcs = null;
- }
-
- if (errorOccured)
- {
- // TODO: use the right type of exception...
- oldTcs.SetException(new Exception("Write failed"));
- }
- else
- {
- // TODO: where does the continuation run?
- oldTcs.SetResult(null);
- }
- }
- catch(Exception e)
+ if (wasError)
{
- Console.WriteLine("Caught exception in a native handler: " + e);
+ unaryResponseTcs.SetException(new RpcException(
+ new Status(StatusCode.Internal, "Internal error occured.")
+ ));
+ return;
}
- }
-
- private void HandleHalfclosed(GRPCOpError error, IntPtr batchContextPtr)
- {
- try
- {
- lock (myLock)
- {
- halfclosed = true;
- ReleaseResourcesIfPossible();
- }
-
- if (error != GRPCOpError.GRPC_OP_OK)
- {
- halfcloseTcs.SetException(new Exception("Halfclose failed"));
-
- }
- else
- {
- halfcloseTcs.SetResult(null);
- }
- }
- catch(Exception e)
+ var status = ctx.GetReceivedStatus();
+ if (status.StatusCode != StatusCode.OK)
{
- Console.WriteLine("Caught exception in a native handler: " + e);
+ unaryResponseTcs.SetException(new RpcException(status));
+ return;
}
- }
-
- private void HandleReadFinished(GRPCOpError error, IntPtr batchContextPtr)
- {
- try
- {
- var ctx = new BatchContextSafeHandleNotOwned(batchContextPtr);
- var payload = ctx.GetReceivedMessage();
-
- TaskCompletionSource<TRead> oldTcs = null;
- IObserver<TRead> observer = null;
-
- Nullable<Status> status = null;
-
- lock (myLock)
- {
- oldTcs = readTcs;
- readTcs = null;
- if (payload == null)
- {
- readingDone = true;
- }
- observer = readObserver;
- status = finishedStatus;
-
- ReleaseResourcesIfPossible();
- }
-
- // TODO: wrap deserialization...
- TRead msg = payload != null ? deserializer(payload) : default(TRead);
- oldTcs.SetResult(msg);
+ // TODO: handle deserialization error
+ TResponse msg;
+ TryDeserialize(ctx.GetReceivedMessage(), out msg);
- // TODO: make sure we deliver reads in the right order.
-
- if (observer != null)
- {
- if (payload != null)
- {
- // TODO: wrap to handle exceptions
- observer.OnNext(msg);
-
- // start a new read
- ReceiveMessageAsync();
- }
- else
- {
- if (!server)
- {
- if (status.HasValue)
- {
- CompleteStreamObserver(status.Value);
- }
- }
- else
- {
- // TODO: wrap to handle exceptions..
- observer.OnCompleted();
- }
- // TODO: completeStreamObserver serverside...
- }
- }
- }
- catch(Exception e)
- {
- Console.WriteLine("Caught exception in a native handler: " + e);
- }
+ unaryResponseTcs.SetResult(msg);
}
- private void HandleFinished(GRPCOpError error, IntPtr batchContextPtr)
+ /// <summary>
+ /// Handles receive status completion for calls with streaming response.
+ /// </summary>
+ private void HandleFinished(bool wasError, BatchContextSafeHandleNotOwned ctx)
{
- try
- {
- var ctx = new BatchContextSafeHandleNotOwned(batchContextPtr);
- var status = ctx.GetReceivedStatus();
-
- bool wasReadingDone;
-
- lock (myLock)
- {
- finished = true;
- finishedStatus = status;
-
- wasReadingDone = readingDone;
-
- ReleaseResourcesIfPossible();
- }
-
- if (wasReadingDone) {
- CompleteStreamObserver(status);
- }
-
- }
- catch(Exception e)
- {
- Console.WriteLine("Caught exception in a native handler: " + e);
- }
- }
+ var status = ctx.GetReceivedStatus();
- private void HandleFinishedServerside(GRPCOpError error, IntPtr batchContextPtr)
- {
- try
+ lock (myLock)
{
- var ctx = new BatchContextSafeHandleNotOwned(batchContextPtr);
-
- lock(myLock)
- {
- finished = true;
-
- // TODO: because of the way server calls are implemented, we need to set
- // reading done to true here. Should be fixed in the future.
- readingDone = true;
-
- ReleaseResourcesIfPossible();
- }
- // TODO: handle error ...
-
- finishedServersideTcs.SetResult(null);
+ finished = true;
+ finishedStatus = status;
+ ReleaseResourcesIfPossible();
}
- catch(Exception e)
- {
- Console.WriteLine("Caught exception in a native handler: " + e);
- }
+
+ CompleteReadObserver();
}
}
}
\ No newline at end of file
diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs b/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs
new file mode 100644
index 0000000000000000000000000000000000000000..44d66b394ca6929bc1fea07c0bf85e0de5e0fcf7
--- /dev/null
+++ b/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs
@@ -0,0 +1,407 @@
+#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.Diagnostics;
+using System.Runtime.CompilerServices;
+using System.Runtime.InteropServices;
+using System.Threading;
+using System.Threading.Tasks;
+using Grpc.Core.Internal;
+using Grpc.Core.Utils;
+
+namespace Grpc.Core.Internal
+{
+ /// <summary>
+ /// Base for handling both client side and server side calls.
+ /// Handles native call lifecycle and provides convenience methods.
+ /// </summary>
+ internal abstract class AsyncCallBase<TWrite, TRead>
+ {
+ readonly Func<TWrite, byte[]> serializer;
+ readonly Func<byte[], TRead> deserializer;
+
+ protected readonly CompletionCallbackDelegate sendFinishedHandler;
+ protected readonly CompletionCallbackDelegate readFinishedHandler;
+ protected readonly CompletionCallbackDelegate halfclosedHandler;
+
+ protected readonly object myLock = new object();
+
+ protected GCHandle gchandle;
+ protected CallSafeHandle call;
+ protected bool disposed;
+
+ protected bool started;
+ protected bool errorOccured;
+ protected bool cancelRequested;
+
+ protected AsyncCompletionDelegate sendCompletionDelegate; // Completion of a pending send or sendclose if not null.
+ protected bool readPending; // True if there is a read in progress.
+ protected bool readingDone;
+ protected bool halfcloseRequested;
+ protected bool halfclosed;
+ protected bool finished; // True if close has been received from the peer.
+
+ // Streaming reads will be delivered to this observer. For a call that only does unary read it may remain null.
+ protected IObserver<TRead> readObserver;
+
+ public AsyncCallBase(Func<TWrite, byte[]> serializer, Func<byte[], TRead> deserializer)
+ {
+ this.serializer = Preconditions.CheckNotNull(serializer);
+ this.deserializer = Preconditions.CheckNotNull(deserializer);
+
+ this.sendFinishedHandler = CreateBatchCompletionCallback(HandleSendFinished);
+ this.readFinishedHandler = CreateBatchCompletionCallback(HandleReadFinished);
+ this.halfclosedHandler = CreateBatchCompletionCallback(HandleHalfclosed);
+ }
+
+ /// <summary>
+ /// Requests cancelling the call.
+ /// </summary>
+ public void Cancel()
+ {
+ lock (myLock)
+ {
+ Preconditions.CheckState(started);
+ cancelRequested = true;
+
+ if (!disposed)
+ {
+ call.Cancel();
+ }
+ }
+ }
+
+ /// <summary>
+ /// Requests cancelling the call with given status.
+ /// </summary>
+ public void CancelWithStatus(Status status)
+ {
+ lock (myLock)
+ {
+ Preconditions.CheckState(started);
+ cancelRequested = true;
+
+ if (!disposed)
+ {
+ call.CancelWithStatus(status);
+ }
+ }
+ }
+
+ protected void InitializeInternal(CallSafeHandle call)
+ {
+ lock (myLock)
+ {
+ // Make sure this object and the delegated held by it will not be garbage collected
+ // before we release this handle.
+ gchandle = GCHandle.Alloc(this);
+ this.call = call;
+ }
+ }
+
+ /// <summary>
+ /// Initiates sending a message. Only once send operation can be active at a time.
+ /// completionDelegate is invoked upon completion.
+ /// </summary>
+ protected void StartSendMessageInternal(TWrite msg, AsyncCompletionDelegate completionDelegate)
+ {
+ byte[] payload = UnsafeSerialize(msg);
+
+ lock (myLock)
+ {
+ Preconditions.CheckNotNull(completionDelegate, "Completion delegate cannot be null");
+ CheckSendingAllowed();
+
+ call.StartSendMessage(payload, sendFinishedHandler);
+ sendCompletionDelegate = completionDelegate;
+ }
+ }
+
+ /// <summary>
+ /// Requests receiving a next message.
+ /// </summary>
+ protected void StartReceiveMessage()
+ {
+ lock (myLock)
+ {
+ Preconditions.CheckState(started);
+ Preconditions.CheckState(!disposed);
+ Preconditions.CheckState(!errorOccured);
+
+ Preconditions.CheckState(!readingDone);
+ Preconditions.CheckState(!readPending);
+
+ call.StartReceiveMessage(readFinishedHandler);
+ readPending = true;
+ }
+ }
+
+ /// <summary>
+ /// Default behavior just completes the read observer, but more sofisticated behavior might be required
+ /// by subclasses.
+ /// </summary>
+ protected virtual void CompleteReadObserver()
+ {
+ FireReadObserverOnCompleted();
+ }
+
+ /// <summary>
+ /// If there are no more pending actions and no new actions can be started, releases
+ /// the underlying native resources.
+ /// </summary>
+ protected bool ReleaseResourcesIfPossible()
+ {
+ if (!disposed && call != null)
+ {
+ if (halfclosed && readingDone && finished)
+ {
+ ReleaseResources();
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private void ReleaseResources()
+ {
+ if (call != null)
+ {
+ call.Dispose();
+ }
+ gchandle.Free();
+ disposed = true;
+ }
+
+ protected void CheckSendingAllowed()
+ {
+ Preconditions.CheckState(started);
+ Preconditions.CheckState(!disposed);
+ Preconditions.CheckState(!errorOccured);
+
+ Preconditions.CheckState(!halfcloseRequested, "Already halfclosed.");
+ Preconditions.CheckState(sendCompletionDelegate == null, "Only one write can be pending at a time");
+ }
+
+ protected byte[] UnsafeSerialize(TWrite msg)
+ {
+ return serializer(msg);
+ }
+
+ protected bool TrySerialize(TWrite msg, out byte[] payload)
+ {
+ try
+ {
+ payload = serializer(msg);
+ return true;
+ }
+ catch(Exception)
+ {
+ Console.WriteLine("Exception occured while trying to serialize message");
+ payload = null;
+ return false;
+ }
+ }
+
+ protected bool TryDeserialize(byte[] payload, out TRead msg)
+ {
+ try
+ {
+ msg = deserializer(payload);
+ return true;
+ }
+ catch(Exception)
+ {
+ Console.WriteLine("Exception occured while trying to deserialize message");
+ msg = default(TRead);
+ return false;
+ }
+ }
+
+ protected void FireReadObserverOnNext(TRead value)
+ {
+ try
+ {
+ readObserver.OnNext(value);
+ }
+ catch(Exception e)
+ {
+ Console.WriteLine("Exception occured while invoking readObserver.OnNext: " + e);
+ }
+ }
+
+ protected void FireReadObserverOnCompleted()
+ {
+ try
+ {
+ readObserver.OnCompleted();
+ }
+ catch(Exception e)
+ {
+ Console.WriteLine("Exception occured while invoking readObserver.OnCompleted: " + e);
+ }
+ }
+
+ protected void FireReadObserverOnError(Exception error)
+ {
+ try
+ {
+ readObserver.OnError(error);
+ }
+ catch(Exception e)
+ {
+ Console.WriteLine("Exception occured while invoking readObserver.OnError: " + e);
+ }
+ }
+
+ protected void FireCompletion(AsyncCompletionDelegate completionDelegate, Exception error)
+ {
+ try
+ {
+ completionDelegate(error);
+ }
+ catch(Exception e)
+ {
+ Console.WriteLine("Exception occured while invoking completion delegate: " + e);
+ }
+ }
+
+ /// <summary>
+ /// Creates completion callback delegate that wraps the batch completion handler in a try catch block to
+ /// prevent propagating exceptions accross managed/unmanaged boundary.
+ /// </summary>
+ protected CompletionCallbackDelegate CreateBatchCompletionCallback(Action<bool, BatchContextSafeHandleNotOwned> handler)
+ {
+ return new CompletionCallbackDelegate( (error, batchContextPtr) => {
+ try
+ {
+ var ctx = new BatchContextSafeHandleNotOwned(batchContextPtr);
+ bool wasError = (error != GRPCOpError.GRPC_OP_OK);
+ handler(wasError, ctx);
+ }
+ catch(Exception e)
+ {
+ Console.WriteLine("Caught exception in a native handler: " + e);
+ }
+ });
+ }
+
+ /// <summary>
+ /// Handles send completion.
+ /// </summary>
+ private void HandleSendFinished(bool wasError, BatchContextSafeHandleNotOwned ctx)
+ {
+ AsyncCompletionDelegate origCompletionDelegate = null;
+ lock (myLock)
+ {
+ origCompletionDelegate = sendCompletionDelegate;
+ sendCompletionDelegate = null;
+
+ ReleaseResourcesIfPossible();
+ }
+
+ if (wasError)
+ {
+ FireCompletion(origCompletionDelegate, new OperationFailedException("Send failed"));
+ }
+ else
+ {
+ FireCompletion(origCompletionDelegate, null);
+ }
+ }
+
+ /// <summary>
+ /// Handles halfclose completion.
+ /// </summary>
+ private void HandleHalfclosed(bool wasError, BatchContextSafeHandleNotOwned ctx)
+ {
+ AsyncCompletionDelegate origCompletionDelegate = null;
+ lock (myLock)
+ {
+ halfclosed = true;
+ origCompletionDelegate = sendCompletionDelegate;
+ sendCompletionDelegate = null;
+
+ ReleaseResourcesIfPossible();
+ }
+
+ if (wasError)
+ {
+ FireCompletion(origCompletionDelegate, new OperationFailedException("Halfclose failed"));
+ }
+ else
+ {
+ FireCompletion(origCompletionDelegate, null);
+ }
+
+ }
+
+ /// <summary>
+ /// Handles streaming read completion.
+ /// </summary>
+ private void HandleReadFinished(bool wasError, BatchContextSafeHandleNotOwned ctx)
+ {
+ var payload = ctx.GetReceivedMessage();
+
+ lock (myLock)
+ {
+ readPending = false;
+ if (payload == null)
+ {
+ readingDone = true;
+ }
+
+ ReleaseResourcesIfPossible();
+ }
+
+ // TODO: handle the case when error occured...
+
+ if (payload != null)
+ {
+ // TODO: handle deserialization error
+ TRead msg;
+ TryDeserialize(payload, out msg);
+
+ FireReadObserverOnNext(msg);
+
+ // Start a new read. The current one has already been delivered,
+ // so correct ordering of reads is assured.
+ StartReceiveMessage();
+ }
+ else
+ {
+ CompleteReadObserver();
+ }
+ }
+ }
+}
\ No newline at end of file
diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs
new file mode 100644
index 0000000000000000000000000000000000000000..d3a2be553fc6ac21cc1f476c5d7f16d9692be0d3
--- /dev/null
+++ b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs
@@ -0,0 +1,125 @@
+#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.Diagnostics;
+using System.Runtime.CompilerServices;
+using System.Runtime.InteropServices;
+using System.Threading;
+using System.Threading.Tasks;
+using Grpc.Core.Internal;
+using Grpc.Core.Utils;
+
+namespace Grpc.Core.Internal
+{
+ /// <summary>
+ /// Handles server side native call lifecycle.
+ /// </summary>
+ internal class AsyncCallServer<TRequest, TResponse> : AsyncCallBase<TResponse, TRequest>
+ {
+ readonly CompletionCallbackDelegate finishedServersideHandler;
+ readonly TaskCompletionSource<object> finishedServersideTcs = new TaskCompletionSource<object>();
+
+ public AsyncCallServer(Func<TResponse, byte[]> serializer, Func<byte[], TRequest> deserializer) : base(serializer, deserializer)
+ {
+ this.finishedServersideHandler = CreateBatchCompletionCallback(HandleFinishedServerside);
+ }
+
+ public void Initialize(CallSafeHandle call)
+ {
+ InitializeInternal(call);
+ }
+
+ /// <summary>
+ /// Starts a server side call. Currently, all server side calls are implemented as duplex
+ /// streaming call and they are adapted to the appropriate streaming arity.
+ /// </summary>
+ public Task ServerSideCallAsync(IObserver<TRequest> readObserver)
+ {
+ lock (myLock)
+ {
+ Preconditions.CheckNotNull(call);
+
+ started = true;
+ this.readObserver = readObserver;
+
+ call.StartServerSide(finishedServersideHandler);
+ StartReceiveMessage();
+ return finishedServersideTcs.Task;
+ }
+ }
+
+ /// <summary>
+ /// Sends a streaming response. Only one pending send action is allowed at any given time.
+ /// completionDelegate is called when the operation finishes.
+ /// </summary>
+ public void StartSendMessage(TResponse msg, AsyncCompletionDelegate completionDelegate)
+ {
+ StartSendMessageInternal(msg, completionDelegate);
+ }
+
+ /// <summary>
+ /// Sends call result status, also indicating server is done with streaming responses.
+ /// 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 completionDelegate)
+ {
+ lock (myLock)
+ {
+ Preconditions.CheckNotNull(completionDelegate, "Completion delegate cannot be null");
+ CheckSendingAllowed();
+
+ call.StartSendStatusFromServer(status, halfclosedHandler);
+ halfcloseRequested = true;
+ sendCompletionDelegate = completionDelegate;
+ }
+ }
+
+ /// <summary>
+ /// Handles the server side close completion.
+ /// </summary>
+ private void HandleFinishedServerside(bool wasError, BatchContextSafeHandleNotOwned ctx)
+ {
+ lock (myLock)
+ {
+ finished = true;
+
+ ReleaseResourcesIfPossible();
+ }
+ // TODO: handle error ...
+
+ finishedServersideTcs.SetResult(null);
+ }
+ }
+}
\ No newline at end of file
diff --git a/src/csharp/Grpc.Core/Internal/AsyncCompletion.cs b/src/csharp/Grpc.Core/Internal/AsyncCompletion.cs
new file mode 100644
index 0000000000000000000000000000000000000000..b78bb497fa71efdc0a01340f50bc990a8c454ecd
--- /dev/null
+++ b/src/csharp/Grpc.Core/Internal/AsyncCompletion.cs
@@ -0,0 +1,95 @@
+#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.Diagnostics;
+using System.Runtime.CompilerServices;
+using System.Runtime.InteropServices;
+using System.Threading;
+using System.Threading.Tasks;
+using Grpc.Core.Internal;
+using Grpc.Core.Utils;
+
+namespace Grpc.Core.Internal
+{
+ /// <summary>
+ /// If error != null, there's been an error or operation has been cancelled.
+ /// </summary>
+ internal delegate void AsyncCompletionDelegate(Exception error);
+
+ /// <summary>
+ /// Helper for transforming AsyncCompletionDelegate into full-fledged Task.
+ /// </summary>
+ internal class AsyncCompletionTaskSource
+ {
+ readonly TaskCompletionSource<object> tcs = new TaskCompletionSource<object>();
+ readonly AsyncCompletionDelegate completionDelegate;
+
+ public AsyncCompletionTaskSource()
+ {
+ completionDelegate = new AsyncCompletionDelegate(HandleCompletion);
+ }
+
+ public Task Task
+ {
+ get
+ {
+ return tcs.Task;
+ }
+ }
+
+ public AsyncCompletionDelegate CompletionDelegate
+ {
+ get
+ {
+ return completionDelegate;
+ }
+ }
+
+ private void HandleCompletion(Exception error)
+ {
+ if (error == null)
+ {
+ tcs.SetResult(null);
+ return;
+ }
+ if (error is OperationCanceledException)
+ {
+ tcs.SetCanceled();
+ return;
+ }
+ tcs.SetException(error);
+ }
+ }
+
+}
\ 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 1c0bc98f062db57fd27bca50e8ade5d966e32047..61566b54072bbb8ab11f9bde2240d4b3760bd0a9 100644
--- a/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs
+++ b/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs
@@ -1,5 +1,4 @@
#region Copyright notice and license
-
// Copyright 2015, Google Inc.
// All rights reserved.
//
@@ -30,7 +29,6 @@
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#endregion
-
using System;
using System.Diagnostics;
using System.Runtime.InteropServices;
@@ -38,14 +36,12 @@ using Grpc.Core;
namespace Grpc.Core.Internal
{
- //TODO: rename the delegate
- internal delegate void CompletionCallbackDelegate(GRPCOpError error, IntPtr batchContextPtr);
-
+ internal delegate void CompletionCallbackDelegate(GRPCOpError error,IntPtr batchContextPtr);
/// <summary>
/// grpc_call from <grpc/grpc.h>
/// </summary>
- internal class CallSafeHandle : SafeHandleZeroIsInvalid
- {
+ internal class CallSafeHandle : SafeHandleZeroIsInvalid
+ {
const UInt32 GRPC_WRITE_BUFFER_HINT = 1;
[DllImport("grpc_csharp_ext.dll")]
@@ -59,22 +55,22 @@ namespace Grpc.Core.Internal
[DllImport("grpc_csharp_ext.dll")]
static extern GRPCCallError grpcsharp_call_start_unary(CallSafeHandle call,
- [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback,
- byte[] send_buffer, UIntPtr send_buffer_len);
+ [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback,
+ byte[] send_buffer, UIntPtr send_buffer_len);
[DllImport("grpc_csharp_ext.dll")]
static extern void grpcsharp_call_blocking_unary(CallSafeHandle call, CompletionQueueSafeHandle dedicatedCq,
- [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback,
- byte[] send_buffer, UIntPtr send_buffer_len);
+ [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback,
+ byte[] send_buffer, UIntPtr send_buffer_len);
[DllImport("grpc_csharp_ext.dll")]
static extern GRPCCallError grpcsharp_call_start_client_streaming(CallSafeHandle call,
- [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback);
+ [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback);
[DllImport("grpc_csharp_ext.dll")]
static extern GRPCCallError grpcsharp_call_start_server_streaming(CallSafeHandle call,
- [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback,
- byte[] send_buffer, UIntPtr send_buffer_len);
+ [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback,
+ byte[] send_buffer, UIntPtr send_buffer_len);
[DllImport("grpc_csharp_ext.dll")]
static extern GRPCCallError grpcsharp_call_start_duplex_streaming(CallSafeHandle call,
@@ -82,28 +78,27 @@ namespace Grpc.Core.Internal
[DllImport("grpc_csharp_ext.dll")]
static extern GRPCCallError grpcsharp_call_send_message(CallSafeHandle call,
- [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback,
- byte[] send_buffer, UIntPtr send_buffer_len);
+ [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback,
+ byte[] send_buffer, UIntPtr send_buffer_len);
[DllImport("grpc_csharp_ext.dll")]
static extern GRPCCallError grpcsharp_call_send_close_from_client(CallSafeHandle call,
- [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback);
+ [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback);
[DllImport("grpc_csharp_ext.dll")]
static extern GRPCCallError grpcsharp_call_send_status_from_server(CallSafeHandle call, [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback, StatusCode statusCode, string statusMessage);
[DllImport("grpc_csharp_ext.dll")]
static extern GRPCCallError grpcsharp_call_recv_message(CallSafeHandle call,
- [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback);
+ [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback);
[DllImport("grpc_csharp_ext.dll")]
static extern GRPCCallError grpcsharp_call_start_serverside(CallSafeHandle call,
- [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback);
+ [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback);
[DllImport("grpc_csharp_ext.dll")]
static extern void grpcsharp_call_destroy(IntPtr call);
-
private CallSafeHandle()
{
}
@@ -115,12 +110,12 @@ namespace Grpc.Core.Internal
public void StartUnary(byte[] payload, CompletionCallbackDelegate callback)
{
- AssertCallOk(grpcsharp_call_start_unary(this, callback, payload, new UIntPtr((ulong) payload.Length)));
+ AssertCallOk(grpcsharp_call_start_unary(this, callback, payload, new UIntPtr((ulong)payload.Length)));
}
public void BlockingUnary(CompletionQueueSafeHandle dedicatedCq, byte[] payload, CompletionCallbackDelegate callback)
{
- grpcsharp_call_blocking_unary(this, dedicatedCq, callback, payload, new UIntPtr((ulong) payload.Length));
+ grpcsharp_call_blocking_unary(this, dedicatedCq, callback, payload, new UIntPtr((ulong)payload.Length));
}
public void StartClientStreaming(CompletionCallbackDelegate callback)
@@ -130,7 +125,7 @@ namespace Grpc.Core.Internal
public void StartServerStreaming(byte[] payload, CompletionCallbackDelegate callback)
{
- AssertCallOk(grpcsharp_call_start_server_streaming(this, callback, payload, new UIntPtr((ulong) payload.Length)));
+ AssertCallOk(grpcsharp_call_start_server_streaming(this, callback, payload, new UIntPtr((ulong)payload.Length)));
}
public void StartDuplexStreaming(CompletionCallbackDelegate callback)
@@ -140,7 +135,7 @@ namespace Grpc.Core.Internal
public void StartSendMessage(byte[] payload, CompletionCallbackDelegate callback)
{
- AssertCallOk(grpcsharp_call_send_message(this, callback, payload, new UIntPtr((ulong) payload.Length)));
+ AssertCallOk(grpcsharp_call_send_message(this, callback, payload, new UIntPtr((ulong)payload.Length)));
}
public void StartSendCloseFromClient(CompletionCallbackDelegate callback)
@@ -173,19 +168,20 @@ namespace Grpc.Core.Internal
AssertCallOk(grpcsharp_call_cancel_with_status(this, status.StatusCode, status.Detail));
}
- protected override bool ReleaseHandle()
- {
+ protected override bool ReleaseHandle()
+ {
grpcsharp_call_destroy(handle);
- return true;
- }
+ return true;
+ }
private static void AssertCallOk(GRPCCallError callError)
{
Trace.Assert(callError == GRPCCallError.GRPC_CALL_OK, "Status not GRPC_CALL_OK");
}
- private static UInt32 GetFlags(bool buffered) {
+ private static UInt32 GetFlags(bool buffered)
+ {
return buffered ? 0 : GRPC_WRITE_BUFFER_HINT;
}
- }
+ }
}
\ No newline at end of file
diff --git a/src/csharp/Grpc.Core/Internal/ClientStreamingInputObserver.cs b/src/csharp/Grpc.Core/Internal/ClientStreamingInputObserver.cs
index fb59e86e2d7f8e0d017abc4b26910ac8cc2fcb65..286c54f2c47811f458d4b1543b06007e7b836378 100644
--- a/src/csharp/Grpc.Core/Internal/ClientStreamingInputObserver.cs
+++ b/src/csharp/Grpc.Core/Internal/ClientStreamingInputObserver.cs
@@ -1,5 +1,4 @@
#region Copyright notice and license
-
// Copyright 2015, Google Inc.
// All rights reserved.
//
@@ -28,40 +27,40 @@
// 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 Grpc.Core.Internal;
namespace Grpc.Core.Internal
{
internal class ClientStreamingInputObserver<TWrite, TRead> : IObserver<TWrite>
- {
+ {
readonly AsyncCall<TWrite, TRead> call;
public ClientStreamingInputObserver(AsyncCall<TWrite, TRead> call)
- {
+ {
this.call = call;
- }
-
- public void OnCompleted()
- {
+ }
+ public void OnCompleted()
+ {
+ var taskSource = new AsyncCompletionTaskSource();
+ call.StartSendCloseFromClient(taskSource.CompletionDelegate);
// TODO: how bad is the Wait here?
- call.SendCloseFromClientAsync().Wait();
- }
+ taskSource.Task.Wait();
+ }
- public void OnError(Exception error)
- {
- throw new InvalidOperationException("This should never be called.");
- }
+ public void OnError(Exception error)
+ {
+ throw new InvalidOperationException("This should never be called.");
+ }
- public void OnNext(TWrite value)
- {
+ public void OnNext(TWrite value)
+ {
+ var taskSource = new AsyncCompletionTaskSource();
+ call.StartSendMessage(value, taskSource.CompletionDelegate);
// TODO: how bad is the Wait here?
- call.SendMessageAsync(value).Wait();
- }
- }
+ taskSource.Task.Wait();
+ }
+ }
}
-
diff --git a/src/csharp/Grpc.Core/Internal/CompletionQueueSafeHandle.cs b/src/csharp/Grpc.Core/Internal/CompletionQueueSafeHandle.cs
index 3f01fdbfd05c70d4ac130681d6af1ab91a6530c7..6bff923c55229166189b497d666b797436dc5d3b 100644
--- a/src/csharp/Grpc.Core/Internal/CompletionQueueSafeHandle.cs
+++ b/src/csharp/Grpc.Core/Internal/CompletionQueueSafeHandle.cs
@@ -1,5 +1,4 @@
#region Copyright notice and license
-
// Copyright 2015, Google Inc.
// All rights reserved.
//
@@ -28,9 +27,7 @@
// 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.InteropServices;
using System.Threading.Tasks;
@@ -40,8 +37,8 @@ namespace Grpc.Core.Internal
/// <summary>
/// grpc_completion_queue from <grpc/grpc.h>
/// </summary>
- internal class CompletionQueueSafeHandle : SafeHandleZeroIsInvalid
- {
+ internal class CompletionQueueSafeHandle : SafeHandleZeroIsInvalid
+ {
[DllImport("grpc_csharp_ext.dll")]
static extern CompletionQueueSafeHandle grpcsharp_completion_queue_create();
@@ -73,11 +70,11 @@ namespace Grpc.Core.Internal
grpcsharp_completion_queue_shutdown(this);
}
- protected override bool ReleaseHandle()
+ protected override bool ReleaseHandle()
{
grpcsharp_completion_queue_destroy(handle);
- return true;
- }
- }
+ return true;
+ }
+ }
}
diff --git a/src/csharp/Grpc.Core/Internal/ServerStreamingOutputObserver.cs b/src/csharp/Grpc.Core/Internal/ServerStreamingOutputObserver.cs
index 08d99214754bd9316a75f5a252fc0861694bbe7a..9873dc9c71077b0d8650199e288e743f8f348896 100644
--- a/src/csharp/Grpc.Core/Internal/ServerStreamingOutputObserver.cs
+++ b/src/csharp/Grpc.Core/Internal/ServerStreamingOutputObserver.cs
@@ -1,5 +1,4 @@
#region Copyright notice and license
-
// Copyright 2015, Google Inc.
// All rights reserved.
//
@@ -28,9 +27,7 @@
// 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 Grpc.Core.Internal;
@@ -40,32 +37,36 @@ namespace Grpc.Core.Internal
/// Observer that writes all arriving messages to a call abstraction (in blocking fashion)
/// and then halfcloses the call. Used for server-side call handling.
/// </summary>
- internal class ServerStreamingOutputObserver<TWrite, TRead> : IObserver<TWrite>
- {
- readonly AsyncCall<TWrite, TRead> call;
+ internal class ServerStreamingOutputObserver<TRequest, TResponse> : IObserver<TResponse>
+ {
+ readonly AsyncCallServer<TRequest, TResponse> call;
- public ServerStreamingOutputObserver(AsyncCall<TWrite, TRead> call)
- {
+ public ServerStreamingOutputObserver(AsyncCallServer<TRequest, TResponse> call)
+ {
this.call = call;
- }
+ }
- public void OnCompleted()
- {
+ public void OnCompleted()
+ {
+ var taskSource = new AsyncCompletionTaskSource();
+ call.StartSendStatusFromServer(new Status(StatusCode.OK, ""), taskSource.CompletionDelegate);
// TODO: how bad is the Wait here?
- call.SendStatusFromServerAsync(new Status(StatusCode.OK, "")).Wait();
- }
+ taskSource.Task.Wait();
+ }
- public void OnError(Exception error)
- {
+ public void OnError(Exception error)
+ {
// TODO: implement this...
- throw new InvalidOperationException("This should never be called.");
- }
+ throw new InvalidOperationException("This should never be called.");
+ }
- public void OnNext(TWrite value)
- {
+ public void OnNext(TResponse value)
+ {
+ var taskSource = new AsyncCompletionTaskSource();
+ call.StartSendMessage(value, taskSource.CompletionDelegate);
// TODO: how bad is the Wait here?
- call.SendMessageAsync(value).Wait();
- }
- }
+ taskSource.Task.Wait();
+ }
+ }
}
diff --git a/src/csharp/Grpc.Core/Internal/Timespec.cs b/src/csharp/Grpc.Core/Internal/Timespec.cs
index b191ecde94cc6e3e96cf966de8fe6813ab5b2628..e6efd66f13b9eb717d573f4469927d56c749d9f7 100644
--- a/src/csharp/Grpc.Core/Internal/Timespec.cs
+++ b/src/csharp/Grpc.Core/Internal/Timespec.cs
@@ -1,5 +1,4 @@
#region Copyright notice and license
-
// Copyright 2015, Google Inc.
// All rights reserved.
//
@@ -28,21 +27,19 @@
// 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.InteropServices;
using System.Threading;
namespace Grpc.Core.Internal
{
- /// <summary>
- /// gpr_timespec from grpc/support/time.h
- /// </summary>
- [StructLayout(LayoutKind.Sequential)]
- internal struct Timespec
- {
+ /// <summary>
+ /// gpr_timespec from grpc/support/time.h
+ /// </summary>
+ [StructLayout(LayoutKind.Sequential)]
+ internal struct Timespec
+ {
const int nanosPerSecond = 1000 * 1000 * 1000;
const int nanosPerTick = 100;
@@ -54,23 +51,22 @@ namespace Grpc.Core.Internal
[DllImport("grpc_csharp_ext.dll")]
static extern int gprsharp_sizeof_timespec();
-
// TODO: revisit this.
- // NOTE: on linux 64bit sizeof(gpr_timespec) = 16, on windows 32bit sizeof(gpr_timespec) = 8
+ // 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;
- public System.IntPtr tv_nsec;
+ public System.IntPtr tv_sec;
+ public System.IntPtr tv_nsec;
- /// <summary>
- /// Timespec a long time in the future.
- /// </summary>
- public static Timespec InfFuture
- {
- get
- {
+ /// <summary>
+ /// Timespec a long time in the future.
+ /// </summary>
+ public static Timespec InfFuture
+ {
+ get
+ {
return gprsharp_inf_future();
- }
- }
+ }
+ }
public static Timespec Now
{
@@ -92,7 +88,8 @@ namespace Grpc.Core.Internal
/// Creates a GPR deadline from current instant and given timeout.
/// </summary>
/// <returns>The from timeout.</returns>
- public static Timespec DeadlineFromTimeout(TimeSpan timeout) {
+ public static Timespec DeadlineFromTimeout(TimeSpan timeout)
+ {
if (timeout == Timeout.InfiniteTimeSpan)
{
return Timespec.InfFuture;
@@ -100,7 +97,8 @@ namespace Grpc.Core.Internal
return Timespec.Now.Add(timeout);
}
- public Timespec Add(TimeSpan timeSpan) {
+ public Timespec Add(TimeSpan timeSpan)
+ {
long nanos = tv_nsec.ToInt64() + (timeSpan.Ticks % TimeSpan.TicksPerSecond) * nanosPerTick;
long overflow_sec = (nanos > nanosPerSecond) ? 1 : 0;
@@ -109,6 +107,6 @@ namespace Grpc.Core.Internal
result.tv_sec = new IntPtr(tv_sec.ToInt64() + (timeSpan.Ticks / TimeSpan.TicksPerSecond) + overflow_sec);
return result;
}
- }
+ }
}
diff --git a/src/csharp/Grpc.Core/OperationFailedException.cs b/src/csharp/Grpc.Core/OperationFailedException.cs
new file mode 100644
index 0000000000000000000000000000000000000000..34a8c95a85c3790f49582bdcd0e1d627e837a4c7
--- /dev/null
+++ b/src/csharp/Grpc.Core/OperationFailedException.cs
@@ -0,0 +1,48 @@
+#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;
+
+namespace Grpc.Core
+{
+ /// <summary>
+ /// Thrown when gRPC operation fails.
+ /// </summary>
+ public class OperationFailedException : Exception
+ {
+ public OperationFailedException(string message) : base(message)
+ {
+ }
+ }
+}
+
diff --git a/src/csharp/Grpc.Core/ServerCallHandler.cs b/src/csharp/Grpc.Core/ServerCallHandler.cs
index 289f97aecee2777f6a7718bed807d1cacb3278c4..3eb8422f575e3b4f745e03f13856c254d76de10b 100644
--- a/src/csharp/Grpc.Core/ServerCallHandler.cs
+++ b/src/csharp/Grpc.Core/ServerCallHandler.cs
@@ -32,7 +32,9 @@
#endregion
using System;
+using System.Linq;
using Grpc.Core.Internal;
+using Grpc.Core.Utils;
namespace Grpc.Core
{
@@ -54,17 +56,17 @@ namespace Grpc.Core
public void StartCall(string methodName, CallSafeHandle call, CompletionQueueSafeHandle cq)
{
- var asyncCall = new AsyncCall<TResponse, TRequest>(
+ var asyncCall = new AsyncCallServer<TRequest, TResponse>(
method.ResponseMarshaller.Serializer,
method.RequestMarshaller.Deserializer);
- asyncCall.InitializeServer(call);
+ asyncCall.Initialize(call);
- var finishedTask = asyncCall.ServerSideUnaryRequestCallAsync();
+ var requestObserver = new RecordingObserver<TRequest>();
+ var finishedTask = asyncCall.ServerSideCallAsync(requestObserver);
- var request = asyncCall.ReceiveMessageAsync().Result;
-
- var responseObserver = new ServerStreamingOutputObserver<TResponse, TRequest>(asyncCall);
+ var request = requestObserver.ToList().Result.Single();
+ var responseObserver = new ServerStreamingOutputObserver<TRequest, TResponse>(asyncCall);
handler(request, responseObserver);
finishedTask.Wait();
@@ -85,15 +87,15 @@ namespace Grpc.Core
public void StartCall(string methodName, CallSafeHandle call, CompletionQueueSafeHandle cq)
{
- var asyncCall = new AsyncCall<TResponse, TRequest>(
+ var asyncCall = new AsyncCallServer<TRequest, TResponse>(
method.ResponseMarshaller.Serializer,
method.RequestMarshaller.Deserializer);
- asyncCall.InitializeServer(call);
+ asyncCall.Initialize(call);
- var responseObserver = new ServerStreamingOutputObserver<TResponse, TRequest>(asyncCall);
+ var responseObserver = new ServerStreamingOutputObserver<TRequest,TResponse>(asyncCall);
var requestObserver = handler(responseObserver);
- var finishedTask = asyncCall.ServerSideStreamingRequestCallAsync(requestObserver);
+ var finishedTask = asyncCall.ServerSideCallAsync(requestObserver);
finishedTask.Wait();
}
}
@@ -103,17 +105,15 @@ namespace Grpc.Core
public void StartCall(string methodName, CallSafeHandle call, CompletionQueueSafeHandle cq)
{
// We don't care about the payload type here.
- AsyncCall<byte[], byte[]> asyncCall = new AsyncCall<byte[], byte[]>(
+ var asyncCall = new AsyncCallServer<byte[], byte[]>(
(payload) => payload, (payload) => payload);
+ asyncCall.Initialize(call);
- asyncCall.InitializeServer(call);
-
- var finishedTask = asyncCall.ServerSideStreamingRequestCallAsync(new NullObserver<byte[]>());
+ var finishedTask = asyncCall.ServerSideCallAsync(new NullObserver<byte[]>());
- // TODO: this makes the call finish before all reads can be done which causes trouble
- // in AsyncCall.HandleReadFinished callback. Revisit this.
- asyncCall.SendStatusFromServerAsync(new Status(StatusCode.Unimplemented, "No such method.")).Wait();
+ // TODO: check result of the completion status.
+ asyncCall.StartSendStatusFromServer(new Status(StatusCode.Unimplemented, "No such method."), new AsyncCompletionDelegate((error) => {}));
finishedTask.Wait();
}
diff --git a/src/csharp/Grpc.Core/Status.cs b/src/csharp/Grpc.Core/Status.cs
index 5ea1df7b481c2bec149fb0550f04b152a427b336..080bbdc2f5b3544b5917126837490732ae1a3f9f 100644
--- a/src/csharp/Grpc.Core/Status.cs
+++ b/src/csharp/Grpc.Core/Status.cs
@@ -1,5 +1,4 @@
#region Copyright notice and license
-
// Copyright 2015, Google Inc.
// All rights reserved.
//
@@ -28,7 +27,6 @@
// 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;
@@ -36,34 +34,40 @@ using System.Runtime.InteropServices;
namespace Grpc.Core
{
- /// <summary>
- /// Represents RPC result.
- /// </summary>
- public struct Status
- {
- readonly StatusCode statusCode;
- readonly string detail;
+ /// <summary>
+ /// Represents RPC result.
+ /// </summary>
+ public struct Status
+ {
+ readonly StatusCode statusCode;
+ readonly string detail;
- public Status(StatusCode statusCode, string detail)
- {
- this.statusCode = statusCode;
- this.detail = detail;
- }
+ public Status(StatusCode statusCode, string detail)
+ {
+ this.statusCode = statusCode;
+ this.detail = detail;
+ }
- public StatusCode StatusCode
- {
- get
- {
- return statusCode;
- }
- }
+ /// <summary>
+ /// Gets the gRPC status code. OK indicates success, all other values indicate an error.
+ /// </summary>
+ public StatusCode StatusCode
+ {
+ get
+ {
+ return statusCode;
+ }
+ }
- public string Detail
- {
- get
- {
- return detail;
- }
- }
- }
+ /// <summary>
+ /// Gets the detail.
+ /// </summary>
+ public string Detail
+ {
+ get
+ {
+ return detail;
+ }
+ }
+ }
}
diff --git a/src/csharp/Grpc.Core/Utils/Preconditions.cs b/src/csharp/Grpc.Core/Utils/Preconditions.cs
new file mode 100644
index 0000000000000000000000000000000000000000..b17ce4211783701b6b4ab8d6e9fb84057fe0fc75
--- /dev/null
+++ b/src/csharp/Grpc.Core/Utils/Preconditions.cs
@@ -0,0 +1,113 @@
+#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.Threading.Tasks;
+using System.Collections.Generic;
+using System.Collections.Concurrent;
+using System.Diagnostics;
+
+namespace Grpc.Core.Utils
+{
+ public static class Preconditions
+ {
+ /// <summary>
+ /// Throws ArgumentException if condition is false.
+ /// </summary>
+ public static void CheckArgument(bool condition)
+ {
+ if (!condition)
+ {
+ throw new ArgumentException();
+ }
+ }
+
+ /// <summary>
+ /// Throws ArgumentException with given message if condition is false.
+ /// </summary>
+ public static void CheckArgument(bool condition, string errorMessage)
+ {
+ if (!condition)
+ {
+ throw new ArgumentException(errorMessage);
+ }
+ }
+
+ /// <summary>
+ /// Throws NullReferenceException if reference is null.
+ /// </summary>
+ public static T CheckNotNull<T> (T reference)
+ {
+ if (reference == null)
+ {
+ throw new NullReferenceException();
+ }
+ return reference;
+ }
+
+ /// <summary>
+ /// Throws NullReferenceException with given message if reference is null.
+ /// </summary>
+ public static T CheckNotNull<T> (T reference, string errorMessage)
+ {
+ if (reference == null)
+ {
+ throw new NullReferenceException(errorMessage);
+ }
+ return reference;
+ }
+
+ /// <summary>
+ /// Throws InvalidOperationException if condition is false.
+ /// </summary>
+ public static void CheckState(bool condition)
+ {
+ if (!condition)
+ {
+ throw new InvalidOperationException();
+ }
+ }
+
+ /// <summary>
+ /// Throws InvalidOperationException with given message if condition is false.
+ /// </summary>
+ public static void CheckState(bool condition, string errorMessage)
+ {
+ if (!condition)
+ {
+ throw new InvalidOperationException(errorMessage);
+ }
+ }
+ }
+}
+
diff --git a/src/csharp/Grpc.Examples.MathClient/MathClient.cs b/src/csharp/Grpc.Examples.MathClient/MathClient.cs
index 95a4678bb8f63e623bd0c3a3c78688922fe7e07a..f5956bd33e008cdafe2e0116d76640645e793570 100644
--- a/src/csharp/Grpc.Examples.MathClient/MathClient.cs
+++ b/src/csharp/Grpc.Examples.MathClient/MathClient.cs
@@ -1,5 +1,4 @@
#region Copyright notice and license
-
// Copyright 2015, Google Inc.
// All rights reserved.
//
@@ -28,9 +27,7 @@
// 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.InteropServices;
using System.Threading;
@@ -38,25 +35,25 @@ using Grpc.Core;
namespace math
{
- class MathClient
+ class MathClient
{
- public static void Main (string[] args)
- {
+ public static void Main(string[] args)
+ {
GrpcEnvironment.Initialize();
- using (Channel channel = new Channel("127.0.0.1:23456"))
- {
- MathGrpc.IMathServiceClient stub = new MathGrpc.MathServiceClientStub(channel);
- MathExamples.DivExample(stub);
+ using (Channel channel = new Channel("127.0.0.1:23456"))
+ {
+ MathGrpc.IMathServiceClient stub = new MathGrpc.MathServiceClientStub(channel);
+ MathExamples.DivExample(stub);
MathExamples.FibExample(stub);
- MathExamples.SumExample(stub);
+ MathExamples.SumExample(stub);
- MathExamples.DivManyExample(stub);
- }
+ MathExamples.DivManyExample(stub);
+ }
GrpcEnvironment.Shutdown();
- }
- }
+ }
+ }
}
diff --git a/src/csharp/Grpc.Examples/MathExamples.cs b/src/csharp/Grpc.Examples/MathExamples.cs
index 97c91b1b1b586248b03e4d94200e7a9629627fef..134270f6f76c8e20c367d132c9e5bffe009df3db 100644
--- a/src/csharp/Grpc.Examples/MathExamples.cs
+++ b/src/csharp/Grpc.Examples/MathExamples.cs
@@ -1,5 +1,4 @@
#region Copyright notice and license
-
// Copyright 2015, Google Inc.
// All rights reserved.
//
@@ -28,7 +27,6 @@
// 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;
@@ -39,59 +37,63 @@ using Grpc.Core.Utils;
namespace math
{
- public static class MathExamples
- {
- public static void DivExample(MathGrpc.IMathServiceClient stub)
- {
- DivReply result = stub.Div(new DivArgs.Builder { Dividend = 10, Divisor = 3 }.Build());
- Console.WriteLine("Div Result: " + result);
- }
-
- public static void DivAsyncExample(MathGrpc.IMathServiceClient stub)
- {
- Task<DivReply> call = stub.DivAsync(new DivArgs.Builder { Dividend = 4, Divisor = 5 }.Build());
- DivReply result = call.Result;
- Console.WriteLine(result);
- }
-
- public static void DivAsyncWithCancellationExample(MathGrpc.IMathServiceClient stub)
- {
- Task<DivReply> call = stub.DivAsync(new DivArgs.Builder { Dividend = 4, Divisor = 5 }.Build());
- DivReply result = call.Result;
- Console.WriteLine(result);
- }
-
- public static void FibExample(MathGrpc.IMathServiceClient stub)
- {
+ public static class MathExamples
+ {
+ public static void DivExample(MathGrpc.IMathServiceClient stub)
+ {
+ DivReply result = stub.Div(new DivArgs.Builder { Dividend = 10, Divisor = 3 }.Build());
+ Console.WriteLine("Div Result: " + result);
+ }
+
+ public static void DivAsyncExample(MathGrpc.IMathServiceClient stub)
+ {
+ Task<DivReply> call = stub.DivAsync(new DivArgs.Builder { Dividend = 4, Divisor = 5 }.Build());
+ DivReply result = call.Result;
+ Console.WriteLine(result);
+ }
+
+ public static void DivAsyncWithCancellationExample(MathGrpc.IMathServiceClient stub)
+ {
+ Task<DivReply> call = stub.DivAsync(new DivArgs.Builder { Dividend = 4, Divisor = 5 }.Build());
+ DivReply result = call.Result;
+ Console.WriteLine(result);
+ }
+
+ public static void FibExample(MathGrpc.IMathServiceClient stub)
+ {
var recorder = new RecordingObserver<Num>();
stub.Fib(new FibArgs.Builder { Limit = 5 }.Build(), recorder);
- List<Num> numbers = recorder.ToList().Result;
+ List<Num> numbers = recorder.ToList().Result;
Console.WriteLine("Fib Result: " + string.Join("|", recorder.ToList().Result));
- }
+ }
- public static void SumExample(MathGrpc.IMathServiceClient stub)
- {
- List<Num> numbers = new List<Num>{new Num.Builder { Num_ = 1 }.Build(),
- new Num.Builder { Num_ = 2 }.Build(),
- new Num.Builder { Num_ = 3 }.Build()};
+ public static void SumExample(MathGrpc.IMathServiceClient stub)
+ {
+ List<Num> numbers = new List<Num>
+ {new Num.Builder { Num_ = 1 }.Build(),
+ new Num.Builder { Num_ = 2 }.Build(),
+ new Num.Builder { Num_ = 3 }.Build()
+ };
var res = stub.Sum();
- foreach (var num in numbers) {
+ foreach (var num in numbers)
+ {
res.Inputs.OnNext(num);
}
res.Inputs.OnCompleted();
- Console.WriteLine("Sum Result: " + res.Task.Result);
- }
+ Console.WriteLine("Sum Result: " + res.Task.Result);
+ }
- public static void DivManyExample(MathGrpc.IMathServiceClient stub)
- {
- List<DivArgs> divArgsList = new List<DivArgs>{
- new DivArgs.Builder { Dividend = 10, Divisor = 3 }.Build(),
- new DivArgs.Builder { Dividend = 100, Divisor = 21 }.Build(),
- new DivArgs.Builder { Dividend = 7, Divisor = 2 }.Build()
- };
+ public static void DivManyExample(MathGrpc.IMathServiceClient stub)
+ {
+ List<DivArgs> divArgsList = new List<DivArgs>
+ {
+ new DivArgs.Builder { Dividend = 10, Divisor = 3 }.Build(),
+ new DivArgs.Builder { Dividend = 100, Divisor = 21 }.Build(),
+ new DivArgs.Builder { Dividend = 7, Divisor = 2 }.Build()
+ };
var recorder = new RecordingObserver<DivReply>();
@@ -102,30 +104,30 @@ namespace math
}
inputs.OnCompleted();
- Console.WriteLine("DivMany Result: " + string.Join("|", recorder.ToList().Result));
- }
+ Console.WriteLine("DivMany Result: " + string.Join("|", recorder.ToList().Result));
+ }
- public static void DependendRequestsExample(MathGrpc.IMathServiceClient stub)
- {
- var numberList = new List<Num>
- { new Num.Builder{ Num_ = 1 }.Build(),
- new Num.Builder{ Num_ = 2 }.Build(), new Num.Builder{ Num_ = 3 }.Build()
- };
+ public static void DependendRequestsExample(MathGrpc.IMathServiceClient stub)
+ {
+ var numberList = new List<Num>
+ { new Num.Builder{ Num_ = 1 }.Build(),
+ new Num.Builder{ Num_ = 2 }.Build(), new Num.Builder{ Num_ = 3 }.Build()
+ };
- numberList.ToObservable();
+ numberList.ToObservable();
- //IObserver<Num> numbers;
- //Task<Num> call = stub.Sum(out numbers);
- //foreach (var num in numberList)
- //{
- // numbers.OnNext(num);
- //}
- //numbers.OnCompleted();
+ //IObserver<Num> numbers;
+ //Task<Num> call = stub.Sum(out numbers);
+ //foreach (var num in numberList)
+ //{
+ // numbers.OnNext(num);
+ //}
+ //numbers.OnCompleted();
- //Num sum = call.Result;
+ //Num sum = call.Result;
- //DivReply result = stub.Div(new DivArgs.Builder { Dividend = sum.Num_, Divisor = numberList.Count }.Build());
- }
- }
+ //DivReply result = stub.Div(new DivArgs.Builder { Dividend = sum.Num_, Divisor = numberList.Count }.Build());
+ }
+ }
}
diff --git a/src/ruby/grpc.gemspec b/src/ruby/grpc.gemspec
index ed26fef4a97a3e381d30cf9503dc62f2a240b34e..45cbacfeb04de3cf6ca43bbf2ee227cf1861e4e4 100755
--- a/src/ruby/grpc.gemspec
+++ b/src/ruby/grpc.gemspec
@@ -21,14 +21,10 @@ Gem::Specification.new do |s|
s.require_paths = ['lib']
s.platform = Gem::Platform::RUBY
- s.add_dependency 'faraday', '~> 0.9'
s.add_dependency 'google-protobuf', '~> 3.0.0alpha.1.1'
s.add_dependency 'googleauth', '~> 0.1'
s.add_dependency 'logging', '~> 1.8'
- s.add_dependency 'jwt', '~> 1.2.1'
s.add_dependency 'minitest', '~> 5.4' # reqd for interop tests
- s.add_dependency 'multi_json', '1.10.1'
- s.add_dependency 'signet', '~> 0.6.0'
s.add_dependency 'xray', '~> 1.1'
s.add_development_dependency 'bundler', '~> 1.7'
diff --git a/src/ruby/lib/grpc/generic/client_stub.rb b/src/ruby/lib/grpc/generic/client_stub.rb
index f234984eec3a4f4bde3d0d7cfe5528e56c1fcd65..01328d4a5bc68895716942b1f360ed0c9b911d7e 100644
--- a/src/ruby/lib/grpc/generic/client_stub.rb
+++ b/src/ruby/lib/grpc/generic/client_stub.rb
@@ -39,6 +39,25 @@ module GRPC
# Default deadline is 5 seconds.
DEFAULT_DEADLINE = 5
+ # setup_channel is used by #initialize to constuct a channel from its
+ # arguments.
+ def self.setup_channel(alt_chan, host, creds, **kw)
+ unless alt_chan.nil?
+ fail(TypeError, '!Channel') unless alt_chan.is_a?(Core::Channel)
+ return alt_chan
+ end
+ return Core::Channel.new(host, kw) if creds.nil?
+ fail(TypeError, '!Credentials') unless creds.is_a?(Core::Credentials)
+ Core::Channel.new(host, kw, creds)
+ end
+
+ # check_update_metadata is used by #initialize verify that it's a Proc.
+ def self.check_update_metadata(update_metadata)
+ return update_metadata if update_metadata.nil?
+ fail(TypeError, '!is_a?Proc') unless update_metadata.is_a?(Proc)
+ update_metadata
+ end
+
# Creates a new ClientStub.
#
# Minimally, a stub is created with the just the host of the gRPC service
@@ -73,40 +92,17 @@ module GRPC
# @param update_metadata a func that updates metadata as described above
# @param kw [KeywordArgs]the channel arguments
def initialize(host, q,
- channel_override:nil,
+ channel_override: nil,
deadline: DEFAULT_DEADLINE,
creds: nil,
update_metadata: nil,
**kw)
- unless q.is_a? Core::CompletionQueue
- fail(ArgumentError, 'not a CompletionQueue')
- end
+ fail(TypeError, '!CompletionQueue') unless q.is_a?(Core::CompletionQueue)
@queue = q
-
- # set the channel instance
- if !channel_override.nil?
- ch = channel_override
- fail(ArgumentError, 'not a Channel') unless ch.is_a? Core::Channel
- else
- if creds.nil?
- ch = Core::Channel.new(host, kw)
- elsif !creds.is_a?(Core::Credentials)
- fail(ArgumentError, 'not a Credentials')
- else
- ch = Core::Channel.new(host, kw, creds)
- end
- end
- @ch = ch
-
- @update_metadata = nil
- unless update_metadata.nil?
- unless update_metadata.is_a? Proc
- fail(ArgumentError, 'update_metadata is not a Proc')
- end
- @update_metadata = update_metadata
- end
-
- @host = host
+ @ch = ClientStub.setup_channel(channel_override, host, creds, **kw)
+ @update_metadata = ClientStub.check_update_metadata(update_metadata)
+ alt_host = kw[Core::Channel::SSL_TARGET]
+ @host = alt_host.nil? ? host : alt_host
@deadline = deadline
end
@@ -400,12 +396,7 @@ module GRPC
# @param deadline [TimeConst]
def new_active_call(ch, marshal, unmarshal, deadline = nil)
absolute_deadline = Core::TimeConsts.from_relative_time(deadline)
- # It should be OK to to pass the hostname:port to create_call, but at
- # the moment this fails a security check. This will be corrected.
- #
- # TODO: # remove this after create_call is updated
- host = @host.split(':')[0]
- call = @ch.create_call(ch, host, absolute_deadline)
+ call = @ch.create_call(ch, @host, absolute_deadline)
ActiveCall.new(call, @queue, marshal, unmarshal, absolute_deadline,
started: false)
end
diff --git a/test/compiler/python_plugin_test.py b/test/compiler/python_plugin_test.py
index f16682862c32a76ac635f2e95390dede819140f3..9cf3c624c05d3be3731bc1c60f164971b0da7a8c 100644
--- a/test/compiler/python_plugin_test.py
+++ b/test/compiler/python_plugin_test.py
@@ -32,8 +32,10 @@ import contextlib
import errno
import itertools
import os
+import shutil
import subprocess
import sys
+import tempfile
import time
import unittest
@@ -55,8 +57,8 @@ DOES_NOT_MATTER_DELAY = 0
NO_DELAY = 0
LONG_DELAY = 1
-# Assigned in __main__.
-_build_mode = None
+# Build mode environment variable set by tools/run_tests/run_tests.py.
+_build_mode = os.environ['CONFIG']
class _ServicerMethods(object):
@@ -227,24 +229,26 @@ class PythonPluginTest(unittest.TestCase):
protoc_command = 'protoc'
# Ensure that the output directory exists.
- outdir = '../../gens/test/compiler/python'
- try:
- os.makedirs(outdir)
- except OSError as exception:
- if exception.errno != errno.EEXIST:
- raise
+ self.outdir = tempfile.mkdtemp()
# Invoke protoc with the plugin.
cmd = [
protoc_command,
'--plugin=protoc-gen-python-grpc=%s' % protoc_plugin_filename,
'-I %s' % os.path.dirname(test_proto_filename),
- '--python_out=%s' % outdir,
- '--python-grpc_out=%s' % outdir,
+ '--python_out=%s' % self.outdir,
+ '--python-grpc_out=%s' % self.outdir,
os.path.basename(test_proto_filename),
]
subprocess.call(' '.join(cmd), shell=True)
- sys.path.append(outdir)
+ sys.path.append(self.outdir)
+
+ def tearDown(self):
+ try:
+ shutil.rmtree(self.outdir)
+ except OSError as exc:
+ if exc.errno != errno.ENOENT:
+ raise
# TODO(atash): Figure out which of theses tests is hanging flakily with small
# probability.
@@ -296,6 +300,8 @@ class PythonPluginTest(unittest.TestCase):
with self.assertRaises(exceptions.ExpirationError):
response_future.result()
+ @unittest.skip('TODO(atash,nathaniel): figure out why this flakily hangs '
+ 'forever and fix.')
def testUnaryCallAsyncCancelled(self):
import test_pb2 # pylint: disable=g-import-not-at-top
request = test_pb2.SimpleRequest(response_size=13)
@@ -325,6 +331,8 @@ class PythonPluginTest(unittest.TestCase):
expected_response, response = check
self.assertEqual(expected_response, response)
+ @unittest.skip('TODO(atash,nathaniel): figure out why this flakily hangs '
+ 'forever and fix.')
def testStreamingOutputCallExpired(self):
import test_pb2 # pylint: disable=g-import-not-at-top
request = StreamingOutputRequest(test_pb2)
@@ -335,6 +343,8 @@ class PythonPluginTest(unittest.TestCase):
with self.assertRaises(exceptions.ExpirationError):
list(responses)
+ @unittest.skip('TODO(atash,nathaniel): figure out why this flakily hangs '
+ 'forever and fix.')
def testStreamingOutputCallCancelled(self):
import test_pb2 # pylint: disable=g-import-not-at-top
request = StreamingOutputRequest(test_pb2)
@@ -359,6 +369,8 @@ class PythonPluginTest(unittest.TestCase):
with self.assertRaises(exceptions.ServicerError):
next(responses)
+ @unittest.skip('TODO(atash,nathaniel): figure out why this flakily hangs '
+ 'forever and fix.')
def testStreamingInputCall(self):
import test_pb2 # pylint: disable=g-import-not-at-top
with _CreateService(test_pb2, NO_DELAY) as (servicer, stub, unused_server):
@@ -426,6 +438,8 @@ class PythonPluginTest(unittest.TestCase):
expected_response, response = check
self.assertEqual(expected_response, response)
+ @unittest.skip('TODO(atash,nathaniel): figure out why this flakily hangs '
+ 'forever and fix.')
def testFullDuplexCallExpired(self):
import test_pb2 # pylint: disable=g-import-not-at-top
request = FullDuplexRequest(test_pb2)
@@ -436,6 +450,8 @@ class PythonPluginTest(unittest.TestCase):
with self.assertRaises(exceptions.ExpirationError):
list(responses)
+ @unittest.skip('TODO(atash,nathaniel): figure out why this flakily hangs '
+ 'forever and fix.')
def testFullDuplexCallCancelled(self):
import test_pb2 # pylint: disable=g-import-not-at-top
with _CreateService(test_pb2, NO_DELAY) as (servicer, stub, unused_server):
@@ -459,6 +475,8 @@ class PythonPluginTest(unittest.TestCase):
with self.assertRaises(exceptions.ServicerError):
next(responses)
+ @unittest.skip('TODO(atash,nathaniel): figure out why this flakily hangs '
+ 'forever and fix.')
def testHalfDuplexCall(self):
import test_pb2 # pylint: disable=g-import-not-at-top
with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as (
@@ -502,14 +520,4 @@ class PythonPluginTest(unittest.TestCase):
if __name__ == '__main__':
os.chdir(os.path.dirname(sys.argv[0]))
- parser = argparse.ArgumentParser(
- description='Run Python compiler plugin test.')
- parser.add_argument(
- '--build_mode', dest='build_mode', type=str, default='dbg',
- help='The build mode of the targets to test, e.g. "dbg", "opt", "asan", '
- 'etc.')
- parser.add_argument('--port', dest='port', type=int, default=0)
- args, remainder = parser.parse_known_args()
- _build_mode = args.build_mode
- sys.argv[1:] = remainder
unittest.main()
diff --git a/tools/run_tests/python_tests.json b/tools/run_tests/python_tests.json
index 9e5b1365e6af4bd294893bdb693613482b6a9204..4b43ee8357cc15b611ccb042d949390148193e4e 100755
--- a/tools/run_tests/python_tests.json
+++ b/tools/run_tests/python_tests.json
@@ -1,18 +1,50 @@
[
- "grpc._adapter._blocking_invocation_inline_service_test",
- "grpc._adapter._c_test",
- "grpc._adapter._event_invocation_synchronous_event_service_test",
- "grpc._adapter._future_invocation_asynchronous_event_service_test",
- "grpc._adapter._links_test",
- "grpc._adapter._lonely_rear_link_test",
- "grpc._adapter._low_test",
- "grpc.early_adopter.implementations_test",
- "grpc.framework.assembly.implementations_test",
- "grpc.framework.base.packets.implementations_test",
- "grpc.framework.face.blocking_invocation_inline_service_test",
- "grpc.framework.face.event_invocation_synchronous_event_service_test",
- "grpc.framework.face.future_invocation_asynchronous_event_service_test",
- "grpc.framework.foundation._later_test",
- "grpc.framework.foundation._logging_pool_test"
+ {
+ "file": "test/compiler/python_plugin_test.py"
+ },
+ {
+ "module": "grpc._adapter._blocking_invocation_inline_service_test"
+ },
+ {
+ "module": "grpc._adapter._c_test"
+ },
+ {
+ "module": "grpc._adapter._event_invocation_synchronous_event_service_test"
+ },
+ {
+ "module": "grpc._adapter._future_invocation_asynchronous_event_service_test"
+ },
+ {
+ "module": "grpc._adapter._links_test"
+ },
+ {
+ "module": "grpc._adapter._lonely_rear_link_test"
+ },
+ {
+ "module": "grpc._adapter._low_test"
+ },
+ {
+ "module": "grpc.early_adopter.implementations_test"
+ },
+ {
+ "module": "grpc.framework.assembly.implementations_test"
+ },
+ {
+ "module": "grpc.framework.base.packets.implementations_test"
+ },
+ {
+ "module": "grpc.framework.face.blocking_invocation_inline_service_test"
+ },
+ {
+ "module": "grpc.framework.face.event_invocation_synchronous_event_service_test"
+ },
+ {
+ "module": "grpc.framework.face.future_invocation_asynchronous_event_service_test"
+ },
+ {
+ "module": "grpc.framework.foundation._later_test"
+ },
+ {
+ "module": "grpc.framework.foundation._logging_pool_test"
+ }
]
-
diff --git a/tools/run_tests/run_python.sh b/tools/run_tests/run_python.sh
index 403862b0a0565c317ad7ac1347dc5f42e1c69365..fa1497aee478a8e2b262effc1f729a03704eb517 100755
--- a/tools/run_tests/run_python.sh
+++ b/tools/run_tests/run_python.sh
@@ -36,4 +36,4 @@ cd $(dirname $0)/../..
root=`pwd`
export LD_LIBRARY_PATH=$root/libs/opt
source python2.7_virtual_environment/bin/activate
-python2.7 -B -m $*
+python2.7 -B $*
diff --git a/tools/run_tests/run_tests.py b/tools/run_tests/run_tests.py
index baad727e51ed8022572335b2245cea79d24f6059..aee19cdc42d6ec9b106ce39d045dc679922584a7 100755
--- a/tools/run_tests/run_tests.py
+++ b/tools/run_tests/run_tests.py
@@ -170,11 +170,16 @@ class PythonLanguage(object):
self._tests = json.load(f)
def test_specs(self, config, travis):
- return [config.job_spec(['tools/run_tests/run_python.sh', test], None)
- for test in self._tests]
+ modules = [config.job_spec(['tools/run_tests/run_python.sh', '-m',
+ test['module']], None)
+ for test in self._tests if 'module' in test]
+ files = [config.job_spec(['tools/run_tests/run_python.sh',
+ test['file']], None)
+ for test in self._tests if 'file' in test]
+ return files + modules
def make_targets(self):
- return ['static_c']
+ return ['static_c', 'grpc_python_plugin']
def build_steps(self):
return [['tools/run_tests/build_python.sh']]