v0.40.0

Version 0.40.0

This release focuses on improving stability during reconnection.

Note:

• The version introduces breaking changes.
• The API reconnect_stream API has been removed because auto-reconnection is now built into the client.
• Removed message_context.consumer.get_stream(message_context.subscriber_name) in favour of message_context.stream

Install:

pip install rstream==0.40.0

Auto-Reconnection

The reconnection is now part of the library, supported by a recovery strategy.

It is possible to customise the recovery strategy by recovery_strategy attribute.
The default behaviour is BackOffRecoveryStrategy, enabled by default.
You can disable it by BackOffRecoveryStrategy(False)

Breaking change: The API reconnect_stream is removed. It is not necessary anymore.

It is possible to check the reconnection status with the logs:

DEBUG:root:[Consumer, backOff recovery strategy] stream: super_cluster_stream-1 exist True. Attempt:1, reason:Exception
DEBUG:root:[Consumer, backOff recovery strategy] stream: super_cluster_stream-1 exists, trying to recover. Attempt:1, reason:Exception
DEBUG:root:[Consumer, backOff recovery strategy] stream: super_cluster_stream-1 recovered successfully. Attempt:1, reason:Exception
DEBUG:root:[Consumer, backOff recovery strategy] stream: super_cluster_stream-2 exist True. Attempt:1, reason:Exception
DEBUG:root:[Consumer, backOff recovery strategy] stream: super_cluster_stream-2 exists, trying to recover. Attempt:1, reason:Exception
DEBUG:root:[Consumer, backOff recovery strategy] stream: super_cluster_stream-2 recovered successfully. Attempt:1, reason:Exception
DEBUG:root:[Consumer, backOff recovery strategy] stream: super_cluster_stream-0 exist True. Attempt:1, reason:Exception

Consumer

The subscribe function now returns the subscription_id that you can use to unsubscribe.

        sub_id = await consumer.subscribe(stream, lambda x, y: None)
        await consumer.unsubscribe(sub_id)

Breaking change: The library no longer creates the subscriber name automatically. If you pass the subscriber name, it won't change at the library.

Publisher

Breaking change: The library no longer creates the producer name automatically. If you pass the producer name, the library automatically enables the deduplication.

Limitation

The version introduces a consumer limitation to guarantee the library's consistency.
The consumer can consume a stream only one time, ex:

✅ await consumer.subscribe( stream="my_stream"...) # ok
✅ await consumer.subscribe( stream="my_stream_2"...) # ok
✅ await consumer.subscribe( stream="my_stream"_3...) # ok
🚫 await consumer.subscribe( stream="my_stream"...) # that's not possible. will reaise StreamAlreadySubscribed exception

Dependencies

Update all the dependencies #243

What's Changed

• Update dependencies by @Gsantomaggio in #243
• Restore flake8 test by @Gsantomaggio in #244
• Restore mypy and fix the hints by @Gsantomaggio in #245
• fix subscribers list by @Gsantomaggio in #246
• Refactor the producers list replace the reference with the id as map key by @Gsantomaggio in #249
• Implement Auto recovery connection for consumer by @Gsantomaggio in #250
• Implement producer recovery connection by @Gsantomaggio in #251
• improve reconnection with metadata update handling by @Gsantomaggio in #252
• bump to 0.40.0-rc.1 by @Gsantomaggio in #253

Full Changelog: 0.31.0...v0.40.0