migration to new C API

src/csharp/GrpcApi/MathGrpc.cs

@@ -81,7 +81,7 @@ namespace math
 
             Task<DivReply> DivAsync(DivArgs request, CancellationToken token = default(CancellationToken));
 
-            Task Fib(FibArgs request, IObserver<Num> responseObserver, CancellationToken token = default(CancellationToken));
+            void Fib(FibArgs request, IObserver<Num> responseObserver, CancellationToken token = default(CancellationToken));
 
             ClientStreamingAsyncResult<Num, Num> Sum(CancellationToken token = default(CancellationToken));
 
@@ -109,10 +109,10 @@ namespace math
                 return Calls.AsyncUnaryCall(call, request, token);
             }
 
-            public Task Fib(FibArgs request, IObserver<Num> responseObserver, CancellationToken token = default(CancellationToken))
+            public void Fib(FibArgs request, IObserver<Num> responseObserver, CancellationToken token = default(CancellationToken))
             {
                 var call = new Google.GRPC.Core.Call<FibArgs, Num>(fibMethod, channel);
-                return Calls.AsyncServerStreamingCall(call, request, responseObserver, token);
+                Calls.AsyncServerStreamingCall(call, request, responseObserver, token);
             }
 
             public ClientStreamingAsyncResult<Num, Num> Sum(CancellationToken token = default(CancellationToken))

src/csharp/GrpcApi/TestServiceGrpc.cs

@@ -99,7 +99,7 @@ namespace grpc.testing
 
             Task<SimpleResponse> UnaryCallAsync(SimpleRequest request, CancellationToken token = default(CancellationToken));
 
-            Task StreamingOutputCall(StreamingOutputCallRequest request, IObserver<StreamingOutputCallResponse> responseObserver, CancellationToken token = default(CancellationToken));
+            void StreamingOutputCall(StreamingOutputCallRequest request, IObserver<StreamingOutputCallResponse> responseObserver, CancellationToken token = default(CancellationToken));
 
             ClientStreamingAsyncResult<StreamingInputCallRequest, StreamingInputCallResponse> StreamingInputCall(CancellationToken token = default(CancellationToken));
 
@@ -141,9 +141,9 @@ namespace grpc.testing
                 return Calls.AsyncUnaryCall(call, request, token);
             }
 
-            public Task StreamingOutputCall(StreamingOutputCallRequest request, IObserver<StreamingOutputCallResponse> responseObserver, CancellationToken token = default(CancellationToken)) {
+            public void StreamingOutputCall(StreamingOutputCallRequest request, IObserver<StreamingOutputCallResponse> responseObserver, CancellationToken token = default(CancellationToken)) {
                 var call = new Google.GRPC.Core.Call<StreamingOutputCallRequest, StreamingOutputCallResponse>(streamingOutputCallMethod, channel);
-                return Calls.AsyncServerStreamingCall(call, request, responseObserver, token);
+                Calls.AsyncServerStreamingCall(call, request, responseObserver, token);
             }
 
             public ClientStreamingAsyncResult<StreamingInputCallRequest, StreamingInputCallResponse> StreamingInputCall(CancellationToken token = default(CancellationToken))

src/csharp/GrpcCore/Calls.cs

@@ -47,50 +47,42 @@ namespace Google.GRPC.Core
     {
         public static TResponse BlockingUnaryCall<TRequest, TResponse>(Call<TRequest, TResponse> call, TRequest req, CancellationToken token)
         {
-            //TODO: implement this in real synchronous style once new GRPC C core API is available.
-            return AsyncUnaryCall(call, req, token).Result;
+            //TODO: implement this in real synchronous style.
+            try {
+                return AsyncUnaryCall(call, req, token).Result;
+            } catch(AggregateException ae) {
+                foreach (var e in ae.InnerExceptions)
+                {
+                    if (e is RpcException)
+                    {
+                        throw e;
+                    }
+                }
+                throw;
+            }
         }
 
         public static async Task<TResponse> AsyncUnaryCall<TRequest, TResponse>(Call<TRequest, TResponse> call, TRequest req, CancellationToken token)
         {
             var asyncCall = new AsyncCall<TRequest, TResponse>(call.RequestSerializer, call.ResponseDeserializer);
-            asyncCall.Initialize(call.Channel, call.MethodName);
-            asyncCall.Start(false, GetCompletionQueue());
-
-            await asyncCall.WriteAsync(req);
-            await asyncCall.WritesCompletedAsync();
-
-            TResponse response = await asyncCall.ReadAsync();
-
-            Status status = await asyncCall.Finished;
-
-            if (status.StatusCode != StatusCode.GRPC_STATUS_OK)
-            {
-                throw new RpcException(status);
-            }
-            return response;
+            asyncCall.Initialize(call.Channel, GetCompletionQueue(), call.MethodName);
+            return await asyncCall.UnaryCallAsync(req);
         }
 
-        public static async Task AsyncServerStreamingCall<TRequest, TResponse>(Call<TRequest, TResponse> call, TRequest req, IObserver<TResponse> outputs, CancellationToken token)
+        public static void AsyncServerStreamingCall<TRequest, TResponse>(Call<TRequest, TResponse> call, TRequest req, IObserver<TResponse> outputs, CancellationToken token)
         {
             var asyncCall = new AsyncCall<TRequest, TResponse>(call.RequestSerializer, call.ResponseDeserializer);
-            asyncCall.Initialize(call.Channel, call.MethodName);
-            asyncCall.Start(false, GetCompletionQueue());
 
-            asyncCall.StartReadingToStream(outputs);
-
-            await asyncCall.WriteAsync(req);
-            await asyncCall.WritesCompletedAsync();
+            asyncCall.Initialize(call.Channel, GetCompletionQueue(), call.MethodName);
+            asyncCall.StartServerStreamingCall(req, outputs);
         }
 
         public static ClientStreamingAsyncResult<TRequest, TResponse> AsyncClientStreamingCall<TRequest, TResponse>(Call<TRequest, TResponse> call, CancellationToken token)
         {
             var asyncCall = new AsyncCall<TRequest, TResponse>(call.RequestSerializer, call.ResponseDeserializer);
-            asyncCall.Initialize(call.Channel, call.MethodName);
-            asyncCall.Start(false, GetCompletionQueue());
-
-            var task = asyncCall.ReadAsync();
-            var inputs = new StreamingInputObserver<TRequest, TResponse>(asyncCall);
+            asyncCall.Initialize(call.Channel, GetCompletionQueue(), call.MethodName);
+            var task = asyncCall.ClientStreamingCallAsync();
+            var inputs = new ClientStreamingInputObserver<TRequest, TResponse>(asyncCall);
             return new ClientStreamingAsyncResult<TRequest, TResponse>(task, inputs);
         }
 
@@ -102,12 +94,10 @@ namespace Google.GRPC.Core
         public static IObserver<TRequest> DuplexStreamingCall<TRequest, TResponse>(Call<TRequest, TResponse> call, IObserver<TResponse> outputs, CancellationToken token)
         {
             var asyncCall = new AsyncCall<TRequest, TResponse>(call.RequestSerializer, call.ResponseDeserializer);
-            asyncCall.Initialize(call.Channel, call.MethodName);
-            asyncCall.Start(false, GetCompletionQueue());
+            asyncCall.Initialize(call.Channel, GetCompletionQueue(), call.MethodName);
 
-            asyncCall.StartReadingToStream(outputs);
-            var inputs = new StreamingInputObserver<TRequest, TResponse>(asyncCall);
-            return inputs;
+            asyncCall.StartDuplexStreamingCall(outputs);
+            return new ClientStreamingInputObserver<TRequest, TResponse>(asyncCall);
         }
 
         private static CompletionQueueSafeHandle GetCompletionQueue() {

src/csharp/GrpcCore/GrpcCore.csproj

@@ -47,21 +47,21 @@
     <Compile Include="Internal\ChannelSafeHandle.cs" />
     <Compile Include="Internal\CompletionQueueSafeHandle.cs" />
     <Compile Include="Internal\Enums.cs" />
-    <Compile Include="Internal\Event.cs" />
     <Compile Include="Internal\SafeHandleZeroIsInvalid.cs" />
     <Compile Include="Internal\Timespec.cs" />
     <Compile Include="Internal\GrpcThreadPool.cs" />
     <Compile Include="Internal\AsyncCall.cs" />
     <Compile Include="Internal\ServerSafeHandle.cs" />
-    <Compile Include="Internal\StreamingInputObserver.cs" />
     <Compile Include="Method.cs" />
     <Compile Include="ServerCalls.cs" />
     <Compile Include="ServerCallHandler.cs" />
-    <Compile Include="Internal\ServerWritingObserver.cs" />
     <Compile Include="Marshaller.cs" />
     <Compile Include="ServerServiceDefinition.cs" />
     <Compile Include="Utils\RecordingObserver.cs" />
     <Compile Include="Utils\RecordingQueue.cs" />
+    <Compile Include="Internal\BatchContextSafeHandle.cs" />
+    <Compile Include="Internal\ClientStreamingInputObserver.cs" />
+    <Compile Include="Internal\ServerStreamingOutputObserver.cs" />
   </ItemGroup>
   <Import Project="$(MSBuildBinPath)\Microsoft.CSharp.targets" />
   <ItemGroup>

src/csharp/GrpcCore/Internal/AsyncCall.cs

@@ -2,11 +2,11 @@
 
 // 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
@@ -16,7 +16,7 @@
 //     * 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
@@ -41,39 +41,28 @@ using Google.GRPC.Core.Internal;
 
 namespace Google.GRPC.Core.Internal
 {
-    /// <summary>
-    /// Listener for call events that can be delivered from a completion queue.
-    /// </summary>
-    internal interface ICallEventListener {
-
-        void OnClientMetadata();
-
-        void OnRead(byte[] payload);
-
-        void OnWriteAccepted(GRPCOpError error);
-
-        void OnFinishAccepted(GRPCOpError error);
-
-        // ignore the status on server
-        void OnFinished(Status status);
-    }
-
     /// <summary>
     /// Handle native call lifecycle and provides convenience methods.
     /// </summary>
-    internal class AsyncCall<TWrite, TRead>: ICallEventListener, IDisposable
+    internal class AsyncCall<TWrite, TRead> : IDisposable
     {
         readonly Func<TWrite, byte[]> serializer;
         readonly Func<byte[], TRead> deserializer;
 
-        // TODO: make sure the delegate doesn't get garbage collected while
+        // TODO: make sure the delegate doesn't get garbage collected while 
         // native callbacks are in the completion queue.
-        readonly EventCallbackDelegate callbackHandler;
+        readonly CompletionCallbackDelegate unaryResponseHandler;
+        readonly CompletionCallbackDelegate finishedHandler;
+        readonly CompletionCallbackDelegate writeFinishedHandler;
+        readonly CompletionCallbackDelegate readFinishedHandler;
+        readonly CompletionCallbackDelegate halfclosedHandler;
+        readonly CompletionCallbackDelegate finishedServersideHandler;
 
         object myLock = new object();
         bool disposed;
         CallSafeHandle call;
 
+        bool server;
         bool started;
         bool errorOccured;
 
@@ -85,54 +74,25 @@ namespace Google.GRPC.Core.Internal
 
         TaskCompletionSource<object> writeTcs;
         TaskCompletionSource<TRead> readTcs;
+
+        TaskCompletionSource<object> finishedServersideTcs = new TaskCompletionSource<object>();
         TaskCompletionSource<object> halfcloseTcs = new TaskCompletionSource<object>();
         TaskCompletionSource<Status> finishedTcs = new TaskCompletionSource<Status>();
 
+        TaskCompletionSource<TRead> unaryResponseTcs;
+
         IObserver<TRead> readObserver;
 
         public AsyncCall(Func<TWrite, byte[]> serializer, Func<byte[], TRead> deserializer)
         {
             this.serializer = serializer;
             this.deserializer = deserializer;
-            this.callbackHandler = HandleEvent;
-        }
-
-        public Task WriteAsync(TWrite msg)
-        {
-            return StartWrite(msg, false).Task;
-        }
-
-        public Task WritesCompletedAsync()
-        {
-            WritesDone();
-            return halfcloseTcs.Task;
-        }
-
-        public Task WriteStatusAsync(Status status)
-        {
-            WriteStatus(status);
-            return halfcloseTcs.Task;
-        }
-
-        public Task<TRead> ReadAsync()
-        {
-            return StartRead().Task;
-        }
-
-        public Task Halfclosed
-        {
-            get
-            {
-                return halfcloseTcs.Task;
-            }
-        }
-
-        public Task<Status> Finished
-        {
-            get
-            {
-                return finishedTcs.Task;
-            }
+            this.unaryResponseHandler = HandleUnaryResponseCompletion;
+            this.finishedHandler = HandleFinished;
+            this.writeFinishedHandler = HandleWriteFinished;
+            this.readFinishedHandler = HandleReadFinished;
+            this.halfclosedHandler = HandleHalfclosed;
+            this.finishedServersideHandler = HandleFinishedServerside;
         }
 
         /// <summary>
@@ -147,14 +107,14 @@ namespace Google.GRPC.Core.Internal
                     throw new InvalidOperationException("Already registered an observer.");
                 }
                 this.readObserver = readObserver;
-                StartRead();
+                ReceiveMessageAsync();
             }
         }
 
-        public void Initialize(Channel channel, String methodName) {
+        public void Initialize(Channel channel, CompletionQueueSafeHandle cq, String methodName) {
             lock (myLock)
             {
-               this.call = CallSafeHandle.Create(channel.Handle, methodName, channel.Target, Timespec.InfFuture);
+                this.call = CallSafeHandle.Create(channel.Handle, cq, methodName, channel.Target, Timespec.InfFuture);
             }
         }
 
@@ -163,42 +123,75 @@ namespace Google.GRPC.Core.Internal
             lock(myLock)
             {
                 this.call = call;
+                started = true;
+                server = true;
             }
         }
 
-        // Client only
-        public void Start(bool buffered, CompletionQueueSafeHandle cq)
+
+        public Task<TRead> UnaryCallAsync(TWrite msg)
         {
             lock (myLock)
             {
-                if (started)
-                {
-                    throw new InvalidOperationException("Already started.");
-                }
-
-                call.Invoke(cq, buffered, callbackHandler, callbackHandler);
                 started = true;
+                halfcloseRequested = true;
+
+                // TODO: handle serialization error...
+                byte[] payload = serializer(msg);
+
+                unaryResponseTcs = new TaskCompletionSource<TRead>();
+                call.StartUnary(payload, unaryResponseHandler);
+
+                return unaryResponseTcs.Task;
             }
         }
 
-        // Server only
-        public void Accept(CompletionQueueSafeHandle cq)
+        public Task<TRead> ClientStreamingCallAsync()
         {
             lock (myLock)
             {
-                if (started)
-                {
-                    throw new InvalidOperationException("Already started.");
-                }
+                started = true;
+
+                unaryResponseTcs = new TaskCompletionSource<TRead>();
+                call.StartClientStreaming(unaryResponseHandler);
+
+                return unaryResponseTcs.Task;
+            }
+        }
 
-                call.ServerAccept(cq, callbackHandler);
-                call.ServerEndInitialMetadata(0);
+        public void StartServerStreamingCall(TWrite msg, IObserver<TRead> readObserver)
+        {
+            lock (myLock)
+            {
                 started = true;
+                halfcloseRequested = true;
+        
+                this.readObserver = readObserver;
+
+                // TODO: handle serialization error...
+                byte[] payload = serializer(msg);
+        
+                call.StartServerStreaming(payload, finishedHandler);
+
+                ReceiveMessageAsync();
             }
         }
 
-        public TaskCompletionSource<object> StartWrite(TWrite msg, bool buffered)
+        public void StartDuplexStreamingCall(IObserver<TRead> readObserver)
         {
+            lock (myLock)
+            {
+                started = true;
+
+                this.readObserver = readObserver;
+
+                call.StartDuplexStreaming(finishedHandler);
+
+                ReceiveMessageAsync();
+            }
+        }
+
+        public Task SendMessageAsync(TWrite msg) {
             lock (myLock)
             {
                 CheckStarted();
@@ -219,14 +212,13 @@ namespace Google.GRPC.Core.Internal
                 // TODO: wrap serialization...
                 byte[] payload = serializer(msg);
 
-                call.StartWrite(payload, buffered, callbackHandler);
+                call.StartSendMessage(payload, writeFinishedHandler);
                 writeTcs = new TaskCompletionSource<object>();
-                return writeTcs;
+                return writeTcs.Task;
             }
         }
 
-        // client only
-        public void WritesDone()
+        public Task SendCloseFromClientAsync()
         {
             lock (myLock)
             {
@@ -240,13 +232,13 @@ namespace Google.GRPC.Core.Internal
                     throw new InvalidOperationException("Already halfclosed.");
                 }
 
-                call.WritesDone(callbackHandler);
+                call.StartSendCloseFromClient(halfclosedHandler);
                 halfcloseRequested = true;
+                return halfcloseTcs.Task;
             }
         }
 
-        // server only
-        public void WriteStatus(Status status)
+        public Task SendStatusFromServerAsync(Status status)
         {
             lock (myLock)
             {
@@ -260,12 +252,13 @@ namespace Google.GRPC.Core.Internal
                     throw new InvalidOperationException("Already halfclosed.");
                 }
 
-                call.StartWriteStatus(status, callbackHandler);
+                call.StartSendStatusFromServer(status, halfclosedHandler);
                 halfcloseRequested = true;
+                return halfcloseTcs.Task;
             }
         }
 
