KafkaConsumer: expose event sequence
#97
Conversation
felixschlegel
force-pushed
the
fs-kafka-consumer-event-sequence
branch
from
c23cea8 to
e7ca788
Compare
felixschlegel
force-pushed
the
fs-kafka-consumer-event-sequence
branch
2 times, most recently
from
73fb645 to
8ac4610
Compare
| /// - config: The ``KafkaConsumerConfiguration`` for configuring the ``KafkaConsumer``. | ||
| /// - logger: A logger. |
| // MARK: - KafkaConsumerEvents | ||
|
|
||
| /// `AsyncSequence` implementation for handling ``KafkaConsumerEvent``s emitted by Kafka. | ||
| public struct KafkaConsumerEvents: AsyncSequence { |
There was a problem hiding this comment.
We should make this Sendable
There was a problem hiding this comment.
Right, also applied this to KafkaProducerEvents
README.md
Outdated
| @@ -88,7 +88,7 @@ var config = KafkaConsumerConfiguration( | |||
| ) | |||
| config.bootstrapServers = [broker] | |||
|
|
|||
| let consumer = try KafkaConsumer( | |||
| let consumer = try KafkaConsumer.makeConsumer( | |||
There was a problem hiding this comment.
Overall, I still think we should make the init() public so that in the case where users are not interested in event they can just call KafkaProducer() and KafkaConsumer().
There was a problem hiding this comment.
Right, also applied this to KafkaProducer.makeProducer
felixschlegel
force-pushed
the
fs-kafka-consumer-event-sequence
branch
from
d62516a to
e53ac31
Compare
Motivation: Like in `KafkaProducer` (swift-server#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
Modifications: * rename to `KafkaConsumerEvents.AsyncIterator` * rename to `KafkaConsumerMessages.AsyncIterator`
Modifications: * make `KafkaConsumerEvents` `Sendable` * make `KafkaProducerEvents` `Sendable` * replace `KafkaConsumer.makeConsumer` with `convenience init` * replace `KafkaProducer.makeProducer` with `convenience init`
felixschlegel
force-pushed
the
fs-kafka-consumer-event-sequence
branch
from
e53ac31 to
bb53e8c
Compare
| logger: logger | ||
| ) | ||
|
|
||
| let sourceAndSequence = NIOAsyncSequenceProducer.makeSequence( |
There was a problem hiding this comment.
I am not sure I fully understand an idea here: it seems like nothing is supposed to be supplied in this source...
Shouldn't be sourceAndSequence.source assigned anywhere?
There was a problem hiding this comment.
Yes, the primary intention of this PR was to expose the KafkaConsumerEvents itself without serving any events at the moment.
Following onto that PR events like rebalance etc. can be easily added into that existing KafkaConsumerEvents sequence.
Motivation:
Like in
KafkaProducer(#96), we want to expose an asynchronoussequence that is able to emit all sorts of events in the future (e.g.
rebalance events for the
KafkaConsumer.Modifications:
KafkaConsumerEvent(currently empty)KafkaConsumer.initprivateKafkaConsumer.createConsumerandKafkaConsumer.createConsumerWithEventsAsyncSequence:KafkaConsumerEvents