ETCD Watch Function (#7)

* Create watch message in etcd.proto and generated files

* Watch client WIP

* Added watch function and unit test

* re-add header

* Update watch to have WatchAsyncSequence

* Create WatchEvent type for Element typealias

* Updated example

* Update test and WatchAsyncSequence

* add documentation

* Updated PR comments and added documentation

* Add license info

* Update kv to keyValue

* Update tests and documentation

* Update test

---------

Co-authored-by: “Ayushi <“[email protected]”>

Author: ayushi2103

Sources/ETCD/ETCDClient.swift

@@ -23,6 +23,7 @@ public final class EtcdClient: @unchecked Sendable {
     private var group: EventLoopGroup
     private var connection: ClientConnection
     private var client: Etcdserverpb_KVNIOClient
+    private var watchClient: Etcdserverpb_WatchAsyncClient
 
     /// Initialize a new ETCD Connection.
     ///
@@ -37,6 +38,7 @@ public final class EtcdClient: @unchecked Sendable {
         self.connection = ClientConnection.insecure(group: self.group)
             .connect(host: host, port: port)
         self.client = Etcdserverpb_KVNIOClient(channel: self.connection)
+        self.watchClient = Etcdserverpb_WatchAsyncClient(channel: self.connection)
     }
 
     /// Sets a value for a specified key in the ETCD server.
@@ -120,5 +122,24 @@ public final class EtcdClient: @unchecked Sendable {
     public func put(_ key: String, value: String) async throws {
         try await put(key.utf8, value: value.utf8)
     }
-
+    
+    /// Watches a specified key in the ETCD server.
+    ///
+    /// - Parameters:
+    ///   - key: The key for which the value is watched. Parameter is of type Sequence<UInt8>.
+    ///   - operation: The operation to be run on the WatchAsyncSequence for the key.
+    public func watch<Result>(_ key: some Sequence<UInt8>, _ operation: (WatchAsyncSequence) async throws -> Result) async throws -> Result {
+        let request = [Etcdserverpb_WatchRequest.with { $0.createRequest.key = Data(key) }]
+        let watchAsyncSequence = WatchAsyncSequence(grpcAsyncSequence: watchClient.watch(request))
+        return try await operation(watchAsyncSequence)
+    }
+    
+    /// Watches a specified key in the ETCD server.
+    ///
+    /// - Parameters:
+    ///   - key: The key for which the value is watched. Parameter is of type String.
+    ///   - operation: The operation to be run on the WatchAsyncSequence for the key.
+    public func watch<Result>(_ key: String, _ operation: (WatchAsyncSequence) async throws -> Result) async throws -> Result {
+        try await self.watch(key.utf8, operation)
+    }
 }

Sources/ETCD/KeyValue.swift

@@ -0,0 +1,52 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the swift-etcd-client-gsoc open source project
+//
+// Copyright (c) 2024 Apple Inc. and the swift-etcd-client-gsoc project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of swift-etcd-client-gsoc project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+import Foundation
+
+/// Struct representing a key-value pair in etcd server.
+public struct KeyValue {
+    public var key: Data
+    public var createRevision: Int
+    public var modRevision: Int
+    public var version: Int
+    public var value: Data
+    public var lease: Int
+
+    init(protoKeyValue: Etcdserverpb_KeyValue) {
+        self.key = protoKeyValue.key
+        self.createRevision = Int(protoKeyValue.createRevision)
+        self.modRevision = Int(protoKeyValue.modRevision)
+        self.version = Int(protoKeyValue.version)
+        self.value = protoKeyValue.value
+        self.lease = Int(protoKeyValue.lease)
+    }
+    
+    /// Initialize a new KeyValue.
+    ///
+    /// - Parameters:
+    ///   - key: key in bytes. An empty key is not allowed.
+    ///   - create_revision: revision of the last creation on the key.
+    ///   - mod_revision: revision of the last modification on the key.
+    ///   - version: version is the version of the key. A deletion resets the version to zero and any modification of the key increases its version.
+    ///   - value: value in bytes.
+    ///   - lease: the ID of the lease attached to the key. If lease is 0, then no lease is attached to the key.
+    public init(key: Data, createRevision: Int, modRevision: Int, version: Int, value: Data, lease: Int) {
+        self.key = key
+        self.createRevision = createRevision
+        self.modRevision = modRevision
+        self.version = version
+        self.value = value
+        self.lease = lease
+    }
+}

Sources/ETCD/WatchAsyncSequence.swift

@@ -0,0 +1,58 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the swift-etcd-client-gsoc open source project
+//
+// Copyright (c) 2024 Apple Inc. and the swift-etcd-client-gsoc project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of swift-etcd-client-gsoc project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+import GRPC
+
+public struct WatchAsyncSequence: AsyncSequence {
+    public typealias Element = [WatchEvent]
+
+    private let grpcAsyncSequence: GRPCAsyncResponseStream<Etcdserverpb_WatchResponse>
+
+    init(grpcAsyncSequence: GRPCAsyncResponseStream<Etcdserverpb_WatchResponse>) {
+        self.grpcAsyncSequence = grpcAsyncSequence
+    }
+
+    public func makeAsyncIterator() -> AsyncIterator {
+        .init(grpcIterator: self.grpcAsyncSequence.makeAsyncIterator())
+    }
+
+    public struct AsyncIterator: AsyncIteratorProtocol {
+        private var grpcIterator: GRPCAsyncResponseStream<Etcdserverpb_WatchResponse>.AsyncIterator?
+
+        fileprivate init(grpcIterator: GRPCAsyncResponseStream<Etcdserverpb_WatchResponse>.AsyncIterator) {
+            self.grpcIterator = grpcIterator
+        }
+
+        public mutating func next() async throws -> Element? {
+            while let response = try await self.grpcIterator?.next() {
+                if response.created {
+                    // We receive this after setting up the watch and need to wait for the next
+                    // response that contains an event
+                    precondition(response.events.isEmpty, "Expected no watch events on created response")
+                    continue
+                }
+
+                if response.canceled {
+                    // We got cancelled and have to return nil now
+                    self.grpcIterator = nil
+                    return nil
+                }
+
+                let events = response.events.map { WatchEvent(protoEvent: $0) }
+                return events
+            }
+            return nil;
+        }
+    }
+}

Sources/ETCD/WatchEvent.swift

@@ -0,0 +1,39 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the swift-etcd-client-gsoc open source project
+//
+// Copyright (c) 2024 Apple Inc. and the swift-etcd-client-gsoc project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of swift-etcd-client-gsoc project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+import Foundation
+
+public struct WatchEvent {
+    public var keyValue: KeyValue
+    public var previousKeyValue: KeyValue?
+
+    init(protoEvent: Etcdserverpb_Event) {
+        self.keyValue = KeyValue(protoKeyValue: protoEvent.kv)
+        if let protoPrevKV = protoEvent.hasPrevKv ? protoEvent.prevKv : nil {
+            self.previousKeyValue = KeyValue(protoKeyValue: protoPrevKV)
+        } else {
+            self.previousKeyValue = nil
+        }
+    }
+    
+    /// Struct representing a watch event in etcd.
+    ///
+    /// - Parameters:
+    ///   - keyValue: keyValue representing a KeyValue to watch.
+    ///   - previousKeyValue: previousKeyValue representing a KeyValue that was previously received.
+    public init(keyValue: KeyValue, previousKeyValue: KeyValue?) {
+        self.keyValue = keyValue
+        self.previousKeyValue = previousKeyValue
+    }
+}