-        public TaskCompletionSource<TRead> StartRead()
+        public Task<TRead> ReceiveMessageAsync()
         {
             lock (myLock)
             {
@@ -285,10 +278,19 @@ namespace Google.GRPC.Core.Internal
                     throw new InvalidOperationException("Only one read can be pending at a time");
                 }
 
-                call.StartRead(callbackHandler);
+                call.StartReceiveMessage(readFinishedHandler);
 
                 readTcs = new TaskCompletionSource<TRead>();
-                return readTcs;
+                return readTcs.Task;
+            }
+        }
+
+        internal Task StartServerSide()
+        {
+            lock (myLock)
+            {
+                call.StartServerSide(finishedServersideHandler);
+                return finishedServersideTcs.Task;
             }
         }
 
@@ -317,107 +319,7 @@ namespace Google.GRPC.Core.Internal
             // grpc_call_cancel_with_status is threadsafe
             call.CancelWithStatus(status);
         }
-
-        public void OnClientMetadata()
-        {
-            // TODO: implement....
-        }
-
-        public void OnRead(byte[] payload)
-        {
-            TaskCompletionSource<TRead> oldTcs = null;
-            IObserver<TRead> observer = null;
-            lock (myLock)
-            {
-                oldTcs = readTcs;
-                readTcs = null;
-                if (payload == null)
-                {
-                    doneWithReading = true;
-                }
-                observer = readObserver;
-            }
-
-            // TODO: wrap deserialization...
-            TRead msg = payload != null ? deserializer(payload) : default(TRead);
-
-            oldTcs.SetResult(msg);
-
-            // TODO: make sure we deliver reads in the right order.
-
-            if (observer != null)
-            {
-                if (payload != null)
-                {
-                    // TODO: wrap to handle exceptions
-                    observer.OnNext(msg);
-
-                    // start a new read
-                    StartRead();
-                }
-                else
-                {
-                    // TODO: wrap to handle exceptions;
-                    observer.OnCompleted();
-                }
-
-            }
-        }
-
-        public void OnWriteAccepted(GRPCOpError error)
-        {
-            TaskCompletionSource<object> oldTcs = null;
-            lock (myLock)
-            {
-                UpdateErrorOccured(error);
-                oldTcs = writeTcs;
-                writeTcs = null;
-            }
-
-            if (errorOccured)
-            {
-                // TODO: use the right type of exception...
-                oldTcs.SetException(new Exception("Write failed"));
-            }
-            else
-            {
-                // TODO: where does the continuation run?
-                oldTcs.SetResult(null);
-            }
-        }
-
-        public void OnFinishAccepted(GRPCOpError error)
-        {
-            lock (myLock)
-            {
-                UpdateErrorOccured(error);
-                halfclosed = true;
-            }
-
-            if (errorOccured)
-            {
-                halfcloseTcs.SetException(new Exception("Halfclose failed"));
-
-            }
-            else
-            {
-                halfcloseTcs.SetResult(null);
-            }
-
-        }
-
-        public void OnFinished(Status status)
-        {
-            lock (myLock)
-            {
-                finishedStatus = status;
-
-                DisposeResourcesIfNeeded();
-            }
-            finishedTcs.SetResult(status);
-
-        }
-
+       
         public void Dispose()
         {
             Dispose(true);
@@ -434,7 +336,7 @@ namespace Google.GRPC.Core.Internal
                     {
                         call.Dispose();
                     }
-                }
+                } 
                 disposed = true;
             }
         }
@@ -489,38 +391,195 @@ namespace Google.GRPC.Core.Internal
             }
         }
 
