Adapt new config structs in legacy implementation

Modifications:

* `ClientConfig` protocol with protocol extension for properties used in
  both consumer and producer
* `ConsumerConfig` and `ProducerConfig` structs
* added initializer to `KafkaConfig` that allows conversion from
  `ClientConfig` to `KafkaConfig`

Author: felixschlegel

Sources/SwiftKafka/KafkaConfig.swift

@@ -46,6 +46,18 @@ public struct KafkaConfig: Hashable, Equatable {
             self.opaque = opaque
         }
 
+        // TODO: docc
+        convenience init(clientConfig: any ClientConfig) throws {
+            self.init(
+                pointer: rd_kafka_conf_new(),
+                opaque: nil
+            )
+
+            try clientConfig.properties.forEach { key, value in
+                try self.set(value, forKey: key)
+            }
+        }
+
         /// Initialize internal `KafkaConfig` object with default configuration.
         convenience init() {
             self.init(
@@ -160,6 +172,11 @@ public struct KafkaConfig: Hashable, Equatable {
         self._internal = .init()
     }
 
+    // TODO: docc new public interface for config
+    init(clientConfig: any ClientConfig) throws {
+        self._internal = try .init(clientConfig: clientConfig)
+    }
+
     /// Retrieve value of configuration property for `key`
     public func value(forKey key: String) -> String? {
         return self._internal.value(forKey: key)

Sources/SwiftKafka/KafkaConsumer.swift

@@ -193,6 +193,39 @@ public final class KafkaConsumer {
         )
     }
 
+    // MARK: - Initialisers with new config
+
+    // TODO: refactor into single initializer for KafkaConsumer and determine assignment/subscription via config
+
+    public convenience init(
+        topics: [String], // TODO: pass topics via config?
+        config: ConsumerConfig = ConsumerConfig(),
+        logger: Logger
+    ) throws {
+        try self.init(
+            topics: topics,
+            groupID: config.groupID,
+            config: KafkaConfig(clientConfig: config),
+            logger: logger
+        )
+    }
+
+    public convenience init(
+        topic: String, // TODO: pass topic via config?
+        partition: KafkaPartition,
+        offset: Int64,
+        config: ConsumerConfig = ConsumerConfig(),
+        logger: Logger
+    ) throws {
+        try self.init(
+            topic: topic,
+            partition: partition,
+            offset: offset,
+            config: KafkaConfig(clientConfig: config),
+            logger: logger
+        )
+    }
+
     /// Subscribe to the given list of `topics`.
     /// The partition assignment happens automatically using `KafkaConsumer`'s consumer group.
     /// - Parameter topics: An array of topic names to subscribe to.

Sources/SwiftKafka/KafkaProducer.swift

@@ -136,6 +136,16 @@ public actor KafkaProducer {
         }
     }
 
+    // MARK: - Initialiser with new config
+
+    public init(
+        config: ProducerConfig = ProducerConfig(),
+        topicConfig: KafkaTopicConfig = KafkaTopicConfig(), // TODO: new topic config
+        logger: Logger
+    ) async throws {
+        try await self.init(config: KafkaConfig(clientConfig: config), topicConfig: topicConfig, logger: logger)
+    }
+
     /// Method to shutdown the ``KafkaProducer``.
     ///
     /// This method flushes any buffered messages and waits until a callback is received for all of them.