Merge pull request #3029 from jtattermusch/csharp_responseheaders_and_tests

c# response headers support and code simplification

src/csharp/.gitignore

@@ -5,4 +5,5 @@ test-results
 packages
 Grpc.v12.suo
 TestResult.xml
+/TestResults
 *.nupkg

src/csharp/Grpc.Core.Tests/Grpc.Core.Tests.csproj

@@ -65,6 +65,7 @@
     </Compile>
     <Compile Include="ClientBaseTest.cs" />
     <Compile Include="ShutdownTest.cs" />
+    <Compile Include="Internal\AsyncCallTest.cs" />
     <Compile Include="Properties\AssemblyInfo.cs" />
     <Compile Include="ClientServerTest.cs" />
     <Compile Include="ServerTest.cs" />

src/csharp/Grpc.Core.Tests/GrpcEnvironmentTest.cs

@@ -53,7 +53,7 @@ namespace Grpc.Core.Tests
         {
             var env1 = GrpcEnvironment.AddRef();
             var env2 = GrpcEnvironment.AddRef();
-            Assert.IsTrue(object.ReferenceEquals(env1, env2));
+            Assert.AreSame(env1, env2);
             GrpcEnvironment.Release();
             GrpcEnvironment.Release();
         }
@@ -61,18 +61,21 @@ namespace Grpc.Core.Tests
         [Test]
         public void InitializeAfterShutdown()
         {
+            Assert.AreEqual(0, GrpcEnvironment.GetRefCount());
+
             var env1 = GrpcEnvironment.AddRef();
             GrpcEnvironment.Release();
 
             var env2 = GrpcEnvironment.AddRef();
             GrpcEnvironment.Release();
 
-            Assert.IsFalse(object.ReferenceEquals(env1, env2));
+            Assert.AreNotSame(env1, env2);
         }
 
         [Test]
         public void ReleaseWithoutAddRef()
         {
+            Assert.AreEqual(0, GrpcEnvironment.GetRefCount());
             Assert.Throws(typeof(InvalidOperationException), () => GrpcEnvironment.Release());
         }
 

src/csharp/Grpc.Core.Tests/Internal/AsyncCallTest.cs

@@ -0,0 +1,222 @@
+#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.InteropServices;
+using System.Threading.Tasks;
+
+using Grpc.Core.Internal;
+using NUnit.Framework;
+
+namespace Grpc.Core.Internal.Tests
+{
+    public class AsyncCallTest
+    {
+        Channel channel;
+        FakeNativeCall fakeCall;
+        AsyncCall<string, string> asyncCall;
+
+        [SetUp]
+        public void Init()
+        {
+            channel = new Channel("localhost", Credentials.Insecure);
+
+            fakeCall = new FakeNativeCall();
+
+            var callDetails = new CallInvocationDetails<string, string>(channel, "someMethod", null, Marshallers.StringMarshaller, Marshallers.StringMarshaller, new CallOptions());
+            asyncCall = new AsyncCall<string, string>(callDetails, fakeCall);
+        }
+
+        [TearDown]
+        public void Cleanup()
+        {
+            channel.ShutdownAsync().Wait();
+        }
+
+        [Test]
+        public void AsyncUnary_CompletionSuccess()
+        {
+            var resultTask = asyncCall.UnaryCallAsync("abc");
+            fakeCall.UnaryResponseClientHandler(true, new ClientSideStatus(Status.DefaultSuccess, new Metadata()), new byte[] { 1, 2, 3 }, new Metadata());
+            Assert.IsTrue(resultTask.IsCompleted);
+            Assert.IsTrue(fakeCall.IsDisposed);
+            Assert.AreEqual(Status.DefaultSuccess, asyncCall.GetStatus());
+        }
+
+        [Test]
+        public void AsyncUnary_CompletionFailure()
+        {
+            var resultTask = asyncCall.UnaryCallAsync("abc");
+            fakeCall.UnaryResponseClientHandler(false, new ClientSideStatus(new Status(StatusCode.Internal, ""), null), new byte[] { 1, 2, 3 }, new Metadata());
+
+            Assert.IsTrue(resultTask.IsCompleted);
+            Assert.IsTrue(fakeCall.IsDisposed);
+
+            Assert.AreEqual(StatusCode.Internal, asyncCall.GetStatus().StatusCode);
+            Assert.IsNull(asyncCall.GetTrailers());
+            var ex = Assert.Throws<RpcException>(() => resultTask.GetAwaiter().GetResult());
+            Assert.AreEqual(StatusCode.Internal, ex.Status.StatusCode);
+        }
+
+        internal class FakeNativeCall : INativeCall
+        {
+            public UnaryResponseClientHandler UnaryResponseClientHandler
+            {
+                get;
+                set;
+            }
+
+            public ReceivedStatusOnClientHandler ReceivedStatusOnClientHandler
+            {
+                get;
+                set;
+            }
+
+            public ReceivedMessageHandler ReceivedMessageHandler
+            {
+                get;
+                set;
+            }
+
+            public ReceivedResponseHeadersHandler ReceivedResponseHeadersHandler
+            {
+                get;
+                set;
+            }
+
+            public SendCompletionHandler SendCompletionHandler
+            {
+                get;
+                set;
+            }
+
+            public ReceivedCloseOnServerHandler ReceivedCloseOnServerHandler
+            {
+                get;
+                set;
+            }
+
+            public bool IsCancelled
+            {
+                get;
+                set;
+            }
+
+            public bool IsDisposed
+            {
+                get;
+                set;
+            }
+
+            public void Cancel()
+            {
+                IsCancelled = true;
+            }
+
+            public void CancelWithStatus(Status status)
+            {
+                IsCancelled = true;
+            }
+
+            public string GetPeer()
+            {
+                return "PEER";
+            }
+
+            public void StartUnary(UnaryResponseClientHandler callback, byte[] payload, MetadataArraySafeHandle metadataArray, WriteFlags writeFlags)
+            {
+                UnaryResponseClientHandler = callback;
+            }
+
+            public void StartUnary(BatchContextSafeHandle ctx, byte[] payload, MetadataArraySafeHandle metadataArray, WriteFlags writeFlags)
+            {
+                throw new NotImplementedException();
+            }
+
+            public void StartClientStreaming(UnaryResponseClientHandler callback, MetadataArraySafeHandle metadataArray)
+            {
+                UnaryResponseClientHandler = callback;
+            }
+
+            public void StartServerStreaming(ReceivedStatusOnClientHandler callback, byte[] payload, MetadataArraySafeHandle metadataArray, WriteFlags writeFlags)
+            {
+                ReceivedStatusOnClientHandler = callback;
+            }
+
+            public void StartDuplexStreaming(ReceivedStatusOnClientHandler callback, MetadataArraySafeHandle metadataArray)
+            {
+                ReceivedStatusOnClientHandler = callback;
+            }
+
+            public void StartReceiveMessage(ReceivedMessageHandler callback)
+            {
+                ReceivedMessageHandler = callback;
+            }
+
+            public void StartReceiveInitialMetadata(ReceivedResponseHeadersHandler callback)
+            {
+                ReceivedResponseHeadersHandler = callback;
+            }
+
+            public void StartSendInitialMetadata(SendCompletionHandler callback, MetadataArraySafeHandle metadataArray)
+            {
+                SendCompletionHandler = callback;
+            }
+
+            public void StartSendMessage(SendCompletionHandler callback, byte[] payload, WriteFlags writeFlags, bool sendEmptyInitialMetadata)
+            {
+                SendCompletionHandler = callback;
+            }
+
+            public void StartSendCloseFromClient(SendCompletionHandler callback)
+            {
+                SendCompletionHandler = callback;
+            }
+
+            public void StartSendStatusFromServer(SendCompletionHandler callback, Status status, MetadataArraySafeHandle metadataArray, bool sendEmptyInitialMetadata)
+            {
+                SendCompletionHandler = callback;
+            }
+
+            public void StartServerSide(ReceivedCloseOnServerHandler callback)
+            {
+                ReceivedCloseOnServerHandler = callback;
+            }
+
+            public void Dispose()
+            {
+                IsDisposed = true;
+            }
+        }
+    }
+}