-        private void HandleEvent(IntPtr eventPtr) {
+        private void CompleteStreamObserver(Status status) {
+            if (status.StatusCode != StatusCode.GRPC_STATUS_OK)
+            {
+                // TODO: wrap to handle exceptions;
+                readObserver.OnError(new RpcException(status));
+            } else {
+                // TODO: wrap to handle exceptions;
+                readObserver.OnCompleted();
+            }
+        }
+
+        private void HandleUnaryResponseCompletion(GRPCOpError error, IntPtr batchContextPtr) {
+            try {
+
+                TaskCompletionSource<TRead> tcs;
+                lock(myLock) {
+                    tcs = unaryResponseTcs;
+                }
+
+                // we're done with this call, get rid of the native object.
+                call.Dispose();
+
+                var ctx = new BatchContextSafeHandleNotOwned(batchContextPtr);
+
+                if (error != GRPCOpError.GRPC_OP_OK) {
+                    tcs.SetException(new RpcException(
+                        new Status(StatusCode.GRPC_STATUS_INTERNAL, "Internal error occured.")
+                    ));
+                    return;
+                }
+
+                var status = ctx.GetReceivedStatus();
+                if (status.StatusCode != StatusCode.GRPC_STATUS_OK) {
+                    tcs.SetException(new RpcException(status));
+                    return;
+                }
+
+                // TODO: handle deserialize error...
+                var msg = deserializer(ctx.GetReceivedMessage());
+                tcs.SetResult(msg);
+            } catch(Exception e) {
+                Console.WriteLine("Caught exception in a native handler: " + e);
+            }
+        }
+
+        private void HandleWriteFinished(GRPCOpError error, IntPtr batchContextPtr) {
+            try {
+
+                TaskCompletionSource<object> oldTcs = null;
+                lock (myLock)
+                {
+                    UpdateErrorOccured(error);
+                    oldTcs = writeTcs;
+                    writeTcs = null;
+                }
+
+                if (errorOccured)
+                {
+                    // TODO: use the right type of exception...
+                    oldTcs.SetException(new Exception("Write failed"));
+                }
+                else
+                {
+                    // TODO: where does the continuation run?
+                    oldTcs.SetResult(null);
+                }
+
+            } catch(Exception e) {
+                Console.WriteLine("Caught exception in a native handler: " + e);
+            }
+        }
+
+        private void HandleHalfclosed(GRPCOpError error, IntPtr batchContextPtr) {
+            try {
+                lock (myLock)
+                {
+                    UpdateErrorOccured(error);
+                    halfclosed = true;
+                }
+
+                if (errorOccured)
+                {
+                    halfcloseTcs.SetException(new Exception("Halfclose failed"));
+
+                }
+                else
+                {
+                    halfcloseTcs.SetResult(null);
+                }
+            } catch(Exception e) {
+                Console.WriteLine("Caught exception in a native handler: " + e);
+            }
+        }
+
+        private void HandleReadFinished(GRPCOpError error, IntPtr batchContextPtr) {
             try {
-                var ev = new EventSafeHandleNotOwned(eventPtr);
-                switch (ev.GetCompletionType())
+
+                var ctx = new BatchContextSafeHandleNotOwned(batchContextPtr);
+                var payload = ctx.GetReceivedMessage();
+
+                TaskCompletionSource<TRead> oldTcs = null;
+                IObserver<TRead> observer = null;
+
+                Nullable<Status> status = null;
+
+                lock (myLock)
                 {
-                case GRPCCompletionType.GRPC_CLIENT_METADATA_READ:
-                    OnClientMetadata();
-                    break;
+                    oldTcs = readTcs;
+                    readTcs = null;
+                    if (payload == null)
+                    {
+                        doneWithReading = true;
+                    }
+                    observer = readObserver;
+                    status = finishedStatus;
+                }
+
+                // TODO: wrap deserialization...
+                TRead msg = payload != null ? deserializer(payload) : default(TRead);
 
-                case GRPCCompletionType.GRPC_READ:
-                    byte[] payload = ev.GetReadData();
-                    OnRead(payload);
-                    break;
+                oldTcs.SetResult(msg);
 
-                case GRPCCompletionType.GRPC_WRITE_ACCEPTED:
-                    OnWriteAccepted(ev.GetWriteAccepted());
-                    break;
+                // TODO: make sure we deliver reads in the right order.
 
-                case GRPCCompletionType.GRPC_FINISH_ACCEPTED:
-                    OnFinishAccepted(ev.GetFinishAccepted());
-                    break;
+                if (observer != null) {
+                    if (payload != null)
+                    {
+                        // TODO: wrap to handle exceptions
+                        observer.OnNext(msg);
+
+                        // start a new read
+                        ReceiveMessageAsync();
+                    }
+                    else
+                    {
+                        if (!server) {
+                            if (status.HasValue) {
+                                CompleteStreamObserver(status.Value);
+                            }
+                        } else {
+                            // TODO: wrap to handle exceptions..
+                            observer.OnCompleted();
+                        }
+                        // TODO: completeStreamObserver serverside...
+                    }
+               }
+            } catch(Exception e) {
+                Console.WriteLine("Caught exception in a native handler: " + e);
+            }
+        }
+
+        private void HandleFinished(GRPCOpError error, IntPtr batchContextPtr) {
+            try {
+                var ctx = new BatchContextSafeHandleNotOwned(batchContextPtr);
+                var status = ctx.GetReceivedStatus();
+
+                bool wasDoneWithReading;
+
+                lock (myLock)
+                {
+                    finishedStatus = status;
 
-                case GRPCCompletionType.GRPC_FINISHED:
-                    OnFinished(ev.GetFinished());
-                    break;
+                    DisposeResourcesIfNeeded();
 
-                default:
-                    throw new ArgumentException("Unexpected completion type");
+                    wasDoneWithReading = doneWithReading;
                 }
+
+                if (wasDoneWithReading) {
+                    CompleteStreamObserver(status);
+                }
+
+            } catch(Exception e) {
+                Console.WriteLine("Caught exception in a native handler: " + e);
+            }
+        }
+
+        private void HandleFinishedServerside(GRPCOpError error, IntPtr batchContextPtr) {
+            try {
+                var ctx = new BatchContextSafeHandleNotOwned(batchContextPtr);
+
+                // TODO: handle error ...
+
+                finishedServersideTcs.SetResult(null);
+
+                call.Dispose();
+
             } catch(Exception e) {
                 Console.WriteLine("Caught exception in a native handler: " + e);
             }
         }
     }
-}
+}

src/csharp/GrpcCore/Internal/BatchContextSafeHandle.cs

@@ -0,0 +1,96 @@
+#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 Google.GRPC.Core;
+
+namespace Google.GRPC.Core.Internal
+{
+    /// <summary>
+    /// Not owned version of 
+    /// grpcsharp_batch_context
+    /// </summary>
+    internal class BatchContextSafeHandleNotOwned : SafeHandleZeroIsInvalid
+    {
+        [DllImport("grpc_csharp_ext.dll")]
+        static extern IntPtr grpcsharp_batch_context_recv_message_length(BatchContextSafeHandleNotOwned ctx);
+
+        [DllImport("grpc_csharp_ext.dll")]
+        static extern void grpcsharp_batch_context_recv_message_to_buffer(BatchContextSafeHandleNotOwned ctx, byte[] buffer, UIntPtr bufferLen);
+
+        [DllImport("grpc_csharp_ext.dll")]
+        static extern StatusCode grpcsharp_batch_context_recv_status_on_client_status(BatchContextSafeHandleNotOwned ctx);
+
+        [DllImport("grpc_csharp_ext.dll")]
+        static extern IntPtr grpcsharp_batch_context_recv_status_on_client_details(BatchContextSafeHandleNotOwned ctx);  // returns const char*
+
+        [DllImport("grpc_csharp_ext.dll")]
+        static extern CallSafeHandle grpcsharp_batch_context_server_rpc_new_call(BatchContextSafeHandleNotOwned ctx);
+
+        [DllImport("grpc_csharp_ext.dll")]
+        static extern IntPtr grpcsharp_batch_context_server_rpc_new_method(BatchContextSafeHandleNotOwned ctx);  // returns const char*
+
+        public BatchContextSafeHandleNotOwned(IntPtr handle) : base(false)
+        {
+            SetHandle(handle);
+        }
+
+        public Status GetReceivedStatus()
+        {
+            // TODO: can the native method return string directly?
+            string details = Marshal.PtrToStringAnsi(grpcsharp_batch_context_recv_status_on_client_details(this));
+            return new Status(grpcsharp_batch_context_recv_status_on_client_status(this), details);
+        }
+
+        public byte[] GetReceivedMessage()
+        {
+            IntPtr len = grpcsharp_batch_context_recv_message_length(this);
+            if (len == new IntPtr(-1))
+            {
+                return null;
+            }
+            byte[] data = new byte[(int) len];
+            grpcsharp_batch_context_recv_message_to_buffer(this, data, new UIntPtr((ulong)data.Length));
+            return data;
+        }
+
+        public CallSafeHandle GetServerRpcNewCall() {
+            return grpcsharp_batch_context_server_rpc_new_call(this);
+        }
+
+        public string GetServerRpcNewMethod() {
+            return Marshal.PtrToStringAnsi(grpcsharp_batch_context_server_rpc_new_method(this));
+        }
+    }
+}

src/csharp/GrpcCore/Internal/CallSafeHandle.cs

@@ -2,11 +2,11 @@
 
 // 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
@@ -16,7 +16,7 @@
 //     * 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
