Basic implementation of the KafkaClient class (#18)

* Basic implementation of the `KafkaClient` class

Motivation:

The `KafkaClient` class is the base class for the future
implementations of `KafkaProducer` and `KafkaConsumer`. It handles the
connection to the Kafka C library.

Modifications:

* removed the `ErrorPrinter` example code
* added a basic implementation for `KafkaConfig`
* added a basic implementation for `KafkaError`
* deleted boilerplate `SwiftKafka` `struct`
* implemented `KafkaClient`
* added `swift-log` to project
* added DocC documentatio to `KafkaClient`

Result:

There is a `KafkaClient` class that is configurable using a preliminary
implementation of `KafkaConfig`.

* Fixed memory issue for KafkaConfig pointer duplicate objects

Modifications:

* created a private initializer for KafkaConfig that takes an
  OpaquePointer as its argmument
* implemented the createDuplicate() function that creates a duplicate
  object of the current KafkaConfig object so we can make use of
KafkaConfigs deinitializer and avoid memory leaks

Result:

No memory issues when duplicating KafkaConfigs

* Improved KafkaClient class

Modifications:

* changed access level of KafkaClient to internal as we do not want to
  subclass it in our public classes KafkaConsumer and KafkaProducer
* made var's let's and moved their initialization to initialization
* take Logger as an argument rather than creating our own Logger

* Added missing DocC comments

* Small changes to KafkaClient

Modifications:

* made the `KafkaClient` class `final`
* got rid of leading _ for internal (not in a Swift sense) variables
* added missing DocC comments

* Made `KafkaConfig` a `struct` with copy-on-write mechanism

Motivation:

Modifications:

* added a `description` property to the preliminary `KafkaError` type
* throw a more descriptive error when `KafkaClient` initialization fails
* replace `subscript` in `KafkaConfig` with explicit `set` and `get`
  functions
* `KafkaConfig` is now a `struct` that contains an internal class and a
* made `KafkaConfig` conform to `Equatable`
* wrote tests for `KafkaConfig`

* Refactoring

Modifications:

* changed /** DocC comments to /// comments
* made `KafkaConfig` conform to `Hashable`

* Refactored KafkaConfig's conformance to Hashable, Equatable

Author: felixschlegel

Package.swift

@@ -22,20 +22,17 @@ let package = Package(
             name: "SwiftKafka",
             targets: ["SwiftKafka"]
         ),
-        .executable(
-            name: "ErrorPrinter",
-            targets: ["ErrorPrinter"]
-        ),
     ],
-    dependencies: [],
+    dependencies: [
+        .package(url: "https://github.com/apple/swift-log.git", from: "1.0.0"),
+    ],
     targets: [
         .target(
             name: "SwiftKafka",
-            dependencies: []
-        ),
-        .executableTarget(
-            name: "ErrorPrinter",
-            dependencies: ["Crdkafka"]
+            dependencies: [
+                "Crdkafka",
+                .product(name: "Logging", package: "swift-log"),
+            ]
         ),
         .systemLibrary(
             name: "Crdkafka",

Sources/ErrorPrinter/ErrorPrinter.swift

@@ -0,0 +1,63 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the swift-kafka-gsoc open source project
+//
+// Copyright (c) 2022 Apple Inc. and the swift-kafka-gsoc project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of swift-kafka-gsoc project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+import Crdkafka
+import Logging
+
+/// Base class for ``KafkaProducer`` and ``KafkaConsumer``,
+/// which is used to handle the connection to the Kafka ecosystem.
+final class KafkaClient {
+    // Default size for Strings returned from C API
+    static let stringSize = 1024
+
+    /// A logger.
+    private let logger: Logger
+
+    /// A client is either a `.producer` or a `.consumer`
+    private let clientType: rd_kafka_type_t
+    /// The configuration object of the client
+    private let config: KafkaConfig
+    /// Handle for the C library's Kafka instance
+    private let kafkaHandle: OpaquePointer
+
+    /// Determines if client is a producer or a consumer
+    enum `Type` {
+        case producer
+        case consumer
+    }
+
+    init(type: Type, config: KafkaConfig, logger: Logger) throws {
+        self.logger = logger
+        self.clientType = type == .producer ? RD_KAFKA_PRODUCER : RD_KAFKA_CONSUMER
+        self.config = config
+
+        let errorChars = UnsafeMutablePointer<CChar>.allocate(capacity: KafkaClient.stringSize)
+        defer { errorChars.deallocate() }
+
+        guard let handle = rd_kafka_new(
+            clientType,
+            config.pointer,
+            errorChars,
+            KafkaClient.stringSize
+        ) else {
+            let errorString = String(cString: errorChars)
+            throw KafkaError(description: errorString)
+        }
+        self.kafkaHandle = handle
+    }
+
+    deinit {
+        rd_kafka_destroy(kafkaHandle)
+    }
+}

Sources/SwiftKafka/KafkaClient.swift

@@ -0,0 +1,120 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the swift-kafka-gsoc open source project
+//
+// Copyright (c) 2022 Apple Inc. and the swift-kafka-gsoc project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of swift-kafka-gsoc project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+import Crdkafka
+
+/// Used to configure producers and consumers.
+/// `KafkaConfig` is a `struct` that points to a configuration in memory.
+/// Once a property of the `KafkaConfig` is changed, a duplicate in-memory config is created using the
+/// copy-on-write mechanism.
+/// For more information on how to configure Kafka, see
+/// [all available configurations](https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md).
+public struct KafkaConfig: Hashable, Equatable {
+    private final class _Internal: Hashable, Equatable {
+        /// Pointer to the `rd_kafka_conf_t` object managed by `librdkafka`
+        private(set) var pointer: OpaquePointer
+
+        /// Initialize internal `KafkaConfig` object with default configuration
+        init() {
+            self.pointer = rd_kafka_conf_new()
+        }
+
+        /// Initialize internal `KafkaConfig` object through a given `rd_kafka_conf_t` pointer
+        init(pointer: OpaquePointer) {
+            self.pointer = pointer
+        }
+
+        deinit {
+            rd_kafka_conf_destroy(pointer)
+        }
+
+        func value(forKey key: String) -> String? {
+            let value = UnsafeMutablePointer<CChar>.allocate(capacity: KafkaClient.stringSize)
+            defer { value.deallocate() }
+
+            var valueSize = KafkaClient.stringSize
+            let configResult = rd_kafka_conf_get(
+                pointer,
+                key,
+                value,
+                &valueSize
+            )
+
+            if configResult == RD_KAFKA_CONF_OK {
+                return String(cString: value)
+            }
+            return nil
+        }
+
+        func set(_ value: String, forKey key: String) throws {
+            let errorChars = UnsafeMutablePointer<CChar>.allocate(capacity: KafkaClient.stringSize)
+            defer { errorChars.deallocate() }
+
+            let configResult = rd_kafka_conf_set(
+                pointer,
+                key,
+                value,
+                errorChars,
+                KafkaClient.stringSize
+            )
+
+            if configResult != RD_KAFKA_CONF_OK {
+                let errorString = String(cString: errorChars)
+                throw KafkaError(description: errorString)
+            }
+        }
+
+        func createDuplicate() -> _Internal {
+            let duplicatePointer: OpaquePointer = rd_kafka_conf_dup(self.pointer)
+            return .init(pointer: duplicatePointer)
+        }
+
+        // MARK: Hashable
+
+        func hash(into hasher: inout Hasher) {
+            hasher.combine(self.pointer)
+        }
+
+        // MARK: Equatable
+
+        static func == (lhs: _Internal, rhs: _Internal) -> Bool {
+            return lhs.pointer == rhs.pointer
+        }
+    }
+
+    private var _internal: _Internal
+
+    public init() {
+        self._internal = .init()
+    }
+
+    var pointer: OpaquePointer {
+        return self._internal.pointer
+    }
+
+    /// Retrieve value of configuration property for `key`
+    public func value(forKey key: String) -> String? {
+        return self._internal.value(forKey: key)
+    }
+
+    /// Set configuration `value` for `key`
+    public mutating func set(_ value: String, forKey key: String) throws {
+        // Copy-on-write mechanism
+        if !isKnownUniquelyReferenced(&(self._internal)) {
+            self._internal = self._internal.createDuplicate()
+        }
+
+        try self._internal.set(value, forKey: key)
+    }
+}

Sources/SwiftKafka/KafkaConfig.swift

@@ -12,8 +12,7 @@
 //
 //===----------------------------------------------------------------------===//
 
-public struct SwiftKafka {
-    public private(set) var text = "Hello, World!"
-
-    public init() {}
+public struct KafkaError: Error {
+    // Preliminary Implementation
+    public let description: String
 }

Sources/SwiftKafka/SwiftKafka.swift renamed to Sources/SwiftKafka/KafkaError.swift

@@ -0,0 +1,68 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the swift-kafka-gsoc open source project
+//
+// Copyright (c) 2022 Apple Inc. and the swift-kafka-gsoc project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of swift-kafka-gsoc project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+@testable import SwiftKafka
+import XCTest
+
+final class KafkaConfigTests: XCTestCase {
+    func testSettingCorrectValueWorks() throws {
+        var config = KafkaConfig()
+
+        try config.set("ssl", forKey: "security.protocol")
+
+        XCTAssertEqual("ssl", config.value(forKey: "security.protocol"))
+    }
+
+    func testSettingWrongKeyFails() {
+        var config = KafkaConfig()
+
+        XCTAssertThrowsError(try config.set("ssl", forKey: "not.a.valid.key"))
+    }
+
+    func testSettingWrongValueFails() {
+        var config = KafkaConfig()
+
+        XCTAssertThrowsError(try config.set("not_a_protocol", forKey: "security.protocol"))
+    }
+
+    func testGetterHasNoSideEffects() {
+        let configA = KafkaConfig()
+        let configB = configA
+
+        _ = configA.value(forKey: "security.protocol")
+        _ = configB.value(forKey: "security.protocol")
+
+        XCTAssertTrue(configA == configB)
+    }
+
+    func testCopyOnWriteWorks() throws {
+        var configA = KafkaConfig()
+        let configB = configA
+        let configC = configA
+
+        // Check if all configs have the default value set
+        [configA, configB, configC].forEach {
+            XCTAssertEqual("plaintext", $0.value(forKey: "security.protocol"))
+        }
+
+        try configA.set("ssl", forKey: "security.protocol")
+
+        XCTAssertEqual("ssl", configA.value(forKey: "security.protocol"))
+        XCTAssertEqual("plaintext", configB.value(forKey: "security.protocol"))
+        XCTAssertEqual("plaintext", configC.value(forKey: "security.protocol"))
+        XCTAssertNotEqual(configA, configB)
+        XCTAssertNotEqual(configA, configC)
+        XCTAssertEqual(configB, configC)
+    }
+}

Tests/SwiftKafkaTests/KafkaConfigTests.swift

@@ -15,11 +15,4 @@
 @testable import SwiftKafka
 import XCTest
 
-final class SwiftKafkaTests: XCTestCase {
-    func testExample() throws {
-        // This is an example of a functional test case.
-        // Use XCTAssert and related functions to verify your tests produce the correct
-        // results.
-        XCTAssertEqual(SwiftKafka().text, "Hello, World!")
-    }
-}
+final class SwiftKafkaTests: XCTestCase {}

Tests/SwiftKafkaTests/SwiftKafkaTests.swift

@@ -10,4 +10,4 @@ WORKDIR /swift-kafka-gsoc
 
 COPY . /swift-kafka-gsoc
 
-CMD ["swift", "run", "ErrorPrinter"]
+CMD ["swift", "test"]