Add a connection pool delegate

Sources/GRPC/ConnectionManager+Delegates.swift

@@ -35,6 +35,12 @@ internal protocol ConnectionManagerConnectivityDelegate {
 }
 
 internal protocol ConnectionManagerHTTP2Delegate {
+  /// An HTTP/2 stream was opened.
+  ///
+  /// - Parameters:
+  ///   - connectionManager: The connection manager reporting the opened stream.
+  func streamOpened(_ connectionManager: ConnectionManager)
+
   /// An HTTP/2 stream was closed.
   ///
   /// - Parameters:

Sources/GRPC/ConnectionManager.swift

@@ -242,6 +242,17 @@ internal final class ConnectionManager {
     }
   }
 
+  /// Returns whether the state is 'shutdown'.
+  private var isShutdown: Bool {
+    self.eventLoop.assertInEventLoop()
+    switch self.state {
+    case .shutdown:
+      return true
+    case .idle, .connecting, .transientFailure, .active, .ready:
+      return false
+    }
+  }
+
   /// Returns the `HTTP2StreamMultiplexer` from the 'ready' state or `nil` if it is not available.
   private var multiplexer: HTTP2StreamMultiplexer? {
     self.eventLoop.assertInEventLoop()
@@ -254,6 +265,17 @@ internal final class ConnectionManager {
     }
   }
 
+  private var channel: Channel? {
+    self.eventLoop.assertInEventLoop()
+    switch self.state {
+    case let .ready(state):
+      return state.channel
+
+    case .idle, .connecting, .transientFailure, .active, .shutdown:
+      return nil
+    }
+  }
+
   /// The `EventLoop` that the managed connection will run on.
   internal let eventLoop: EventLoop
 
@@ -807,6 +829,11 @@ internal final class ConnectionManager {
     }
   }
 
+  internal func streamOpened() {
+    self.eventLoop.assertInEventLoop()
+    self.http2Delegate?.streamOpened(self)
+  }
+
   internal func streamClosed() {
     self.eventLoop.assertInEventLoop()
     self.http2Delegate?.streamClosed(self)
@@ -1001,12 +1028,23 @@ extension ConnectionManager {
       return self.manager.isIdle
     }
 
+    /// Returne `true` if the connection is in the shutdown state.
+    internal var isShutdown: Bool {
+      return self.manager.isShutdown
+    }
+
     /// Returns the `multiplexer` from a connection in the `ready` state or `nil` if it is any
     /// other state.
     internal var multiplexer: HTTP2StreamMultiplexer? {
       return self.manager.multiplexer
     }
 
+    /// Returns the `channel` from a connection in the `ready` state or `nil` if it is any
+    /// other state.
+    internal var channel: Channel? {
+      return self.manager.channel
+    }
+
     // Start establishing a connection. Must only be called when `isIdle` is `true`.
     internal func startConnecting() {
       self.manager.startConnecting()

Sources/GRPC/ConnectionPool/ConnectionManagerID.swift

@@ -15,7 +15,7 @@
  */
 
 @usableFromInline
-internal struct ConnectionManagerID: Hashable, CustomStringConvertible {
+internal struct ConnectionManagerID: Hashable, CustomStringConvertible, GRPCSendable {
   @usableFromInline
   internal let _id: ObjectIdentifier
 

Sources/GRPC/ConnectionPool/ConnectionPool+PerConnectionState.swift

@@ -26,8 +26,32 @@ extension ConnectionPool {
     @usableFromInline
     internal var _availability: StreamAvailability?
 
+    @usableFromInline
+    internal var isQuiescing: Bool {
+      get {
+        return self._availability?.isQuiescing ?? false
+      }
+      set {
+        self._availability?.isQuiescing = true
+      }
+    }
+
     @usableFromInline
     internal struct StreamAvailability {
+      @usableFromInline
+      struct Utilization {
+        @usableFromInline
+        var used: Int
+        @usableFromInline
+        var capacity: Int
+
+        @usableFromInline
+        init(used: Int, capacity: Int) {
+          self.used = used
+          self.capacity = capacity
+        }
+      }
+
       @usableFromInline
       var multiplexer: HTTP2StreamMultiplexer
       /// Maximum number of available streams.
@@ -36,24 +60,39 @@ extension ConnectionPool {
       /// Number of streams reserved.
       @usableFromInline
       var reserved: Int = 0
+      /// Number of streams opened.
+      @usableFromInline
+      var open: Int = 0
+      @usableFromInline
+      var isQuiescing = false
       /// Number of available streams.
       @usableFromInline
       var available: Int {
-        return self.maxAvailable - self.reserved
+        return self.isQuiescing ? 0 : self.maxAvailable - self.reserved
       }
 
       /// Increment the reserved streams and return the multiplexer.
       @usableFromInline
       mutating func reserve() -> HTTP2StreamMultiplexer {
+        assert(!self.isQuiescing)
         self.reserved += 1
         return self.multiplexer
       }
 
+      @usableFromInline
+      mutating func opened() -> Utilization {
+        self.open += 1
+        return .init(used: self.open, capacity: self.maxAvailable)
+      }
+
       /// Decrement the reserved streams by one.
       @usableFromInline
-      mutating func `return`() {
+      mutating func `return`() -> Utilization {
         self.reserved -= 1
+        self.open -= 1
         assert(self.reserved >= 0)
+        assert(self.open >= 0)
+        return .init(used: self.open, capacity: self.maxAvailable)
       }
     }
 
@@ -92,10 +131,15 @@ extension ConnectionPool {
       return self._availability?.reserve()
     }
 
+    @usableFromInline
+    internal mutating func openedStream() -> PerConnectionState.StreamAvailability.Utilization? {
+      return self._availability?.opened()
+    }
+
     /// Return a reserved stream to the connection.
     @usableFromInline
-    internal mutating func returnStream() {
-      self._availability?.return()
+    internal mutating func returnStream() -> PerConnectionState.StreamAvailability.Utilization? {
+      return self._availability?.return()
     }
 
     /// Update the maximum concurrent streams available on the connection, marking it as available