Reduce cost of responding to unary calls. (#932)

Motivation:

When looking at the unary response path I noticed that there was quite a
lot of flushing in the callback chain for sending unary responses. In
particular, as this callback chain performed only synchronous operations
(`.map` and `.recover`) there was no need for multiple flushes: the
final `.whenSuccess` covered all code paths and had an unconditional
`flush` in it, so that flush would cover the other operations.

However, on further inspection I realised that I could refactor the
entire code path to use a single Future callback. This ends up being a
substantial win. Unnecessary future callbacks put pressure on the
allocator, as they tend to have to allocate closure contexts (if the
closure captures, as all of these did) as well as additional Future
objects for return values. Reducing unnecessary chaining can therefore
be a tidy win when it happens in hot paths.

Modifications:

1. Replace a `writeAndFlush` with a `write(..., promise: nil)`, removing
   both an unnecessary flush and a future allocation.
2. Condense a long callback chain into a single function using
   `.whenComplete`.

Results:

The previous code allocated 4 closure contexts and 4 futures, and
flushed twice. The new code allocates 1 closure context, zero futures,
and flushes once. A tidy little profit: 4.7% improvement in unary call
microbenchmarks.

Sources/GRPC/ServerCallContexts/UnaryResponseCallContext.swift

@@ -69,29 +69,25 @@ open class UnaryResponseCallContextImpl<ResponsePayload>: UnaryResponseCallConte
     super.init(eventLoop: channel.eventLoop, request: request, logger: logger)
 
     responsePromise.futureResult
-      // Send the response provided to the promise.
-      .map { responseMessage -> EventLoopFuture<Void> in
-        return self.channel.writeAndFlush(NIOAny(WrappedResponse.message(.init(responseMessage, compressed: self.compressionEnabled))))
-      }
-      .map { _ in
-        GRPCStatusAndMetadata(status: self.responseStatus, metadata: nil)
-      }
-      // Ensure that any error provided can be transformed to `GRPCStatus`, using "internal server error" as a fallback.
-      .recover { [weak errorDelegate] error in
-        errorDelegate?.observeRequestHandlerError(error, request: request)
-        
-        if let transformed: GRPCStatusAndMetadata = errorDelegate?.transformRequestHandlerError(error, request: request) {
-          return transformed
-        }
-        
-        if let grpcStatusTransformable = error as? GRPCStatusTransformable {
-          return GRPCStatusAndMetadata(status: grpcStatusTransformable.makeGRPCStatus(), metadata: nil)
+      .whenComplete { [self, weak errorDelegate] result in
+        let statusAndMetadata: GRPCStatusAndMetadata
+
+        switch result {
+        case .success(let responseMessage):
+          self.channel.write(NIOAny(WrappedResponse.message(.init(responseMessage, compressed: self.compressionEnabled))), promise: nil)
+          statusAndMetadata = GRPCStatusAndMetadata(status: self.responseStatus, metadata: nil)
+        case .failure(let error):
+          errorDelegate?.observeRequestHandlerError(error, request: request)
+
+          if let transformed: GRPCStatusAndMetadata = errorDelegate?.transformRequestHandlerError(error, request: request) {
+            statusAndMetadata = transformed
+          } else if let grpcStatusTransformable = error as? GRPCStatusTransformable {
+            statusAndMetadata = GRPCStatusAndMetadata(status: grpcStatusTransformable.makeGRPCStatus(), metadata: nil)
+          } else {
+            statusAndMetadata = GRPCStatusAndMetadata(status: .processingError, metadata: nil)
+          }
         }
-        
-        return GRPCStatusAndMetadata(status: .processingError, metadata: nil)
-      }
-      // Finish the call by returning the final status.
-      .whenSuccess { statusAndMetadata in
+
         if let metadata = statusAndMetadata.metadata {
           self.trailingMetadata.add(contentsOf: metadata)
         }