Expose ResourceWatch providing error and event callbacks

Author: iabudiab

Sources/SwiftkubeClient/Client/ClusterScopedGenericKubernetesClient.swift

@@ -55,8 +55,36 @@ public extension ClusterScopedGenericKubernetesClient where Resource: ReadableRe
 	/// event paired with the corresponding resource as a pair to the `eventHandler`.
 	///
 	/// - Returns: A cancellable `HTTPClient.Task` instance, representing a streaming connetion to the API server.
-	func watch(options: [ListOption]? = nil, eventHandler: @escaping ResourceWatch<Resource>.EventHandler) throws -> HTTPClient.Task<Void> {
-		try super.watch(in: .allNamespaces, options: options, using: ResourceWatch<Resource>(eventHandler))
+	func watch(
+		options: [ListOption]? = nil,
+		eventHandler: @escaping ResourceWatch<Resource>.EventHandler
+	) throws -> HTTPClient.Task<Void> {
+		let resourceWatch = ResourceWatch<Resource>(onError: nil, onEvent: eventHandler)
+		return try watch(options: options, resourceWatch: resourceWatch)
+	}
+
+	/// Watches cluster-scoped resources.
+	///
+	/// Watching resources opens a persistent connection to the API server. The connection is represented by a `HTTPClient.Task` instance, that acts
+	/// as an active "subscription" to the events stream. The task can be cancelled any time to stop the watch.
+	///
+	/// ```swift
+	/// let task: HTTPClient.Task<Void> = client.namespaces.watch() { (event, namespace) in
+	///    print("\(event): \(namespace)")
+	///	}
+	///
+	///	task.cancel()
+	/// ```
+	///
+	/// - Parameter eventHandler: A `ResourceWatch` instance, which is used as a callback for new events. The clients sends each
+	/// event paired with the corresponding resource as a pair to the `eventHandler`.   Errors are sent to the `errorHandler`.
+	///
+	/// - Returns: A cancellable `HTTPClient.Task` instance, representing a streaming connetion to the API server.
+	func watch(
+		options: [ListOption]? = nil,
+		resourceWatch: ResourceWatch<Resource>
+	) throws -> HTTPClient.Task<Void> {
+		try super.watch(in: .allNamespaces, options: options, using: resourceWatch)
 	}
 }
 

Sources/SwiftkubeClient/Client/GenericKubernetesClient.swift