@@ -38,8 +38,8 @@ using Google.GRPC.Core;
 
 namespace Google.GRPC.Core.Internal
 {
-    // TODO: we need to make sure that the delegates are not collected before invoked.
-    internal delegate void EventCallbackDelegate(IntPtr eventPtr);
+    //TODO: rename the delegate
+    internal delegate void CompletionCallbackDelegate(GRPCOpError error, IntPtr batchContextPtr);
 
     /// <summary>
     /// grpc_call from <grpc/grpc.h>
@@ -49,142 +49,108 @@ namespace Google.GRPC.Core.Internal
         const UInt32 GRPC_WRITE_BUFFER_HINT = 1;
 
         [DllImport("grpc_csharp_ext.dll")]
-        static extern CallSafeHandle grpcsharp_channel_create_call_old(ChannelSafeHandle channel, string method, string host, Timespec deadline);
+        static extern CallSafeHandle grpcsharp_channel_create_call(ChannelSafeHandle channel, CompletionQueueSafeHandle cq, string method, string host, Timespec deadline);
 
         [DllImport("grpc_csharp_ext.dll")]
-        static extern GRPCCallError grpcsharp_call_add_metadata(CallSafeHandle call, IntPtr metadata, UInt32 flags);
+        static extern GRPCCallError grpcsharp_call_cancel(CallSafeHandle call);
 
         [DllImport("grpc_csharp_ext.dll")]
-        static extern GRPCCallError grpcsharp_call_invoke_old(CallSafeHandle call, CompletionQueueSafeHandle cq, IntPtr metadataReadTag, IntPtr finishedTag, UInt32 flags);
-
-        [DllImport("grpc_csharp_ext.dll", EntryPoint = "grpcsharp_call_invoke_old")]
-        static extern GRPCCallError grpcsharp_call_invoke_old_CALLBACK(CallSafeHandle call, CompletionQueueSafeHandle cq,
-                                                              [MarshalAs(UnmanagedType.FunctionPtr)] EventCallbackDelegate metadataReadCallback,
-                                                              [MarshalAs(UnmanagedType.FunctionPtr)] EventCallbackDelegate finishedCallback,
-                                                              UInt32 flags);
+        static extern GRPCCallError grpcsharp_call_cancel_with_status(CallSafeHandle call, StatusCode status, string description);
 
         [DllImport("grpc_csharp_ext.dll")]
-        static extern GRPCCallError grpcsharp_call_server_accept_old(CallSafeHandle call, CompletionQueueSafeHandle completionQueue, IntPtr finishedTag);
-
-        [DllImport("grpc_csharp_ext.dll", EntryPoint = "grpcsharp_call_server_accept_old")]
-        static extern GRPCCallError grpcsharp_call_server_accept_old_CALLBACK(CallSafeHandle call, CompletionQueueSafeHandle completionQueue, [MarshalAs(UnmanagedType.FunctionPtr)] EventCallbackDelegate finishedCallback);
+        static extern GRPCCallError grpcsharp_call_start_unary(CallSafeHandle call,
+                                                                        [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback,
+                                                                        byte[] send_buffer, UIntPtr send_buffer_len);
 
         [DllImport("grpc_csharp_ext.dll")]
-        static extern GRPCCallError grpcsharp_call_server_end_initial_metadata_old(CallSafeHandle call, UInt32 flags);
+        static extern GRPCCallError grpcsharp_call_start_client_streaming(CallSafeHandle call,
+                                                                      [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback);
 
         [DllImport("grpc_csharp_ext.dll")]
-        static extern GRPCCallError grpcsharp_call_cancel(CallSafeHandle call);
+        static extern GRPCCallError grpcsharp_call_start_server_streaming(CallSafeHandle call,
+                                                                      [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback,
+                                                                      byte[] send_buffer, UIntPtr send_buffer_len);
 
         [DllImport("grpc_csharp_ext.dll")]
-        static extern GRPCCallError grpcsharp_call_cancel_with_status(CallSafeHandle call, StatusCode status, string description);
+        static extern GRPCCallError grpcsharp_call_start_duplex_streaming(CallSafeHandle call,
+                                                                          [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback);
 
         [DllImport("grpc_csharp_ext.dll")]
-        static extern GRPCCallError grpcsharp_call_start_write_status_old(CallSafeHandle call, StatusCode statusCode, string statusMessage, IntPtr tag);
-
-        [DllImport("grpc_csharp_ext.dll", EntryPoint = "grpcsharp_call_start_write_status_old")]
-        static extern GRPCCallError grpcsharp_call_start_write_status_old_CALLBACK(CallSafeHandle call, StatusCode statusCode, string statusMessage, [MarshalAs(UnmanagedType.FunctionPtr)] EventCallbackDelegate callback);
+        static extern GRPCCallError grpcsharp_call_send_message(CallSafeHandle call,
+                                                                      [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback,
+                                                                      byte[] send_buffer, UIntPtr send_buffer_len);
 
         [DllImport("grpc_csharp_ext.dll")]
-        static extern GRPCCallError grpcsharp_call_writes_done_old(CallSafeHandle call, IntPtr tag);
-
-        [DllImport("grpc_csharp_ext.dll", EntryPoint = "grpcsharp_call_writes_done_old")]
-        static extern GRPCCallError grpcsharp_call_writes_done_old_CALLBACK(CallSafeHandle call, [MarshalAs(UnmanagedType.FunctionPtr)] EventCallbackDelegate callback);
+        static extern GRPCCallError grpcsharp_call_send_close_from_client(CallSafeHandle call,
+                                                                             [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback);
 
         [DllImport("grpc_csharp_ext.dll")]
-        static extern GRPCCallError grpcsharp_call_start_read_old(CallSafeHandle call, IntPtr tag);
-
-        [DllImport("grpc_csharp_ext.dll", EntryPoint = "grpcsharp_call_start_read_old")]
-        static extern GRPCCallError grpcsharp_call_start_read_old_CALLBACK(CallSafeHandle call, [MarshalAs(UnmanagedType.FunctionPtr)] EventCallbackDelegate callback);
+        static extern GRPCCallError grpcsharp_call_send_status_from_server(CallSafeHandle call, [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback, StatusCode statusCode, string statusMessage);
 
         [DllImport("grpc_csharp_ext.dll")]
-        static extern void grpcsharp_call_start_write_from_copied_buffer(CallSafeHandle call,
-                                                                    byte[] buffer, UIntPtr length,
-                                                                    IntPtr tag, UInt32 flags);
+        static extern GRPCCallError grpcsharp_call_recv_message(CallSafeHandle call,
+                                                               [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback);
 
-        [DllImport("grpc_csharp_ext.dll", EntryPoint = "grpcsharp_call_start_write_from_copied_buffer")]
-        static extern void grpcsharp_call_start_write_from_copied_buffer_CALLBACK(CallSafeHandle call,
-                                                                             byte[] buffer, UIntPtr length,
-                                                                             [MarshalAs(UnmanagedType.FunctionPtr)] EventCallbackDelegate callback,
-                                                                             UInt32 flags);
+        [DllImport("grpc_csharp_ext.dll")]
+        static extern GRPCCallError grpcsharp_call_start_serverside(CallSafeHandle call,
+                                                                [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback);
 
-		[DllImport("grpc_csharp_ext.dll")]
+        [DllImport("grpc_csharp_ext.dll")]
         static extern void grpcsharp_call_destroy(IntPtr call);
 
-        private CallSafeHandle()
-        {
-        }
-
-        /// <summary>
-        /// Creates a client call.
-        /// </summary>
-        public static CallSafeHandle Create(ChannelSafeHandle channel, string method, string host, Timespec deadline)
-        {
-            return grpcsharp_channel_create_call_old(channel, method, host, deadline);
-        }
-
-        public void Invoke(CompletionQueueSafeHandle cq, IntPtr metadataReadTag, IntPtr finishedTag, bool buffered)
-        {
-            AssertCallOk(grpcsharp_call_invoke_old(this, cq, metadataReadTag, finishedTag, GetFlags(buffered)));
-        }
-
-        public void Invoke(CompletionQueueSafeHandle cq, bool buffered, EventCallbackDelegate metadataReadCallback, EventCallbackDelegate finishedCallback)
-        {
-            AssertCallOk(grpcsharp_call_invoke_old_CALLBACK(this, cq, metadataReadCallback, finishedCallback, GetFlags(buffered)));
-        }
 
-        public void ServerAccept(CompletionQueueSafeHandle cq, IntPtr finishedTag)
+        private CallSafeHandle()
         {
-            AssertCallOk(grpcsharp_call_server_accept_old(this, cq, finishedTag));
         }
 
-        public void ServerAccept(CompletionQueueSafeHandle cq, EventCallbackDelegate callback)
+        public static CallSafeHandle Create(ChannelSafeHandle channel, CompletionQueueSafeHandle cq, string method, string host, Timespec deadline)
         {
-            AssertCallOk(grpcsharp_call_server_accept_old_CALLBACK(this, cq, callback));
+            return grpcsharp_channel_create_call(channel, cq, method, host, deadline);
         }
 
-        public void ServerEndInitialMetadata(UInt32 flags)
+        public void StartUnary(byte[] payload, CompletionCallbackDelegate callback)
         {
-            AssertCallOk(grpcsharp_call_server_end_initial_metadata_old(this, flags));
+            AssertCallOk(grpcsharp_call_start_unary(this, callback, payload, new UIntPtr((ulong) payload.Length)));
         }
 
-        public void StartWrite(byte[] payload, IntPtr tag, bool buffered)
+        public void StartClientStreaming(CompletionCallbackDelegate callback)
         {
-            grpcsharp_call_start_write_from_copied_buffer(this, payload, new UIntPtr((ulong) payload.Length), tag, GetFlags(buffered));
+            AssertCallOk(grpcsharp_call_start_client_streaming(this, callback));
         }
 
-        public void StartWrite(byte[] payload, bool buffered, EventCallbackDelegate callback)
+        public void StartServerStreaming(byte[] payload, CompletionCallbackDelegate callback)
         {
-            grpcsharp_call_start_write_from_copied_buffer_CALLBACK(this, payload, new UIntPtr((ulong) payload.Length), callback, GetFlags(buffered));
+            AssertCallOk(grpcsharp_call_start_server_streaming(this, callback, payload, new UIntPtr((ulong) payload.Length)));
         }
 
-        public void StartWriteStatus(Status status, IntPtr tag)
+        public void StartDuplexStreaming(CompletionCallbackDelegate callback)
         {
-            AssertCallOk(grpcsharp_call_start_write_status_old(this, status.StatusCode, status.Detail, tag));
+            AssertCallOk(grpcsharp_call_start_duplex_streaming(this, callback));
         }
 
-        public void StartWriteStatus(Status status, EventCallbackDelegate callback)
+        public void StartSendMessage(byte[] payload, CompletionCallbackDelegate callback)
         {
-            AssertCallOk(grpcsharp_call_start_write_status_old_CALLBACK(this, status.StatusCode, status.Detail, callback));
+            AssertCallOk(grpcsharp_call_send_message(this, callback, payload, new UIntPtr((ulong) payload.Length)));
         }
 
-        public void WritesDone(IntPtr tag)
+        public void StartSendCloseFromClient(CompletionCallbackDelegate callback)
         {
-            AssertCallOk(grpcsharp_call_writes_done_old(this, tag));
+            AssertCallOk(grpcsharp_call_send_close_from_client(this, callback));
         }
 
-        public void WritesDone(EventCallbackDelegate callback)
+        public void StartSendStatusFromServer(Status status, CompletionCallbackDelegate callback)
         {
-            AssertCallOk(grpcsharp_call_writes_done_old_CALLBACK(this, callback));
+            AssertCallOk(grpcsharp_call_send_status_from_server(this, callback, status.StatusCode, status.Detail));
         }
 
-        public void StartRead(IntPtr tag)
+        public void StartReceiveMessage(CompletionCallbackDelegate callback)
         {
-            AssertCallOk(grpcsharp_call_start_read_old(this, tag));
+            AssertCallOk(grpcsharp_call_recv_message(this, callback));
         }
 
-        public void StartRead(EventCallbackDelegate callback)
+        public void StartServerSide(CompletionCallbackDelegate callback)
         {
-            AssertCallOk(grpcsharp_call_start_read_old_CALLBACK(this, callback));
+            AssertCallOk(grpcsharp_call_start_serverside(this, callback));
         }
 
         public void Cancel()
@@ -212,4 +178,4 @@ namespace Google.GRPC.Core.Internal
             return buffered ? 0 : GRPC_WRITE_BUFFER_HINT;
         }
 	}
-}
+}

src/csharp/GrpcCore/Internal/ClientStreamingInputObserver.cs

@@ -2,11 +2,11 @@
 
 // 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
@@ -16,7 +16,7 @@
 //     * 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
@@ -36,19 +36,20 @@ using Google.GRPC.Core.Internal;
 
 namespace Google.GRPC.Core.Internal
 {
-    internal class StreamingInputObserver<TWrite, TRead> : IObserver<TWrite>
+    internal class ClientStreamingInputObserver<TWrite, TRead> : IObserver<TWrite>
 	{
         readonly AsyncCall<TWrite, TRead> call;
 
-        public StreamingInputObserver(AsyncCall<TWrite, TRead> call)
+        public ClientStreamingInputObserver(AsyncCall<TWrite, TRead> call)
 		{
             this.call = call;
 		}
 
 		public void OnCompleted()
 		{
+
             // TODO: how bad is the Wait here?
-            call.WritesCompletedAsync().Wait();
+            call.SendCloseFromClientAsync().Wait();
 		}
 
 		public void OnError(Exception error)
@@ -59,7 +60,7 @@ namespace Google.GRPC.Core.Internal
 		public void OnNext(TWrite value)
 		{
             // TODO: how bad is the Wait here?
-            call.WriteAsync(value).Wait();
+            call.SendMessageAsync(value).Wait();
 		}
 	}
 }

src/csharp/GrpcCore/Internal/CompletionQueueSafeHandle.cs

@@ -45,12 +45,6 @@ namespace Google.GRPC.Core.Internal
         [DllImport("grpc_csharp_ext.dll")]
         static extern CompletionQueueSafeHandle grpcsharp_completion_queue_create();
 
-        [DllImport("grpc_csharp_ext.dll")]
-        static extern EventSafeHandle grpcsharp_completion_queue_pluck(CompletionQueueSafeHandle cq, IntPtr tag, Timespec deadline);
-
-        [DllImport("grpc_csharp_ext.dll")]
-        static extern EventSafeHandle grpcsharp_completion_queue_next(CompletionQueueSafeHandle cq, Timespec deadline);
-
         [DllImport("grpc_csharp_ext.dll")]
         static extern void grpcsharp_completion_queue_shutdown(CompletionQueueSafeHandle cq);
 
@@ -69,21 +63,11 @@ namespace Google.GRPC.Core.Internal
             return grpcsharp_completion_queue_create();
         }
 
-        public EventSafeHandle Next(Timespec deadline)
-        {
-            return grpcsharp_completion_queue_next(this, deadline);
-        }
-
         public GRPCCompletionType NextWithCallback()
         {
             return grpcsharp_completion_queue_next_with_callback(this);
         }
 
-        public EventSafeHandle Pluck(IntPtr tag, Timespec deadline)
-        {
-            return grpcsharp_completion_queue_pluck(this, tag, deadline);
-        }
-
         public void Shutdown()
         {
             grpcsharp_completion_queue_shutdown(this);

src/csharp/GrpcCore/Internal/Event.cs

@@ -1,224 +0,0 @@
-#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 Google.GRPC.Core;
-
-namespace Google.GRPC.Core.Internal
-{
-    /// <summary>
-    /// grpc_event from grpc/grpc.h
-    /// </summary>
-    internal class EventSafeHandle : SafeHandleZeroIsInvalid
-    {
-        [DllImport("grpc_csharp_ext.dll")]
-        static extern void grpcsharp_event_finish(IntPtr ev);
-
-        [DllImport("grpc_csharp_ext.dll")]
-        static extern GRPCCompletionType grpcsharp_event_type(EventSafeHandle ev);
-
-        [DllImport("grpc_csharp_ext.dll")]
-        static extern CallSafeHandle grpcsharp_event_call(EventSafeHandle ev);
-
-        [DllImport("grpc_csharp_ext.dll")]
-        static extern GRPCOpError grpcsharp_event_write_accepted(EventSafeHandle ev);
-
-        [DllImport("grpc_csharp_ext.dll")]
-        static extern GRPCOpError grpcsharp_event_finish_accepted(EventSafeHandle ev);
-
-        [DllImport("grpc_csharp_ext.dll")]
-        static extern StatusCode grpcsharp_event_finished_status(EventSafeHandle ev);
-
-        [DllImport("grpc_csharp_ext.dll")]
-        static extern IntPtr grpcsharp_event_finished_details(EventSafeHandle ev);  // returns const char*
-
-        [DllImport("grpc_csharp_ext.dll")]
-        static extern IntPtr grpcsharp_event_read_length(EventSafeHandle ev);
-
-        [DllImport("grpc_csharp_ext.dll")]
-        static extern void grpcsharp_event_read_copy_to_buffer(EventSafeHandle ev, byte[] buffer, UIntPtr bufferLen);
-
-        [DllImport("grpc_csharp_ext.dll")]
-        static extern IntPtr grpcsharp_event_server_rpc_new_method(EventSafeHandle ev); // returns const char*
-
-        public GRPCCompletionType GetCompletionType()
-        {
-            return grpcsharp_event_type(this);
-        }
-
-        public GRPCOpError GetWriteAccepted()
-        {
-            return grpcsharp_event_write_accepted(this);
-        }
-
-        public GRPCOpError GetFinishAccepted()
-        {
-            return grpcsharp_event_finish_accepted(this);
-        }
-
-        public Status GetFinished()
-        {
-            // TODO: can the native method return string directly?
-            string details = Marshal.PtrToStringAnsi(grpcsharp_event_finished_details(this));
-            return new Status(grpcsharp_event_finished_status(this), details);
-        }
-
-        public byte[] GetReadData()
-        {
-            IntPtr len = grpcsharp_event_read_length(this);
-            if (len == new IntPtr(-1))
-            {
-                return null;
-            }
-            byte[] data = new byte[(int) len];
-            grpcsharp_event_read_copy_to_buffer(this, data, new UIntPtr((ulong)data.Length));
-            return data;
-        }
-
-        public CallSafeHandle GetCall() {
-            return grpcsharp_event_call(this);
-        }
-
-        public string GetServerRpcNewMethod() {
-            // TODO: can the native method return string directly?
-            return Marshal.PtrToStringAnsi(grpcsharp_event_server_rpc_new_method(this));
-        }
-
-        //TODO: client_metadata_read event type
-
-        protected override bool ReleaseHandle()
-        {
-            grpcsharp_event_finish(handle);
-            return true;
-        }
-    }
-
-    // TODO: this is basically c&p of EventSafeHandle. Unify!
-    /// <summary>
-    /// Not owned version of
-    /// grpc_event from grpc/grpc.h
-    /// </summary>
-    internal class EventSafeHandleNotOwned : SafeHandleZeroIsInvalid
-    {
-        [DllImport("grpc_csharp_ext.dll")]
-        static extern void grpcsharp_event_finish(IntPtr ev);
-
-        [DllImport("grpc_csharp_ext.dll")]
-        static extern GRPCCompletionType grpcsharp_event_type(EventSafeHandleNotOwned ev);
-
-        [DllImport("grpc_csharp_ext.dll")]
-        static extern CallSafeHandle grpcsharp_event_call(EventSafeHandleNotOwned ev);
-
-        [DllImport("grpc_csharp_ext.dll")]
-        static extern GRPCOpError grpcsharp_event_write_accepted(EventSafeHandleNotOwned ev);
-
-        [DllImport("grpc_csharp_ext.dll")]
-        static extern GRPCOpError grpcsharp_event_finish_accepted(EventSafeHandleNotOwned ev);
-
-        [DllImport("grpc_csharp_ext.dll")]
-        static extern StatusCode grpcsharp_event_finished_status(EventSafeHandleNotOwned ev);
-
-        [DllImport("grpc_csharp_ext.dll")]
-        static extern IntPtr grpcsharp_event_finished_details(EventSafeHandleNotOwned ev);  // returns const char*
-
-        [DllImport("grpc_csharp_ext.dll")]
-        static extern IntPtr grpcsharp_event_read_length(EventSafeHandleNotOwned ev);
-
-        [DllImport("grpc_csharp_ext.dll")]
-        static extern void grpcsharp_event_read_copy_to_buffer(EventSafeHandleNotOwned ev, byte[] buffer, UIntPtr bufferLen);
-
-        [DllImport("grpc_csharp_ext.dll")]
-        static extern IntPtr grpcsharp_event_server_rpc_new_method(EventSafeHandleNotOwned ev); // returns const char*
-
-        public EventSafeHandleNotOwned() : base(false)
-        {
-        }
-
-        public EventSafeHandleNotOwned(IntPtr handle) : base(false)
-        {
-            SetHandle(handle);
-        }
-
-        public GRPCCompletionType GetCompletionType()
-        {
-            return grpcsharp_event_type(this);
-        }
-
-        public GRPCOpError GetWriteAccepted()
-        {
-            return grpcsharp_event_write_accepted(this);
-        }
-
-        public GRPCOpError GetFinishAccepted()
-        {
-            return grpcsharp_event_finish_accepted(this);
-        }
-
-        public Status GetFinished()
-        {
-            // TODO: can the native method return string directly?
-            string details = Marshal.PtrToStringAnsi(grpcsharp_event_finished_details(this));
-            return new Status(grpcsharp_event_finished_status(this), details);
-        }
-
-        public byte[] GetReadData()
-        {
-            IntPtr len = grpcsharp_event_read_length(this);
-            if (len == new IntPtr(-1))
-            {
-                return null;
-            }
-            byte[] data = new byte[(int) len];
-            grpcsharp_event_read_copy_to_buffer(this, data, new UIntPtr((ulong)data.Length));
-            return data;
-        }
-
-        public CallSafeHandle GetCall() {
-            return grpcsharp_event_call(this);
-        }
-
-        public string GetServerRpcNewMethod() {
-            // TODO: can the native method return string directly?
-            return Marshal.PtrToStringAnsi(grpcsharp_event_server_rpc_new_method(this));
-        }
-
-        //TODO: client_metadata_read event type
-
-        protected override bool ReleaseHandle()
-        {
-            grpcsharp_event_finish(handle);
-            return true;
-        }
-    }
-}

src/csharp/GrpcCore/Internal/GrpcThreadPool.cs

@@ -48,7 +48,6 @@ namespace Google.GRPC.Core.Internal
         readonly object myLock = new object();
         readonly List<Thread> threads = new List<Thread>();
         readonly int poolSize;
-        readonly Action<EventSafeHandle> eventHandler;
 
         CompletionQueueSafeHandle cq;
 
@@ -56,11 +55,6 @@ namespace Google.GRPC.Core.Internal
             this.poolSize = poolSize;
         }
 
-        internal GrpcThreadPool(int poolSize, Action<EventSafeHandle> eventHandler) {
-            this.poolSize = poolSize;
-            this.eventHandler = eventHandler;
-        }
-
         public void Start() {
 
             lock (myLock)
@@ -104,34 +98,19 @@ namespace Google.GRPC.Core.Internal
             }
         }
 
-        private Thread CreateAndStartThread(int i) {
-            Action body;
-            if (eventHandler != null)
-            {
-                body = ThreadBodyWithHandler;
-            }
-            else
-            {
-                body = ThreadBodyNoHandler;
-            }
-            var thread = new Thread(new ThreadStart(body));
+        private Thread CreateAndStartThread(int i)
+        {
+            var thread = new Thread(new ThreadStart(RunHandlerLoop));
             thread.IsBackground = false;
             thread.Start();
-            if (eventHandler != null)
-            {
-                thread.Name = "grpc_server_newrpc " + i;
-            }
-            else
-            {
-                thread.Name = "grpc " + i;
-            }
+            thread.Name = "grpc " + i;
             return thread;
         }
 
         /// <summary>
         /// Body of the polling thread.
         /// </summary>
-        private void ThreadBodyNoHandler()
+        private void RunHandlerLoop()
         {
             GRPCCompletionType completionType;
             do
@@ -140,22 +119,6 @@ namespace Google.GRPC.Core.Internal
             } while(completionType != GRPCCompletionType.GRPC_QUEUE_SHUTDOWN);
             Console.WriteLine("Completion queue has shutdown successfully, thread " + Thread.CurrentThread.Name + " exiting.");
         }
-
-        /// <summary>
-        /// Body of the polling thread.
-        /// </summary>
-        private void ThreadBodyWithHandler()
-        {
-            GRPCCompletionType completionType;
-            do
-            {
-                using (EventSafeHandle ev = cq.Next(Timespec.InfFuture)) {
-                    completionType = ev.GetCompletionType();
-                    eventHandler(ev);
-                }
-            } while(completionType != GRPCCompletionType.GRPC_QUEUE_SHUTDOWN);
-            Console.WriteLine("Completion queue has shutdown successfully, thread " + Thread.CurrentThread.Name + " exiting.");
-        }
     }
 
 }

src/csharp/GrpcCore/Internal/SafeHandleZeroIsInvalid.cs

@@ -56,6 +56,12 @@ namespace Google.GRPC.Core.Internal
                 return handle == IntPtr.Zero;
             }
         }
