KafkaConsumer: expose event sequence (#97)

* `KafkaConsumer`: expose event sequence

Motivation:

Like in `KafkaProducer` (#96), we want to expose an asynchronous
sequence that is able to emit all sorts of events in the future (e.g.
rebalance events for the `KafkaConsumer`.

Modifications:

* add a new type `KafkaConsumerEvent` (currently empty)
* make `KafkaConsumer.init` private
* add factory methods `KafkaConsumer.createConsumer` and
  `KafkaConsumer.createConsumerWithEvents`
* create a new `AsyncSequence`: `KafkaConsumerEvents`
* update README
* update tests

* `AsyncIterator` rename

Modifications:

* rename to `KafkaConsumerEvents.AsyncIterator`
* rename to `KafkaConsumerMessages.AsyncIterator`

* Review Gus: missing documentation

* Review Franz

Modifications:

* make `KafkaConsumerEvents` `Sendable`
* make `KafkaProducerEvents` `Sendable`
* replace `KafkaConsumer.makeConsumer` with `convenience init`
* replace `KafkaProducer.makeProducer` with `convenience init`

* Fix post-rebase issue

* Remove outdated comments

Author: felixschlegel

Sources/SwiftKafka/KafkaConsumer.swift

@@ -35,6 +35,29 @@ extension KafkaConsumerCloseOnTerminate: NIOAsyncSequenceProducerDelegate {
     }
 }
 
+// MARK: - KafkaConsumerEvents
+
+/// `AsyncSequence` implementation for handling ``KafkaConsumerEvent``s emitted by Kafka.
+public struct KafkaConsumerEvents: Sendable, AsyncSequence {
+    public typealias Element = KafkaConsumerEvent
+    typealias BackPressureStrategy = NIOAsyncSequenceProducerBackPressureStrategies.NoBackPressure
+    typealias WrappedSequence = NIOAsyncSequenceProducer<Element, BackPressureStrategy, KafkaConsumerCloseOnTerminate>
+    let wrappedSequence: WrappedSequence
+
+    /// `AsynceIteratorProtocol` implementation for handling ``KafkaConsumerEvent``s emitted by Kafka.
+    public struct AsyncIterator: AsyncIteratorProtocol {
+        var wrappedIterator: WrappedSequence.AsyncIterator
+
+        public mutating func next() async -> Element? {
+            await self.wrappedIterator.next()
+        }
+    }
+
+    public func makeAsyncIterator() -> AsyncIterator {
+        return AsyncIterator(wrappedIterator: self.wrappedSequence.makeAsyncIterator())
+    }
+}
+
 // MARK: - KafkaConsumerMessages
 
 /// `AsyncSequence` implementation for handling messages received from the Kafka cluster (``KafkaConsumerMessage``).
@@ -52,7 +75,7 @@ public struct KafkaConsumerMessages: Sendable, AsyncSequence {
     let wrappedSequence: WrappedSequence
 
     /// `AsynceIteratorProtocol` implementation for handling messages received from the Kafka cluster (``KafkaConsumerMessage``).
-    public struct ConsumerMessagesAsyncIterator: AsyncIteratorProtocol {
+    public struct AsyncIterator: AsyncIteratorProtocol {
         let stateMachine: NIOLockedValueBox<KafkaConsumer.StateMachine>
         var wrappedIterator: WrappedSequence.AsyncIterator?
 
@@ -80,8 +103,8 @@ public struct KafkaConsumerMessages: Sendable, AsyncSequence {
         }
     }
 
-    public func makeAsyncIterator() -> ConsumerMessagesAsyncIterator {
-        return ConsumerMessagesAsyncIterator(
+    public func makeAsyncIterator() -> AsyncIterator {
+        return AsyncIterator(
             stateMachine: self.stateMachine,
             wrappedIterator: self.wrappedSequence.makeAsyncIterator()
         )
@@ -108,34 +131,27 @@ public final class KafkaConsumer: Sendable, Service {
     /// `AsyncSequence` that returns all ``KafkaConsumerMessage`` objects that the consumer receives.
     public let messages: KafkaConsumerMessages
 
+    // Private initializer, use factory method or convenience init to create KafkaConsumer
     /// Initialize a new ``KafkaConsumer``.
     /// To listen to incoming messages, please subscribe to a list of topics using ``subscribe(topics:)``
     /// or assign the consumer to a particular topic + partition pair using ``assign(topic:partition:offset:)``.
-    /// - Parameter config: The ``KafkaConsumerConfiguration`` for configuring the ``KafkaConsumer``.
-    /// - Parameter logger: A logger.
+    ///
+    /// - Parameters:
+    ///     - client: Client used for handling the connection to the Kafka cluster.
+    ///     - stateMachine: The state machine containing the state of the ``KafkaConsumer``.
+    ///     - config: The ``KafkaConsumerConfiguration`` for configuring the ``KafkaConsumer``.
+    ///     - logger: A logger.
     /// - Throws: A ``KafkaError`` if the initialization failed.
-    public init(
+    private init(
+        client: RDKafkaClient,
+        stateMachine: NIOLockedValueBox<StateMachine>,
         config: KafkaConsumerConfiguration,
         logger: Logger
     ) throws {
         self.config = config
+        self.stateMachine = stateMachine
         self.logger = logger
 
-        var subscribedEvents: [RDKafkaEvent] = [.log, .fetch]
-        // Only listen to offset commit events when autoCommit is false
-        if self.config.enableAutoCommit == false {
-            subscribedEvents.append(.offsetCommit)
-        }
-
-        let client = try RDKafkaClient.makeClient(
-            type: .consumer,
-            configDictionary: config.dictionary,
-            events: subscribedEvents,
-            logger: logger
-        )
-
-        self.stateMachine = NIOLockedValueBox(StateMachine(logger: self.logger))
-
         let sourceAndSequence = NIOThrowingAsyncSequenceProducer.makeSequence(
             elementType: KafkaConsumerMessage.self,
             backPressureStrategy: NIOAsyncSequenceProducerBackPressureStrategies.NoBackPressure(),
@@ -166,6 +182,91 @@ public final class KafkaConsumer: Sendable, Service {
         }
     }
 
+    /// Initialize a new ``KafkaConsumer``.
+    ///
+    /// This creates a consumer without that does not listen to any events other than consumer messages.
+    ///
+    /// - Parameters:
+    ///     - config: The ``KafkaConsumerConfiguration`` for configuring the ``KafkaConsumer``.
+    ///     - logger: A logger.
+    /// - Returns: The newly created ``KafkaConsumer``.
+    /// - Throws: A ``KafkaError`` if the initialization failed.
+    public convenience init(
+        config: KafkaConsumerConfiguration,
+        logger: Logger
+    ) throws {
+        let stateMachine = NIOLockedValueBox(StateMachine(logger: logger))
+
+        var subscribedEvents: [RDKafkaEvent] = [.log, .fetch]
+        // Only listen to offset commit events when autoCommit is false
+        if config.enableAutoCommit == false {
+            subscribedEvents.append(.offsetCommit)
+        }
+
+        let client = try RDKafkaClient.makeClient(
+            type: .consumer,
+            configDictionary: config.dictionary,
+            events: subscribedEvents,
+            logger: logger
+        )
+
+        try self.init(
+            client: client,
+            stateMachine: stateMachine,
+            config: config,
+            logger: logger
+        )
+    }
+
+    /// Initialize a new ``KafkaConsumer`` and a ``KafkaConsumerEvents`` asynchronous sequence.
+    ///
+    /// Use the asynchronous sequence to consume events.
+    ///
+    /// - Important: When the asynchronous sequence is deinited the producer will be shutdown and disallow sending more messages.
+    /// Additionally, make sure to consume the asynchronous sequence otherwise the events will be buffered in memory indefinitely.
+    ///
+    /// - Parameters:
+    ///     - config: The ``KafkaConsumerConfiguration`` for configuring the ``KafkaConsumer``.
+    ///     - logger: A logger.
+    /// - Returns: A tuple containing the created ``KafkaConsumer`` and the ``KafkaConsumerEvents``
+    /// `AsyncSequence` used for receiving message events.
+    /// - Throws: A ``KafkaError`` if the initialization failed.
+    public static func makeConsumerWithEvents(
+        config: KafkaConsumerConfiguration,
+        logger: Logger
+    ) throws -> (KafkaConsumer, KafkaConsumerEvents) {
+        let stateMachine = NIOLockedValueBox(StateMachine(logger: logger))
+
+        var subscribedEvents: [RDKafkaEvent] = [.log, .fetch]
+        // Only listen to offset commit events when autoCommit is false
+        if config.enableAutoCommit == false {
+            subscribedEvents.append(.offsetCommit)
+        }
+
+        let client = try RDKafkaClient.makeClient(
+            type: .consumer,
+            configDictionary: config.dictionary,
+            events: subscribedEvents,
+            logger: logger
+        )
+
+        let consumer = try KafkaConsumer(
+            client: client,
+            stateMachine: stateMachine,
+            config: config,
+            logger: logger
+        )
+
+        let sourceAndSequence = NIOAsyncSequenceProducer.makeSequence(
+            elementType: KafkaConsumerEvent.self,
+            backPressureStrategy: NIOAsyncSequenceProducerBackPressureStrategies.NoBackPressure(),
+            delegate: KafkaConsumerCloseOnTerminate(stateMachine: stateMachine)
+        )
+
+        let eventsSequence = KafkaConsumerEvents(wrappedSequence: sourceAndSequence.sequence)
+        return (consumer, eventsSequence)
+    }
+
     /// 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/KafkaConsumerEvent.swift

@@ -0,0 +1,28 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the swift-kafka-gsoc open source project
+//
+// Copyright (c) 2023 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
+//
+//===----------------------------------------------------------------------===//
+
+/// An enumeration representing events that can be received through the ``KafkaConsumerEvents`` asynchronous sequence.
+public enum KafkaConsumerEvent: Sendable, Hashable {
+    /// - Important: Always provide a `default` case when switiching over this `enum`.
+    case DO_NOT_SWITCH_OVER_THIS_EXHAUSITVELY
+
+    internal init(_ event: RDKafkaClient.KafkaEvent) {
+        switch event {
+        case .deliveryReport:
+            fatalError("Cannot cast \(event) to KafkaConsumerEvent")
+        case .consumerMessages:
+            fatalError("Consumer messages should be handled in the KafkaConsumerMessages asynchronous sequence")
+        }
+    }
+}

Sources/SwiftKafka/KafkaProducer.swift

@@ -44,7 +44,7 @@ extension KafkaProducerCloseOnTerminate: NIOAsyncSequenceProducerDelegate {
 // MARK: - KafkaProducerEvents
 
 /// `AsyncSequence` implementation for handling ``KafkaProducerEvent``s emitted by Kafka.
-public struct KafkaProducerEvents: AsyncSequence {
+public struct KafkaProducerEvents: Sendable, AsyncSequence {
     public typealias Element = KafkaProducerEvent
     typealias BackPressureStrategy = NIOAsyncSequenceProducerBackPressureStrategies.NoBackPressure
     typealias WrappedSequence = NIOAsyncSequenceProducer<Element, BackPressureStrategy, KafkaProducerCloseOnTerminate>
@@ -85,7 +85,7 @@ public final class KafkaProducer: Service, Sendable {
     /// Topic configuration that is used when a new topic has to be created by the producer.
     private let topicConfig: KafkaTopicConfiguration
 
-    // Private initializer, use factory methods to create KafkaProducer
+    // Private initializer, use factory method or convenience init to create KafkaProducer
     /// Initialize a new ``KafkaProducer``.
     ///
     /// - Parameter stateMachine: The ``KafkaProducer/StateMachine`` instance associated with the ``KafkaProducer``.///
@@ -104,18 +104,18 @@ public final class KafkaProducer: Service, Sendable {
 
     /// Initialize a new ``KafkaProducer``.
     ///
-    /// This factory method creates a producer without listening for events.
+    /// This creates a producer without listening for events.
     ///
     /// - Parameter config: The ``KafkaProducerConfiguration`` for configuring the ``KafkaProducer``.
     /// - Parameter topicConfig: The ``KafkaTopicConfiguration`` used for newly created topics.
     /// - Parameter logger: A logger.
     /// - Returns: The newly created ``KafkaProducer``.
     /// - Throws: A ``KafkaError`` if initializing the producer failed.
-    public static func makeProducer(
+    public convenience init(
         config: KafkaProducerConfiguration = KafkaProducerConfiguration(),
         topicConfig: KafkaTopicConfiguration = KafkaTopicConfiguration(),
         logger: Logger
-    ) throws -> KafkaProducer {
+    ) throws {
         let stateMachine = NIOLockedValueBox(StateMachine(logger: logger))
 
         let client = try RDKafkaClient.makeClient(
@@ -125,20 +125,18 @@ public final class KafkaProducer: Service, Sendable {
             logger: logger
         )
 
-        let producer = try KafkaProducer(
-            stateMachine: stateMachine,
-            config: config,
-            topicConfig: topicConfig
-        )
-
         stateMachine.withLockedValue {
             $0.initialize(
                 client: client,
                 source: nil
             )
         }
 
-        return producer
+        try self.init(
+            stateMachine: stateMachine,
+            config: config,
+            topicConfig: topicConfig
+        )
     }
 
     /// Initialize a new ``KafkaProducer`` and a ``KafkaProducerEvents`` asynchronous sequence.

Tests/SwiftKafkaTests/KafkaProducerTests.swift

@@ -246,7 +246,7 @@ final class KafkaProducerTests: XCTestCase {
         var config = KafkaProducerConfiguration()
         config.bootstrapServers = []
 
-        let producer = try KafkaProducer.makeProducer(config: config, logger: mockLogger)
+        let producer = try KafkaProducer(config: config, logger: mockLogger)
 
         let serviceGroup = ServiceGroup(
             services: [producer],