src/csharp/Grpc.Core.Tests/ResponseHeadersTest.cs

@@ -32,13 +32,16 @@
 #endregion
 
 using System;
+using System.Collections.Generic;
 using System.Diagnostics;
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
+
 using Grpc.Core;
 using Grpc.Core.Internal;
 using Grpc.Core.Utils;
+
 using NUnit.Framework;
 
 namespace Grpc.Core.Tests
@@ -73,6 +76,80 @@ namespace Grpc.Core.Tests
             server.ShutdownAsync().Wait();
         }
 
+        [Test]
+        public async Task ResponseHeadersAsync_UnaryCall()
+        {
+            helper.UnaryHandler = new UnaryServerMethod<string, string>(async (request, context) =>
+            {
+                await context.WriteResponseHeadersAsync(headers);
+                return "PASS";
+            });
+
+            var call = Calls.AsyncUnaryCall(helper.CreateUnaryCall(), "");
+            var responseHeaders = await call.ResponseHeadersAsync;
+
+            Assert.AreEqual(headers.Count, responseHeaders.Count);
+            Assert.AreEqual("ascii-header", responseHeaders[0].Key);
+            Assert.AreEqual("abcdefg", responseHeaders[0].Value);
+
+            Assert.AreEqual("PASS", await call.ResponseAsync);
+        }
+
+        [Test]
+        public async Task ResponseHeadersAsync_ClientStreamingCall()
+        {
+            helper.ClientStreamingHandler = new ClientStreamingServerMethod<string, string>(async (requestStream, context) =>
+            {
+                await context.WriteResponseHeadersAsync(headers);
+                return "PASS";
+            });
+
+            var call = Calls.AsyncClientStreamingCall(helper.CreateClientStreamingCall());
+            await call.RequestStream.CompleteAsync();
+            var responseHeaders = await call.ResponseHeadersAsync;
+
+            Assert.AreEqual("ascii-header", responseHeaders[0].Key);
+            Assert.AreEqual("PASS", await call.ResponseAsync);
+        }
+
+        [Test]
+        public async Task ResponseHeadersAsync_ServerStreamingCall()
+        {
+            helper.ServerStreamingHandler = new ServerStreamingServerMethod<string, string>(async (request, responseStream, context) =>
+            {
+                await context.WriteResponseHeadersAsync(headers);
+                await responseStream.WriteAsync("PASS");
+            });
+
+            var call = Calls.AsyncServerStreamingCall(helper.CreateServerStreamingCall(), "");
+            var responseHeaders = await call.ResponseHeadersAsync;
+
+            Assert.AreEqual("ascii-header", responseHeaders[0].Key);
+            CollectionAssert.AreEqual(new[] { "PASS" }, await call.ResponseStream.ToListAsync());
+        }
+
+        [Test]
+        public async Task ResponseHeadersAsync_DuplexStreamingCall()
+        {
+            helper.DuplexStreamingHandler = new DuplexStreamingServerMethod<string, string>(async (requestStream, responseStream, context) =>
+            {
+                await context.WriteResponseHeadersAsync(headers);
+                while (await requestStream.MoveNext())
+                {
+                    await responseStream.WriteAsync(requestStream.Current);
+                }
+            });
+
+            var call = Calls.AsyncDuplexStreamingCall(helper.CreateDuplexStreamingCall());
+            var responseHeaders = await call.ResponseHeadersAsync;
+
+            var messages = new[] { "PASS" };
+            await call.RequestStream.WriteAllAsync(messages);
+
+            Assert.AreEqual("ascii-header", responseHeaders[0].Key);
+            CollectionAssert.AreEqual(messages, await call.ResponseStream.ToListAsync());
+        }
+
         [Test]
         public void WriteResponseHeaders_NullNotAllowed()
         {

src/csharp/Grpc.Core/AsyncClientStreamingCall.cs

@@ -44,14 +44,16 @@ namespace Grpc.Core
     {
         readonly IClientStreamWriter<TRequest> requestStream;
         readonly Task<TResponse> responseAsync;
+        readonly Task<Metadata> responseHeadersAsync;
         readonly Func<Status> getStatusFunc;
         readonly Func<Metadata> getTrailersFunc;
         readonly Action disposeAction;
 
-        public AsyncClientStreamingCall(IClientStreamWriter<TRequest> requestStream, Task<TResponse> responseAsync, Func<Status> getStatusFunc, Func<Metadata> getTrailersFunc, Action disposeAction)
+        public AsyncClientStreamingCall(IClientStreamWriter<TRequest> requestStream, Task<TResponse> responseAsync, Task<Metadata> responseHeadersAsync, Func<Status> getStatusFunc, Func<Metadata> getTrailersFunc, Action disposeAction)
         {
             this.requestStream = requestStream;
             this.responseAsync = responseAsync;
+            this.responseHeadersAsync = responseHeadersAsync;
             this.getStatusFunc = getStatusFunc;
             this.getTrailersFunc = getTrailersFunc;
             this.disposeAction = disposeAction;
@@ -68,6 +70,17 @@ namespace Grpc.Core
             }
         }
 
+        /// <summary>
+        /// Asynchronous access to response headers.
+        /// </summary>
+        public Task<Metadata> ResponseHeadersAsync
+        {
+            get
+            {
+                return this.responseHeadersAsync;
+            }
+        }
+
         /// <summary>
         /// Async stream to send streaming requests.
         /// </summary>

src/csharp/Grpc.Core/AsyncDuplexStreamingCall.cs

@@ -32,6 +32,7 @@
 #endregion
 
 using System;
+using System.Threading.Tasks;
 
 namespace Grpc.Core
 {
@@ -42,14 +43,16 @@ namespace Grpc.Core
     {
         readonly IClientStreamWriter<TRequest> requestStream;
         readonly IAsyncStreamReader<TResponse> responseStream;
+        readonly Task<Metadata> responseHeadersAsync;
         readonly Func<Status> getStatusFunc;
         readonly Func<Metadata> getTrailersFunc;
         readonly Action disposeAction;
 
-        public AsyncDuplexStreamingCall(IClientStreamWriter<TRequest> requestStream, IAsyncStreamReader<TResponse> responseStream, Func<Status> getStatusFunc, Func<Metadata> getTrailersFunc, Action disposeAction)
+        public AsyncDuplexStreamingCall(IClientStreamWriter<TRequest> requestStream, IAsyncStreamReader<TResponse> responseStream, Task<Metadata> responseHeadersAsync, Func<Status> getStatusFunc, Func<Metadata> getTrailersFunc, Action disposeAction)
         {
             this.requestStream = requestStream;
             this.responseStream = responseStream;
+            this.responseHeadersAsync = responseHeadersAsync;
             this.getStatusFunc = getStatusFunc;
             this.getTrailersFunc = getTrailersFunc;
             this.disposeAction = disposeAction;
@@ -77,6 +80,17 @@ namespace Grpc.Core
             }
         }
 
+        /// <summary>
+        /// Asynchronous access to response headers.
+        /// </summary>
+        public Task<Metadata> ResponseHeadersAsync
+        {
+            get
+            {
+                return this.responseHeadersAsync;
+            }
+        }
+
         /// <summary>
         /// Gets the call status if the call has already finished.
         /// Throws InvalidOperationException otherwise.

src/csharp/Grpc.Core/AsyncServerStreamingCall.cs

@@ -32,6 +32,7 @@
 #endregion
 
 using System;
+using System.Threading.Tasks;
 
 namespace Grpc.Core
 {
@@ -41,13 +42,15 @@ namespace Grpc.Core
     public sealed class AsyncServerStreamingCall<TResponse> : IDisposable
     {
         readonly IAsyncStreamReader<TResponse> responseStream;
+        readonly Task<Metadata> responseHeadersAsync;
         readonly Func<Status> getStatusFunc;
         readonly Func<Metadata> getTrailersFunc;
         readonly Action disposeAction;
 
-        public AsyncServerStreamingCall(IAsyncStreamReader<TResponse> responseStream, Func<Status> getStatusFunc, Func<Metadata> getTrailersFunc, Action disposeAction)
+        public AsyncServerStreamingCall(IAsyncStreamReader<TResponse> responseStream, Task<Metadata> responseHeadersAsync, Func<Status> getStatusFunc, Func<Metadata> getTrailersFunc, Action disposeAction)
         {
             this.responseStream = responseStream;
+            this.responseHeadersAsync = responseHeadersAsync;
             this.getStatusFunc = getStatusFunc;
             this.getTrailersFunc = getTrailersFunc;
             this.disposeAction = disposeAction;
@@ -64,6 +67,17 @@ namespace Grpc.Core
             }
         }
 
+        /// <summary>
+        /// Asynchronous access to response headers.
+        /// </summary>
+        public Task<Metadata> ResponseHeadersAsync
+        {
+            get
+            {
+                return this.responseHeadersAsync;
+            }
+        }
+
         /// <summary>
         /// Gets the call status if the call has already finished.
         /// Throws InvalidOperationException otherwise.

src/csharp/Grpc.Core/AsyncUnaryCall.cs

@@ -43,13 +43,15 @@ namespace Grpc.Core
     public sealed class AsyncUnaryCall<TResponse> : IDisposable
     {
         readonly Task<TResponse> responseAsync;
+        readonly Task<Metadata> responseHeadersAsync;
         readonly Func<Status> getStatusFunc;
         readonly Func<Metadata> getTrailersFunc;
         readonly Action disposeAction;
 
-        public AsyncUnaryCall(Task<TResponse> responseAsync, Func<Status> getStatusFunc, Func<Metadata> getTrailersFunc, Action disposeAction)
+        public AsyncUnaryCall(Task<TResponse> responseAsync, Task<Metadata> responseHeadersAsync, Func<Status> getStatusFunc, Func<Metadata> getTrailersFunc, Action disposeAction)
         {
             this.responseAsync = responseAsync;
+            this.responseHeadersAsync = responseHeadersAsync;
             this.getStatusFunc = getStatusFunc;
             this.getTrailersFunc = getTrailersFunc;
             this.disposeAction = disposeAction;
@@ -66,6 +68,17 @@ namespace Grpc.Core
             }
         }
 
+        /// <summary>
+        /// Asynchronous access to response headers.
+        /// </summary>
+        public Task<Metadata> ResponseHeadersAsync
+        {
+            get
+            {
+                return this.responseHeadersAsync;
+            }
+        }
+
         /// <summary>
         /// Allows awaiting this object directly.
         /// </summary>

src/csharp/Grpc.Core/Calls.cs

@@ -74,7 +74,7 @@ namespace Grpc.Core
         {
             var asyncCall = new AsyncCall<TRequest, TResponse>(call);
             var asyncResult = asyncCall.UnaryCallAsync(req);
-            return new AsyncUnaryCall<TResponse>(asyncResult, asyncCall.GetStatus, asyncCall.GetTrailers, asyncCall.Cancel);
+            return new AsyncUnaryCall<TResponse>(asyncResult, asyncCall.ResponseHeadersAsync, asyncCall.GetStatus, asyncCall.GetTrailers, asyncCall.Cancel);
         }
 
         /// <summary>
@@ -93,7 +93,7 @@ namespace Grpc.Core
             var asyncCall = new AsyncCall<TRequest, TResponse>(call);
             asyncCall.StartServerStreamingCall(req);
             var responseStream = new ClientResponseStream<TRequest, TResponse>(asyncCall);
-            return new AsyncServerStreamingCall<TResponse>(responseStream, asyncCall.GetStatus, asyncCall.GetTrailers, asyncCall.Cancel);
+            return new AsyncServerStreamingCall<TResponse>(responseStream, asyncCall.ResponseHeadersAsync, asyncCall.GetStatus, asyncCall.GetTrailers, asyncCall.Cancel);
         }
 
         /// <summary>
@@ -110,7 +110,7 @@ namespace Grpc.Core
             var asyncCall = new AsyncCall<TRequest, TResponse>(call);
             var resultTask = asyncCall.ClientStreamingCallAsync();
             var requestStream = new ClientRequestStream<TRequest, TResponse>(asyncCall);
-            return new AsyncClientStreamingCall<TRequest, TResponse>(requestStream, resultTask, asyncCall.GetStatus, asyncCall.GetTrailers, asyncCall.Cancel);
+            return new AsyncClientStreamingCall<TRequest, TResponse>(requestStream, resultTask, asyncCall.ResponseHeadersAsync, asyncCall.GetStatus, asyncCall.GetTrailers, asyncCall.Cancel);
         }
 
         /// <summary>
@@ -130,7 +130,7 @@ namespace Grpc.Core
             asyncCall.StartDuplexStreamingCall();
             var requestStream = new ClientRequestStream<TRequest, TResponse>(asyncCall);
             var responseStream = new ClientResponseStream<TRequest, TResponse>(asyncCall);
-            return new AsyncDuplexStreamingCall<TRequest, TResponse>(requestStream, responseStream, asyncCall.GetStatus, asyncCall.GetTrailers, asyncCall.Cancel);
+            return new AsyncDuplexStreamingCall<TRequest, TResponse>(requestStream, responseStream, asyncCall.ResponseHeadersAsync, asyncCall.GetStatus, asyncCall.GetTrailers, asyncCall.Cancel);
         }
     }
 }

