Implement connectivity state property and observers

Sources/CgRPC/shim/cgrpc.h

@@ -147,6 +147,11 @@ cgrpc_completion_queue *cgrpc_channel_completion_queue(cgrpc_channel *channel);
 
 grpc_connectivity_state cgrpc_channel_check_connectivity_state(
     cgrpc_channel *channel, int try_to_connect);
+void cgrpc_channel_watch_connectivity_state(cgrpc_channel *channel,
+                                            cgrpc_completion_queue *completion_queue,
+                                            grpc_connectivity_state last_observed_state,
+                                            double deadline,
+                                            void *tag);
 
 // server support
 cgrpc_server *cgrpc_server_create(const char *address);
@@ -159,6 +164,7 @@ void cgrpc_server_start(cgrpc_server *s);
 cgrpc_completion_queue *cgrpc_server_get_completion_queue(cgrpc_server *s);
 
 // completion queues
+cgrpc_completion_queue *cgrpc_completion_queue_create_for_next();
 grpc_event cgrpc_completion_queue_get_next_event(cgrpc_completion_queue *cq,
                                                  double timeout);
 void cgrpc_completion_queue_drain(cgrpc_completion_queue *cq);

Sources/CgRPC/shim/channel.c

@@ -103,3 +103,8 @@ cgrpc_completion_queue *cgrpc_channel_completion_queue(cgrpc_channel *channel) {
 grpc_connectivity_state cgrpc_channel_check_connectivity_state(cgrpc_channel *channel, int try_to_connect) {
   return grpc_channel_check_connectivity_state(channel->channel, try_to_connect);
 }
+
+void cgrpc_channel_watch_connectivity_state(cgrpc_channel *channel, cgrpc_completion_queue *completion_queue, grpc_connectivity_state last_observed_state, double deadline, void *tag) {
+  gpr_timespec deadline_seconds = cgrpc_deadline_in_seconds_from_now(deadline);
+  return grpc_channel_watch_connectivity_state(channel->channel, last_observed_state, deadline_seconds, completion_queue, tag);
+}

Sources/CgRPC/shim/completion_queue.c

@@ -18,6 +18,10 @@
 
 #include <stdio.h>
 
+grpc_completion_queue *cgrpc_completion_queue_create_for_next() {
+  return grpc_completion_queue_create_for_next(NULL);
+}
+
 grpc_event cgrpc_completion_queue_get_next_event(grpc_completion_queue *cq, double timeout) {
   gpr_timespec deadline = cgrpc_deadline_in_seconds_from_now(timeout);
   if (timeout < 0) {

Sources/SwiftGRPC/Core/Channel.swift

@@ -15,6 +15,7 @@
  */
 #if SWIFT_PACKAGE
   import CgRPC
+  import Dispatch
 #endif
 import Foundation
 
@@ -31,10 +32,9 @@ public class Channel {
 
   /// Default host to use for new calls
   public var host: String
-
-  public var connectivityState: ConnectivityState? {
-    return ConnectivityState.fromCEnum(cgrpc_channel_check_connectivity_state(underlyingChannel, 0))
-  }
+
+  /// Connectivity state observers
+  private var connectivityObservers: [ConnectivityObserver] = []
 
   /// Initializes a gRPC channel
   ///
@@ -47,8 +47,7 @@ public class Channel {
     } else {
       underlyingChannel = cgrpc_channel_create(address)
     }
-    completionQueue = CompletionQueue(
-      underlyingCompletionQueue: cgrpc_channel_completion_queue(underlyingChannel), name: "Client")
+    completionQueue = CompletionQueue(underlyingCompletionQueue: cgrpc_channel_completion_queue(underlyingChannel), name: "Client")
     completionQueue.run() // start a loop that watches the channel's completion queue
   }
 
@@ -59,13 +58,13 @@ public class Channel {
   /// - Parameter host: an optional hostname override
   public init(address: String, certificates: String, host: String?) {
     self.host = address
-    underlyingChannel = cgrpc_channel_create_secure(address, certificates, host)
-    completionQueue = CompletionQueue(
-      underlyingCompletionQueue: cgrpc_channel_completion_queue(underlyingChannel), name: "Client")
+    underlyingChannel = cgrpc_channel_create_secure(address, certificates, &argumentValues, Int32(arguments.count))
+    completionQueue = CompletionQueue(underlyingCompletionQueue: cgrpc_channel_completion_queue(underlyingChannel), name: "Client")
     completionQueue.run() // start a loop that watches the channel's completion queue
   }
 
   deinit {
+    connectivityObservers.forEach { $0.polling = false }
     cgrpc_channel_destroy(underlyingChannel)
     completionQueue.shutdown()
   }
@@ -81,4 +80,139 @@ public class Channel {
     let underlyingCall = cgrpc_channel_create_call(underlyingChannel, method, host, timeout)!
     return Call(underlyingCall: underlyingCall, owned: true, completionQueue: completionQueue)
   }
+
+  public func connectivityState(tryToConnect: Bool = false) -> ConnectivityState {
+    return ConnectivityState.connectivityState(cgrpc_channel_check_connectivity_state(underlyingChannel, tryToConnect ? 1 : 0))
+  }
+
+  public func subscribe(callback: @escaping (ConnectivityState) -> ()) {
+    let observer = ConnectivityObserver(underlyingChannel: underlyingChannel, callback: callback)
+    observer.polling = true
+    connectivityObservers.append(observer)
+  }
+}
+
+private extension Channel {
+  class ConnectivityObserver {
+    private let completionQueue: CompletionQueue
+    private let underlyingChannel: UnsafeMutableRawPointer
+    private let underlyingCompletionQueue: UnsafeMutableRawPointer
+    private let callback: (ConnectivityState) -> ()
+    private var lastState: ConnectivityState
+    private let queue: OperationQueue
+
+    var polling: Bool = false {
+      didSet {
+        queue.addOperation { [weak self] in
+          guard let `self` = self else { return }
+
+          if self.polling == true && oldValue == false {
+            self.run()
+          } else if self.polling == false && oldValue == true {
+            self.shutdown()
+          }
+        }
+      }
+    }
+
+    init(underlyingChannel: UnsafeMutableRawPointer, callback: @escaping (ConnectivityState) -> ()) {
+      self.underlyingChannel = underlyingChannel
+      self.underlyingCompletionQueue = cgrpc_completion_queue_create_for_next()
+      self.completionQueue = CompletionQueue(underlyingCompletionQueue: self.underlyingCompletionQueue, name: "Connectivity State")
+      self.callback = callback
+      self.lastState = ConnectivityState.connectivityState(cgrpc_channel_check_connectivity_state(self.underlyingChannel, 0))
+
+      queue = OperationQueue()
+      queue.maxConcurrentOperationCount = 1
+      queue.qualityOfService = .background
+    }
+
+    deinit {
+      shutdown()
+    }
+
+    private func run() {
+      DispatchQueue.global().async { [weak self] in
+        guard let `self` = self else { return }
+
+        while self.polling {
+          guard let underlyingState = self.lastState.underlyingState else { return }
+
+          let deadline: TimeInterval = 0.2
+          cgrpc_channel_watch_connectivity_state(self.underlyingChannel, self.underlyingCompletionQueue, underlyingState, deadline, nil)
+          let event = self.completionQueue.wait(timeout: deadline)
+
+          if event.success == 1 {
+            let newState = ConnectivityState.connectivityState(cgrpc_channel_check_connectivity_state(self.underlyingChannel, 1))
+
+            guard newState != self.lastState else { continue }
+            defer { self.lastState = newState }
+
+            self.callback(newState)
+          }
+        }
+      }
+    }
+
+    private func shutdown() {
+      completionQueue.shutdown()
+    }
+  }
+}
+
+extension Channel {
+  public enum ConnectivityState {
+    /// Channel has just been initialized
+    case initialized
+    /// Channel is idle
+    case idle
+    /// Channel is connecting
+    case connecting
+    /// Channel is ready for work
+    case ready
+    /// Channel has seen a failure but expects to recover
+    case transientFailure
+    /// Channel has seen a failure that it cannot recover from
+    case shutdown
+    /// Channel connectivity state is unknown
+    case unknown
+
+    fileprivate static func connectivityState(_ value: grpc_connectivity_state) -> ConnectivityState {
+      switch value {
+      case GRPC_CHANNEL_INIT:
+        return .initialized
+      case GRPC_CHANNEL_IDLE:
+        return .idle
+      case GRPC_CHANNEL_CONNECTING:
+        return .connecting
+      case GRPC_CHANNEL_READY:
+        return .ready
+      case GRPC_CHANNEL_TRANSIENT_FAILURE:
+        return .transientFailure
+      case GRPC_CHANNEL_SHUTDOWN:
+        return .shutdown
+      default:
+        return .unknown
+      }
+    }
+
+    fileprivate var underlyingState: grpc_connectivity_state? {
+      switch self {
+      case .initialized:
+        return GRPC_CHANNEL_INIT
+      case .idle:
+        return GRPC_CHANNEL_IDLE
+      case .connecting:
+        return GRPC_CHANNEL_CONNECTING
+      case .ready:
+        return GRPC_CHANNEL_READY
+      case .transientFailure:
+        return GRPC_CHANNEL_TRANSIENT_FAILURE
+      case .shutdown:
+        return GRPC_CHANNEL_SHUTDOWN
+      default:
+        return nil
+      }
+    }
+  }
 }