+
+        protected override bool ReleaseHandle()
+        {
+            // handle is not owned.
+            return true;
+        }
     }
 }
 

src/csharp/GrpcCore/Internal/ServerSafeHandle.cs

@@ -38,13 +38,16 @@ using System.Collections.Concurrent;
 
 namespace Google.GRPC.Core.Internal
 {
+    // TODO: we need to make sure that the delegates are not collected before invoked.
+    internal delegate void ServerShutdownCallbackDelegate(IntPtr eventPtr);
+
     /// <summary>
     /// grpc_server from grpc/grpc.h
     /// </summary>
     internal sealed class ServerSafeHandle : SafeHandleZeroIsInvalid
     {
-        [DllImport("grpc_csharp_ext.dll", EntryPoint = "grpcsharp_server_request_call_old")]
-        static extern GRPCCallError grpcsharp_server_request_call_old_CALLBACK(ServerSafeHandle server, [MarshalAs(UnmanagedType.FunctionPtr)] EventCallbackDelegate callback);
+        [DllImport("grpc_csharp_ext.dll")]
+        static extern GRPCCallError grpcsharp_server_request_call(ServerSafeHandle server, CompletionQueueSafeHandle cq, [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback);
 
         [DllImport("grpc_csharp_ext.dll")]
         static extern ServerSafeHandle grpcsharp_server_create(CompletionQueueSafeHandle cq, IntPtr args);
@@ -63,8 +66,9 @@ namespace Google.GRPC.Core.Internal
         [DllImport("grpc_csharp_ext.dll")]
         static extern void grpcsharp_server_shutdown(ServerSafeHandle server);
 
+        // TODO: get rid of the old callback style
         [DllImport("grpc_csharp_ext.dll", EntryPoint = "grpcsharp_server_shutdown_and_notify")]
-        static extern void grpcsharp_server_shutdown_and_notify_CALLBACK(ServerSafeHandle server, [MarshalAs(UnmanagedType.FunctionPtr)] EventCallbackDelegate callback);
+        static extern void grpcsharp_server_shutdown_and_notify_CALLBACK(ServerSafeHandle server, [MarshalAs(UnmanagedType.FunctionPtr)] ServerShutdownCallbackDelegate callback);
 
         [DllImport("grpc_csharp_ext.dll")]
         static extern void grpcsharp_server_destroy(IntPtr server);
@@ -95,14 +99,14 @@ namespace Google.GRPC.Core.Internal
             grpcsharp_server_shutdown(this);
         }
 
-        public void ShutdownAndNotify(EventCallbackDelegate callback)
+        public void ShutdownAndNotify(ServerShutdownCallbackDelegate callback)
         {
             grpcsharp_server_shutdown_and_notify_CALLBACK(this, callback);
         }
 
-        public GRPCCallError RequestCall(EventCallbackDelegate callback)
+        public GRPCCallError RequestCall(CompletionQueueSafeHandle cq, CompletionCallbackDelegate callback)
         {
-            return grpcsharp_server_request_call_old_CALLBACK(this, callback);
+            return grpcsharp_server_request_call(this, cq, callback);
         }
 
         protected override bool ReleaseHandle()

src/csharp/GrpcCore/Internal/ServerStreamingOutputObserver.cs

@@ -40,11 +40,11 @@ namespace Google.GRPC.Core.Internal
     /// Observer that writes all arriving messages to a call abstraction (in blocking fashion)
     /// and then halfcloses the call. Used for server-side call handling.
     /// </summary>
-    internal class ServerWritingObserver<TWrite, TRead> : IObserver<TWrite>
+    internal class ServerStreamingOutputObserver<TWrite, TRead> : IObserver<TWrite>
 	{
         readonly AsyncCall<TWrite, TRead> call;
 
-        public ServerWritingObserver(AsyncCall<TWrite, TRead> call)
+        public ServerStreamingOutputObserver(AsyncCall<TWrite, TRead> call)
 		{
             this.call = call;
 		}
@@ -52,19 +52,19 @@ namespace Google.GRPC.Core.Internal
 		public void OnCompleted()
 		{
             // TODO: how bad is the Wait here?
-            call.WriteStatusAsync(new Status(StatusCode.GRPC_STATUS_OK, "")).Wait();
+            call.SendStatusFromServerAsync(new Status(StatusCode.GRPC_STATUS_OK, "")).Wait();
 		}
 
 		public void OnError(Exception error)
 		{
-            // TODO: handle this...
+            // TODO: implement this...
 			throw new InvalidOperationException("This should never be called.");
 		}
 
 		public void OnNext(TWrite value)
 		{
             // TODO: how bad is the Wait here?
-            call.WriteAsync(value).Wait();
+            call.SendMessageAsync(value).Wait();
 		}
 	}
 }

src/csharp/GrpcCore/Server.cs

@@ -49,8 +49,8 @@ namespace Google.GRPC.Core
     {
         // TODO: make sure the delegate doesn't get garbage collected while
         // native callbacks are in the completion queue.
-        readonly EventCallbackDelegate newRpcHandler;
-        readonly EventCallbackDelegate serverShutdownHandler;
+        readonly ServerShutdownCallbackDelegate serverShutdownHandler;
+        readonly CompletionCallbackDelegate newServerRpcHandler;
 
         readonly BlockingCollection<NewRpcInfo> newRpcQueue = new BlockingCollection<NewRpcInfo>();
         readonly ServerSafeHandle handle;
@@ -61,9 +61,8 @@ namespace Google.GRPC.Core
 
         public Server()
         {
-            // TODO: what is the tag for server shutdown?
             this.handle = ServerSafeHandle.NewServer(GetCompletionQueue(), IntPtr.Zero);
-            this.newRpcHandler = HandleNewRpc;
+            this.newServerRpcHandler = HandleNewServerRpc;
             this.serverShutdownHandler = HandleServerShutdown;
         }
 
@@ -99,7 +98,7 @@ namespace Google.GRPC.Core
             {
                 var rpcInfo = newRpcQueue.Take();
 
-                Console.WriteLine("Server received RPC " + rpcInfo.Method);
+                //Console.WriteLine("Server received RPC " + rpcInfo.Method);
 
                 IServerCallHandler callHandler;
                 if (!callHandlers.TryGetValue(rpcInfo.Method, out callHandler))
@@ -138,23 +137,25 @@ namespace Google.GRPC.Core
 
         private void AllowOneRpc()
         {
-            AssertCallOk(handle.RequestCall(newRpcHandler));
+            AssertCallOk(handle.RequestCall(GetCompletionQueue(), newServerRpcHandler));
         }
 
-        private void HandleNewRpc(IntPtr eventPtr)
-        {
-            try
-            {
-                var ev = new EventSafeHandleNotOwned(eventPtr);
-                var rpcInfo = new NewRpcInfo(ev.GetCall(), ev.GetServerRpcNewMethod());
+        private void HandleNewServerRpc(GRPCOpError error, IntPtr batchContextPtr) {
+            try {
+                var ctx = new BatchContextSafeHandleNotOwned(batchContextPtr);
+
+                if (error != GRPCOpError.GRPC_OP_OK) {
+                    // TODO: handle error
+                }
+
+                var rpcInfo = new NewRpcInfo(ctx.GetServerRpcNewCall(), ctx.GetServerRpcNewMethod());
 
                 // after server shutdown, the callback returns with null call
                 if (!rpcInfo.Call.IsInvalid) {
                     newRpcQueue.Add(rpcInfo);
                 }
-            }
-            catch (Exception e)
-            {
+
+            } catch(Exception e) {
                 Console.WriteLine("Caught exception in a native handler: " + e);
             }
         }

src/csharp/GrpcCore/ServerCallHandler.cs

@@ -59,15 +59,16 @@ namespace Google.GRPC.Core
                 method.RequestMarshaller.Deserializer);
 
             asyncCall.InitializeServer(call);
-            asyncCall.Accept(cq);
+           
+            var finishedTask = asyncCall.StartServerSide();
 
-            var request = asyncCall.ReadAsync().Result;
+            var request = asyncCall.ReceiveMessageAsync().Result;
 
-            var responseObserver = new ServerWritingObserver<TResponse, TRequest>(asyncCall);
+            var responseObserver = new ServerStreamingOutputObserver<TResponse, TRequest>(asyncCall);
             handler(request, responseObserver);
 
-            asyncCall.Halfclosed.Wait();
-            asyncCall.Finished.Wait();
+            finishedTask.Wait();
+
         }
     }
 
@@ -89,16 +90,16 @@ namespace Google.GRPC.Core
                 method.RequestMarshaller.Deserializer);
 
             asyncCall.InitializeServer(call);
-            asyncCall.Accept(cq);
 
-            var responseObserver = new ServerWritingObserver<TResponse, TRequest>(asyncCall);
+            var finishedTask = asyncCall.StartServerSide();
+
+            var responseObserver = new ServerStreamingOutputObserver<TResponse, TRequest>(asyncCall);
             var requestObserver = handler(responseObserver);
 
             // feed the requests
             asyncCall.StartReadingToStream(requestObserver);
 
-            asyncCall.Halfclosed.Wait();
-            asyncCall.Finished.Wait();
+            finishedTask.Wait();
         }
     }
 