src/csharp/Grpc.Core/Channel.cs

@@ -58,7 +58,6 @@ namespace Grpc.Core
         readonly List<ChannelOption> options;
 
         bool shutdownRequested;
-        bool disposed;
 
         /// <summary>
         /// Creates a channel that connects to a specific host.

src/csharp/Grpc.Core/Grpc.Core.csproj

@@ -49,6 +49,7 @@
     <Compile Include="AsyncDuplexStreamingCall.cs" />
     <Compile Include="AsyncServerStreamingCall.cs" />
     <Compile Include="IClientStreamWriter.cs" />
+    <Compile Include="Internal\INativeCall.cs" />
     <Compile Include="IServerStreamWriter.cs" />
     <Compile Include="IAsyncStreamWriter.cs" />
     <Compile Include="IAsyncStreamReader.cs" />

src/csharp/Grpc.Core/GrpcEnvironment.cs

@@ -102,6 +102,14 @@ namespace Grpc.Core
             }
         }
 
+        internal static int GetRefCount()
+        {
+            lock (staticLock)
+            {
+                return refCount;
+            }
+        }
+
         /// <summary>
         /// Gets application-wide logger used by gRPC.
         /// </summary>
@@ -177,7 +185,6 @@ namespace Grpc.Core
             return Marshal.PtrToStringAnsi(ptr);
         }
 
