diff --git a/src/csharp/.gitignore b/src/csharp/.gitignore
index dbaf60de0cf6dfbeb27fa4948a42a70982b10ef6..c064ff1fa6ddddd02590f05d90a7dcf45931c925 100644
--- a/src/csharp/.gitignore
+++ b/src/csharp/.gitignore
@@ -1,4 +1,5 @@
*.userprefs
+*.csproj.user
StyleCop.Cache
test-results
packages
diff --git a/src/csharp/Grpc.Core.Tests/ClientServerTest.cs b/src/csharp/Grpc.Core.Tests/ClientServerTest.cs
index 3da9e33e5364b1a5ad92fa677ea13df9773bd2d1..9e510e82a6192d5482adfdf47526becb28ca3829 100644
--- a/src/csharp/Grpc.Core.Tests/ClientServerTest.cs
+++ b/src/csharp/Grpc.Core.Tests/ClientServerTest.cs
@@ -44,16 +44,18 @@ namespace Grpc.Core.Tests
{
public class ClientServerTest
{
- string host = "localhost";
+ const string Host = "localhost";
+ const string ServiceName = "/tests.Test";
- string serviceName = "/tests.Test";
-
- Method<string, string> unaryEchoStringMethod = new Method<string, string>(
+ static readonly Method<string, string> UnaryEchoStringMethod = new Method<string, string>(
MethodType.Unary,
"/tests.Test/UnaryEchoString",
Marshallers.StringMarshaller,
Marshallers.StringMarshaller);
+ static readonly ServerServiceDefinition ServiceDefinition = ServerServiceDefinition.CreateBuilder(ServiceName)
+ .AddMethod(UnaryEchoStringMethod, HandleUnaryEchoString).Build();
+
[TestFixtureSetUp]
public void Init()
{
@@ -69,21 +71,39 @@ namespace Grpc.Core.Tests
[Test]
public void UnaryCall()
{
- Server server = new Server();
- server.AddServiceDefinition(
- ServerServiceDefinition.CreateBuilder(serviceName)
- .AddMethod(unaryEchoStringMethod, HandleUnaryEchoString).Build());
-
- int port = server.AddListeningPort(host + ":0");
+ var server = new Server();
+ server.AddServiceDefinition(ServiceDefinition);
+ int port = server.AddListeningPort(Host + ":0");
server.Start();
- using (Channel channel = new Channel(host + ":" + port))
+ using (Channel channel = new Channel(Host + ":" + port))
{
- var call = new Call<string, string>(serviceName, unaryEchoStringMethod, channel, Metadata.Empty);
-
+ var call = new Call<string, string>(ServiceName, UnaryEchoStringMethod, channel, Metadata.Empty);
Assert.AreEqual("ABC", Calls.BlockingUnaryCall(call, "ABC", default(CancellationToken)));
+ }
- Assert.AreEqual("abcdef", Calls.BlockingUnaryCall(call, "abcdef", default(CancellationToken)));
+ server.ShutdownAsync().Wait();
+ }
+
+ [Test]
+ public void CallOnDisposedChannel()
+ {
+ var server = new Server();
+ server.AddServiceDefinition(ServiceDefinition);
+ int port = server.AddListeningPort(Host + ":0");
+ server.Start();
+
+ Channel channel = new Channel(Host + ":" + port);
+ channel.Dispose();
+
+ var call = new Call<string, string>(ServiceName, UnaryEchoStringMethod, channel, Metadata.Empty);
+ try
+ {
+ Calls.BlockingUnaryCall(call, "ABC", default(CancellationToken));
+ Assert.Fail();
+ }
+ catch (ObjectDisposedException e)
+ {
}
server.ShutdownAsync().Wait();
@@ -92,18 +112,15 @@ namespace Grpc.Core.Tests
[Test]
public void UnaryCallPerformance()
{
- Server server = new Server();
- server.AddServiceDefinition(
- ServerServiceDefinition.CreateBuilder(serviceName)
- .AddMethod(unaryEchoStringMethod, HandleUnaryEchoString).Build());
-
- int port = server.AddListeningPort(host + ":0");
+ var server = new Server();
+ server.AddServiceDefinition(ServiceDefinition);
+ int port = server.AddListeningPort(Host + ":0");
server.Start();
- using (Channel channel = new Channel(host + ":" + port))
+ using (Channel channel = new Channel(Host + ":" + port))
{
- var call = new Call<string, string>(serviceName, unaryEchoStringMethod, channel, Metadata.Empty);
- BenchmarkUtil.RunBenchmark(100, 1000,
+ var call = new Call<string, string>(ServiceName, UnaryEchoStringMethod, channel, Metadata.Empty);
+ BenchmarkUtil.RunBenchmark(100, 100,
() => { Calls.BlockingUnaryCall(call, "ABC", default(CancellationToken)); });
}
@@ -113,17 +130,14 @@ namespace Grpc.Core.Tests
[Test]
public void UnknownMethodHandler()
{
- Server server = new Server();
- server.AddServiceDefinition(
- ServerServiceDefinition.CreateBuilder(serviceName).Build());
-
- int port = server.AddListeningPort(host + ":0");
+ var server = new Server();
+ server.AddServiceDefinition(ServerServiceDefinition.CreateBuilder(ServiceName).Build());
+ int port = server.AddListeningPort(Host + ":0");
server.Start();
- using (Channel channel = new Channel(host + ":" + port))
+ using (Channel channel = new Channel(Host + ":" + port))
{
- var call = new Call<string, string>(serviceName, unaryEchoStringMethod, channel, Metadata.Empty);
-
+ var call = new Call<string, string>(ServiceName, UnaryEchoStringMethod, channel, Metadata.Empty);
try
{
Calls.BlockingUnaryCall(call, "ABC", default(CancellationToken));
@@ -138,10 +152,12 @@ namespace Grpc.Core.Tests
server.ShutdownAsync().Wait();
}
- private void HandleUnaryEchoString(string request, IObserver<string> responseObserver)
+ /// <summary>
+ /// Handler for unaryEchoString method.
+ /// </summary>
+ private static Task<string> HandleUnaryEchoString(string request)
{
- responseObserver.OnNext(request);
- responseObserver.OnCompleted();
+ return Task.FromResult(request);
}
}
}
diff --git a/src/csharp/Grpc.Core/Internal/ServerStreamingOutputObserver.cs b/src/csharp/Grpc.Core/AsyncClientStreamingCall.cs
similarity index 53%
rename from src/csharp/Grpc.Core/Internal/ServerStreamingOutputObserver.cs
rename to src/csharp/Grpc.Core/AsyncClientStreamingCall.cs
index 97b62d05692046bc826cfbda7a21a8c9bb436e52..e81ce01ebbc969ab599a239ff56f8a155947b605 100644
--- a/src/csharp/Grpc.Core/Internal/ServerStreamingOutputObserver.cs
+++ b/src/csharp/Grpc.Core/AsyncClientStreamingCall.cs
@@ -1,4 +1,5 @@
#region Copyright notice and license
+
// Copyright 2015, Google Inc.
// All rights reserved.
//
@@ -27,45 +28,74 @@
// 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;
+using System.Runtime.CompilerServices;
+using System.Threading.Tasks;
-namespace Grpc.Core.Internal
+namespace Grpc.Core
{
/// <summary>
- /// Observer that writes all arriving messages to a call abstraction (in blocking fashion)
- /// and then halfcloses the call. Used for server-side call handling.
+ /// Return type for client streaming calls.
/// </summary>
- internal class ServerStreamingOutputObserver<TRequest, TResponse> : IObserver<TResponse>
+ public struct AsyncClientStreamingCall<TRequest, TResponse>
{
- readonly AsyncCallServer<TRequest, TResponse> call;
+ readonly IClientStreamWriter<TRequest> requestStream;
+ readonly Task<TResponse> result;
+
+ public AsyncClientStreamingCall(IClientStreamWriter<TRequest> requestStream, Task<TResponse> result)
+ {
+ this.requestStream = requestStream;
+ this.result = result;
+ }
+
+ /// <summary>
+ /// Writes a request to RequestStream.
+ /// </summary>
+ public Task Write(TRequest message)
+ {
+ return requestStream.Write(message);
+ }
- public ServerStreamingOutputObserver(AsyncCallServer<TRequest, TResponse> call)
+ /// <summary>
+ /// Closes the RequestStream.
+ /// </summary>
+ public Task Close()
{
- this.call = call;
+ return requestStream.Close();
}
- public void OnCompleted()
+ /// <summary>
+ /// Asynchronous call result.
+ /// </summary>
+ public Task<TResponse> Result
{
- var taskSource = new AsyncCompletionTaskSource();
- call.StartSendStatusFromServer(new Status(StatusCode.OK, ""), taskSource.CompletionDelegate);
- // TODO: how bad is the Wait here?
- taskSource.Task.Wait();
+ get
+ {
+ return this.result;
+ }
}
- public void OnError(Exception error)
+ /// <summary>
+ /// Async stream to send streaming requests.
+ /// </summary>
+ public IClientStreamWriter<TRequest> RequestStream
{
- // TODO: implement this...
- throw new InvalidOperationException("This should never be called.");
+ get
+ {
+ return requestStream;
+ }
}
- public void OnNext(TResponse value)
+ /// <summary>
+ /// Allows awaiting this object directly.
+ /// </summary>
+ /// <returns></returns>
+ public TaskAwaiter<TResponse> GetAwaiter()
{
- var taskSource = new AsyncCompletionTaskSource();
- call.StartSendMessage(value, taskSource.CompletionDelegate);
- // TODO: how bad is the Wait here?
- taskSource.Task.Wait();
+ return result.GetAwaiter();
}
}
}
diff --git a/src/csharp/Grpc.Core/AsyncDuplexStreamingCall.cs b/src/csharp/Grpc.Core/AsyncDuplexStreamingCall.cs
new file mode 100644
index 0000000000000000000000000000000000000000..1cb30f47795aed61baaee934445183df0f607b02
--- /dev/null
+++ b/src/csharp/Grpc.Core/AsyncDuplexStreamingCall.cs
@@ -0,0 +1,101 @@
+#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 bidirectional streaming calls.
+ /// </summary>
+ public struct AsyncDuplexStreamingCall<TRequest, TResponse>
+ {
+ readonly IClientStreamWriter<TRequest> requestStream;
+ readonly IAsyncStreamReader<TResponse> responseStream;
+
+ public AsyncDuplexStreamingCall(IClientStreamWriter<TRequest> requestStream, IAsyncStreamReader<TResponse> responseStream)
+ {
+ this.requestStream = requestStream;
+ this.responseStream = responseStream;
+ }
+
+ /// <summary>
+ /// Writes a request to RequestStream.
+ /// </summary>
+ public Task Write(TRequest message)
+ {
+ return requestStream.Write(message);
+ }
+
+ /// <summary>
+ /// Closes the RequestStream.
+ /// </summary>
+ public Task Close()
+ {
+ return requestStream.Close();
+ }
+
+ /// <summary>
+ /// Reads a response from ResponseStream.
+ /// </summary>
+ /// <returns></returns>
+ public Task<TResponse> ReadNext()
+ {
+ return responseStream.ReadNext();
+ }
+
+ /// <summary>
+ /// Async stream to read streaming responses.
+ /// </summary>
+ public IAsyncStreamReader<TResponse> ResponseStream
+ {
+ get
+ {
+ return responseStream;
+ }
+ }
+
+ /// <summary>
+ /// Async stream to send streaming requests.
+ /// </summary>
+ public IClientStreamWriter<TRequest> RequestStream
+ {
+ get
+ {
+ return requestStream;
+ }
+ }
+ }
+}
diff --git a/src/csharp/Grpc.Core/Utils/RecordingQueue.cs b/src/csharp/Grpc.Core/AsyncServerStreamingCall.cs
similarity index 64%
rename from src/csharp/Grpc.Core/Utils/RecordingQueue.cs
rename to src/csharp/Grpc.Core/AsyncServerStreamingCall.cs
index 9749168af02731f0e02f1a497ece3371080eb4a6..d614916fb7c1fe3c082bcabbe77b153aa77f0c10 100644
--- a/src/csharp/Grpc.Core/Utils/RecordingQueue.cs
+++ b/src/csharp/Grpc.Core/AsyncServerStreamingCall.cs
@@ -32,51 +32,40 @@
#endregion
using System;
-using System.Collections.Concurrent;
-using System.Collections.Generic;
+using System.Runtime.CompilerServices;
using System.Threading.Tasks;
-namespace Grpc.Core.Utils
+namespace Grpc.Core
{
- // TODO: replace this by something that implements IAsyncEnumerator.
/// <summary>
- /// Observer that allows us to await incoming messages one-by-one.
- /// The implementation is not ideal and class will be probably replaced
- /// by something more versatile in the future.
+ /// Return type for server streaming calls.
/// </summary>
- public class RecordingQueue<T> : IObserver<T>
+ public struct AsyncServerStreamingCall<TResponse>
{
- readonly BlockingCollection<T> queue = new BlockingCollection<T>();
- TaskCompletionSource<object> tcs = new TaskCompletionSource<object>();
+ readonly IAsyncStreamReader<TResponse> responseStream;
- public void OnCompleted()
+ public AsyncServerStreamingCall(IAsyncStreamReader<TResponse> responseStream)
{
- tcs.SetResult(null);
+ this.responseStream = responseStream;
}
- public void OnError(Exception error)
+ /// <summary>
+ /// Reads the next response from ResponseStream
+ /// </summary>
+ /// <returns></returns>
+ public Task<TResponse> ReadNext()
{
- tcs.SetException(error);
+ return responseStream.ReadNext();
}
- public void OnNext(T value)
- {
- queue.Add(value);
- }
-
- public BlockingCollection<T> Queue
- {
- get
- {
- return queue;
- }
- }
-
- public Task Finished
+ /// <summary>
+ /// Async stream to read streaming responses.
+ /// </summary>
+ public IAsyncStreamReader<TResponse> ResponseStream
{
get
{
- return tcs.Task;
+ return responseStream;
}
}
}
diff --git a/src/csharp/Grpc.Core/Call.cs b/src/csharp/Grpc.Core/Call.cs
index fe5f40f5e9020abb93ae713699194f792589de27..070dfb569d6477da2fff25106f6606ba9ed87cd5 100644
--- a/src/csharp/Grpc.Core/Call.cs
+++ b/src/csharp/Grpc.Core/Call.cs
@@ -37,6 +37,9 @@ using Grpc.Core.Utils;
namespace Grpc.Core
{
+ /// <summary>
+ /// Abstraction of a call to be invoked on a client.
+ /// </summary>
public class Call<TRequest, TResponse>
{
readonly string name;
diff --git a/src/csharp/Grpc.Core/Calls.cs b/src/csharp/Grpc.Core/Calls.cs
index 280387b323bfc808bf61549c9269654bf28ce0ff..9365ccd9fb671c9d9d304e62d100fd9582ead079 100644
--- a/src/csharp/Grpc.Core/Calls.cs
+++ b/src/csharp/Grpc.Core/Calls.cs
@@ -39,7 +39,7 @@ using Grpc.Core.Internal;
namespace Grpc.Core
{
/// <summary>
- /// Helper methods for generated stubs to make RPC calls.
+ /// Helper methods for generated client stubs to make RPC calls.
/// </summary>
public static class Calls
{
@@ -56,35 +56,37 @@ namespace Grpc.Core
return await asyncCall.UnaryCallAsync(req, call.Headers);
}
- public static void AsyncServerStreamingCall<TRequest, TResponse>(Call<TRequest, TResponse> call, TRequest req, IObserver<TResponse> outputs, CancellationToken token)
+ public static AsyncServerStreamingCall<TResponse> AsyncServerStreamingCall<TRequest, TResponse>(Call<TRequest, TResponse> call, TRequest req, CancellationToken token)
{
var asyncCall = new AsyncCall<TRequest, TResponse>(call.RequestMarshaller.Serializer, call.ResponseMarshaller.Deserializer);
asyncCall.Initialize(call.Channel, GetCompletionQueue(), call.Name);
- asyncCall.StartServerStreamingCall(req, outputs, call.Headers);
+ asyncCall.StartServerStreamingCall(req, call.Headers);
+ var responseStream = new ClientResponseStream<TRequest, TResponse>(asyncCall);
+ return new AsyncServerStreamingCall<TResponse>(responseStream);
}
- public static ClientStreamingAsyncResult<TRequest, TResponse> AsyncClientStreamingCall<TRequest, TResponse>(Call<TRequest, TResponse> call, CancellationToken token)
+ public static AsyncClientStreamingCall<TRequest, TResponse> AsyncClientStreamingCall<TRequest, TResponse>(Call<TRequest, TResponse> call, CancellationToken token)
{
var asyncCall = new AsyncCall<TRequest, TResponse>(call.RequestMarshaller.Serializer, call.ResponseMarshaller.Deserializer);
asyncCall.Initialize(call.Channel, GetCompletionQueue(), call.Name);
- var task = asyncCall.ClientStreamingCallAsync(call.Headers);
- var inputs = new ClientStreamingInputObserver<TRequest, TResponse>(asyncCall);
- return new ClientStreamingAsyncResult<TRequest, TResponse>(task, inputs);
+ var resultTask = asyncCall.ClientStreamingCallAsync(call.Headers);
+ var requestStream = new ClientRequestStream<TRequest, TResponse>(asyncCall);
+ return new AsyncClientStreamingCall<TRequest, TResponse>(requestStream, resultTask);
}
- public static TResponse BlockingClientStreamingCall<TRequest, TResponse>(Call<TRequest, TResponse> call, IObservable<TRequest> inputs, CancellationToken token)
- {
- throw new NotImplementedException();
- }
-
- public static IObserver<TRequest> DuplexStreamingCall<TRequest, TResponse>(Call<TRequest, TResponse> call, IObserver<TResponse> outputs, CancellationToken token)
+ public static AsyncDuplexStreamingCall<TRequest, TResponse> AsyncDuplexStreamingCall<TRequest, TResponse>(Call<TRequest, TResponse> call, CancellationToken token)
{
var asyncCall = new AsyncCall<TRequest, TResponse>(call.RequestMarshaller.Serializer, call.ResponseMarshaller.Deserializer);
asyncCall.Initialize(call.Channel, GetCompletionQueue(), call.Name);
- asyncCall.StartDuplexStreamingCall(outputs, call.Headers);
- return new ClientStreamingInputObserver<TRequest, TResponse>(asyncCall);
+ asyncCall.StartDuplexStreamingCall(call.Headers);
+ var requestStream = new ClientRequestStream<TRequest, TResponse>(asyncCall);
+ var responseStream = new ClientResponseStream<TRequest, TResponse>(asyncCall);
+ return new AsyncDuplexStreamingCall<TRequest, TResponse>(requestStream, responseStream);
}
+ /// <summary>
+ /// Gets shared completion queue used for async calls.
+ /// </summary>
private static CompletionQueueSafeHandle GetCompletionQueue()
{
return GrpcEnvironment.ThreadPool.CompletionQueue;
diff --git a/src/csharp/Grpc.Core/Channel.cs b/src/csharp/Grpc.Core/Channel.cs
index 3a42dac1d7a54fe8610b5153f69faec12785bbe3..b47d810672031c2d32a8e27fdbf40a6f6fb82509 100644
--- a/src/csharp/Grpc.Core/Channel.cs
+++ b/src/csharp/Grpc.Core/Channel.cs
@@ -66,14 +66,6 @@ namespace Grpc.Core
this.target = GetOverridenTarget(target, channelArgs);
}
- internal ChannelSafeHandle Handle
- {
- get
- {
- return this.handle;
- }
- }
-
public string Target
{
get
@@ -88,6 +80,14 @@ namespace Grpc.Core
GC.SuppressFinalize(this);
}
+ internal ChannelSafeHandle Handle
+ {
+ get
+ {
+ return this.handle;
+ }
+ }
+
protected virtual void Dispose(bool disposing)
{
if (handle != null && !handle.IsInvalid)
diff --git a/src/csharp/Grpc.Core/Credentials.cs b/src/csharp/Grpc.Core/Credentials.cs
index 15dd3ef3216e2a14c6b2cb79a11ed7763bc97533..e64c1e3dc1da7ac16f4c06499fbb793396001bbf 100644
--- a/src/csharp/Grpc.Core/Credentials.cs
+++ b/src/csharp/Grpc.Core/Credentials.cs
@@ -37,7 +37,7 @@ using Grpc.Core.Internal;
namespace Grpc.Core
{
/// <summary>
- /// Client-side credentials.
+ /// Client-side credentials. Used for creation of a secure channel.
/// </summary>
public abstract class Credentials
{
diff --git a/src/csharp/Grpc.Core/Grpc.Core.csproj b/src/csharp/Grpc.Core/Grpc.Core.csproj
index 0b85392e15a9d54ca936ff432b61273e2da50ed5..fee742b2201b20daac908b42cbb68f9290e7f123 100644
--- a/src/csharp/Grpc.Core/Grpc.Core.csproj
+++ b/src/csharp/Grpc.Core/Grpc.Core.csproj
@@ -39,12 +39,18 @@
</Reference>
</ItemGroup>
<ItemGroup>
+ <Compile Include="AsyncDuplexStreamingCall.cs" />
+ <Compile Include="AsyncServerStreamingCall.cs" />
+ <Compile Include="IClientStreamWriter.cs" />
+ <Compile Include="IServerStreamWriter.cs" />
+ <Compile Include="IAsyncStreamWriter.cs" />
+ <Compile Include="IAsyncStreamReader.cs" />
<Compile Include="Internal\GrpcLog.cs" />
<Compile Include="Properties\AssemblyInfo.cs" />
<Compile Include="RpcException.cs" />
<Compile Include="Calls.cs" />
<Compile Include="Call.cs" />
- <Compile Include="ClientStreamingAsyncResult.cs" />
+ <Compile Include="AsyncClientStreamingCall.cs" />
<Compile Include="GrpcEnvironment.cs" />
<Compile Include="Status.cs" />
<Compile Include="StatusCode.cs" />
@@ -59,14 +65,10 @@
<Compile Include="Internal\GrpcThreadPool.cs" />
<Compile Include="Internal\ServerSafeHandle.cs" />
<Compile Include="Method.cs" />
- <Compile Include="ServerCalls.cs" />
<Compile Include="Internal\ServerCallHandler.cs" />
<Compile Include="Marshaller.cs" />
<Compile Include="ServerServiceDefinition.cs" />
- <Compile Include="Utils\RecordingObserver.cs" />
- <Compile Include="Utils\RecordingQueue.cs" />
- <Compile Include="Internal\ClientStreamingInputObserver.cs" />
- <Compile Include="Internal\ServerStreamingOutputObserver.cs" />
+ <Compile Include="Utils\AsyncStreamExtensions.cs" />
<Compile Include="Internal\BatchContextSafeHandleNotOwned.cs" />
<Compile Include="Utils\BenchmarkUtil.cs" />
<Compile Include="Utils\ExceptionHelper.cs" />
@@ -86,6 +88,12 @@
<Compile Include="Internal\MetadataArraySafeHandle.cs" />
<Compile Include="Stub\AbstractStub.cs" />
<Compile Include="Stub\StubConfiguration.cs" />
+ <Compile Include="Internal\ServerCalls.cs" />
+ <Compile Include="ServerMethods.cs" />
+ <Compile Include="Internal\ClientRequestStream.cs" />
+ <Compile Include="Internal\ClientResponseStream.cs" />
+ <Compile Include="Internal\ServerRequestStream.cs" />
+ <Compile Include="Internal\ServerResponseStream.cs" />
</ItemGroup>
<ItemGroup>
<None Include="packages.config" />
diff --git a/src/csharp/Grpc.Core/IAsyncStreamReader.cs b/src/csharp/Grpc.Core/IAsyncStreamReader.cs
new file mode 100644
index 0000000000000000000000000000000000000000..61cf57f7e0d3c4ad3f4f8a4892e88fc547c005a1
--- /dev/null
+++ b/src/csharp/Grpc.Core/IAsyncStreamReader.cs
@@ -0,0 +1,54 @@
+#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.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace Grpc.Core
+{
+ /// <summary>
+ /// A stream of messages to be read.
+ /// </summary>
+ /// <typeparam name="T"></typeparam>
+ public interface IAsyncStreamReader<T>
+ {
+ /// <summary>
+ /// Reads a single message. Returns default(T) if the last message was already read.
+ /// A following read can only be started when the previous one finishes.
+ /// </summary>
+ Task<T> ReadNext();
+ }
+}
diff --git a/src/csharp/Grpc.Core/IAsyncStreamWriter.cs b/src/csharp/Grpc.Core/IAsyncStreamWriter.cs
new file mode 100644
index 0000000000000000000000000000000000000000..724bae8f3130eb7a97115b271b558cbdc38f2932
--- /dev/null
+++ b/src/csharp/Grpc.Core/IAsyncStreamWriter.cs
@@ -0,0 +1,54 @@
+#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.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace Grpc.Core
+{
+ /// <summary>
+ /// A writable stream of messages.
+ /// </summary>
+ /// <typeparam name="T"></typeparam>
+ public interface IAsyncStreamWriter<T>
+ {
+ /// <summary>
+ /// Writes a single message. Only one write can be pending at a time.
+ /// </summary>
+ /// <param name="message">the message to be written. Cannot be null.</param>
+ Task Write(T message);
+ }
+}
diff --git a/src/csharp/Grpc.Core/IClientStreamWriter.cs b/src/csharp/Grpc.Core/IClientStreamWriter.cs
new file mode 100644
index 0000000000000000000000000000000000000000..6da42e9ccc5c030a6eed980d4d962ad14f25d729
--- /dev/null
+++ b/src/csharp/Grpc.Core/IClientStreamWriter.cs
@@ -0,0 +1,53 @@
+#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.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace Grpc.Core
+{
+ /// <summary>
+ /// Client-side writable stream of messages with Close capability.
+ /// </summary>
+ /// <typeparam name="T"></typeparam>
+ public interface IClientStreamWriter<T> : IAsyncStreamWriter<T>
+ {
+ /// <summary>
+ /// Closes the stream. Can only be called once there is no pending write. No writes should follow calling this.
+ /// </summary>
+ Task Close();
+ }
+}
diff --git a/src/csharp/Grpc.Core/IServerStreamWriter.cs b/src/csharp/Grpc.Core/IServerStreamWriter.cs
new file mode 100644
index 0000000000000000000000000000000000000000..e76397d8a0d9200964df735cf9ff0292f87ac56c
--- /dev/null
+++ b/src/csharp/Grpc.Core/IServerStreamWriter.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;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace Grpc.Core
+{
+ /// <summary>
+ /// A writable stream of messages that is used in server-side handlers.
+ /// </summary>
+ public interface IServerStreamWriter<T> : IAsyncStreamWriter<T>
+ {
+ }
+}
diff --git a/src/csharp/Grpc.Core/Internal/AsyncCall.cs b/src/csharp/Grpc.Core/Internal/AsyncCall.cs
index bc72cb78de38d856111d6247ca75f799e390326b..fd94771ddd9d19101b5f75bc21fb58f53d336082 100644
--- a/src/csharp/Grpc.Core/Internal/AsyncCall.cs
+++ b/src/csharp/Grpc.Core/Internal/AsyncCall.cs
@@ -43,7 +43,7 @@ using Grpc.Core.Utils;
namespace Grpc.Core.Internal
{
/// <summary>
- /// Handles client side native call lifecycle.
+ /// Manages client side native call lifecycle.
/// </summary>
internal class AsyncCall<TRequest, TResponse> : AsyncCallBase<TRequest, TResponse>
{
@@ -160,7 +160,7 @@ namespace Grpc.Core.Internal
/// <summary>
/// Starts a unary request - streamed response call.
/// </summary>
- public void StartServerStreamingCall(TRequest msg, IObserver<TResponse> readObserver, Metadata headers)
+ public void StartServerStreamingCall(TRequest msg, Metadata headers)
{
lock (myLock)
{
@@ -169,17 +169,13 @@ namespace Grpc.Core.Internal
started = true;
halfcloseRequested = true;
halfclosed = true; // halfclose not confirmed yet, but it will be once finishedHandler is called.
-
- this.readObserver = readObserver;
byte[] payload = UnsafeSerialize(msg);
-
+
using (var metadataArray = MetadataArraySafeHandle.Create(headers))
{
call.StartServerStreaming(payload, finishedHandler, metadataArray);
}
-
- StartReceiveMessage();
}
}
@@ -187,7 +183,7 @@ namespace Grpc.Core.Internal
/// Starts a streaming request - streaming response call.
/// Use StartSendMessage and StartSendCloseFromClient to stream requests.
/// </summary>
- public void StartDuplexStreamingCall(IObserver<TResponse> readObserver, Metadata headers)
+ public void StartDuplexStreamingCall(Metadata headers)
{
lock (myLock)
{
@@ -195,14 +191,10 @@ namespace Grpc.Core.Internal
started = true;
- this.readObserver = readObserver;
-
using (var metadataArray = MetadataArraySafeHandle.Create(headers))
{
call.StartDuplexStreaming(finishedHandler, metadataArray);
}
-
- StartReceiveMessage();
}
}
@@ -210,17 +202,26 @@ namespace Grpc.Core.Internal
/// 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)
+ public void StartSendMessage(TRequest msg, AsyncCompletionDelegate<object> completionDelegate)
{
StartSendMessageInternal(msg, completionDelegate);
}
+ /// <summary>
+ /// Receives a streaming response. Only one pending read action is allowed at any given time.
+ /// completionDelegate is called when the operation finishes.
+ /// </summary>
+ public void StartReadMessage(AsyncCompletionDelegate<TResponse> completionDelegate)
+ {
+ StartReadMessageInternal(completionDelegate);
+ }
+
/// <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)
+ public void StartSendCloseFromClient(AsyncCompletionDelegate<object> completionDelegate)
{
lock (myLock)
{
@@ -235,12 +236,12 @@ namespace Grpc.Core.Internal
}
/// <summary>
- /// On client-side, we only fire readObserver.OnCompleted once all messages have been read
+ /// On client-side, we only fire readCompletionDelegate once all messages have been read
/// and status has been received.
/// </summary>
- protected override void CompleteReadObserver()
+ protected override void ProcessLastRead(AsyncCompletionDelegate<TResponse> completionDelegate)
{
- if (readingDone && finishedStatus.HasValue)
+ if (completionDelegate != null && readingDone && finishedStatus.HasValue)
{
bool shouldComplete;
lock (myLock)
@@ -254,11 +255,11 @@ namespace Grpc.Core.Internal
var status = finishedStatus.Value;
if (status.StatusCode != StatusCode.OK)
{
- FireReadObserverOnError(new RpcException(status));
+ FireCompletion(completionDelegate, default(TResponse), new RpcException(status));
}
else
{
- FireReadObserverOnCompleted();
+ FireCompletion(completionDelegate, default(TResponse), null);
}
}
}
@@ -304,15 +305,18 @@ namespace Grpc.Core.Internal
{
var status = ctx.GetReceivedStatus();
+ AsyncCompletionDelegate<TResponse> origReadCompletionDelegate = null;
lock (myLock)
{
finished = true;
finishedStatus = status;
+ origReadCompletionDelegate = readCompletionDelegate;
+
ReleaseResourcesIfPossible();
}
- CompleteReadObserver();
+ ProcessLastRead(origReadCompletionDelegate);
}
}
}
\ No newline at end of file
diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs b/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs
index 15b0cfe2495705346a8ec4cfacc651ae3d5aab7b..2bde4b37202442e2b23f77a97140badd8397a88d 100644
--- a/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs
+++ b/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs
@@ -44,7 +44,7 @@ namespace Grpc.Core.Internal
{
/// <summary>
/// Base for handling both client side and server side calls.
- /// Handles native call lifecycle and provides convenience methods.
+ /// Manages native call lifecycle and provides convenience methods.
/// </summary>
internal abstract class AsyncCallBase<TWrite, TRead>
{
@@ -65,16 +65,14 @@ namespace Grpc.Core.Internal
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 AsyncCompletionDelegate<object> sendCompletionDelegate; // Completion of a pending send or sendclose if not null.
+ protected AsyncCompletionDelegate<TRead> readCompletionDelegate; // Completion of a pending send or sendclose if not null.
+
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);
@@ -131,10 +129,10 @@ namespace Grpc.Core.Internal
}
/// <summary>
- /// Initiates sending a message. Only once send operation can be active at a time.
+ /// Initiates sending a message. Only one send operation can be active at a time.
/// completionDelegate is invoked upon completion.
/// </summary>
- protected void StartSendMessageInternal(TWrite msg, AsyncCompletionDelegate completionDelegate)
+ protected void StartSendMessageInternal(TWrite msg, AsyncCompletionDelegate<object> completionDelegate)
{
byte[] payload = UnsafeSerialize(msg);
@@ -149,31 +147,29 @@ namespace Grpc.Core.Internal
}
/// <summary>
- /// Requests receiving a next message.
+ /// Initiates reading a message. Only one read operation can be active at a time.
+ /// completionDelegate is invoked upon completion.
/// </summary>
- protected void StartReceiveMessage()
+ protected void StartReadMessageInternal(AsyncCompletionDelegate<TRead> completionDelegate)
{
lock (myLock)
{
- Preconditions.CheckState(started);
- Preconditions.CheckState(!disposed);
- Preconditions.CheckState(!errorOccured);
-
- Preconditions.CheckState(!readingDone);
- Preconditions.CheckState(!readPending);
+ Preconditions.CheckNotNull(completionDelegate, "Completion delegate cannot be null");
+ CheckReadingAllowed();
call.StartReceiveMessage(readFinishedHandler);
- readPending = true;
+ readCompletionDelegate = completionDelegate;
}
}
+ // TODO(jtattermusch): find more fitting name for this method.
/// <summary>
/// Default behavior just completes the read observer, but more sofisticated behavior might be required
/// by subclasses.
/// </summary>
- protected virtual void CompleteReadObserver()
+ protected virtual void ProcessLastRead(AsyncCompletionDelegate<TRead> completionDelegate)
{
- FireReadObserverOnCompleted();
+ FireCompletion(completionDelegate, default(TRead), null);
}
/// <summary>
@@ -213,6 +209,16 @@ namespace Grpc.Core.Internal
Preconditions.CheckState(sendCompletionDelegate == null, "Only one write can be pending at a time");
}
+ protected void CheckReadingAllowed()
+ {
+ Preconditions.CheckState(started);
+ Preconditions.CheckState(!disposed);
+ Preconditions.CheckState(!errorOccured);
+
+ Preconditions.CheckState(!readingDone, "Stream has already been closed.");
+ Preconditions.CheckState(readCompletionDelegate == null, "Only one write can be pending at a time");
+ }
+
protected byte[] UnsafeSerialize(TWrite msg)
{
return serializer(msg);
@@ -248,47 +254,11 @@ namespace Grpc.Core.Internal
}
}
- protected void FireReadObserverOnNext(TRead value)
- {
- try
- {
- readObserver.OnNext(value);
- }
- catch (Exception e)
- {
- Console.WriteLine("Exception occured while invoking readObserver.OnNext: " + e);
- }
- }
-
- protected void FireReadObserverOnCompleted()
+ protected void FireCompletion<T>(AsyncCompletionDelegate<T> completionDelegate, T value, Exception error)
{
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);
+ completionDelegate(value, error);
}
catch (Exception e)
{
@@ -322,7 +292,7 @@ namespace Grpc.Core.Internal
/// </summary>
private void HandleSendFinished(bool wasError, BatchContextSafeHandleNotOwned ctx)
{
- AsyncCompletionDelegate origCompletionDelegate = null;
+ AsyncCompletionDelegate<object> origCompletionDelegate = null;
lock (myLock)
{
origCompletionDelegate = sendCompletionDelegate;
@@ -333,11 +303,11 @@ namespace Grpc.Core.Internal
if (wasError)
{
- FireCompletion(origCompletionDelegate, new OperationFailedException("Send failed"));
+ FireCompletion(origCompletionDelegate, null, new OperationFailedException("Send failed"));
}
else
{
- FireCompletion(origCompletionDelegate, null);
+ FireCompletion(origCompletionDelegate, null, null);
}
}
@@ -346,7 +316,7 @@ namespace Grpc.Core.Internal
/// </summary>
private void HandleHalfclosed(bool wasError, BatchContextSafeHandleNotOwned ctx)
{
- AsyncCompletionDelegate origCompletionDelegate = null;
+ AsyncCompletionDelegate<object> origCompletionDelegate = null;
lock (myLock)
{
halfclosed = true;
@@ -358,11 +328,11 @@ namespace Grpc.Core.Internal
if (wasError)
{
- FireCompletion(origCompletionDelegate, new OperationFailedException("Halfclose failed"));
+ FireCompletion(origCompletionDelegate, null, new OperationFailedException("Halfclose failed"));
}
else
{
- FireCompletion(origCompletionDelegate, null);
+ FireCompletion(origCompletionDelegate, null, null);
}
}
@@ -373,11 +343,19 @@ namespace Grpc.Core.Internal
{
var payload = ctx.GetReceivedMessage();
+ AsyncCompletionDelegate<TRead> origCompletionDelegate = null;
lock (myLock)
{
- readPending = false;
- if (payload == null)
+ origCompletionDelegate = readCompletionDelegate;
+ if (payload != null)
{
+ readCompletionDelegate = null;
+ }
+ else
+ {
+ // This was the last read. Keeping the readCompletionDelegate
+ // to be either fired by this handler or by client-side finished
+ // handler.
readingDone = true;
}
@@ -392,15 +370,11 @@ namespace Grpc.Core.Internal
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();
+ FireCompletion(origCompletionDelegate, msg, null);
}
else
{
- CompleteReadObserver();
+ ProcessLastRead(origCompletionDelegate);
}
}
}
diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs
index d3a2be553fc6ac21cc1f476c5d7f16d9692be0d3..25dc15bbc0bf9ee7aba90fd79d4fd405c065bc07 100644
--- a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs
+++ b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs
@@ -43,7 +43,7 @@ using Grpc.Core.Utils;
namespace Grpc.Core.Internal
{
/// <summary>
- /// Handles server side native call lifecycle.
+ /// Manages server side native call lifecycle.
/// </summary>
internal class AsyncCallServer<TRequest, TResponse> : AsyncCallBase<TResponse, TRequest>
{
@@ -61,20 +61,17 @@ namespace Grpc.Core.Internal
}
/// <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.
+ /// Starts a server side call.
/// </summary>
- public Task ServerSideCallAsync(IObserver<TRequest> readObserver)
+ public Task ServerSideCallAsync()
{
lock (myLock)
{
Preconditions.CheckNotNull(call);
started = true;
- this.readObserver = readObserver;
call.StartServerSide(finishedServersideHandler);
- StartReceiveMessage();
return finishedServersideTcs.Task;
}
}
@@ -83,17 +80,26 @@ namespace Grpc.Core.Internal
/// 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)
+ public void StartSendMessage(TResponse msg, AsyncCompletionDelegate<object> completionDelegate)
{
StartSendMessageInternal(msg, completionDelegate);
}
+ /// <summary>
+ /// Receives a streaming request. Only one pending read action is allowed at any given time.
+ /// completionDelegate is called when the operation finishes.
+ /// </summary>
+ public void StartReadMessage(AsyncCompletionDelegate<TRequest> completionDelegate)
+ {
+ StartReadMessageInternal(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)
+ public void StartSendStatusFromServer(Status status, AsyncCompletionDelegate<object> completionDelegate)
{
lock (myLock)
{
diff --git a/src/csharp/Grpc.Core/Internal/AsyncCompletion.cs b/src/csharp/Grpc.Core/Internal/AsyncCompletion.cs
index 673b527fb20267fb910b322615fabfb368c977e6..c88cae98fe75c326fbe9f4e898bcd7eaae3939c2 100644
--- a/src/csharp/Grpc.Core/Internal/AsyncCompletion.cs
+++ b/src/csharp/Grpc.Core/Internal/AsyncCompletion.cs
@@ -45,22 +45,22 @@ 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);
+ internal delegate void AsyncCompletionDelegate<T>(T result, Exception error);
/// <summary>
/// Helper for transforming AsyncCompletionDelegate into full-fledged Task.
/// </summary>
- internal class AsyncCompletionTaskSource
+ internal class AsyncCompletionTaskSource<T>
{
- readonly TaskCompletionSource<object> tcs = new TaskCompletionSource<object>();
- readonly AsyncCompletionDelegate completionDelegate;
+ readonly TaskCompletionSource<T> tcs = new TaskCompletionSource<T>();
+ readonly AsyncCompletionDelegate<T> completionDelegate;
public AsyncCompletionTaskSource()
{
- completionDelegate = new AsyncCompletionDelegate(HandleCompletion);
+ completionDelegate = new AsyncCompletionDelegate<T>(HandleCompletion);
}
- public Task Task
+ public Task<T> Task
{
get
{
@@ -68,7 +68,7 @@ namespace Grpc.Core.Internal
}
}
- public AsyncCompletionDelegate CompletionDelegate
+ public AsyncCompletionDelegate<T> CompletionDelegate
{
get
{
@@ -76,11 +76,11 @@ namespace Grpc.Core.Internal
}
}
- private void HandleCompletion(Exception error)
+ private void HandleCompletion(T value, Exception error)
{
if (error == null)
{
- tcs.SetResult(null);
+ tcs.SetResult(value);
return;
}
if (error is OperationCanceledException)
diff --git a/src/csharp/Grpc.Core/Internal/ClientStreamingInputObserver.cs b/src/csharp/Grpc.Core/Internal/ClientRequestStream.cs
similarity index 70%
rename from src/csharp/Grpc.Core/Internal/ClientStreamingInputObserver.cs
rename to src/csharp/Grpc.Core/Internal/ClientRequestStream.cs
index 286c54f2c47811f458d4b1543b06007e7b836378..6854922a6f7165f7ece8ab180594d80a8aca3546 100644
--- a/src/csharp/Grpc.Core/Internal/ClientStreamingInputObserver.cs
+++ b/src/csharp/Grpc.Core/Internal/ClientRequestStream.cs
@@ -29,38 +29,35 @@
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#endregion
using System;
+using System.Threading.Tasks;
using Grpc.Core.Internal;
namespace Grpc.Core.Internal
{
- internal class ClientStreamingInputObserver<TWrite, TRead> : IObserver<TWrite>
+ /// <summary>
+ /// Writes requests asynchronously to an underlying AsyncCall object.
+ /// </summary>
+ internal class ClientRequestStream<TRequest, TResponse> : IClientStreamWriter<TRequest>
{
- readonly AsyncCall<TWrite, TRead> call;
+ readonly AsyncCall<TRequest, TResponse> call;
- public ClientStreamingInputObserver(AsyncCall<TWrite, TRead> call)
+ public ClientRequestStream(AsyncCall<TRequest, TResponse> call)
{
this.call = call;
}
- public void OnCompleted()
+ public Task Write(TRequest message)
{
- var taskSource = new AsyncCompletionTaskSource();
- call.StartSendCloseFromClient(taskSource.CompletionDelegate);
- // TODO: how bad is the Wait here?
- taskSource.Task.Wait();
+ var taskSource = new AsyncCompletionTaskSource<object>();
+ call.StartSendMessage(message, taskSource.CompletionDelegate);
+ return taskSource.Task;
}
- public void OnError(Exception error)
+ public Task Close()
{
- throw new InvalidOperationException("This should never be called.");
- }
-
- public void OnNext(TWrite value)
- {
- var taskSource = new AsyncCompletionTaskSource();
- call.StartSendMessage(value, taskSource.CompletionDelegate);
- // TODO: how bad is the Wait here?
- taskSource.Task.Wait();
+ var taskSource = new AsyncCompletionTaskSource<object>();
+ call.StartSendCloseFromClient(taskSource.CompletionDelegate);
+ return taskSource.Task;
}
}
}
diff --git a/src/csharp/Grpc.Core/Utils/RecordingObserver.cs b/src/csharp/Grpc.Core/Internal/ClientResponseStream.cs
similarity index 76%
rename from src/csharp/Grpc.Core/Utils/RecordingObserver.cs
rename to src/csharp/Grpc.Core/Internal/ClientResponseStream.cs
index 7b43ab8ad5f20b9fedbd926187e49bfc9ecc1f52..7fa511faa82b8099110c5ec4d0018a4d87551ec3 100644
--- a/src/csharp/Grpc.Core/Utils/RecordingObserver.cs
+++ b/src/csharp/Grpc.Core/Internal/ClientResponseStream.cs
@@ -35,31 +35,22 @@ using System;
using System.Collections.Generic;
using System.Threading.Tasks;
-namespace Grpc.Core.Utils
+namespace Grpc.Core.Internal
{
- public class RecordingObserver<T> : IObserver<T>
+ internal class ClientResponseStream<TRequest, TResponse> : IAsyncStreamReader<TResponse>
{
- TaskCompletionSource<List<T>> tcs = new TaskCompletionSource<List<T>>();
- List<T> data = new List<T>();
+ readonly AsyncCall<TRequest, TResponse> call;
- public void OnCompleted()
+ public ClientResponseStream(AsyncCall<TRequest, TResponse> call)
{
- tcs.SetResult(data);
+ this.call = call;
}
- public void OnError(Exception error)
+ public Task<TResponse> ReadNext()
{
- tcs.SetException(error);
- }
-
- public void OnNext(T value)
- {
- data.Add(value);
- }
-
- public Task<List<T>> ToList()
- {
- return tcs.Task;
+ var taskSource = new AsyncCompletionTaskSource<TResponse>();
+ call.StartReadMessage(taskSource.CompletionDelegate);
+ return taskSource.Task;
}
}
}
diff --git a/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs b/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs
index 25fd4fab8f7f9ec23396bc434b16c4487b0231de..2ae924b4a8869cc8d0b23ca7d1b604054bffbbb7 100644
--- a/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs
+++ b/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs
@@ -33,6 +33,7 @@
using System;
using System.Linq;
+using System.Threading.Tasks;
using Grpc.Core.Internal;
using Grpc.Core.Utils;
@@ -40,96 +41,147 @@ namespace Grpc.Core.Internal
{
internal interface IServerCallHandler
{
- void StartCall(string methodName, CallSafeHandle call, CompletionQueueSafeHandle cq);
+ Task HandleCall(string methodName, CallSafeHandle call, CompletionQueueSafeHandle cq);
}
- internal class UnaryRequestServerCallHandler<TRequest, TResponse> : IServerCallHandler
+ internal class UnaryServerCallHandler<TRequest, TResponse> : IServerCallHandler
{
readonly Method<TRequest, TResponse> method;
- readonly UnaryRequestServerMethod<TRequest, TResponse> handler;
+ readonly UnaryServerMethod<TRequest, TResponse> handler;
- public UnaryRequestServerCallHandler(Method<TRequest, TResponse> method, UnaryRequestServerMethod<TRequest, TResponse> handler)
+ public UnaryServerCallHandler(Method<TRequest, TResponse> method, UnaryServerMethod<TRequest, TResponse> handler)
{
this.method = method;
this.handler = handler;
}
- public void StartCall(string methodName, CallSafeHandle call, CompletionQueueSafeHandle cq)
+ public async Task HandleCall(string methodName, CallSafeHandle call, CompletionQueueSafeHandle cq)
{
var asyncCall = new AsyncCallServer<TRequest, TResponse>(
method.ResponseMarshaller.Serializer,
method.RequestMarshaller.Deserializer);
asyncCall.Initialize(call);
-
- var requestObserver = new RecordingObserver<TRequest>();
- var finishedTask = asyncCall.ServerSideCallAsync(requestObserver);
-
- var request = requestObserver.ToList().Result.Single();
- var responseObserver = new ServerStreamingOutputObserver<TRequest, TResponse>(asyncCall);
- handler(request, responseObserver);
-
- finishedTask.Wait();
+ var finishedTask = asyncCall.ServerSideCallAsync();
+ var requestStream = new ServerRequestStream<TRequest, TResponse>(asyncCall);
+ var responseStream = new ServerResponseStream<TRequest, TResponse>(asyncCall);
+
+ var request = await requestStream.ReadNext();
+ // TODO(jtattermusch): we need to read the full stream so that native callhandle gets deallocated.
+ Preconditions.CheckArgument(await requestStream.ReadNext() == null);
+
+ var result = await handler(request);
+ await responseStream.Write(result);
+ await responseStream.WriteStatus(Status.DefaultSuccess);
+ await finishedTask;
}
}
- internal class StreamingRequestServerCallHandler<TRequest, TResponse> : IServerCallHandler
+ internal class ServerStreamingServerCallHandler<TRequest, TResponse> : IServerCallHandler
{
readonly Method<TRequest, TResponse> method;
- readonly StreamingRequestServerMethod<TRequest, TResponse> handler;
+ readonly ServerStreamingServerMethod<TRequest, TResponse> handler;
- public StreamingRequestServerCallHandler(Method<TRequest, TResponse> method, StreamingRequestServerMethod<TRequest, TResponse> handler)
+ public ServerStreamingServerCallHandler(Method<TRequest, TResponse> method, ServerStreamingServerMethod<TRequest, TResponse> handler)
{
this.method = method;
this.handler = handler;
}
- public void StartCall(string methodName, CallSafeHandle call, CompletionQueueSafeHandle cq)
+ public async Task HandleCall(string methodName, CallSafeHandle call, CompletionQueueSafeHandle cq)
{
var asyncCall = new AsyncCallServer<TRequest, TResponse>(
method.ResponseMarshaller.Serializer,
method.RequestMarshaller.Deserializer);
asyncCall.Initialize(call);
+ var finishedTask = asyncCall.ServerSideCallAsync();
+ var requestStream = new ServerRequestStream<TRequest, TResponse>(asyncCall);
+ var responseStream = new ServerResponseStream<TRequest, TResponse>(asyncCall);
+
+ var request = await requestStream.ReadNext();
+ // TODO(jtattermusch): we need to read the full stream so that native callhandle gets deallocated.
+ Preconditions.CheckArgument(await requestStream.ReadNext() == null);
- var responseObserver = new ServerStreamingOutputObserver<TRequest, TResponse>(asyncCall);
- var requestObserver = handler(responseObserver);
- var finishedTask = asyncCall.ServerSideCallAsync(requestObserver);
- finishedTask.Wait();
+ await handler(request, responseStream);
+ await responseStream.WriteStatus(Status.DefaultSuccess);
+ await finishedTask;
}
}
- internal class NoSuchMethodCallHandler : IServerCallHandler
+ internal class ClientStreamingServerCallHandler<TRequest, TResponse> : IServerCallHandler
{
- public void StartCall(string methodName, CallSafeHandle call, CompletionQueueSafeHandle cq)
- {
- // We don't care about the payload type here.
- var asyncCall = new AsyncCallServer<byte[], byte[]>(
- (payload) => payload, (payload) => payload);
-
- asyncCall.Initialize(call);
+ readonly Method<TRequest, TResponse> method;
+ readonly ClientStreamingServerMethod<TRequest, TResponse> handler;
- var finishedTask = asyncCall.ServerSideCallAsync(new NullObserver<byte[]>());
+ public ClientStreamingServerCallHandler(Method<TRequest, TResponse> method, ClientStreamingServerMethod<TRequest, TResponse> handler)
+ {
+ this.method = method;
+ this.handler = handler;
+ }
- // TODO: check result of the completion status.
- asyncCall.StartSendStatusFromServer(new Status(StatusCode.Unimplemented, "No such method."), new AsyncCompletionDelegate((error) => { }));
+ public async Task HandleCall(string methodName, CallSafeHandle call, CompletionQueueSafeHandle cq)
+ {
+ var asyncCall = new AsyncCallServer<TRequest, TResponse>(
+ method.ResponseMarshaller.Serializer,
+ method.RequestMarshaller.Deserializer);
- finishedTask.Wait();
+ asyncCall.Initialize(call);
+ var finishedTask = asyncCall.ServerSideCallAsync();
+ var requestStream = new ServerRequestStream<TRequest, TResponse>(asyncCall);
+ var responseStream = new ServerResponseStream<TRequest, TResponse>(asyncCall);
+
+ var result = await handler(requestStream);
+ await responseStream.Write(result);
+ await responseStream.WriteStatus(Status.DefaultSuccess);
+ await finishedTask;
}
}
- internal class NullObserver<T> : IObserver<T>
+ internal class DuplexStreamingServerCallHandler<TRequest, TResponse> : IServerCallHandler
{
- public void OnCompleted()
+ readonly Method<TRequest, TResponse> method;
+ readonly DuplexStreamingServerMethod<TRequest, TResponse> handler;
+
+ public DuplexStreamingServerCallHandler(Method<TRequest, TResponse> method, DuplexStreamingServerMethod<TRequest, TResponse> handler)
{
+ this.method = method;
+ this.handler = handler;
}
- public void OnError(Exception error)
+ public async Task HandleCall(string methodName, CallSafeHandle call, CompletionQueueSafeHandle cq)
{
+ var asyncCall = new AsyncCallServer<TRequest, TResponse>(
+ method.ResponseMarshaller.Serializer,
+ method.RequestMarshaller.Deserializer);
+
+ asyncCall.Initialize(call);
+ var finishedTask = asyncCall.ServerSideCallAsync();
+ var requestStream = new ServerRequestStream<TRequest, TResponse>(asyncCall);
+ var responseStream = new ServerResponseStream<TRequest, TResponse>(asyncCall);
+
+ await handler(requestStream, responseStream);
+ await responseStream.WriteStatus(Status.DefaultSuccess);
+ await finishedTask;
}
+ }
- public void OnNext(T value)
+ internal class NoSuchMethodCallHandler : IServerCallHandler
+ {
+ public async Task HandleCall(string methodName, CallSafeHandle call, CompletionQueueSafeHandle cq)
{
+ // We don't care about the payload type here.
+ var asyncCall = new AsyncCallServer<byte[], byte[]>(
+ (payload) => payload, (payload) => payload);
+
+ asyncCall.Initialize(call);
+ var finishedTask = asyncCall.ServerSideCallAsync();
+ var requestStream = new ServerRequestStream<byte[], byte[]>(asyncCall);
+ var responseStream = new ServerResponseStream<byte[], byte[]>(asyncCall);
+ await responseStream.WriteStatus(new Status(StatusCode.Unimplemented, "No such method."));
+ // TODO(jtattermusch): if we don't read what client has sent, the server call never gets disposed.
+ await requestStream.ToList();
+ await finishedTask;
}
}
}
diff --git a/src/csharp/Grpc.Core/Internal/ServerCalls.cs b/src/csharp/Grpc.Core/Internal/ServerCalls.cs
new file mode 100644
index 0000000000000000000000000000000000000000..5c6b335c7f1c9ee5d01a39d8d3c3b844a4d9c2dc
--- /dev/null
+++ b/src/csharp/Grpc.Core/Internal/ServerCalls.cs
@@ -0,0 +1,63 @@
+#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;
+using System.Threading.Tasks;
+using Grpc.Core;
+
+namespace Grpc.Core.Internal
+{
+ internal static class ServerCalls
+ {
+ public static IServerCallHandler UnaryCall<TRequest, TResponse>(Method<TRequest, TResponse> method, UnaryServerMethod<TRequest, TResponse> handler)
+ {
+ return new UnaryServerCallHandler<TRequest, TResponse>(method, handler);
+ }
+
+ public static IServerCallHandler ClientStreamingCall<TRequest, TResponse>(Method<TRequest, TResponse> method, ClientStreamingServerMethod<TRequest, TResponse> handler)
+ {
+ return new ClientStreamingServerCallHandler<TRequest, TResponse>(method, handler);
+ }
+
+ public static IServerCallHandler ServerStreamingCall<TRequest, TResponse>(Method<TRequest, TResponse> method, ServerStreamingServerMethod<TRequest, TResponse> handler)
+ {
+ return new ServerStreamingServerCallHandler<TRequest, TResponse>(method, handler);
+ }
+
+ public static IServerCallHandler DuplexStreamingCall<TRequest, TResponse>(Method<TRequest, TResponse> method, DuplexStreamingServerMethod<TRequest, TResponse> handler)
+ {
+ return new DuplexStreamingServerCallHandler<TRequest, TResponse>(method, handler);
+ }
+ }
+}
diff --git a/src/csharp/Grpc.Core/ClientStreamingAsyncResult.cs b/src/csharp/Grpc.Core/Internal/ServerRequestStream.cs
similarity index 71%
rename from src/csharp/Grpc.Core/ClientStreamingAsyncResult.cs
rename to src/csharp/Grpc.Core/Internal/ServerRequestStream.cs
index 65bedb0a33f6a96234acf8d0aa6600caae5cd1bd..aa311059c32b1d3b0358c71d508262fc20aec427 100644
--- a/src/csharp/Grpc.Core/ClientStreamingAsyncResult.cs
+++ b/src/csharp/Grpc.Core/Internal/ServerRequestStream.cs
@@ -32,38 +32,25 @@
#endregion
using System;
+using System.Collections.Generic;
using System.Threading.Tasks;
-namespace Grpc.Core
+namespace Grpc.Core.Internal
{
- /// <summary>
- /// Return type for client streaming async method.
- /// </summary>
- public struct ClientStreamingAsyncResult<TRequest, TResponse>
+ internal class ServerRequestStream<TRequest, TResponse> : IAsyncStreamReader<TRequest>
{
- readonly Task<TResponse> task;
- readonly IObserver<TRequest> inputs;
+ readonly AsyncCallServer<TRequest, TResponse> call;
- public ClientStreamingAsyncResult(Task<TResponse> task, IObserver<TRequest> inputs)
+ public ServerRequestStream(AsyncCallServer<TRequest, TResponse> call)
{
- this.task = task;
- this.inputs = inputs;
+ this.call = call;
}
- public Task<TResponse> Task
+ public Task<TRequest> ReadNext()
{
- get
- {
- return this.task;
- }
- }
-
- public IObserver<TRequest> Inputs
- {
- get
- {
- return this.inputs;
- }
+ var taskSource = new AsyncCompletionTaskSource<TRequest>();
+ call.StartReadMessage(taskSource.CompletionDelegate);
+ return taskSource.Task;
}
}
}
diff --git a/src/csharp/Grpc.Core/ServerCalls.cs b/src/csharp/Grpc.Core/Internal/ServerResponseStream.cs
similarity index 64%
rename from src/csharp/Grpc.Core/ServerCalls.cs
rename to src/csharp/Grpc.Core/Internal/ServerResponseStream.cs
index dcae99446fe3e1474843f56a40119110bf18b151..686017c048dcf1ad3532f33df886d8eb7693c7fb 100644
--- a/src/csharp/Grpc.Core/ServerCalls.cs
+++ b/src/csharp/Grpc.Core/Internal/ServerResponseStream.cs
@@ -1,5 +1,4 @@
#region Copyright notice and license
-
// Copyright 2015, Google Inc.
// All rights reserved.
//
@@ -28,30 +27,38 @@
// 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 Grpc.Core.Internal;
-namespace Grpc.Core
+namespace Grpc.Core.Internal
{
- // TODO: perhaps add also serverSideStreaming and clientSideStreaming
-
- public delegate void UnaryRequestServerMethod<TRequest, TResponse>(TRequest request, IObserver<TResponse> responseObserver);
+ /// <summary>
+ /// Writes responses asynchronously to an underlying AsyncCallServer object.
+ /// </summary>
+ internal class ServerResponseStream<TRequest, TResponse> : IServerStreamWriter<TResponse>
+ {
+ readonly AsyncCallServer<TRequest, TResponse> call;
- public delegate IObserver<TRequest> StreamingRequestServerMethod<TRequest, TResponse>(IObserver<TResponse> responseObserver);
+ public ServerResponseStream(AsyncCallServer<TRequest, TResponse> call)
+ {
+ this.call = call;
+ }
- internal static class ServerCalls
- {
- public static IServerCallHandler UnaryRequestCall<TRequest, TResponse>(Method<TRequest, TResponse> method, UnaryRequestServerMethod<TRequest, TResponse> handler)
+ public Task Write(TResponse message)
{
- return new UnaryRequestServerCallHandler<TRequest, TResponse>(method, handler);
+ var taskSource = new AsyncCompletionTaskSource<object>();
+ call.StartSendMessage(message, taskSource.CompletionDelegate);
+ return taskSource.Task;
}
- public static IServerCallHandler StreamingRequestCall<TRequest, TResponse>(Method<TRequest, TResponse> method, StreamingRequestServerMethod<TRequest, TResponse> handler)
+ public Task WriteStatus(Status status)
{
- return new StreamingRequestServerCallHandler<TRequest, TResponse>(method, handler);
+ var taskSource = new AsyncCompletionTaskSource<object>();
+ call.StartSendStatusFromServer(status, taskSource.CompletionDelegate);
+ return taskSource.Task;
}
}
}
diff --git a/src/csharp/Grpc.Core/Method.cs b/src/csharp/Grpc.Core/Method.cs
index 4f97eeef37f1fb77b7fcb5bff71bebdce779faf0..156e780c7dd1d995cf1a9c6aa3388b56336b61f3 100644
--- a/src/csharp/Grpc.Core/Method.cs
+++ b/src/csharp/Grpc.Core/Method.cs
@@ -35,12 +35,15 @@ using System;
namespace Grpc.Core
{
+ /// <summary>
+ /// Method types supported by gRPC.
+ /// </summary>
public enum MethodType
{
- Unary,
- ClientStreaming,
- ServerStreaming,
- DuplexStreaming
+ Unary, // Unary request, unary response.
+ ClientStreaming, // Streaming request, unary response.
+ ServerStreaming, // Unary request, streaming response.
+ DuplexStreaming // Streaming request, streaming response.
}
/// <summary>
diff --git a/src/csharp/Grpc.Core/Server.cs b/src/csharp/Grpc.Core/Server.cs
index e686cdddef782c6bbfd16175f5778d6753d970ba..a3000cee462131ce90cc619e715bba07df6007bf 100644
--- a/src/csharp/Grpc.Core/Server.cs
+++ b/src/csharp/Grpc.Core/Server.cs
@@ -181,7 +181,7 @@ namespace Grpc.Core
/// <summary>
/// Selects corresponding handler for given call and handles the call.
/// </summary>
- private void InvokeCallHandler(CallSafeHandle call, string method)
+ private async Task InvokeCallHandler(CallSafeHandle call, string method)
{
try
{
@@ -190,7 +190,7 @@ namespace Grpc.Core
{
callHandler = new NoSuchMethodCallHandler();
}
- callHandler.StartCall(method, call, GetCompletionQueue());
+ await callHandler.HandleCall(method, call, GetCompletionQueue());
}
catch (Exception e)
{
@@ -218,7 +218,7 @@ namespace Grpc.Core
// after server shutdown, the callback returns with null call
if (!call.IsInvalid)
{
- Task.Run(() => InvokeCallHandler(call, method));
+ Task.Run(async () => await InvokeCallHandler(call, method));
}
AllowOneRpc();
diff --git a/src/csharp/Grpc.Core/ServerMethods.cs b/src/csharp/Grpc.Core/ServerMethods.cs
new file mode 100644
index 0000000000000000000000000000000000000000..6646bb5a89b4f82daee7ce4c1f869a17f9753b85
--- /dev/null
+++ b/src/csharp/Grpc.Core/ServerMethods.cs
@@ -0,0 +1,61 @@
+#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;
+using System.Threading.Tasks;
+
+using Grpc.Core.Internal;
+
+namespace Grpc.Core
+{
+ /// <summary>
+ /// Server-side handler for unary call.
+ /// </summary>
+ public delegate Task<TResponse> UnaryServerMethod<TRequest, TResponse>(TRequest request);
+
+ /// <summary>
+ /// Server-side handler for client streaming call.
+ /// </summary>
+ public delegate Task<TResponse> ClientStreamingServerMethod<TRequest, TResponse>(IAsyncStreamReader<TRequest> requestStream);
+
+ /// <summary>
+ /// Server-side handler for server streaming call.
+ /// </summary>
+ public delegate Task ServerStreamingServerMethod<TRequest, TResponse>(TRequest request, IServerStreamWriter<TResponse> responseStream);
+
+ /// <summary>
+ /// Server-side handler for bidi streaming call.
+ /// </summary>
+ public delegate Task DuplexStreamingServerMethod<TRequest, TResponse>(IAsyncStreamReader<TRequest> requestStream, IServerStreamWriter<TResponse> responseStream);
+}
diff --git a/src/csharp/Grpc.Core/ServerServiceDefinition.cs b/src/csharp/Grpc.Core/ServerServiceDefinition.cs
index f08c7d88f3ff9034ed9d9097f33b73eb2ba7e3ab..01b1dc8f7bdceaf82169b4a1c13e0dbd77ef9a3f 100644
--- a/src/csharp/Grpc.Core/ServerServiceDefinition.cs
+++ b/src/csharp/Grpc.Core/ServerServiceDefinition.cs
@@ -75,17 +75,33 @@ namespace Grpc.Core
public Builder AddMethod<TRequest, TResponse>(
Method<TRequest, TResponse> method,
- UnaryRequestServerMethod<TRequest, TResponse> handler)
+ UnaryServerMethod<TRequest, TResponse> handler)
{
- callHandlers.Add(GetFullMethodName(serviceName, method.Name), ServerCalls.UnaryRequestCall(method, handler));
+ callHandlers.Add(GetFullMethodName(serviceName, method.Name), ServerCalls.UnaryCall(method, handler));
return this;
}
public Builder AddMethod<TRequest, TResponse>(
Method<TRequest, TResponse> method,
- StreamingRequestServerMethod<TRequest, TResponse> handler)
+ ClientStreamingServerMethod<TRequest, TResponse> handler)
{
- callHandlers.Add(GetFullMethodName(serviceName, method.Name), ServerCalls.StreamingRequestCall(method, handler));
+ callHandlers.Add(GetFullMethodName(serviceName, method.Name), ServerCalls.ClientStreamingCall(method, handler));
+ return this;
+ }
+
+ public Builder AddMethod<TRequest, TResponse>(
+ Method<TRequest, TResponse> method,
+ ServerStreamingServerMethod<TRequest, TResponse> handler)
+ {
+ callHandlers.Add(GetFullMethodName(serviceName, method.Name), ServerCalls.ServerStreamingCall(method, handler));
+ return this;
+ }
+
+ public Builder AddMethod<TRequest, TResponse>(
+ Method<TRequest, TResponse> method,
+ DuplexStreamingServerMethod<TRequest, TResponse> handler)
+ {
+ callHandlers.Add(GetFullMethodName(serviceName, method.Name), ServerCalls.DuplexStreamingCall(method, handler));
return this;
}
diff --git a/src/csharp/Grpc.Core/Status.cs b/src/csharp/Grpc.Core/Status.cs
index 7d76aec4d1198dda8969c79cede6ff0dbe3e556b..b5881706941defb7e8796ec1ebd5b07684446267 100644
--- a/src/csharp/Grpc.Core/Status.cs
+++ b/src/csharp/Grpc.Core/Status.cs
@@ -39,6 +39,11 @@ namespace Grpc.Core
/// </summary>
public struct Status
{
+ /// <summary>
+ /// Default result of a successful RPC. StatusCode=OK, empty details message.
+ /// </summary>
+ public static readonly Status DefaultSuccess = new Status(StatusCode.OK, "");
+
readonly StatusCode statusCode;
readonly string detail;
diff --git a/src/csharp/Grpc.Core/Utils/AsyncStreamExtensions.cs b/src/csharp/Grpc.Core/Utils/AsyncStreamExtensions.cs
new file mode 100644
index 0000000000000000000000000000000000000000..f915155f8a4094ee8c0b6531017a3df7f16b49e7
--- /dev/null
+++ b/src/csharp/Grpc.Core/Utils/AsyncStreamExtensions.cs
@@ -0,0 +1,111 @@
+#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.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+
+namespace Grpc.Core.Utils
+{
+ /// <summary>
+ /// Extension methods that simplify work with gRPC streaming calls.
+ /// </summary>
+ public static class AsyncStreamExtensions
+ {
+ /// <summary>
+ /// Reads the entire stream and executes an async action for each element.
+ /// </summary>
+ public static async Task ForEach<T>(this IAsyncStreamReader<T> streamReader, Func<T, Task> asyncAction)
+ where T : class
+ {
+ while (true)
+ {
+ var elem = await streamReader.ReadNext();
+ if (elem == null)
+ {
+ break;
+ }
+ await asyncAction(elem);
+ }
+ }
+
+ /// <summary>
+ /// Reads the entire stream and creates a list containing all the elements read.
+ /// </summary>
+ public static async Task<List<T>> ToList<T>(this IAsyncStreamReader<T> streamReader)
+ where T : class
+ {
+ var result = new List<T>();
+ while (true)
+ {
+ var elem = await streamReader.ReadNext();
+ if (elem == null)
+ {
+ break;
+ }
+ result.Add(elem);
+ }
+ return result;
+ }
+
+ /// <summary>
+ /// Writes all elements from given enumerable to the stream.
+ /// Closes the stream afterwards unless close = false.
+ /// </summary>
+ public static async Task WriteAll<T>(this IClientStreamWriter<T> streamWriter, IEnumerable<T> elements, bool close = true)
+ where T : class
+ {
+ foreach (var element in elements)
+ {
+ await streamWriter.Write(element);
+ }
+ if (close)
+ {
+ await streamWriter.Close();
+ }
+ }
+
+ /// <summary>
+ /// Writes all elements from given enumerable to the stream.
+ /// </summary>
+ public static async Task WriteAll<T>(this IServerStreamWriter<T> streamWriter, IEnumerable<T> elements)
+ where T : class
+ {
+ foreach (var element in elements)
+ {
+ await streamWriter.Write(element);
+ }
+ }
+ }
+}
diff --git a/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs b/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs
index fa5d6688a6bdcaf1ddec742b69cf9c213c02ee3a..332795e0e54c1516ee421fc033db554d88ded72b 100644
--- a/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs
+++ b/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs
@@ -63,7 +63,7 @@ namespace math.Tests
server.Start();
channel = new Channel(host + ":" + port);
- // TODO: get rid of the custom header here once we have dedicated tests
+ // TODO(jtattermusch): get rid of the custom header here once we have dedicated tests
// for header support.
var stubConfig = new StubConfiguration((headerBuilder) =>
{
@@ -97,55 +97,67 @@ namespace math.Tests
Assert.AreEqual(0, response.Remainder);
}
- // TODO: test division by zero
+ // TODO(jtattermusch): test division by zero
[Test]
public void DivAsync()
{
- DivReply response = client.DivAsync(new DivArgs.Builder { Dividend = 10, Divisor = 3 }.Build()).Result;
- Assert.AreEqual(3, response.Quotient);
- Assert.AreEqual(1, response.Remainder);
+ Task.Run(async () =>
+ {
+ DivReply response = await client.DivAsync(new DivArgs.Builder { Dividend = 10, Divisor = 3 }.Build());
+ Assert.AreEqual(3, response.Quotient);
+ Assert.AreEqual(1, response.Remainder);
+ }).Wait();
}
[Test]
public void Fib()
{
- var recorder = new RecordingObserver<Num>();
- client.Fib(new FibArgs.Builder { Limit = 6 }.Build(), recorder);
+ Task.Run(async () =>
+ {
+ var call = client.Fib(new FibArgs.Builder { Limit = 6 }.Build());
- CollectionAssert.AreEqual(new List<long> { 1, 1, 2, 3, 5, 8 },
- recorder.ToList().Result.ConvertAll((n) => n.Num_));
+ var responses = await call.ResponseStream.ToList();
+ CollectionAssert.AreEqual(new List<long> { 1, 1, 2, 3, 5, 8 },
+ responses.ConvertAll((n) => n.Num_));
+ }).Wait();
}
// TODO: test Fib with limit=0 and cancellation
[Test]
public void Sum()
{
- var clientStreamingResult = client.Sum();
- var numList = new List<long> { 10, 20, 30 }.ConvertAll(
- n => Num.CreateBuilder().SetNum_(n).Build());
- numList.Subscribe(clientStreamingResult.Inputs);
-
- Assert.AreEqual(60, clientStreamingResult.Task.Result.Num_);
+ Task.Run(async () =>
+ {
+ var call = client.Sum();
+ var numbers = new List<long> { 10, 20, 30 }.ConvertAll(
+ n => Num.CreateBuilder().SetNum_(n).Build());
+
+ await call.RequestStream.WriteAll(numbers);
+ var result = await call.Result;
+ Assert.AreEqual(60, result.Num_);
+ }).Wait();
}
[Test]
public void DivMany()
{
- List<DivArgs> divArgsList = new List<DivArgs>
+ Task.Run(async () =>
{
- 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>();
- var requestObserver = client.DivMany(recorder);
- divArgsList.Subscribe(requestObserver);
- var result = recorder.ToList().Result;
-
- CollectionAssert.AreEqual(new long[] { 3, 4, 3 }, result.ConvertAll((divReply) => divReply.Quotient));
- CollectionAssert.AreEqual(new long[] { 1, 16, 1 }, result.ConvertAll((divReply) => divReply.Remainder));
+ var 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 call = client.DivMany();
+ await call.RequestStream.WriteAll(divArgsList);
+ var result = await call.ResponseStream.ToList();
+
+ CollectionAssert.AreEqual(new long[] { 3, 4, 3 }, result.ConvertAll((divReply) => divReply.Quotient));
+ CollectionAssert.AreEqual(new long[] { 1, 16, 1 }, result.ConvertAll((divReply) => divReply.Remainder));
+ }).Wait();
}
}
}
diff --git a/src/csharp/Grpc.Examples/MathExamples.cs b/src/csharp/Grpc.Examples/MathExamples.cs
index 032372b2a119f25509b6765560ceee1329c2b06b..dba5a7736cd4f8855c8e3717a80e6e3bcce8d894 100644
--- a/src/csharp/Grpc.Examples/MathExamples.cs
+++ b/src/csharp/Grpc.Examples/MathExamples.cs
@@ -61,9 +61,8 @@ namespace math
public static async Task FibExample(MathGrpc.IMathServiceClient stub)
{
- var recorder = new RecordingObserver<Num>();
- stub.Fib(new FibArgs.Builder { Limit = 5 }.Build(), recorder);
- List<Num> result = await recorder.ToList();
+ var call = stub.Fib(new FibArgs.Builder { Limit = 5 }.Build());
+ List<Num> result = await call.ResponseStream.ToList();
Console.WriteLine("Fib Result: " + string.Join("|", result));
}
@@ -76,9 +75,9 @@ namespace math
new Num.Builder { Num_ = 3 }.Build()
};
- var clientStreamingResult = stub.Sum();
- numbers.Subscribe(clientStreamingResult.Inputs);
- Console.WriteLine("Sum Result: " + await clientStreamingResult.Task);
+ var call = stub.Sum();
+ await call.RequestStream.WriteAll(numbers);
+ Console.WriteLine("Sum Result: " + await call.Result);
}
public static async Task DivManyExample(MathGrpc.IMathServiceClient stub)
@@ -89,12 +88,9 @@ namespace math
new DivArgs.Builder { Dividend = 100, Divisor = 21 }.Build(),
new DivArgs.Builder { Dividend = 7, Divisor = 2 }.Build()
};
-
- var recorder = new RecordingObserver<DivReply>();
- var inputs = stub.DivMany(recorder);
- divArgsList.Subscribe(inputs);
- var result = await recorder.ToList();
- Console.WriteLine("DivMany Result: " + string.Join("|", result));
+ var call = stub.DivMany();
+ await call.RequestStream.WriteAll(divArgsList);
+ Console.WriteLine("DivMany Result: " + string.Join("|", await call.ResponseStream.ToList()));
}
public static async Task DependendRequestsExample(MathGrpc.IMathServiceClient stub)
@@ -106,9 +102,9 @@ namespace math
new Num.Builder { Num_ = 3 }.Build()
};
- var clientStreamingResult = stub.Sum();
- numbers.Subscribe(clientStreamingResult.Inputs);
- Num sum = await clientStreamingResult.Task;
+ var sumCall = stub.Sum();
+ await sumCall.RequestStream.WriteAll(numbers);
+ Num sum = await sumCall.Result;
DivReply result = await stub.DivAsync(new DivArgs.Builder { Dividend = sum.Num_, Divisor = numbers.Count }.Build());
Console.WriteLine("Avg Result: " + result);
diff --git a/src/csharp/Grpc.Examples/MathGrpc.cs b/src/csharp/Grpc.Examples/MathGrpc.cs
index 24e6a1de8e397bf2ae48f968862260bb96c4edd4..60408b901848e94be15f2d14021594c2bccf54d6 100644
--- a/src/csharp/Grpc.Examples/MathGrpc.cs
+++ b/src/csharp/Grpc.Examples/MathGrpc.cs
@@ -82,11 +82,11 @@ namespace math
Task<DivReply> DivAsync(DivArgs request, CancellationToken token = default(CancellationToken));
- void Fib(FibArgs request, IObserver<Num> responseObserver, CancellationToken token = default(CancellationToken));
+ AsyncServerStreamingCall<Num> Fib(FibArgs request, CancellationToken token = default(CancellationToken));
- ClientStreamingAsyncResult<Num, Num> Sum(CancellationToken token = default(CancellationToken));
+ AsyncClientStreamingCall<Num, Num> Sum(CancellationToken token = default(CancellationToken));
- IObserver<DivArgs> DivMany(IObserver<DivReply> responseObserver, CancellationToken token = default(CancellationToken));
+ AsyncDuplexStreamingCall<DivArgs, DivReply> DivMany(CancellationToken token = default(CancellationToken));
}
public class MathServiceClientStub : AbstractStub<MathServiceClientStub, StubConfiguration>, IMathServiceClient
@@ -111,35 +111,35 @@ namespace math
return Calls.AsyncUnaryCall(call, request, token);
}
- public void Fib(FibArgs request, IObserver<Num> responseObserver, CancellationToken token = default(CancellationToken))
+ public AsyncServerStreamingCall<Num> Fib(FibArgs request, CancellationToken token = default(CancellationToken))
{
var call = CreateCall(ServiceName, FibMethod);
- Calls.AsyncServerStreamingCall(call, request, responseObserver, token);
+ return Calls.AsyncServerStreamingCall(call, request, token);
}
- public ClientStreamingAsyncResult<Num, Num> Sum(CancellationToken token = default(CancellationToken))
+ public AsyncClientStreamingCall<Num, Num> Sum(CancellationToken token = default(CancellationToken))
{
var call = CreateCall(ServiceName, SumMethod);
return Calls.AsyncClientStreamingCall(call, token);
}
- public IObserver<DivArgs> DivMany(IObserver<DivReply> responseObserver, CancellationToken token = default(CancellationToken))
+ public AsyncDuplexStreamingCall<DivArgs, DivReply> DivMany(CancellationToken token = default(CancellationToken))
{
var call = CreateCall(ServiceName, DivManyMethod);
- return Calls.DuplexStreamingCall(call, responseObserver, token);
+ return Calls.AsyncDuplexStreamingCall(call, token);
}
}
// server-side interface
public interface IMathService
{
- void Div(DivArgs request, IObserver<DivReply> responseObserver);
+ Task<DivReply> Div(DivArgs request);
- void Fib(FibArgs request, IObserver<Num> responseObserver);
+ Task Fib(FibArgs request, IServerStreamWriter<Num> responseStream);
- IObserver<Num> Sum(IObserver<Num> responseObserver);
+ Task<Num> Sum(IAsyncStreamReader<Num> requestStream);
- IObserver<DivArgs> DivMany(IObserver<DivReply> responseObserver);
+ Task DivMany(IAsyncStreamReader<DivArgs> requestStream, IServerStreamWriter<DivReply> responseStream);
}
public static ServerServiceDefinition BindService(IMathService serviceImpl)
diff --git a/src/csharp/Grpc.Examples/MathServiceImpl.cs b/src/csharp/Grpc.Examples/MathServiceImpl.cs
index 0b2357e0fade71eab84e6e23dede9b24b6e73ef7..83ec2a8c3df20fcda82be12209088e0812adadc6 100644
--- a/src/csharp/Grpc.Examples/MathServiceImpl.cs
+++ b/src/csharp/Grpc.Examples/MathServiceImpl.cs
@@ -36,6 +36,7 @@ using System.Collections.Generic;
using System.Reactive.Linq;
using System.Threading;
using System.Threading.Tasks;
+using Grpc.Core;
using Grpc.Core.Utils;
namespace math
@@ -45,18 +46,16 @@ namespace math
/// </summary>
public class MathServiceImpl : MathGrpc.IMathService
{
- public void Div(DivArgs request, IObserver<DivReply> responseObserver)
+ public Task<DivReply> Div(DivArgs request)
{
- var response = DivInternal(request);
- responseObserver.OnNext(response);
- responseObserver.OnCompleted();
+ return Task.FromResult(DivInternal(request));
}
- public void Fib(FibArgs request, IObserver<Num> responseObserver)
+ public async Task Fib(FibArgs request, IServerStreamWriter<Num> responseStream)
{
if (request.Limit <= 0)
{
- // TODO: support cancellation....
+ // TODO(jtattermusch): support cancellation
throw new NotImplementedException("Not implemented yet");
}
@@ -64,34 +63,27 @@ namespace math
{
foreach (var num in FibInternal(request.Limit))
{
- responseObserver.OnNext(num);
+ await responseStream.Write(num);
}
- responseObserver.OnCompleted();
}
}
- public IObserver<Num> Sum(IObserver<Num> responseObserver)
+ public async Task<Num> Sum(IAsyncStreamReader<Num> requestStream)
{
- var recorder = new RecordingObserver<Num>();
- Task.Factory.StartNew(() =>
+ long sum = 0;
+ await requestStream.ForEach(async num =>
{
- List<Num> inputs = recorder.ToList().Result;
-
- long sum = 0;
- foreach (Num num in inputs)
- {
- sum += num.Num_;
- }
-
- responseObserver.OnNext(Num.CreateBuilder().SetNum_(sum).Build());
- responseObserver.OnCompleted();
+ sum += num.Num_;
});
- return recorder;
+ return Num.CreateBuilder().SetNum_(sum).Build();
}
- public IObserver<DivArgs> DivMany(IObserver<DivReply> responseObserver)
+ public async Task DivMany(IAsyncStreamReader<DivArgs> requestStream, IServerStreamWriter<DivReply> responseStream)
{
- return new DivObserver(responseObserver);
+ await requestStream.ForEach(async divArgs =>
+ {
+ await responseStream.Write(DivInternal(divArgs));
+ });
}
static DivReply DivInternal(DivArgs args)
@@ -114,31 +106,6 @@ namespace math
b = temp + b;
yield return new Num.Builder { Num_ = a }.Build();
}
- }
-
- private class DivObserver : IObserver<DivArgs>
- {
- readonly IObserver<DivReply> responseObserver;
-
- public DivObserver(IObserver<DivReply> responseObserver)
- {
- this.responseObserver = responseObserver;
- }
-
- public void OnCompleted()
- {
- responseObserver.OnCompleted();
- }
-
- public void OnError(Exception error)
- {
- throw new NotImplementedException();
- }
-
- public void OnNext(DivArgs value)
- {
- responseObserver.OnNext(DivInternal(value));
- }
- }
+ }
}
}
diff --git a/src/csharp/Grpc.IntegrationTesting/InteropClient.cs b/src/csharp/Grpc.IntegrationTesting/InteropClient.cs
index 1fbae374b1d61372f48371b197a81fbbb797280b..573ab3045226ca07ec292b2a6a11cc82811426fa 100644
--- a/src/csharp/Grpc.IntegrationTesting/InteropClient.cs
+++ b/src/csharp/Grpc.IntegrationTesting/InteropClient.cs
@@ -34,6 +34,7 @@
using System;
using System.Collections.Generic;
using System.Text.RegularExpressions;
+using System.Threading.Tasks;
using Google.ProtocolBuffers;
using grpc.testing;
@@ -199,113 +200,115 @@ namespace Grpc.IntegrationTesting
public static void RunClientStreaming(TestServiceGrpc.ITestServiceClient client)
{
- Console.WriteLine("running client_streaming");
+ Task.Run(async () =>
+ {
+ Console.WriteLine("running client_streaming");
- var bodySizes = new List<int> { 27182, 8, 1828, 45904 };
+ var bodySizes = new List<int> { 27182, 8, 1828, 45904 }.ConvertAll((size) => StreamingInputCallRequest.CreateBuilder().SetPayload(CreateZerosPayload(size)).Build());
- var context = client.StreamingInputCall();
- foreach (var size in bodySizes)
- {
- context.Inputs.OnNext(
- StreamingInputCallRequest.CreateBuilder().SetPayload(CreateZerosPayload(size)).Build());
- }
- context.Inputs.OnCompleted();
+ var call = client.StreamingInputCall();
+ await call.RequestStream.WriteAll(bodySizes);
- var response = context.Task.Result;
- Assert.AreEqual(74922, response.AggregatedPayloadSize);
- Console.WriteLine("Passed!");
+ var response = await call.Result;
+ Assert.AreEqual(74922, response.AggregatedPayloadSize);
+ Console.WriteLine("Passed!");
+ }).Wait();
}
public static void RunServerStreaming(TestServiceGrpc.ITestServiceClient client)
{
- Console.WriteLine("running server_streaming");
+ Task.Run(async () =>
+ {
+ Console.WriteLine("running server_streaming");
- var bodySizes = new List<int> { 31415, 9, 2653, 58979 };
+ var bodySizes = new List<int> { 31415, 9, 2653, 58979 };
- var request = StreamingOutputCallRequest.CreateBuilder()
+ var request = StreamingOutputCallRequest.CreateBuilder()
.SetResponseType(PayloadType.COMPRESSABLE)
.AddRangeResponseParameters(bodySizes.ConvertAll(
- (size) => ResponseParameters.CreateBuilder().SetSize(size).Build()))
+ (size) => ResponseParameters.CreateBuilder().SetSize(size).Build()))
.Build();
- var recorder = new RecordingObserver<StreamingOutputCallResponse>();
- client.StreamingOutputCall(request, recorder);
+ var call = client.StreamingOutputCall(request);
- var responseList = recorder.ToList().Result;
-
- foreach (var res in responseList)
- {
- Assert.AreEqual(PayloadType.COMPRESSABLE, res.Payload.Type);
- }
- CollectionAssert.AreEqual(bodySizes, responseList.ConvertAll((item) => item.Payload.Body.Length));
- Console.WriteLine("Passed!");
+ var responseList = await call.ResponseStream.ToList();
+ foreach (var res in responseList)
+ {
+ Assert.AreEqual(PayloadType.COMPRESSABLE, res.Payload.Type);
+ }
+ CollectionAssert.AreEqual(bodySizes, responseList.ConvertAll((item) => item.Payload.Body.Length));
+ Console.WriteLine("Passed!");
+ }).Wait();
}
public static void RunPingPong(TestServiceGrpc.ITestServiceClient client)
{
- Console.WriteLine("running ping_pong");
+ Task.Run(async () =>
+ {
+ Console.WriteLine("running ping_pong");
- var recorder = new RecordingQueue<StreamingOutputCallResponse>();
- var inputs = client.FullDuplexCall(recorder);
+ var call = client.FullDuplexCall();
- StreamingOutputCallResponse response;
+ StreamingOutputCallResponse response;
- inputs.OnNext(StreamingOutputCallRequest.CreateBuilder()
+ await call.RequestStream.Write(StreamingOutputCallRequest.CreateBuilder()
.SetResponseType(PayloadType.COMPRESSABLE)
.AddResponseParameters(ResponseParameters.CreateBuilder().SetSize(31415))
.SetPayload(CreateZerosPayload(27182)).Build());
- response = recorder.Queue.Take();
- Assert.AreEqual(PayloadType.COMPRESSABLE, response.Payload.Type);
- Assert.AreEqual(31415, response.Payload.Body.Length);
+ response = await call.ResponseStream.ReadNext();
+ Assert.AreEqual(PayloadType.COMPRESSABLE, response.Payload.Type);
+ Assert.AreEqual(31415, response.Payload.Body.Length);
- inputs.OnNext(StreamingOutputCallRequest.CreateBuilder()
+ await call.RequestStream.Write(StreamingOutputCallRequest.CreateBuilder()
.SetResponseType(PayloadType.COMPRESSABLE)
.AddResponseParameters(ResponseParameters.CreateBuilder().SetSize(9))
.SetPayload(CreateZerosPayload(8)).Build());
- response = recorder.Queue.Take();
- Assert.AreEqual(PayloadType.COMPRESSABLE, response.Payload.Type);
- Assert.AreEqual(9, response.Payload.Body.Length);
+ response = await call.ResponseStream.ReadNext();
+ Assert.AreEqual(PayloadType.COMPRESSABLE, response.Payload.Type);
+ Assert.AreEqual(9, response.Payload.Body.Length);
- inputs.OnNext(StreamingOutputCallRequest.CreateBuilder()
+ await call.RequestStream.Write(StreamingOutputCallRequest.CreateBuilder()
.SetResponseType(PayloadType.COMPRESSABLE)
.AddResponseParameters(ResponseParameters.CreateBuilder().SetSize(2653))
.SetPayload(CreateZerosPayload(1828)).Build());
- response = recorder.Queue.Take();
- Assert.AreEqual(PayloadType.COMPRESSABLE, response.Payload.Type);
- Assert.AreEqual(2653, response.Payload.Body.Length);
+ response = await call.ResponseStream.ReadNext();
+ Assert.AreEqual(PayloadType.COMPRESSABLE, response.Payload.Type);
+ Assert.AreEqual(2653, response.Payload.Body.Length);
- inputs.OnNext(StreamingOutputCallRequest.CreateBuilder()
+ await call.RequestStream.Write(StreamingOutputCallRequest.CreateBuilder()
.SetResponseType(PayloadType.COMPRESSABLE)
.AddResponseParameters(ResponseParameters.CreateBuilder().SetSize(58979))
.SetPayload(CreateZerosPayload(45904)).Build());
- response = recorder.Queue.Take();
- Assert.AreEqual(PayloadType.COMPRESSABLE, response.Payload.Type);
- Assert.AreEqual(58979, response.Payload.Body.Length);
+ response = await call.ResponseStream.ReadNext();
+ Assert.AreEqual(PayloadType.COMPRESSABLE, response.Payload.Type);
+ Assert.AreEqual(58979, response.Payload.Body.Length);
- inputs.OnCompleted();
+ await call.RequestStream.Close();
- recorder.Finished.Wait();
- Assert.AreEqual(0, recorder.Queue.Count);
+ response = await call.ResponseStream.ReadNext();
+ Assert.AreEqual(null, response);
- Console.WriteLine("Passed!");
+ Console.WriteLine("Passed!");
+ }).Wait();
}
public static void RunEmptyStream(TestServiceGrpc.ITestServiceClient client)
{
- Console.WriteLine("running empty_stream");
-
- var recorder = new RecordingObserver<StreamingOutputCallResponse>();
- var inputs = client.FullDuplexCall(recorder);
- inputs.OnCompleted();
+ Task.Run(async () =>
+ {
+ Console.WriteLine("running empty_stream");
+ var call = client.FullDuplexCall();
+ await call.Close();
- var responseList = recorder.ToList().Result;
- Assert.AreEqual(0, responseList.Count);
+ var responseList = await call.ResponseStream.ToList();
+ Assert.AreEqual(0, responseList.Count);
- Console.WriteLine("Passed!");
+ Console.WriteLine("Passed!");
+ }).Wait();
}
public static void RunServiceAccountCreds(TestServiceGrpc.ITestServiceClient client)
diff --git a/src/csharp/Grpc.IntegrationTesting/InteropClientServerTest.cs b/src/csharp/Grpc.IntegrationTesting/InteropClientServerTest.cs
index 1e76d3df2134873a5ad79295377a3647b7793490..e929b76b5ed15ca71cfd6b955ab02d0f837a852a 100644
--- a/src/csharp/Grpc.IntegrationTesting/InteropClientServerTest.cs
+++ b/src/csharp/Grpc.IntegrationTesting/InteropClientServerTest.cs
@@ -87,7 +87,7 @@ namespace Grpc.IntegrationTesting
[Test]
public void LargeUnary()
{
- InteropClient.RunEmptyUnary(client);
+ InteropClient.RunLargeUnary(client);
}
[Test]
diff --git a/src/csharp/Grpc.IntegrationTesting/TestServiceGrpc.cs b/src/csharp/Grpc.IntegrationTesting/TestServiceGrpc.cs
index f63e0361a451b36ade4d15c91064629b0fe8a2b6..d1f8aa12c78187f806a317ee455b5967f0f30f1e 100644
--- a/src/csharp/Grpc.IntegrationTesting/TestServiceGrpc.cs
+++ b/src/csharp/Grpc.IntegrationTesting/TestServiceGrpc.cs
@@ -100,13 +100,13 @@ namespace grpc.testing
Task<SimpleResponse> UnaryCallAsync(SimpleRequest request, CancellationToken token = default(CancellationToken));
- void StreamingOutputCall(StreamingOutputCallRequest request, IObserver<StreamingOutputCallResponse> responseObserver, CancellationToken token = default(CancellationToken));
+ AsyncServerStreamingCall<StreamingOutputCallResponse> StreamingOutputCall(StreamingOutputCallRequest request, CancellationToken token = default(CancellationToken));
- ClientStreamingAsyncResult<StreamingInputCallRequest, StreamingInputCallResponse> StreamingInputCall(CancellationToken token = default(CancellationToken));
+ AsyncClientStreamingCall<StreamingInputCallRequest, StreamingInputCallResponse> StreamingInputCall(CancellationToken token = default(CancellationToken));
- IObserver<StreamingOutputCallRequest> FullDuplexCall(IObserver<StreamingOutputCallResponse> responseObserver, CancellationToken token = default(CancellationToken));
+ AsyncDuplexStreamingCall<StreamingOutputCallRequest, StreamingOutputCallResponse> FullDuplexCall(CancellationToken token = default(CancellationToken));
- IObserver<StreamingOutputCallRequest> HalfDuplexCall(IObserver<StreamingOutputCallResponse> responseObserver, CancellationToken token = default(CancellationToken));
+ AsyncDuplexStreamingCall<StreamingOutputCallRequest, StreamingOutputCallResponse> HalfDuplexCall(CancellationToken token = default(CancellationToken));
}
public class TestServiceClientStub : AbstractStub<TestServiceClientStub, StubConfiguration>, ITestServiceClient
@@ -143,45 +143,45 @@ namespace grpc.testing
return Calls.AsyncUnaryCall(call, request, token);
}
- public void StreamingOutputCall(StreamingOutputCallRequest request, IObserver<StreamingOutputCallResponse> responseObserver, CancellationToken token = default(CancellationToken))
+ public AsyncServerStreamingCall<StreamingOutputCallResponse> StreamingOutputCall(StreamingOutputCallRequest request, CancellationToken token = default(CancellationToken))
{
var call = CreateCall(ServiceName, StreamingOutputCallMethod);
- Calls.AsyncServerStreamingCall(call, request, responseObserver, token);
+ return Calls.AsyncServerStreamingCall(call, request, token);
}
- public ClientStreamingAsyncResult<StreamingInputCallRequest, StreamingInputCallResponse> StreamingInputCall(CancellationToken token = default(CancellationToken))
+ public AsyncClientStreamingCall<StreamingInputCallRequest, StreamingInputCallResponse> StreamingInputCall(CancellationToken token = default(CancellationToken))
{
var call = CreateCall(ServiceName, StreamingInputCallMethod);
return Calls.AsyncClientStreamingCall(call, token);
}
- public IObserver<StreamingOutputCallRequest> FullDuplexCall(IObserver<StreamingOutputCallResponse> responseObserver, CancellationToken token = default(CancellationToken))
+ public AsyncDuplexStreamingCall<StreamingOutputCallRequest, StreamingOutputCallResponse> FullDuplexCall(CancellationToken token = default(CancellationToken))
{
var call = CreateCall(ServiceName, FullDuplexCallMethod);
- return Calls.DuplexStreamingCall(call, responseObserver, token);
+ return Calls.AsyncDuplexStreamingCall(call, token);
}
- public IObserver<StreamingOutputCallRequest> HalfDuplexCall(IObserver<StreamingOutputCallResponse> responseObserver, CancellationToken token = default(CancellationToken))
+ public AsyncDuplexStreamingCall<StreamingOutputCallRequest, StreamingOutputCallResponse> HalfDuplexCall(CancellationToken token = default(CancellationToken))
{
var call = CreateCall(ServiceName, HalfDuplexCallMethod);
- return Calls.DuplexStreamingCall(call, responseObserver, token);
+ return Calls.AsyncDuplexStreamingCall(call, token);
}
}
// server-side interface
public interface ITestService
{
- void EmptyCall(Empty request, IObserver<Empty> responseObserver);
+ Task<Empty> EmptyCall(Empty request);
- void UnaryCall(SimpleRequest request, IObserver<SimpleResponse> responseObserver);
+ Task<SimpleResponse> UnaryCall(SimpleRequest request);
- void StreamingOutputCall(StreamingOutputCallRequest request, IObserver<StreamingOutputCallResponse> responseObserver);
+ Task StreamingOutputCall(StreamingOutputCallRequest request, IServerStreamWriter<StreamingOutputCallResponse> responseStream);
- IObserver<StreamingInputCallRequest> StreamingInputCall(IObserver<StreamingInputCallResponse> responseObserver);
+ Task<StreamingInputCallResponse> StreamingInputCall(IAsyncStreamReader<StreamingInputCallRequest> requestStream);
- IObserver<StreamingOutputCallRequest> FullDuplexCall(IObserver<StreamingOutputCallResponse> responseObserver);
+ Task FullDuplexCall(IAsyncStreamReader<StreamingOutputCallRequest> requestStream, IServerStreamWriter<StreamingOutputCallResponse> responseStream);
- IObserver<StreamingOutputCallRequest> HalfDuplexCall(IObserver<StreamingOutputCallResponse> responseObserver);
+ Task HalfDuplexCall(IAsyncStreamReader<StreamingOutputCallRequest> requestStream, IServerStreamWriter<StreamingOutputCallResponse> responseStream);
}
public static ServerServiceDefinition BindService(ITestService serviceImpl)
diff --git a/src/csharp/Grpc.IntegrationTesting/TestServiceImpl.cs b/src/csharp/Grpc.IntegrationTesting/TestServiceImpl.cs
index 661b31b0ee67f909c95aa9a533b97054c900b760..8b0cf3a2d0596c8319f8db6dfd77478701999071 100644
--- a/src/csharp/Grpc.IntegrationTesting/TestServiceImpl.cs
+++ b/src/csharp/Grpc.IntegrationTesting/TestServiceImpl.cs
@@ -36,6 +36,7 @@ using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Google.ProtocolBuffers;
+using Grpc.Core;
using Grpc.Core.Utils;
namespace grpc.testing
@@ -45,88 +46,54 @@ namespace grpc.testing
/// </summary>
public class TestServiceImpl : TestServiceGrpc.ITestService
{
- public void EmptyCall(Empty request, IObserver<Empty> responseObserver)
+ public Task<Empty> EmptyCall(Empty request)
{
- responseObserver.OnNext(Empty.DefaultInstance);
- responseObserver.OnCompleted();
+ return Task.FromResult(Empty.DefaultInstance);
}
- public void UnaryCall(SimpleRequest request, IObserver<SimpleResponse> responseObserver)
+ public Task<SimpleResponse> UnaryCall(SimpleRequest request)
{
var response = SimpleResponse.CreateBuilder()
.SetPayload(CreateZerosPayload(request.ResponseSize)).Build();
- // TODO: check we support ReponseType
- responseObserver.OnNext(response);
- responseObserver.OnCompleted();
+ return Task.FromResult(response);
}
- public void StreamingOutputCall(StreamingOutputCallRequest request, IObserver<StreamingOutputCallResponse> responseObserver)
+ public async Task StreamingOutputCall(StreamingOutputCallRequest request, IServerStreamWriter<StreamingOutputCallResponse> responseStream)
{
foreach (var responseParam in request.ResponseParametersList)
{
var response = StreamingOutputCallResponse.CreateBuilder()
.SetPayload(CreateZerosPayload(responseParam.Size)).Build();
- responseObserver.OnNext(response);
+ await responseStream.Write(response);
}
- responseObserver.OnCompleted();
}
- public IObserver<StreamingInputCallRequest> StreamingInputCall(IObserver<StreamingInputCallResponse> responseObserver)
+ public async Task<StreamingInputCallResponse> StreamingInputCall(IAsyncStreamReader<StreamingInputCallRequest> requestStream)
{
- var recorder = new RecordingObserver<StreamingInputCallRequest>();
- Task.Run(() =>
+ int sum = 0;
+ await requestStream.ForEach(async request =>
{
- int sum = 0;
- foreach (var req in recorder.ToList().Result)
- {
- sum += req.Payload.Body.Length;
- }
- var response = StreamingInputCallResponse.CreateBuilder()
- .SetAggregatedPayloadSize(sum).Build();
- responseObserver.OnNext(response);
- responseObserver.OnCompleted();
+ sum += request.Payload.Body.Length;
});
- return recorder;
+ return StreamingInputCallResponse.CreateBuilder().SetAggregatedPayloadSize(sum).Build();
}
- public IObserver<StreamingOutputCallRequest> FullDuplexCall(IObserver<StreamingOutputCallResponse> responseObserver)
+ public async Task FullDuplexCall(IAsyncStreamReader<StreamingOutputCallRequest> requestStream, IServerStreamWriter<StreamingOutputCallResponse> responseStream)
{
- return new FullDuplexObserver(responseObserver);
- }
-
- public IObserver<StreamingOutputCallRequest> HalfDuplexCall(IObserver<StreamingOutputCallResponse> responseObserver)
- {
- throw new NotImplementedException();
- }
-
- private class FullDuplexObserver : IObserver<StreamingOutputCallRequest>
- {
- readonly IObserver<StreamingOutputCallResponse> responseObserver;
-
- public FullDuplexObserver(IObserver<StreamingOutputCallResponse> responseObserver)
- {
- this.responseObserver = responseObserver;
- }
-
- public void OnCompleted()
+ await requestStream.ForEach(async request =>
{
- responseObserver.OnCompleted();
- }
-
- public void OnError(Exception error)
- {
- throw new NotImplementedException();
- }
-
- public void OnNext(StreamingOutputCallRequest value)
- {
- foreach (var responseParam in value.ResponseParametersList)
+ foreach (var responseParam in request.ResponseParametersList)
{
var response = StreamingOutputCallResponse.CreateBuilder()
.SetPayload(CreateZerosPayload(responseParam.Size)).Build();
- responseObserver.OnNext(response);
+ await responseStream.Write(response);
}
- }
+ });
+ }
+
+ public async Task HalfDuplexCall(IAsyncStreamReader<StreamingOutputCallRequest> requestStream, IServerStreamWriter<StreamingOutputCallResponse> responseStream)
+ {
+ throw new NotImplementedException();
}
private static Payload CreateZerosPayload(int size)