@@ -110,11 +111,14 @@ namespace Google.GRPC.Core
             AsyncCall<byte[], byte[]> asyncCall = new AsyncCall<byte[], byte[]>(
                 (payload) => payload, (payload) => payload);
 
+
             asyncCall.InitializeServer(call);
-            asyncCall.Accept(cq);
-            asyncCall.WriteStatusAsync(new Status(StatusCode.GRPC_STATUS_UNIMPLEMENTED, "No such method.")).Wait();
 
-            asyncCall.Finished.Wait();
+            var finishedTask = asyncCall.StartServerSide();
+
+            asyncCall.SendStatusFromServerAsync(new Status(StatusCode.GRPC_STATUS_UNIMPLEMENTED, "No such method.")).Wait();
+
+            finishedTask.Wait();
         }
     }
 }

src/csharp/GrpcCoreTests/ClientServerTest.cs

@@ -36,6 +36,7 @@ using NUnit.Framework;
 using Google.GRPC.Core;
 using Google.GRPC.Core.Internal;
 using System.Threading;
+using System.Diagnostics;
 using System.Threading.Tasks;
 using Google.GRPC.Core.Utils;
 
@@ -52,7 +53,7 @@ namespace Google.GRPC.Core.Tests
             Marshallers.StringMarshaller);
 
         [Test]