-
         internal static void GrpcNativeInit()
         {
             grpcsharp_init();

src/csharp/Grpc.Core/Internal/AsyncCall.cs

@@ -51,22 +51,35 @@ namespace Grpc.Core.Internal
         static readonly ILogger Logger = GrpcEnvironment.Logger.ForType<AsyncCall<TRequest, TResponse>>();
 
         readonly CallInvocationDetails<TRequest, TResponse> details;
+        readonly INativeCall injectedNativeCall;  // for testing
 
         // Completion of a pending unary response if not null.
         TaskCompletionSource<TResponse> unaryResponseTcs;
 
+        // Indicates that steaming call has finished.
+        TaskCompletionSource<object> streamingCallFinishedTcs = new TaskCompletionSource<object>();
+
+        // Response headers set here once received.
+        TaskCompletionSource<Metadata> responseHeadersTcs = new TaskCompletionSource<Metadata>();
+
         // Set after status is received. Used for both unary and streaming response calls.
         ClientSideStatus? finishedStatus;
 
-        bool readObserverCompleted;  // True if readObserver has already been completed.
-
         public AsyncCall(CallInvocationDetails<TRequest, TResponse> callDetails)
-            : base(callDetails.RequestMarshaller.Serializer, callDetails.ResponseMarshaller.Deserializer)
+            : base(callDetails.RequestMarshaller.Serializer, callDetails.ResponseMarshaller.Deserializer, callDetails.Channel.Environment)
         {
             this.details = callDetails.WithOptions(callDetails.Options.Normalize());
             this.initialMetadataSent = true;  // we always send metadata at the very beginning of the call.
         }
 
+        /// <summary>
+        /// This constructor should only be used for testing.
+        /// </summary>
+        public AsyncCall(CallInvocationDetails<TRequest, TResponse> callDetails, INativeCall injectedNativeCall) : this(callDetails)
+        {
+            this.injectedNativeCall = injectedNativeCall;
+        }
+
         // TODO: this method is not Async, so it shouldn't be in AsyncCall class, but 
         // it is reusing fair amount of code in this class, so we are leaving it here.
         /// <summary>
@@ -100,7 +113,7 @@ namespace Grpc.Core.Internal
                         bool success = (ev.success != 0);
                         try
                         {
-                            HandleUnaryResponse(success, ctx);
+                            HandleUnaryResponse(success, ctx.GetReceivedStatusOnClient(), ctx.GetReceivedMessage(), ctx.GetReceivedInitialMetadata());
                         }
                         catch (Exception e)
                         {
@@ -125,7 +138,7 @@ namespace Grpc.Core.Internal
                 Preconditions.CheckState(!started);
                 started = true;
 
-                Initialize(details.Channel.Environment.CompletionQueue);
+                Initialize(environment.CompletionQueue);
 
                 halfcloseRequested = true;
                 readingDone = true;
@@ -152,7 +165,7 @@ namespace Grpc.Core.Internal
                 Preconditions.CheckState(!started);
                 started = true;
 
-                Initialize(details.Channel.Environment.CompletionQueue);
+                Initialize(environment.CompletionQueue);
 
                 readingDone = true;
 
@@ -176,10 +189,9 @@ namespace Grpc.Core.Internal
                 Preconditions.CheckState(!started);
                 started = true;
 
-                Initialize(details.Channel.Environment.CompletionQueue);
+                Initialize(environment.CompletionQueue);
 
                 halfcloseRequested = true;
-                halfclosed = true;  // halfclose not confirmed yet, but it will be once finishedHandler is called.
 
                 byte[] payload = UnsafeSerialize(msg);
 
@@ -187,6 +199,7 @@ namespace Grpc.Core.Internal
                 {
                     call.StartServerStreaming(HandleFinished, payload, metadataArray, GetWriteFlagsForCall());
                 }
+                call.StartReceiveInitialMetadata(HandleReceivedResponseHeaders);
             }
         }
 
@@ -201,12 +214,13 @@ namespace Grpc.Core.Internal
                 Preconditions.CheckState(!started);
                 started = true;
 
-                Initialize(details.Channel.Environment.CompletionQueue);
+                Initialize(environment.CompletionQueue);
 
                 using (var metadataArray = MetadataArraySafeHandle.Create(details.Options.Headers))
                 {
                     call.StartDuplexStreaming(HandleFinished, metadataArray);
                 }
+                call.StartReceiveInitialMetadata(HandleReceivedResponseHeaders);
             }
         }
 
@@ -247,6 +261,28 @@ namespace Grpc.Core.Internal
             }
         }
 
+        /// <summary>
+        /// Get the task that completes once if streaming call finishes with ok status and throws RpcException with given status otherwise.
+        /// </summary>
+        public Task StreamingCallFinishedTask
+        {
+            get
+            {
+                return streamingCallFinishedTcs.Task;
+            }
+        }
+
+        /// <summary>
+        /// Get the task that completes once response headers are received.
+        /// </summary>
+        public Task<Metadata> ResponseHeadersAsync
+        {
+            get
+            {
+                return responseHeadersTcs.Task;
+            }
+        }
+
         /// <summary>
         /// Gets the resulting status if the call has already finished.
         /// Throws InvalidOperationException otherwise.
@@ -281,36 +317,6 @@ namespace Grpc.Core.Internal
             }
         }
 
