A new version C# API based on async/await

src/csharp/.gitignore

@@ -1,4 +1,5 @@
 *.userprefs
+*.csproj.user
 StyleCop.Cache
 test-results
 packages

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);
         }
     }
 }

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();
         }
     }
 }

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;
+            }
+        }
+    }
+}

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;
             }
         }
     }

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;

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;

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)

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
     {

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" />

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();
+    }
+}

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);
+    }
+}

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();
+    }
+}

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>
+    {
+    }
+}

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);
         }
     }
 }

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);
             }
         }
     }

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)
             {

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)

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;
         }
     }
 }

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;
         }
     }
 }

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;
         }
     }
 }

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);
+        }
+    }
+}

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;
         }
     }
 }

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;
         }
     }
 }

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>

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();

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);
+}

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;
             }
 

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;
 

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);
+            }
+        }
+    }
+}

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();
         }
     }
 }

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);

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)

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));
-            }
-        }
+        }        
     }
 }

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)

src/csharp/Grpc.IntegrationTesting/InteropClientServerTest.cs

@@ -87,7 +87,7 @@ namespace Grpc.IntegrationTesting
         [Test]
         public void LargeUnary()
         {
-            InteropClient.RunEmptyUnary(client);
+            InteropClient.RunLargeUnary(client);
         }
 
         [Test]

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)

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)