Connect EmbeddedChannel when FuzzTesting (grpc#1569)

Motivation:

FuzzTesting uses EmbeddedChannel but never called connect meaning that
channel active was never fired. HTTP/2 recently added activity state
checking and complains that we fire inactive without having activated.

Modifications:

- call connect on the EmbeddedChannel in fuzzing and in tests
- if the connection manager is closed while connecting it waits for the
  connect to resolve before closing the channel, however the channel
  future is completed before channel active is fired so defer closing it
  until the next loop tick.

Result:

Channel active is more frequently fired at the right time.

Author: glbrntt

FuzzTesting/Sources/ServerFuzzerLib/ServerFuzzer.swift

@@ -22,6 +22,8 @@ public func test(_ start: UnsafeRawPointer, _ count: Int) -> CInt {
   let bytes = UnsafeRawBufferPointer(start: start, count: count)
 
   let channel = EmbeddedChannel()
+  try! channel.connect(to: try! SocketAddress(ipAddress: "127.0.0.1", port: 0)).wait()
+
   defer {
     _ = try? channel.finish()
   }

Sources/GRPC/ConnectionManager.swift

@@ -514,9 +514,14 @@ internal final class ConnectionManager {
       state.candidate.whenComplete {
         switch $0 {
         case let .success(channel):
-          // In case we do successfully connect, close immediately.
-          channel.close(mode: .all, promise: nil)
-          promise.completeWith(channel.closeFuture.recoveringFromUncleanShutdown())
+          // In case we do successfully connect, close on the next loop tick. When connecting a
+          // channel NIO will complete the promise for the channel before firing channel active.
+          // That means we may close and fire inactive before active which HTTP/2 will be unhappy
+          // about.
+          self.eventLoop.execute {
+            channel.close(mode: .all, promise: nil)
+            promise.completeWith(channel.closeFuture.recoveringFromUncleanShutdown())
+          }
 
         case .failure:
           // We failed to connect, that's fine we still shutdown successfully.

Tests/GRPCTests/ServerFuzzingRegressionTests.swift

@@ -31,6 +31,7 @@ final class ServerFuzzingRegressionTests: GRPCTestCase {
 
   private func runTest(withInput buffer: ByteBuffer) {
     let channel = EmbeddedChannel()
+    try! channel.connect(to: try! SocketAddress(ipAddress: "127.0.0.1", port: 0)).wait()
     defer {
       _ = try? channel.finish()
     }