-        /// <summary>
-        /// On client-side, we only fire readCompletionDelegate once all messages have been read 
-        /// and status has been received.
-        /// </summary>
-        protected override void ProcessLastRead(AsyncCompletionDelegate<TResponse> completionDelegate)
-        {
-            if (completionDelegate != null && readingDone && finishedStatus.HasValue)
-            {
-                bool shouldComplete;
-                lock (myLock)
-                {
-                    shouldComplete = !readObserverCompleted;
-                    readObserverCompleted = true;
-                }
-
-                if (shouldComplete)
-                {
-                    var status = finishedStatus.Value.Status;
-                    if (status.StatusCode != StatusCode.OK)
-                    {
-                        FireCompletion(completionDelegate, default(TResponse), new RpcException(status));
-                    }
-                    else
-                    {
-                        FireCompletion(completionDelegate, default(TResponse), null);
-                    }
-                }
-            }
-        }
-
         protected override void OnAfterReleaseResources()
         {
             details.Channel.RemoveCallReference(this);
@@ -318,16 +324,24 @@ namespace Grpc.Core.Internal
 
         private void Initialize(CompletionQueueSafeHandle cq)
         {
+            var call = CreateNativeCall(cq);
+            details.Channel.AddCallReference(this);
+            InitializeInternal(call);
+            RegisterCancellationCallback();
+        }
+
+        private INativeCall CreateNativeCall(CompletionQueueSafeHandle cq)
+        {
+            if (injectedNativeCall != null)
+            {
+                return injectedNativeCall;  // allows injecting a mock INativeCall in tests.
+            }
+
             var parentCall = details.Options.PropagationToken != null ? details.Options.PropagationToken.ParentCall : CallSafeHandle.NullInstance;
 
-            var call = details.Channel.Handle.CreateCall(details.Channel.Environment.CompletionRegistry,
+            return details.Channel.Handle.CreateCall(environment.CompletionRegistry,
                 parentCall, ContextPropagationToken.DefaultMask, cq,
                 details.Method, details.Host, Timespec.FromDateTime(details.Options.Deadline.Value));
-
-            details.Channel.AddCallReference(this);
-
-            InitializeInternal(call);
-            RegisterCancellationCallback();
         }
 
         // Make sure that once cancellationToken for this call is cancelled, Cancel() will be called.
@@ -350,31 +364,31 @@ namespace Grpc.Core.Internal
         }
 
         /// <summary>
-        /// Handler for unary response completion.
+        /// Handles receive status completion for calls with streaming response.
         /// </summary>
-        private void HandleUnaryResponse(bool success, BatchContextSafeHandle ctx)
+        private void HandleReceivedResponseHeaders(bool success, Metadata responseHeaders)
         {
-            var fullStatus = ctx.GetReceivedStatusOnClient();
+            responseHeadersTcs.SetResult(responseHeaders);
+        }
 
+        /// <summary>
+        /// Handler for unary response completion.
+        /// </summary>
+        private void HandleUnaryResponse(bool success, ClientSideStatus receivedStatus, byte[] receivedMessage, Metadata responseHeaders)
+        {
             lock (myLock)
             {
                 finished = true;
-                finishedStatus = fullStatus;
-
-                halfclosed = true;
+                finishedStatus = receivedStatus;
 
                 ReleaseResourcesIfPossible();
             }
 
-            if (!success)
-            {
-                unaryResponseTcs.SetException(new RpcException(new Status(StatusCode.Internal, "Internal error occured.")));
-                return;
-            }
+            responseHeadersTcs.SetResult(responseHeaders);
 
-            var status = fullStatus.Status;
+            var status = receivedStatus.Status;
 
-            if (status.StatusCode != StatusCode.OK)
+            if (!success || status.StatusCode != StatusCode.OK)
             {
                 unaryResponseTcs.SetException(new RpcException(status));
                 return;
@@ -382,7 +396,7 @@ namespace Grpc.Core.Internal
 
             // TODO: handle deserialization error
             TResponse msg;
-            TryDeserialize(ctx.GetReceivedMessage(), out msg);
+            TryDeserialize(receivedMessage, out msg);
 
             unaryResponseTcs.SetResult(msg);
         }
@@ -390,22 +404,25 @@ namespace Grpc.Core.Internal
         /// <summary>
         /// Handles receive status completion for calls with streaming response.
         /// </summary>
-        private void HandleFinished(bool success, BatchContextSafeHandle ctx)
+        private void HandleFinished(bool success, ClientSideStatus receivedStatus)
         {
-            var fullStatus = ctx.GetReceivedStatusOnClient();
-
-            AsyncCompletionDelegate<TResponse> origReadCompletionDelegate = null;
             lock (myLock)
             {
                 finished = true;
-                finishedStatus = fullStatus;
-
-                origReadCompletionDelegate = readCompletionDelegate;
+                finishedStatus = receivedStatus;
 
                 ReleaseResourcesIfPossible();
             }
 
-            ProcessLastRead(origReadCompletionDelegate);
+            var status = receivedStatus.Status;
+
+            if (!success || status.StatusCode != StatusCode.OK)
+            {
+                streamingCallFinishedTcs.SetException(new RpcException(status));
+                return;
+            }
+
+            streamingCallFinishedTcs.SetResult(null);
         }
     }
 }

src/csharp/Grpc.Core/Internal/AsyncCallBase.cs

@@ -54,30 +54,30 @@ namespace Grpc.Core.Internal
         readonly Func<TWrite, byte[]> serializer;
         readonly Func<byte[], TRead> deserializer;
 
+        protected readonly GrpcEnvironment environment;
         protected readonly object myLock = new object();
 
-        protected CallSafeHandle call;
+        protected INativeCall call;
         protected bool disposed;
 
         protected bool started;
-        protected bool errorOccured;
         protected bool cancelRequested;
 
         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 readingDone;  // True if last read (i.e. read with null payload) was already received.
+        protected bool halfcloseRequested;  // True if send close have been initiated.
         protected bool finished;  // True if close has been received from the peer.
 
         protected bool initialMetadataSent;
-        protected long streamingWritesCounter;
+        protected long streamingWritesCounter;  // Number of streaming send operations started so far.
 
-        public AsyncCallBase(Func<TWrite, byte[]> serializer, Func<byte[], TRead> deserializer)
+        public AsyncCallBase(Func<TWrite, byte[]> serializer, Func<byte[], TRead> deserializer, GrpcEnvironment environment)
         {
             this.serializer = Preconditions.CheckNotNull(serializer);
             this.deserializer = Preconditions.CheckNotNull(deserializer);
+            this.environment = Preconditions.CheckNotNull(environment);
         }
 
         /// <summary>
@@ -114,7 +114,7 @@ namespace Grpc.Core.Internal
             }
         }
 
-        protected void InitializeInternal(CallSafeHandle call)
+        protected void InitializeInternal(INativeCall call)
         {
             lock (myLock)
             {
@@ -159,16 +159,6 @@ namespace Grpc.Core.Internal
             }
         }
 
-        // 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 ProcessLastRead(AsyncCompletionDelegate<TRead> completionDelegate)
-        {
-            FireCompletion(completionDelegate, default(TRead), null);
-        }
-
         /// <summary>
         /// If there are no more pending actions and no new actions can be started, releases
         /// the underlying native resources.
