Adopt swift-service-lifecycle
#81
Merged
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Modifications:
* add `swift-service-lifecycle` aka `ServicLifecycle` dependency to
`Package.swift`
* add `Sendable` conformance to the following types:
* `KafkaProducerConfiguration`
* `KafkaConsumerConfiguration`
* `KafkaTopicConfiguration`
* `KafkaPartition`
* `KafkaClient`
* `KafkaConsumerMessages`
* `KafkaConsumerMessage`
* `RDKafkaConfig.CapturedClosures`
* `RDKafkaTopicHandles`
* add `@unchecked Sendable` conformance to the following types:
* `KafkaProducer`
* `KafkaConsumer`
* add `Service` conformance to the following types:
* `KafkaProducer`
* `KafkaConsumer`
* `SwiftKafkaTests`:
* use `ServiceGroup`s
* `KafkaProducerTests`:
* use `ServiceGroup`s
* remove `testFlushQueuedProducerMessages` as it relied on the now
`private` implementation detail `triggerGracefulShutdown`
* remove `testProducerNotUsableAfterShutdown` as it relied on the now
`private` implementation detail `triggerGracefulShutdown`
* `testNoMemoryLeakAfterShutdown`: remove obsolete wait for timeout
as flushing in `KafkaProducer` is now non-blocking
* `KafkaConsumer`:
* remove `triggerGracefulShutdown` on `deinit` -> this is now
`ServicLifecycle`'s responsibility
* refactor `RDKafkaConfig` to make `RDKafkaConfig.CapturedClosures`
sendable
Motivation: Exiting the `run()` loop early is an error in `swift-service-lifecycle`. Therefore we **don't** want to invoke `KafkaProducer/KafkaConsumer.triggerGracefulShutdown()` when the our `NIOAsyncSequence`s have stopped being consumed Modifications: * remove `Kafka[Producer|Consumer]ShutdownOnTerminate` behaviour * adjust tests
FranzBusch
requested changes
Jul 4, 2023
README.md
Outdated
| ### Producer API | ||
|
|
||
| The `send(_:)` method of `KafkaProducer` returns a message-id that can later be used to identify the corresponding acknowledgement. Acknowledgements are received through the `acknowledgements` [`AsyncSequence`](https://developer.apple.com/documentation/swift/asyncsequence). Each acknowledgement indicates that producing a message was successful or returns an error. | ||
|
|
||
| ```swift | ||
| let config = KafkaProducerConfiguration(bootstrapServers: ["localhost:9092"]) | ||
|
|
||
| let (producer, acknowledgements) = try KafkaProducer.makeProducerWithAcknowledgements( | ||
| let (producerService, acknowledgements) = try KafkaProducer.makeProducerWithAcknowledgements( |
There was a problem hiding this comment.
Don't think we need the suffix here
Suggested change
| let (producerService, acknowledgements) = try KafkaProducer.makeProducerWithAcknowledgements( | |
| let (producer, acknowledgements) = try KafkaProducer.makeProducerWithAcknowledgements( |
README.md
Outdated
| @@ -74,21 +80,26 @@ let config = KafkaConsumerConfiguration( | |||
| bootstrapServers: ["localhost:9092"] | |||
| ) | |||
|
|
|||
| let consumer = try KafkaConsumer( | |||
| let consumerService = try KafkaConsumer( | |||
README.md
Outdated
| @@ -137,24 +153,29 @@ let config = KafkaConsumerConfiguration( | |||
| bootstrapServers: ["localhost:9092"] | |||
| ) | |||
|
|
|||
| let consumer = try KafkaConsumer( | |||
| let consumerService = try KafkaConsumer( | |||
| @@ -81,14 +44,14 @@ public struct KafkaConsumerMessages: AsyncSequence { | |||
| // MARK: - KafkaConsumer | |||
|
|
|||
| /// Receive messages from the Kafka cluster. | |||
| public final class KafkaConsumer { | |||
| public final class KafkaConsumer: Sendable, Service { // We can do that because our stored propery is protected by a lock | |||
There was a problem hiding this comment.
We can remove this comment. It is not fully true because only the stateMachine is protected and the compiler is enforcing it here and we don't use @unchchecked
Modifications: * remove `*Service` suffix from `consumerService` and `producerService` in tests and README * remove old comment in `KafkaConsumer`
FranzBusch
approved these changes
Jul 5, 2023
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Adapt
swift-service-lifecycleModifications:
swift-service-lifecycleakaServicLifecycledependency toPackage.swiftSendableconformance to the following types:KafkaProducerConfigurationKafkaConsumerConfigurationKafkaTopicConfigurationKafkaPartitionKafkaClientKafkaConsumerMessagesKafkaConsumerMessageRDKafkaConfig.CapturedClosuresRDKafkaTopicHandles@unchecked Sendableconformance to the following types:KafkaProducerKafkaConsumerServiceconformance to the following types:KafkaProducerKafkaConsumerSwiftKafkaTests:ServiceGroupsKafkaProducerTests:ServiceGroupstestFlushQueuedProducerMessagesas it relied on the nowprivateimplementation detailtriggerGracefulShutdowntestProducerNotUsableAfterShutdownas it relied on the nowprivateimplementation detailtriggerGracefulShutdowntestNoMemoryLeakAfterShutdown: remove obsolete wait for timeoutas flushing in
KafkaProduceris now non-blockingKafkaConsumer:triggerGracefulShutdownondeinit-> this is nowServicLifecycle's responsibilityRDKafkaConfigto makeRDKafkaConfig.CapturedClosuressendable
Remove
ShutdownOnTerminateforAsyncSequencesMotivation:
Exiting the
run()loop early is an error inswift-service-lifecycle.Therefore we don't want to invoke
KafkaProducer/KafkaConsumer.triggerGracefulShutdown()when the ourNIOAsyncSequences have stopped being consumedModifications:
Kafka[Producer|Consumer]ShutdownOnTerminatebehaviour