-        public void EmptyCall()
+        public void UnaryCall()
         {
             GrpcEnvironment.Initialize();
 
@@ -69,6 +70,7 @@ namespace Google.GRPC.Core.Tests
                 var call = new Call<string, string>(unaryEchoStringMethod, channel);
 
                 Assert.AreEqual("ABC", Calls.BlockingUnaryCall(call, "ABC", default(CancellationToken)));
+
                 Assert.AreEqual("abcdef", Calls.BlockingUnaryCall(call, "abcdef", default(CancellationToken)));
             }
 
@@ -77,11 +79,72 @@ namespace Google.GRPC.Core.Tests
             GrpcEnvironment.Shutdown();
         }
 
+        [Test]
+        public void UnaryCallPerformance()
+        {
+            GrpcEnvironment.Initialize();
+
+            Server server = new Server();
+            server.AddServiceDefinition(
+                ServerServiceDefinition.CreateBuilder("someService")
+                .AddMethod(unaryEchoStringMethod, HandleUnaryEchoString).Build());
+
+            int port = server.AddPort(host + ":0");
+            server.Start();
+
+            using (Channel channel = new Channel(host + ":" + port))
+            {
+                var call = new Call<string, string>(unaryEchoStringMethod, channel);
+
+                var stopwatch = new Stopwatch();
+                stopwatch.Start();
+                for (int i = 0; i < 1000; i++)
+                {
+                    Calls.BlockingUnaryCall(call, "ABC", default(CancellationToken));
+                }
+                stopwatch.Stop();
+                Console.WriteLine("Elapsed time: " + stopwatch.ElapsedMilliseconds + "ms");
+            }
+
+            server.ShutdownAsync().Wait();
+
+            GrpcEnvironment.Shutdown();
+        }
+
+
+        [Test]
+        public void UnknownMethodHandler()
+        {
+            GrpcEnvironment.Initialize();
+
+            Server server = new Server();
+            server.AddServiceDefinition(
+                ServerServiceDefinition.CreateBuilder("someService").Build());
+
+            int port = server.AddPort(host + ":0");
+            server.Start();
+
+            using (Channel channel = new Channel(host + ":" + port))
+            {
+                var call = new Call<string, string>(unaryEchoStringMethod, channel);
+
+                try {
+                    Calls.BlockingUnaryCall(call, "ABC", default(CancellationToken));
+                    Assert.Fail();
+                } catch(RpcException e) {
+                    Assert.AreEqual(StatusCode.GRPC_STATUS_UNIMPLEMENTED, e.Status.StatusCode);
+                }
+            }
+
+            server.ShutdownAsync().Wait();
+
+            GrpcEnvironment.Shutdown();
+        }
+
         private void HandleUnaryEchoString(string request, IObserver<string> responseObserver) {
             responseObserver.OnNext(request);
             responseObserver.OnCompleted();
         }
-
     }
 }
 

src/csharp/ext/grpc_csharp_ext.c

@@ -32,9 +32,11 @@
  */
 
 #include <grpc/support/port_platform.h>
+#include <grpc/support/alloc.h>
 #include <grpc/grpc.h>
 #include <grpc/support/log.h>
 #include <grpc/support/slice.h>
+#include <grpc/support/string.h>
 
 #include <string.h>
 
@@ -58,6 +60,134 @@ grpc_byte_buffer *string_to_byte_buffer(const char *buffer, size_t len) {
   return bb;
 }
 
+typedef void(GPR_CALLTYPE * callback_funcptr)(grpc_op_error op_error, void *batch_context);
+
+/*
+ * Helper to maintain lifetime of batch op inputs and store batch op outputs.
+ */
+typedef struct gprcsharp_batch_context {
+  grpc_metadata_array send_initial_metadata;
+  grpc_byte_buffer *send_message;
+  struct {
+    grpc_metadata_array trailing_metadata;
+    char *status_details;
+  } send_status_from_server;
+  grpc_metadata_array recv_initial_metadata;
+  grpc_byte_buffer *recv_message;
+  struct {
+    grpc_metadata_array trailing_metadata;
+    grpc_status_code status;
+    char *status_details;
+    size_t status_details_capacity;
+  } recv_status_on_client;
+  int recv_close_on_server_cancelled;
+  struct {
+    grpc_call *call;
+    grpc_call_details call_details;
+    grpc_metadata_array request_metadata;
+  } server_rpc_new;
+
+  /* callback will be called upon completion */
+  callback_funcptr callback;
+
+} grpcsharp_batch_context;
+
+grpcsharp_batch_context *grpcsharp_batch_context_create() {
+  grpcsharp_batch_context *ctx = gpr_malloc(sizeof(grpcsharp_batch_context));
+  memset(ctx, 0, sizeof(grpcsharp_batch_context));
+  return ctx;
+}
+
+/**
+ * Destroys metadata array including keys and values.
+ */
+void grpcsharp_metadata_array_destroy_recursive(grpc_metadata_array *array) {
+  if (!array->metadata) {
+    return;
+  }
+  /* TODO: destroy also keys and values */
+  grpc_metadata_array_destroy(array);
+}
+
+void grpcsharp_batch_context_destroy(grpcsharp_batch_context *ctx) {
+  if (!ctx) {
+    return;
+  }
+  grpcsharp_metadata_array_destroy_recursive(&(ctx->send_initial_metadata));
+
+  grpc_byte_buffer_destroy(ctx->send_message);
+
+  grpcsharp_metadata_array_destroy_recursive(&(ctx->send_status_from_server.trailing_metadata));
+  gpr_free(ctx->send_status_from_server.status_details);
+
+  grpc_metadata_array_destroy(&(ctx->recv_initial_metadata));
+
+  grpc_byte_buffer_destroy(ctx->recv_message);
+
+  grpc_metadata_array_destroy(&(ctx->recv_status_on_client.trailing_metadata));
+  gpr_free((void*) ctx->recv_status_on_client.status_details);
+
+  /* NOTE: ctx->server_rpc_new.call is not destroyed because callback handler is supposed
+     to take its ownership. */
+
+  grpc_call_details_destroy(&(ctx->server_rpc_new.call_details));
+  grpc_metadata_array_destroy(&(ctx->server_rpc_new.request_metadata));
+
+  gpr_free(ctx);
+}
+
+GPR_EXPORT gpr_intptr GPR_CALLTYPE grpcsharp_batch_context_recv_message_length(const grpcsharp_batch_context *ctx) {
+  if (!ctx->recv_message) {
+      return -1;
+    }
+    return grpc_byte_buffer_length(ctx->recv_message);
+}
+
+/*
+ * Copies data from recv_message to a buffer. Fatal error occurs if
+ * buffer is too small.
+ */
+GPR_EXPORT void GPR_CALLTYPE
+grpcsharp_batch_context_recv_message_to_buffer(const grpcsharp_batch_context *ctx, char *buffer,
+                                    size_t buffer_len) {
+  grpc_byte_buffer_reader *reader;
+  gpr_slice slice;
+  size_t offset = 0;
+
+  reader = grpc_byte_buffer_reader_create(ctx->recv_message);
+
+  while (grpc_byte_buffer_reader_next(reader, &slice)) {
+    size_t len = GPR_SLICE_LENGTH(slice);
+    GPR_ASSERT(offset + len <= buffer_len);
+    memcpy(buffer + offset, GPR_SLICE_START_PTR(slice),
+           GPR_SLICE_LENGTH(slice));
+    offset += len;
+    gpr_slice_unref(slice);
+  }
+  grpc_byte_buffer_reader_destroy(reader);
+}
+
+GPR_EXPORT grpc_status_code GPR_CALLTYPE
+grpcsharp_batch_context_recv_status_on_client_status(const grpcsharp_batch_context *ctx) {
+  return ctx->recv_status_on_client.status;
+}
+
+GPR_EXPORT const char *GPR_CALLTYPE
+grpcsharp_batch_context_recv_status_on_client_details(const grpcsharp_batch_context *ctx) {
+  return ctx->recv_status_on_client.status_details;
+}
+
+GPR_EXPORT grpc_call* GPR_CALLTYPE
+grpcsharp_batch_context_server_rpc_new_call(const grpcsharp_batch_context *ctx) {
+  return ctx->server_rpc_new.call;
+}
+
+GPR_EXPORT const char *GPR_CALLTYPE
+grpcsharp_batch_context_server_rpc_new_method(const grpcsharp_batch_context *ctx) {
+  return ctx->server_rpc_new.call_details.method;
+}
+
+
 /* Init & shutdown */
 
 GPR_EXPORT void GPR_CALLTYPE grpcsharp_init(void) { grpc_init(); }