@@ -177,7 +167,7 @@ namespace Grpc.Core.Internal
         {
             if (!disposed && call != null)
             {
-                bool noMoreSendCompletions = halfclosed || (cancelRequested && sendCompletionDelegate == null);
+                bool noMoreSendCompletions = sendCompletionDelegate == null && (halfcloseRequested || cancelRequested || finished);
                 if (noMoreSendCompletions && readingDone && finished)
                 {
                     ReleaseResources();
@@ -204,11 +194,11 @@ namespace Grpc.Core.Internal
         protected void CheckSendingAllowed()
         {
             Preconditions.CheckState(started);
-            Preconditions.CheckState(!errorOccured);
             CheckNotCancelled();
             Preconditions.CheckState(!disposed);
 
             Preconditions.CheckState(!halfcloseRequested, "Already halfclosed.");
+            Preconditions.CheckState(!finished, "Already finished.");
             Preconditions.CheckState(sendCompletionDelegate == null, "Only one write can be pending at a time");
         }
 
@@ -216,7 +206,6 @@ namespace Grpc.Core.Internal
         {
             Preconditions.CheckState(started);
             Preconditions.CheckState(!disposed);
-            Preconditions.CheckState(!errorOccured);
 
             Preconditions.CheckState(!readingDone, "Stream has already been closed.");
             Preconditions.CheckState(readCompletionDelegate == null, "Only one read can be pending at a time");
@@ -280,7 +269,7 @@ namespace Grpc.Core.Internal
         /// <summary>
         /// Handles send completion.
         /// </summary>
-        protected void HandleSendFinished(bool success, BatchContextSafeHandle ctx)
+        protected void HandleSendFinished(bool success)
         {
             AsyncCompletionDelegate<object> origCompletionDelegate = null;
             lock (myLock)
@@ -304,12 +293,11 @@ namespace Grpc.Core.Internal
         /// <summary>
         /// Handles halfclose completion.
         /// </summary>
-        protected void HandleHalfclosed(bool success, BatchContextSafeHandle ctx)
+        protected void HandleHalfclosed(bool success)
         {
             AsyncCompletionDelegate<object> origCompletionDelegate = null;
             lock (myLock)
             {
-                halfclosed = true;
                 origCompletionDelegate = sendCompletionDelegate;
                 sendCompletionDelegate = null;
 
@@ -329,23 +317,17 @@ namespace Grpc.Core.Internal
         /// <summary>
         /// Handles streaming read completion.
         /// </summary>
-        protected void HandleReadFinished(bool success, BatchContextSafeHandle ctx)
+        protected void HandleReadFinished(bool success, byte[] receivedMessage)
         {
-            var payload = ctx.GetReceivedMessage();
-
             AsyncCompletionDelegate<TRead> origCompletionDelegate = null;
             lock (myLock)
             {
                 origCompletionDelegate = readCompletionDelegate;
-                if (payload != null)
-                {
-                    readCompletionDelegate = null;
-                }
-                else
+                readCompletionDelegate = null;
+
+                if (receivedMessage == null)
                 {
-                    // This was the last read. Keeping the readCompletionDelegate
-                    // to be either fired by this handler or by client-side finished
-                    // handler.
+                    // This was the last read.
                     readingDone = true;
                 }
 
@@ -354,17 +336,17 @@ namespace Grpc.Core.Internal
 
             // TODO: handle the case when error occured...
 
-            if (payload != null)
+            if (receivedMessage != null)
             {
                 // TODO: handle deserialization error
                 TRead msg;
-                TryDeserialize(payload, out msg);
+                TryDeserialize(receivedMessage, out msg);
 
                 FireCompletion(origCompletionDelegate, msg, null);
             }
             else
             {
-                ProcessLastRead(origCompletionDelegate);
+                FireCompletion(origCompletionDelegate, default(TRead), null);
             }
         }
     }

src/csharp/Grpc.Core/Internal/AsyncCallServer.cs

@@ -49,12 +49,10 @@ namespace Grpc.Core.Internal
     {
         readonly TaskCompletionSource<object> finishedServersideTcs = new TaskCompletionSource<object>();
         readonly CancellationTokenSource cancellationTokenSource = new CancellationTokenSource();
-        readonly GrpcEnvironment environment;
         readonly Server server;
 
-        public AsyncCallServer(Func<TResponse, byte[]> serializer, Func<byte[], TRequest> deserializer, GrpcEnvironment environment, Server server) : base(serializer, deserializer)
+        public AsyncCallServer(Func<TResponse, byte[]> serializer, Func<byte[], TRequest> deserializer, GrpcEnvironment environment, Server server) : base(serializer, deserializer, environment)
         {
-            this.environment = Preconditions.CheckNotNull(environment);
             this.server = Preconditions.CheckNotNull(server);
         }
 
@@ -185,10 +183,8 @@ namespace Grpc.Core.Internal
         /// <summary>
         /// Handles the server side close completion.
         /// </summary>
-        private void HandleFinishedServerside(bool success, BatchContextSafeHandle ctx)
+        private void HandleFinishedServerside(bool success, bool cancelled)
         {
-            bool cancelled = ctx.GetReceivedCloseOnServerCancelled();
-
             lock (myLock)
             {
                 finished = true;

src/csharp/Grpc.Core/Internal/CallSafeHandle.cs

@@ -40,7 +40,7 @@ namespace Grpc.Core.Internal
     /// <summary>
     /// grpc_call from <grpc/grpc.h>
     /// </summary>
-    internal class CallSafeHandle : SafeHandleZeroIsInvalid
+    internal class CallSafeHandle : SafeHandleZeroIsInvalid, INativeCall
     {
         public static readonly CallSafeHandle NullInstance = new CallSafeHandle();
 
@@ -86,6 +86,10 @@ namespace Grpc.Core.Internal
         static extern GRPCCallError grpcsharp_call_recv_message(CallSafeHandle call,
             BatchContextSafeHandle ctx);
 
+        [DllImport("grpc_csharp_ext.dll")]
+        static extern GRPCCallError grpcsharp_call_recv_initial_metadata(CallSafeHandle call,
+            BatchContextSafeHandle ctx);
+
         [DllImport("grpc_csharp_ext.dll")]
         static extern GRPCCallError grpcsharp_call_start_serverside(CallSafeHandle call,
             BatchContextSafeHandle ctx);
@@ -109,10 +113,10 @@ namespace Grpc.Core.Internal
             this.completionRegistry = completionRegistry;
         }
 
-        public void StartUnary(BatchCompletionDelegate callback, byte[] payload, MetadataArraySafeHandle metadataArray, WriteFlags writeFlags)
+        public void StartUnary(UnaryResponseClientHandler callback, byte[] payload, MetadataArraySafeHandle metadataArray, WriteFlags writeFlags)
         {
             var ctx = BatchContextSafeHandle.Create();
-            completionRegistry.RegisterBatchCompletion(ctx, callback);
+            completionRegistry.RegisterBatchCompletion(ctx, (success, context) => callback(success, context.GetReceivedStatusOnClient(), context.GetReceivedMessage(), context.GetReceivedInitialMetadata()));
             grpcsharp_call_start_unary(this, ctx, payload, new UIntPtr((ulong)payload.Length), metadataArray, writeFlags)
                 .CheckOk();
         }
@@ -123,66 +127,73 @@ namespace Grpc.Core.Internal
                 .CheckOk();
         }
 
-        public void StartClientStreaming(BatchCompletionDelegate callback, MetadataArraySafeHandle metadataArray)
+        public void StartClientStreaming(UnaryResponseClientHandler callback, MetadataArraySafeHandle metadataArray)
         {
             var ctx = BatchContextSafeHandle.Create();
-            completionRegistry.RegisterBatchCompletion(ctx, callback);
+            completionRegistry.RegisterBatchCompletion(ctx, (success, context) => callback(success, context.GetReceivedStatusOnClient(), context.GetReceivedMessage(), context.GetReceivedInitialMetadata()));
             grpcsharp_call_start_client_streaming(this, ctx, metadataArray).CheckOk();
         }
 
-        public void StartServerStreaming(BatchCompletionDelegate callback, byte[] payload, MetadataArraySafeHandle metadataArray, WriteFlags writeFlags)
+        public void StartServerStreaming(ReceivedStatusOnClientHandler callback, byte[] payload, MetadataArraySafeHandle metadataArray, WriteFlags writeFlags)
         {
             var ctx = BatchContextSafeHandle.Create();
-            completionRegistry.RegisterBatchCompletion(ctx, callback);
+            completionRegistry.RegisterBatchCompletion(ctx, (success, context) => callback(success, context.GetReceivedStatusOnClient()));
             grpcsharp_call_start_server_streaming(this, ctx, payload, new UIntPtr((ulong)payload.Length), metadataArray, writeFlags).CheckOk();
         }
 
-        public void StartDuplexStreaming(BatchCompletionDelegate callback, MetadataArraySafeHandle metadataArray)
+        public void StartDuplexStreaming(ReceivedStatusOnClientHandler callback, MetadataArraySafeHandle metadataArray)
         {
             var ctx = BatchContextSafeHandle.Create();
-            completionRegistry.RegisterBatchCompletion(ctx, callback);
+            completionRegistry.RegisterBatchCompletion(ctx, (success, context) => callback(success, context.GetReceivedStatusOnClient()));
             grpcsharp_call_start_duplex_streaming(this, ctx, metadataArray).CheckOk();
         }
 
-        public void StartSendMessage(BatchCompletionDelegate callback, byte[] payload, WriteFlags writeFlags, bool sendEmptyInitialMetadata)
+        public void StartSendMessage(SendCompletionHandler callback, byte[] payload, WriteFlags writeFlags, bool sendEmptyInitialMetadata)
         {
             var ctx = BatchContextSafeHandle.Create();
-            completionRegistry.RegisterBatchCompletion(ctx, callback);
+            completionRegistry.RegisterBatchCompletion(ctx, (success, context) => callback(success));
             grpcsharp_call_send_message(this, ctx, payload, new UIntPtr((ulong)payload.Length), writeFlags, sendEmptyInitialMetadata).CheckOk();
         }
 
-        public void StartSendCloseFromClient(BatchCompletionDelegate callback)
+        public void StartSendCloseFromClient(SendCompletionHandler callback)
         {
             var ctx = BatchContextSafeHandle.Create();
-            completionRegistry.RegisterBatchCompletion(ctx, callback);
+            completionRegistry.RegisterBatchCompletion(ctx, (success, context) => callback(success));
             grpcsharp_call_send_close_from_client(this, ctx).CheckOk();
         }
 
-        public void StartSendStatusFromServer(BatchCompletionDelegate callback, Status status, MetadataArraySafeHandle metadataArray, bool sendEmptyInitialMetadata)
+        public void StartSendStatusFromServer(SendCompletionHandler callback, Status status, MetadataArraySafeHandle metadataArray, bool sendEmptyInitialMetadata)
         {
             var ctx = BatchContextSafeHandle.Create();
-            completionRegistry.RegisterBatchCompletion(ctx, callback);
+            completionRegistry.RegisterBatchCompletion(ctx, (success, context) => callback(success));
             grpcsharp_call_send_status_from_server(this, ctx, status.StatusCode, status.Detail, metadataArray, sendEmptyInitialMetadata).CheckOk();
         }
 
-        public void StartReceiveMessage(BatchCompletionDelegate callback)
+        public void StartReceiveMessage(ReceivedMessageHandler callback)
         {
             var ctx = BatchContextSafeHandle.Create();
-            completionRegistry.RegisterBatchCompletion(ctx, callback);
+            completionRegistry.RegisterBatchCompletion(ctx, (success, context) => callback(success, context.GetReceivedMessage()));
             grpcsharp_call_recv_message(this, ctx).CheckOk();
         }
 
-        public void StartServerSide(BatchCompletionDelegate callback)
+        public void StartReceiveInitialMetadata(ReceivedResponseHeadersHandler callback)
+        {
+            var ctx = BatchContextSafeHandle.Create();
+            completionRegistry.RegisterBatchCompletion(ctx, (success, context) => callback(success, context.GetReceivedInitialMetadata()));
+            grpcsharp_call_recv_initial_metadata(this, ctx).CheckOk();
+        }
+
+        public void StartServerSide(ReceivedCloseOnServerHandler callback)
         {
             var ctx = BatchContextSafeHandle.Create();
-            completionRegistry.RegisterBatchCompletion(ctx, callback);
+            completionRegistry.RegisterBatchCompletion(ctx, (success, context) => callback(success, context.GetReceivedCloseOnServerCancelled()));
             grpcsharp_call_start_serverside(this, ctx).CheckOk();
         }
 
-        public void StartSendInitialMetadata(BatchCompletionDelegate callback, MetadataArraySafeHandle metadataArray)
+        public void StartSendInitialMetadata(SendCompletionHandler callback, MetadataArraySafeHandle metadataArray)
         {
             var ctx = BatchContextSafeHandle.Create();
-            completionRegistry.RegisterBatchCompletion(ctx, callback);
+            completionRegistry.RegisterBatchCompletion(ctx, (success, context) => callback(success));
             grpcsharp_call_send_initial_metadata(this, ctx, metadataArray).CheckOk();
         }
 

src/csharp/Grpc.Core/Internal/ClientResponseStream.cs

@@ -72,7 +72,13 @@ namespace Grpc.Core.Internal
             call.StartReadMessage(taskSource.CompletionDelegate);
             var result = await taskSource.Task;
             this.current = result;
-            return result != null;
+
+            if (result == null)
+            {
+                await call.StreamingCallFinishedTask;
+                return false;
+            }
+            return true;
         }
 
         public void Dispose()

src/csharp/Grpc.Core/Internal/INativeCall.cs

@@ -0,0 +1,85 @@
+#region Copyright notice and license
+// Copyright 2015, Google Inc.
+// All rights reserved.
+// 
+// Redistribution and use in source and binary forms, with or without
+// modification, are permitted provided that the following conditions are
+// met:
+// 
+//     * Redistributions of source code must retain the above copyright
+// notice, this list of conditions and the following disclaimer.
+//     * Redistributions in binary form must reproduce the above
+// copyright notice, this list of conditions and the following disclaimer
+// in the documentation and/or other materials provided with the
+// distribution.
+//     * Neither the name of Google Inc. nor the names of its
+// contributors may be used to endorse or promote products derived from
+// this software without specific prior written permission.
+// 
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+#endregion
+
+using System;
+
+namespace Grpc.Core.Internal
+{
+    internal delegate void UnaryResponseClientHandler(bool success, ClientSideStatus receivedStatus, byte[] receivedMessage, Metadata responseHeaders);
+
+    // Received status for streaming response calls.
+    internal delegate void ReceivedStatusOnClientHandler(bool success, ClientSideStatus receivedStatus);
+
+    internal delegate void ReceivedMessageHandler(bool success, byte[] receivedMessage);
+
+    internal delegate void ReceivedResponseHeadersHandler(bool success, Metadata responseHeaders);
+
+    internal delegate void SendCompletionHandler(bool success);
+
+    internal delegate void ReceivedCloseOnServerHandler(bool success, bool cancelled);
+
+    /// <summary>
+    /// Abstraction of a native call object.
+    /// </summary>
+    internal interface INativeCall : IDisposable
+    {
+        void Cancel();
+
+        void CancelWithStatus(Grpc.Core.Status status);
+
+        string GetPeer();
+
+        void StartUnary(UnaryResponseClientHandler callback, byte[] payload, MetadataArraySafeHandle metadataArray, Grpc.Core.WriteFlags writeFlags);
+
+        void StartUnary(BatchContextSafeHandle ctx, byte[] payload, MetadataArraySafeHandle metadataArray, Grpc.Core.WriteFlags writeFlags);
+
+        void StartClientStreaming(UnaryResponseClientHandler callback, MetadataArraySafeHandle metadataArray);
+
+        void StartServerStreaming(ReceivedStatusOnClientHandler callback, byte[] payload, MetadataArraySafeHandle metadataArray, Grpc.Core.WriteFlags writeFlags);
+
+        void StartDuplexStreaming(ReceivedStatusOnClientHandler callback, MetadataArraySafeHandle metadataArray);
+
+        void StartReceiveMessage(ReceivedMessageHandler callback);
+
+        void StartReceiveInitialMetadata(ReceivedResponseHeadersHandler callback);
+
+        void StartSendInitialMetadata(SendCompletionHandler callback, MetadataArraySafeHandle metadataArray);
+
+        void StartSendMessage(SendCompletionHandler callback, byte[] payload, Grpc.Core.WriteFlags writeFlags, bool sendEmptyInitialMetadata);
+
+        void StartSendCloseFromClient(SendCompletionHandler callback);
+
+        void StartSendStatusFromServer(SendCompletionHandler callback, Grpc.Core.Status status, MetadataArraySafeHandle metadataArray, bool sendEmptyInitialMetadata);
+
+        void StartServerSide(ReceivedCloseOnServerHandler callback);
+    }
+}

src/csharp/ext/grpc_csharp_ext.c

@@ -595,7 +595,7 @@ GPR_EXPORT grpc_call_error GPR_CALLTYPE grpcsharp_call_start_server_streaming(
     grpc_call *call, grpcsharp_batch_context *ctx, const char *send_buffer,
     size_t send_buffer_len, grpc_metadata_array *initial_metadata, gpr_uint32 write_flags) {
   /* TODO: don't use magic number */
-  grpc_op ops[5];
+  grpc_op ops[4];
   ops[0].op = GRPC_OP_SEND_INITIAL_METADATA;
   grpcsharp_metadata_array_move(&(ctx->send_initial_metadata),
                                 initial_metadata);
@@ -615,23 +615,18 @@ GPR_EXPORT grpc_call_error GPR_CALLTYPE grpcsharp_call_start_server_streaming(
   ops[2].flags = 0;
   ops[2].reserved = NULL;
 
-  ops[3].op = GRPC_OP_RECV_INITIAL_METADATA;
-  ops[3].data.recv_initial_metadata = &(ctx->recv_initial_metadata);
-  ops[3].flags = 0;
-  ops[3].reserved = NULL;
-
-  ops[4].op = GRPC_OP_RECV_STATUS_ON_CLIENT;
-  ops[4].data.recv_status_on_client.trailing_metadata =
+  ops[3].op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+  ops[3].data.recv_status_on_client.trailing_metadata =
       &(ctx->recv_status_on_client.trailing_metadata);
-  ops[4].data.recv_status_on_client.status =
+  ops[3].data.recv_status_on_client.status =
       &(ctx->recv_status_on_client.status);
   /* not using preallocation for status_details */
-  ops[4].data.recv_status_on_client.status_details =
+  ops[3].data.recv_status_on_client.status_details =
       &(ctx->recv_status_on_client.status_details);
-  ops[4].data.recv_status_on_client.status_details_capacity =
+  ops[3].data.recv_status_on_client.status_details_capacity =
       &(ctx->recv_status_on_client.status_details_capacity);
-  ops[4].flags = 0;
-  ops[4].reserved = NULL;
+  ops[3].flags = 0;
+  ops[3].reserved = NULL;
 
   return grpc_call_start_batch(call, ops, sizeof(ops) / sizeof(ops[0]), ctx,
                                NULL);
@@ -642,7 +637,7 @@ grpcsharp_call_start_duplex_streaming(grpc_call *call,
                                       grpcsharp_batch_context *ctx,
                                       grpc_metadata_array *initial_metadata) {
   /* TODO: don't use magic number */
-  grpc_op ops[3];
+  grpc_op ops[2];
   ops[0].op = GRPC_OP_SEND_INITIAL_METADATA;
   grpcsharp_metadata_array_move(&(ctx->send_initial_metadata),
                                 initial_metadata);
@@ -652,28 +647,36 @@ grpcsharp_call_start_duplex_streaming(grpc_call *call,
   ops[0].flags = 0;
   ops[0].reserved = NULL;
 
-  ops[1].op = GRPC_OP_RECV_INITIAL_METADATA;
-  ops[1].data.recv_initial_metadata = &(ctx->recv_initial_metadata);
-  ops[1].flags = 0;
-  ops[1].reserved = NULL;
-
-  ops[2].op = GRPC_OP_RECV_STATUS_ON_CLIENT;
-  ops[2].data.recv_status_on_client.trailing_metadata =
+  ops[1].op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+  ops[1].data.recv_status_on_client.trailing_metadata =
       &(ctx->recv_status_on_client.trailing_metadata);
-  ops[2].data.recv_status_on_client.status =
+  ops[1].data.recv_status_on_client.status =
       &(ctx->recv_status_on_client.status);
   /* not using preallocation for status_details */
-  ops[2].data.recv_status_on_client.status_details =
+  ops[1].data.recv_status_on_client.status_details =
       &(ctx->recv_status_on_client.status_details);
-  ops[2].data.recv_status_on_client.status_details_capacity =
+  ops[1].data.recv_status_on_client.status_details_capacity =
       &(ctx->recv_status_on_client.status_details_capacity);
-  ops[2].flags = 0;
-  ops[2].reserved = NULL;
+  ops[1].flags = 0;
+  ops[1].reserved = NULL;
 
   return grpc_call_start_batch(call, ops, sizeof(ops) / sizeof(ops[0]), ctx,
                                NULL);
 }
 
+GPR_EXPORT grpc_call_error GPR_CALLTYPE grpcsharp_call_recv_initial_metadata(
+	grpc_call *call, grpcsharp_batch_context *ctx) {
+	/* TODO: don't use magic number */
+	grpc_op ops[1];
+	ops[0].op = GRPC_OP_RECV_INITIAL_METADATA;
+	ops[0].data.recv_initial_metadata = &(ctx->recv_initial_metadata);
+	ops[0].flags = 0;
+	ops[0].reserved = NULL;
+
+	return grpc_call_start_batch(call, ops, sizeof(ops) / sizeof(ops[0]), ctx,
+		NULL);
+}
+
 GPR_EXPORT grpc_call_error GPR_CALLTYPE
 grpcsharp_call_send_message(grpc_call *call, grpcsharp_batch_context *ctx,
                             const char *send_buffer, size_t send_buffer_len,