Event engine/compatibility (#185)

* Fix compatibility issues between EventEngine and Asyncio subscription manager

* PubNub SDK v7.4.4 release.

---------

Co-authored-by: PubNub Release Bot <[email protected]>

Author: Sebastian Molenda

.github/workflows/run-tests.yml

@@ -77,8 +77,8 @@ jobs:
           cp sdk-specifications/features/encryption/cryptor-module.feature tests/acceptance/encryption
           mkdir tests/acceptance/encryption/assets/
           cp sdk-specifications/features/encryption/assets/* tests/acceptance/encryption/assets/
-          cp sdk-specifications/features/subscribe/event-engine/happy-path.feature tests/acceptance/subscribe/happy-path.feature
-          cp sdk-specifications/features/presence/event-engine/presence-engine.feature  tests/acceptance/subscribe/presence-engine.feature
+          cp sdk-specifications/features/subscribe/event-engine/happy-path_Legacy.feature tests/acceptance/subscribe/happy-path_Legacy.feature
+          cp sdk-specifications/features/presence/event-engine/presence-engine_Legacy.feature  tests/acceptance/subscribe/presence-engine_Legacy.feature
 
           sudo pip3 install -r requirements-dev.txt
           behave --junit tests/acceptance/pam

.pubnub.yml

@@ -1,5 +1,5 @@
 name: python
-version: 7.4.3
+version: 7.4.4
 schema: 1
 scm: github.com/pubnub/python
 sdks:
@@ -18,7 +18,7 @@ sdks:
         distributions:
           - distribution-type: library
             distribution-repository: package
-            package-name: pubnub-7.4.3
+            package-name: pubnub-7.4.4
             location: https://pypi.org/project/pubnub/
             supported-platforms:
               supported-operating-systems:
@@ -97,8 +97,8 @@ sdks:
           -
             distribution-type: library
             distribution-repository: git release
-            package-name: pubnub-7.4.3
-            location: https://github.com/pubnub/python/releases/download/v7.4.3/pubnub-7.4.3.tar.gz
+            package-name: pubnub-7.4.4
+            location: https://github.com/pubnub/python/releases/download/v7.4.4/pubnub-7.4.4.tar.gz
             supported-platforms:
               supported-operating-systems:
                 Linux:
@@ -169,6 +169,11 @@ sdks:
                 license-url: https://github.com/aio-libs/aiohttp/blob/master/LICENSE.txt
                 is-required: Required
 changelog:
+  - date: 2024-04-10
+    version: v7.4.4
+    changes:
+      - type: bug
+        text: "Fix compatibility issues between EventEngine and Asyncio subscription manager."
   - date: 2024-03-28
     version: v7.4.3
     changes:

CHANGELOG.md

@@ -1,3 +1,9 @@
+## v7.4.4
+April 10 2024
+
+#### Fixed
+- Fix compatibility issues between EventEngine and Asyncio subscription manager.
+
 ## v7.4.3
 March 28 2024
 

pubnub/event_engine/effects.py

@@ -58,11 +58,11 @@ def get_new_stop_event(self):
         return event
 
     def calculate_reconnection_delay(self, attempts):
-        if self.reconnection_policy is PNReconnectionPolicy.LINEAR:
+        if self.reconnection_policy is PNReconnectionPolicy.EXPONENTIAL:
+            delay = int(math.pow(2, attempts - 5 * math.floor((attempts - 1) / 5)) - 1)
+        else:
             delay = self.interval
 
-        elif self.reconnection_policy is PNReconnectionPolicy.EXPONENTIAL:
-            delay = int(math.pow(2, attempts - 5 * math.floor((attempts - 1) / 5)) - 1)
         return delay
 
 
@@ -88,9 +88,9 @@ async def handshake_async(self, channels, groups, stop_event, timetoken: int = 0
         request.timetoken(0)
         response = await request.future()
 
-        if isinstance(response, PubNubException):
+        if isinstance(response, Exception):
             self.logger.warning(f'Handshake failed: {str(response)}')
-            handshake_failure = events.HandshakeFailureEvent(str(response), 1, timetoken=timetoken)
+            handshake_failure = events.HandshakeFailureEvent(response, 1, timetoken=timetoken)
             self.event_engine.trigger(handshake_failure)
         elif response.status.error:
             self.logger.warning(f'Handshake failed: {response.status.error_data.__dict__}')
@@ -292,7 +292,7 @@ async def heartbeat(self, channels, groups, stop_event):
             self.logger.warning(f'Heartbeat failed: {str(response)}')
             self.event_engine.trigger(events.HeartbeatFailureEvent(channels=channels, groups=groups,
                                                                    reason=response.status.error_data, attempt=1))
-        elif response.status.error:
+        elif response.status and response.status.error:
             self.logger.warning(f'Heartbeat failed: {response.status.error_data.__dict__}')
             self.event_engine.trigger(events.HeartbeatFailureEvent(channels=channels, groups=groups,
                                                                    reason=response.status.error_data, attempt=1))
@@ -427,5 +427,6 @@ def emit_message(self, invocation: invocations.EmitMessagesInvocation):
     def emit_status(self, invocation: invocations.EmitStatusInvocation):
         pn_status = PNStatus()
         pn_status.category = invocation.status
+        pn_status.operation = invocation.operation
         pn_status.error = False
         self.pubnub._subscription_manager._listener_manager.announce_status(pn_status)

pubnub/event_engine/models/events.py

@@ -102,6 +102,10 @@ class ReconnectEvent(PNEvent):
     pass
 
 
+class UnsubscribeAllEvent(PNEvent):
+    pass
+
+
 """
     Presence Events
 """
@@ -116,7 +120,8 @@ class HeartbeatReconnectEvent(PNEvent):
 
 
 class HeartbeatLeftAllEvent(PNEvent):
-    pass
+    def __init__(self, suppress_leave: bool = False) -> None:
+        self.suppress_leave = suppress_leave
 
 
 class HeartbeatLeftEvent(PNChannelGroupsEvent):

pubnub/event_engine/models/invocations.py

@@ -1,6 +1,6 @@
 from typing import List, Union
 from pubnub.exceptions import PubNubException
-from pubnub.enums import PNStatusCategory
+from pubnub.enums import PNOperationType, PNStatusCategory
 
 
 class PNInvocation:
@@ -90,9 +90,10 @@ def __init__(self, messages: Union[None, List[str]]) -> None:
 
 
 class EmitStatusInvocation(PNEmittableInvocation):
-    def __init__(self, status: Union[None, PNStatusCategory]) -> None:
+    def __init__(self, status: Union[None, PNStatusCategory], operation: Union[None, PNOperationType] = None) -> None:
         super().__init__()
         self.status = status
+        self.operation = operation
 
 
 """