@@ -96,11 +226,18 @@ grpcsharp_completion_queue_destroy(grpc_completion_queue *cq) {
 GPR_EXPORT grpc_completion_type GPR_CALLTYPE
 grpcsharp_completion_queue_next_with_callback(grpc_completion_queue *cq) {
   grpc_event *ev;
+  grpcsharp_batch_context *batch_context;
   grpc_completion_type t;
   void(GPR_CALLTYPE * callback)(grpc_event *);
 
   ev = grpc_completion_queue_next(cq, gpr_inf_future);
   t = ev->type;
+  if (t == GRPC_OP_COMPLETE && ev->tag) {
+    /* NEW API handler */
+    batch_context = (grpcsharp_batch_context *) ev->tag;
+    batch_context->callback(ev->data.op_complete, batch_context);
+    grpcsharp_batch_context_destroy(batch_context);
+  } else
   if (ev->tag) {
     /* call the callback in ev->tag */
     /* C forbids to cast object pointers to function pointers, so
@@ -128,6 +265,12 @@ GPR_EXPORT void GPR_CALLTYPE grpcsharp_channel_destroy(grpc_channel *channel) {
   grpc_channel_destroy(channel);
 }
 
+GPR_EXPORT grpc_call *GPR_CALLTYPE grpcsharp_channel_create_call(grpc_channel *channel, grpc_completion_queue *cq,
+                                  const char *method,
+                                  const char *host, gpr_timespec deadline) {
+  return grpc_channel_create_call(channel, cq, method, host, deadline);
+}
+
 GPR_EXPORT grpc_call *GPR_CALLTYPE
 grpcsharp_channel_create_call_old(grpc_channel *channel, const char *method,
                                   const char *host, gpr_timespec deadline) {
@@ -145,6 +288,12 @@ grpcsharp_event_type(const grpc_event *event) {
   return event->type;
 }
 
+GPR_EXPORT grpc_op_error GPR_CALLTYPE
+grpcsharp_event_op_complete(const grpc_event *event) {
+  GPR_ASSERT(event->type == GRPC_OP_COMPLETE);
+  return event->data.op_complete;
+}
+
 GPR_EXPORT grpc_op_error GPR_CALLTYPE
 grpcsharp_event_write_accepted(const grpc_event *event) {
   GPR_ASSERT(event->type == GRPC_WRITE_ACCEPTED);
@@ -343,3 +492,219 @@ grpcsharp_server_shutdown_and_notify(grpc_server *server, void *tag) {
 GPR_EXPORT void GPR_CALLTYPE grpcsharp_server_destroy(grpc_server *server) {
   grpc_server_destroy(server);
 }
+
+/* New API Experiments */
+
+GPR_EXPORT grpc_call_error GPR_CALLTYPE grpcsharp_call_start_unary(grpc_call *call,
+    callback_funcptr callback,
+    const char *send_buffer, size_t send_buffer_len) {
+  /* TODO: don't use magic number */
+  grpc_op ops[6];
+  grpcsharp_batch_context *ctx = grpcsharp_batch_context_create();
+  ctx->callback = callback;
+
+  /* TODO: implement sending the metadata... */
+  ops[0].op = GRPC_OP_SEND_INITIAL_METADATA;
+  /* ctx->send_initial_metadata is already zeroed out. */
+  ops[0].data.send_initial_metadata.count = 0;
+  ops[0].data.send_initial_metadata.metadata = NULL;
+
+  ops[1].op = GRPC_OP_SEND_MESSAGE;
+  ctx->send_message = string_to_byte_buffer(send_buffer, send_buffer_len);
+  ops[1].data.send_message = ctx->send_message;
+
+  ops[2].op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+
+  ops[3].op = GRPC_OP_RECV_INITIAL_METADATA;
+  ops[3].data.recv_initial_metadata = &(ctx->recv_initial_metadata);
+
+  ops[4].op = GRPC_OP_RECV_MESSAGE;
+  ops[4].data.recv_message = &(ctx->recv_message);
+
+  ops[5].op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+  ops[5].data.recv_status_on_client.trailing_metadata = &(ctx->recv_status_on_client.trailing_metadata);
+  ops[5].data.recv_status_on_client.status = &(ctx->recv_status_on_client.status);
+  /* not using preallocation for status_details */
+  ops[5].data.recv_status_on_client.status_details = &(ctx->recv_status_on_client.status_details);
+  ops[5].data.recv_status_on_client.status_details_capacity = &(ctx->recv_status_on_client.status_details_capacity);
+
+  return grpc_call_start_batch(call, ops, sizeof(ops)/sizeof(ops[0]), ctx);
+}
+
+GPR_EXPORT grpc_call_error GPR_CALLTYPE grpcsharp_call_start_client_streaming(grpc_call *call,
+    callback_funcptr callback) {
+  /* TODO: don't use magic number */
+  grpc_op ops[4];
+  grpcsharp_batch_context *ctx = grpcsharp_batch_context_create();
+  ctx->callback = callback;
+
+  /* TODO: implement sending the metadata... */
+  ops[0].op = GRPC_OP_SEND_INITIAL_METADATA;
+  /* ctx->send_initial_metadata is already zeroed out. */
+  ops[0].data.send_initial_metadata.count = 0;
+  ops[0].data.send_initial_metadata.metadata = NULL;
+
+  ops[1].op = GRPC_OP_RECV_INITIAL_METADATA;
+  ops[1].data.recv_initial_metadata = &(ctx->recv_initial_metadata);
+
+  ops[2].op = GRPC_OP_RECV_MESSAGE;
+  ops[2].data.recv_message = &(ctx->recv_message);
+
+  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[3].data.recv_status_on_client.status = &(ctx->recv_status_on_client.status);
+  /* not using preallocation for status_details */
+  ops[3].data.recv_status_on_client.status_details = &(ctx->recv_status_on_client.status_details);
+  ops[3].data.recv_status_on_client.status_details_capacity = &(ctx->recv_status_on_client.status_details_capacity);
+
+  return grpc_call_start_batch(call, ops, sizeof(ops)/sizeof(ops[0]), ctx);
+}
+
+GPR_EXPORT grpc_call_error GPR_CALLTYPE grpcsharp_call_start_server_streaming(grpc_call *call,
+    callback_funcptr callback,
+    const char *send_buffer, size_t send_buffer_len) {
+  /* TODO: don't use magic number */
+  grpc_op ops[5];
+  grpcsharp_batch_context *ctx = grpcsharp_batch_context_create();
+  ctx->callback = callback;
+
+  /* TODO: implement sending the metadata... */
+  ops[0].op = GRPC_OP_SEND_INITIAL_METADATA;
+  /* ctx->send_initial_metadata is already zeroed out. */
+  ops[0].data.send_initial_metadata.count = 0;
+  ops[0].data.send_initial_metadata.metadata = NULL;
+
+  ops[1].op = GRPC_OP_SEND_MESSAGE;
+  ctx->send_message = string_to_byte_buffer(send_buffer, send_buffer_len);
+  ops[1].data.send_message = ctx->send_message;
+
+  ops[2].op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+
+  ops[3].op = GRPC_OP_RECV_INITIAL_METADATA;
+  ops[3].data.recv_initial_metadata = &(ctx->recv_initial_metadata);
+
+  ops[4].op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+  ops[4].data.recv_status_on_client.trailing_metadata = &(ctx->recv_status_on_client.trailing_metadata);
+  ops[4].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 = &(ctx->recv_status_on_client.status_details);
+  ops[4].data.recv_status_on_client.status_details_capacity = &(ctx->recv_status_on_client.status_details_capacity);
+
+  return grpc_call_start_batch(call, ops, sizeof(ops)/sizeof(ops[0]), ctx);
+}
+
+GPR_EXPORT grpc_call_error GPR_CALLTYPE grpcsharp_call_start_duplex_streaming(grpc_call *call,
+    callback_funcptr callback) {
+  /* TODO: don't use magic number */
+  grpc_op ops[3];
+  grpcsharp_batch_context *ctx = grpcsharp_batch_context_create();
+  ctx->callback = callback;
+
+  /* TODO: implement sending the metadata... */
+  ops[0].op = GRPC_OP_SEND_INITIAL_METADATA;
+  /* ctx->send_initial_metadata is already zeroed out. */
+  ops[0].data.send_initial_metadata.count = 0;
+  ops[0].data.send_initial_metadata.metadata = NULL;
+
+  ops[1].op = GRPC_OP_RECV_INITIAL_METADATA;
+  ops[1].data.recv_initial_metadata = &(ctx->recv_initial_metadata);
+
+  ops[2].op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+  ops[2].data.recv_status_on_client.trailing_metadata = &(ctx->recv_status_on_client.trailing_metadata);
+  ops[2].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 = &(ctx->recv_status_on_client.status_details);
+  ops[2].data.recv_status_on_client.status_details_capacity = &(ctx->recv_status_on_client.status_details_capacity);
+
+  return grpc_call_start_batch(call, ops, sizeof(ops)/sizeof(ops[0]), ctx);
+}
+
+GPR_EXPORT grpc_call_error GPR_CALLTYPE grpcsharp_call_send_message(grpc_call *call,
+    callback_funcptr callback,
+    const char *send_buffer, size_t send_buffer_len) {
+  /* TODO: don't use magic number */
+  grpc_op ops[1];
+  grpcsharp_batch_context *ctx = grpcsharp_batch_context_create();
+  ctx->callback = callback;
+
+  ops[0].op = GRPC_OP_SEND_MESSAGE;
+  ctx->send_message = string_to_byte_buffer(send_buffer, send_buffer_len);
+  ops[0].data.send_message = ctx->send_message;
+
+  return grpc_call_start_batch(call, ops, sizeof(ops)/sizeof(ops[0]), ctx);
+}
+
+GPR_EXPORT grpc_call_error GPR_CALLTYPE grpcsharp_call_send_close_from_client(grpc_call *call,
+    callback_funcptr callback) {
+  /* TODO: don't use magic number */
+  grpc_op ops[1];
+  grpcsharp_batch_context *ctx = grpcsharp_batch_context_create();
+  ctx->callback = callback;
+
+  ops[0].op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+
+  return grpc_call_start_batch(call, ops, sizeof(ops)/sizeof(ops[0]), ctx);
+}
+
+GPR_EXPORT grpc_call_error GPR_CALLTYPE grpcsharp_call_send_status_from_server(grpc_call *call,
+    callback_funcptr callback, grpc_status_code status_code, const char* status_details) {
+  /* TODO: don't use magic number */
+  grpc_op ops[1];
+  grpcsharp_batch_context *ctx = grpcsharp_batch_context_create();
+  ctx->callback = callback;
+
+  ops[0].op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+  ops[0].data.send_status_from_server.status = status_code;
+  ops[0].data.send_status_from_server.status_details = gpr_strdup(status_details);
+  ops[0].data.send_status_from_server.trailing_metadata = NULL;
+  ops[0].data.send_status_from_server.trailing_metadata_count = 0;
+
+  return grpc_call_start_batch(call, ops, sizeof(ops)/sizeof(ops[0]), ctx);
+}
+
+GPR_EXPORT grpc_call_error GPR_CALLTYPE grpcsharp_call_recv_message(grpc_call *call,
+    callback_funcptr callback) {
+  /* TODO: don't use magic number */
+  grpc_op ops[1];
+  grpcsharp_batch_context *ctx = grpcsharp_batch_context_create();
+  ctx->callback = callback;
+
+  ops[0].op = GRPC_OP_RECV_MESSAGE;
+  ops[0].data.recv_message = &(ctx->recv_message);
+  return grpc_call_start_batch(call, ops, sizeof(ops)/sizeof(ops[0]), ctx);
+}
+
+GPR_EXPORT grpc_call_error GPR_CALLTYPE grpcsharp_call_start_serverside(grpc_call *call,
+    callback_funcptr callback) {
+  /* TODO: don't use magic number */
+  grpc_op ops[2];
+
+  grpcsharp_batch_context *ctx = grpcsharp_batch_context_create();
+  ctx->callback = callback;
+
+  ops[0].op = GRPC_OP_SEND_INITIAL_METADATA;
+  ops[0].data.send_initial_metadata.count = 0;
+  ops[0].data.send_initial_metadata.metadata = NULL;
+
+  ops[1].op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+  ops[1].data.recv_close_on_server.cancelled = (&ctx->recv_close_on_server_cancelled);
+
+  return grpc_call_start_batch(call, ops, sizeof(ops)/sizeof(ops[0]), ctx);
+}
+
+
+GPR_EXPORT grpc_call_error GPR_CALLTYPE grpcsharp_server_request_call(grpc_server *server,
+    grpc_completion_queue *cq, callback_funcptr callback) {
+
+  grpcsharp_batch_context *ctx = grpcsharp_batch_context_create();
+  ctx->callback = callback;
+
+  return grpc_server_request_call(server, &(ctx->server_rpc_new.call),
+                                  &(ctx->server_rpc_new.call_details),
+                                  &(ctx->server_rpc_new.request_metadata),
+                                  cq, ctx);
+}
+
+
+
+