@@ -378,7 +378,7 @@ internal extension GenericKubernetesClient {
 	/// - Returns: A cancellable `HTTPClient.Task` instance, representing a streaming connetion to the API server.
 	func watch(in namespace: NamespaceSelector, options: [ListOption]? = nil, using watch: ResourceWatch<Resource>) throws -> HTTPClient.Task<Void> {
 		let request = try makeRequest().toWatch().in(namespace).with(options: options).build()
-		let delegate = WatchDelegate(watch: watch, logger: logger)
+		let delegate = WatchDelegate(watcher: watch, logger: logger)
 
 		return httpClient.execute(request: request, delegate: delegate, logger: logger)
 	}
@@ -407,7 +407,7 @@ internal extension GenericKubernetesClient {
 	/// - Returns: A cancellable `HTTPClient.Task` instance, representing a streaming connetion to the API server.
 	func follow(in namespace: NamespaceSelector, name: String, container: String?, using watch: LogWatch) throws -> HTTPClient.Task<Void> {
 		let request = try makeRequest().toFollow(pod: name, container: container).in(namespace).build()
-		let delegate = WatchDelegate(watch: watch, logger: logger)
+		let delegate = WatchDelegate(watcher: watch, logger: logger)
 
 		return httpClient.execute(request: request, delegate: delegate, logger: logger)
 	}

Sources/SwiftkubeClient/Client/NamespacedGenericKubernetesClient+Pod.swift

@@ -20,8 +20,13 @@ import SwiftkubeModel
 
 public extension NamespacedGenericKubernetesClient where Resource == core.v1.Pod {
 
-	func follow(in namespace: NamespaceSelector? = nil, name: String, container: String?, lineHandler: @escaping LogWatch.LineHandler) throws -> HTTPClient.Task<Void> {
-		try super.follow(in: namespace ?? .namespace(config.namespace), name: name, container: container, using: LogWatch(logger: logger, lineHandler))
+	func follow(
+		in namespace: NamespaceSelector? = nil,
+		name: String, container: String?,
+		lineHandler: @escaping LogWatch.LineHandler
+	) throws -> HTTPClient.Task<Void> {
+		let logWatch = LogWatch(onError: nil, onNext: lineHandler)
+		return try super.follow(in: namespace ?? .namespace(config.namespace), name: name, container: container, using: logWatch)
 	}
 
 	func status(in namespace: NamespaceSelector? = nil, name: String) throws -> EventLoopFuture<core.v1.Pod> {

Sources/SwiftkubeClient/Client/NamespacedGenericKubernetesClient.swift

@@ -63,8 +63,42 @@ public extension NamespacedGenericKubernetesClient where Resource: ReadableResou
 	/// event paired with the corresponding resource as a pair to the `eventHandler`.
 	///
 	/// - Returns: A cancellable `HTTPClient.Task` instance, representing a streaming connetion to the API server.
-	func watch(in namespace: NamespaceSelector? = nil, options: [ListOption]? = nil, eventHandler: @escaping ResourceWatch<Resource>.EventHandler) throws -> HTTPClient.Task<Void> {
-		try super.watch(in: namespace ?? .namespace(config.namespace), options: options, using: ResourceWatch<Resource>(logger: logger, eventHandler))
+	func watch(
+		in namespace: NamespaceSelector? = nil,
+		options: [ListOption]? = nil,
+		eventHandler: @escaping ResourceWatch<Resource>.EventHandler
+	) throws -> HTTPClient.Task<Void> {
+		let resourceWatch = ResourceWatch<Resource>(onError: nil, onEvent: eventHandler)
+		return try watch(in: namespace, options: options, resourceWatch: resourceWatch)
+	}
+
+	/// Watches the API resources in the given namespace.
+	///
+	/// Watching resources opens a persistent connection to the API server. The connection is represented by a `HTTPClient.Task` instance, that acts
+	/// as an active "subscription" to the events stream. The task can be cancelled any time to stop the watch.
+	///
+	/// If the namespace is not specified then the default namespace defined in the `KubernetesClientConfig` will be used instead.
+	///
+	/// ```swift
+	/// let watch = ResourceWatch<core.v1.Pod>(logger: logger, onError: errorHandler) { (event, pod) in
+	///    print("\(event): \(pod)")
+	///	}
+	/// let task: HTTPClient.Task<Void> = client.pods.watch(in: .namespace("default"), resourceWatch: watch)
+	///	task.cancel()
+	/// ```
+	///
+	/// - Parameters:
+	///   - namespace: The namespace for this API request.
+	///   - resourceWatch: A `ResourceWatch` instance, which is used for error and event callbacks. The clients sends each
+	/// event paired with the corresponding resource as a pair to the `eventHandler`. Errors are sent to the `errorHandler`.
+	///
+	/// - Returns: A cancellable `HTTPClient.Task` instance, representing a streaming connetion to the API server.
+	func watch(
+		in namespace: NamespaceSelector? = nil,
+		options: [ListOption]? = nil,
+		resourceWatch: ResourceWatch<Resource>
+	) throws -> HTTPClient.Task<Void> {
+		try super.watch(in: namespace ?? .namespace(config.namespace), options: options, using: resourceWatch)
 	}
 }
 

Sources/SwiftkubeClient/SwiftkubeClient.swift

@@ -30,6 +30,8 @@ public enum SwiftkubeClientError: Error {
 	case decodingError(String)
 	/// Thrown on all  errors returned from the Kubernetes API server.
 	case requestError(meta.v1.Status)
+	/// Thrown when the underlying HTTPClient reports an error.
+	case clientError(Error)
 
 	internal static func methodNotAllowed(_ method: HTTPMethod) -> SwiftkubeClientError {
 		let status = sk.status {

Sources/SwiftkubeClient/Watch/ResourceWatch.swift

@@ -32,28 +32,40 @@ public enum EventType: String, RawRepresentable {
 
 // MARK: - Watcher
 
-protocol Watcher {
-	func handle(payload: Data)
+public protocol Watcher {
+	typealias ErrorHandler = (SwiftkubeClientError) -> Void
+
+	func onError(error: SwiftkubeClientError)
+	func onNext(payload: Data)
 }
 
 // MARK: - ResourceWatch
 
-final public class ResourceWatch<Resource: KubernetesAPIResource>: Watcher {
+open class ResourceWatch<Resource: KubernetesAPIResource>: Watcher {
 
 	public typealias EventHandler = (EventType, Resource) -> Void
 
-	private let decoder = JSONDecoder()
-	private let handler: EventHandler
-	private let logger: Logger
+	private let decoder: JSONDecoder
+	private let errorHandler: ErrorHandler?
+	private let eventHandler: EventHandler
+
+	public init(
+		decoder: JSONDecoder = JSONDecoder(),
+		onError errorHandler: ErrorHandler? = nil,
+		onEvent eventHandler: @escaping EventHandler
+	) {
+		self.decoder = decoder
+		self.errorHandler = errorHandler
+		self.eventHandler = eventHandler
+	}
 
-	init(logger: Logger? = nil, _ handler: @escaping EventHandler) {
-		self.handler = handler
-		self.logger = logger ?? KubernetesClient.loggingDisabled
+	public func onError(error: SwiftkubeClientError) {
+		errorHandler?(error)
 	}
 
-	internal func handle(payload: Data) {
+	public func onNext(payload: Data) {
 		guard let string = String(data: payload, encoding: .utf8) else {
-			logger.warning("Could not deserialize payload")
+			errorHandler?(.decodingError("Could not deserialize payload"))
 			return
 		}
 
@@ -62,45 +74,52 @@ final public class ResourceWatch<Resource: KubernetesAPIResource>: Watcher {
 				let data = line.data(using: .utf8),
 				let event = try? self.decoder.decode(meta.v1.WatchEvent.self, from: data)
 			else {
-				self.logger.warning("Error decoding meta.v1.WatchEvent payload")
+				self.errorHandler?(.decodingError("Error decoding meta.v1.WatchEvent payload"))
 				return
 			}
 
 			guard let eventType = EventType(rawValue: event.type) else {
-				self.logger.warning("Error parsing EventType")
+				self.errorHandler?(.decodingError("Error parsing EventType"))
 				return
 			}
 
 			guard
 				let jsonData = try? JSONSerialization.data(withJSONObject: event.object),
 				let resource = try? self.decoder.decode(Resource.self, from: jsonData)
 			else {
-				self.logger.warning("Error deserializing \(String(describing: Resource.self))")
+				self.errorHandler?(.decodingError("Error deserializing \(String(describing: Resource.self))"))
 				return
 			}
 
-			self.handler(eventType, resource)
+			self.eventHandler(eventType, resource)
 		}
 	}
 }
 
 // MARK: - LogWatch
 
-final public class LogWatch: Watcher {
+open class LogWatch: Watcher {
 
 	public typealias LineHandler = (String) -> Void
 
-	private let logger: Logger
+	private let errorHandler: ErrorHandler?
 	private let lineHandler: LineHandler
 
-	public init(logger: Logger? = nil, _ lineHandler: @escaping LineHandler = { line in print(line) }) {
-		self.logger = logger ?? KubernetesClient.loggingDisabled
+	public init(
+		onError errorHandler: ErrorHandler? = nil,
+		onNext lineHandler: @escaping LineHandler
+	) {
+		self.errorHandler = errorHandler
 		self.lineHandler = lineHandler
 	}
 
-	internal func handle(payload: Data) {
+	public func onError(error: SwiftkubeClientError) {
+		errorHandler?(error)
+	}
+
+	public func onNext(payload: Data) {
 		guard let string = String(data: payload, encoding: .utf8) else {
-			logger.warning("Could not deserialize payload")
+			errorHandler?(.decodingError("Could not deserialize payload"))
 			return
 		}
 
@@ -116,35 +135,30 @@ internal class WatchDelegate: HTTPClientResponseDelegate {
 
 	typealias Response = Void
 
-	private let watch: Watcher
+	private let watcher: Watcher
 	private let logger: Logger
 
-	init(watch: Watcher, logger: Logger) {
-		self.watch = watch
+	init(watcher: Watcher, logger: Logger) {
+		self.watcher = watcher
 		self.logger = logger
 	}
 
-	func didSendRequestHead(task: HTTPClient.Task<Response>, _ head: HTTPRequestHead) {
-		logger.debug("Did send request head: \(head.headers)")
-	}
-
-	func didSendRequestPart(task: HTTPClient.Task<Response>, _ part: IOData) {
-		logger.debug("Did send request part: \(part)")
-	}
-
-	func didSendRequest(task: HTTPClient.Task<Response>) {
-		logger.debug("Did send request: \(task)")
-	}
-
 	func didReceiveHead(task: HTTPClient.Task<Response>, _ head: HTTPResponseHead) -> EventLoopFuture<Void> {
 		logger.debug("Did receive response head: \(head.headers)")
+		switch head.status.code {
+		case HTTPResponseStatus.badRequest.code:
+			watcher.onError(error: .badRequest(head.status.reasonPhrase))
+		default:
+			watcher.onError(error: .emptyResponse)
+		}
+
 		return task.eventLoop.makeSucceededFuture(())
 	}
 
 	func didReceiveBodyPart(task: HTTPClient.Task<Response>, _ buffer: ByteBuffer) -> EventLoopFuture<Void> {
 		logger.debug("Did receive body part: \(task)")
 		let payload = Data(buffer: buffer)
-		watch.handle(payload: payload)
+		watcher.onNext(payload: payload)
 		return task.eventLoop.makeSucceededFuture(())
 	}
 
@@ -154,6 +168,7 @@ internal class WatchDelegate: HTTPClientResponseDelegate {
 	}
 
 	func didReceiveError(task: HTTPClient.Task<Response>, _ error: Error) {
-		logger.warning("Did receive error: \(error.localizedDescription)")
+		logger.debug("Did receive error: \(error.localizedDescription)")
+		watcher.onError(error: .clientError(error))
 	}
 }