diff options
Diffstat (limited to 'src/csharp/Grpc.Core')
28 files changed, 928 insertions, 309 deletions
diff --git a/src/csharp/Grpc.Core/Internal/ServerStreamingOutputObserver.cs b/src/csharp/Grpc.Core/AsyncClientStreamingCall.cs index 97b62d0569..e81ce01ebb 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 0000000000..1cb30f4779 --- /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 index 9749168af0..d614916fb7 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 fe5f40f5e9..070dfb569d 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 280387b323..9365ccd9fb 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 3a42dac1d7..b47d810672 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 15dd3ef321..e64c1e3dc1 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 0b85392e15..fee742b220 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 0000000000..61cf57f7e0 --- /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 0000000000..724bae8f31 --- /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 0000000000..6da42e9ccc --- /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 0000000000..e76397d8a0 --- /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 bc72cb78de..fd94771ddd 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 15b0cfe249..2bde4b3720 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 d3a2be553f..25dc15bbc0 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 673b527fb2..c88cae98fe 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 index 286c54f2c4..6854922a6f 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 index 7b43ab8ad5..7fa511faa8 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 25fd4fab8f..2ae924b4a8 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 0000000000..5c6b335c7f --- /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 index 65bedb0a33..aa311059c3 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 index dcae99446f..686017c048 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 4f97eeef37..156e780c7d 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 e686cdddef..a3000cee46 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 0000000000..6646bb5a89 --- /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 f08c7d88f3..01b1dc8f7b 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 7d76aec4d1..b588170694 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 0000000000..f915155f8a --- /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); + } + } + } +} |