Presence engine - states and events (#178)

* Presence engine - states and events

* Fixes and effects

* Add missing presence_enable switch

* Add ee param to heartbeat, subscribe and leave endpoints

* Presence engine, refactors and everything else including meaning of life

* Add pnpres to channel groups

* rename delayed effects

* Removed code repetitions + improved behave tests

* Fixed Heartbeat Retry

* Fixes

* Rename effects to invocations

* Add support for subscription state

* Fix leaving channels

* Added example

* Fixes in example

* PubNub SDK v7.4.0 release.

---------

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

Author: Sebastian Molenda

.github/workflows/run-tests.yml

@@ -78,6 +78,7 @@ jobs:
           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
 
           sudo pip3 install -r requirements-dev.txt
           behave --junit tests/acceptance/pam

.pubnub.yml

@@ -1,5 +1,5 @@
 name: python
-version: 7.3.2
+version: 7.4.0
 schema: 1
 scm: github.com/pubnub/python
 sdks:
@@ -18,7 +18,7 @@ sdks:
         distributions:
           - distribution-type: library
             distribution-repository: package
-            package-name: pubnub-7.3.2
+            package-name: pubnub-7.4.0
             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.3.2
-            location: https://github.com/pubnub/python/releases/download/v7.3.2/pubnub-7.3.2.tar.gz
+            package-name: pubnub-7.4.0
+            location: https://github.com/pubnub/python/releases/download/v7.4.0/pubnub-7.4.0.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-02-08
+    version: v7.4.0
+    changes:
+      - type: feature
+        text: "Optional Event Engine for Subscribe Loop."
   - date: 2023-11-27
     version: v7.3.2
     changes:

CHANGELOG.md

@@ -1,3 +1,9 @@
+## v7.4.0
+February 08 2024
+
+#### Added
+- Optional Event Engine for Subscribe Loop.
+
 ## v7.3.2
 November 27 2023
 

examples/cli_chat.py

@@ -0,0 +1,63 @@
+import argparse
+import asyncio
+
+from os import getenv
+from pubnub.callbacks import SubscribeCallback
+from pubnub.pubnub_asyncio import EventEngineSubscriptionManager, PubNubAsyncio
+from pubnub.pnconfiguration import PNConfiguration
+
+parser = argparse.ArgumentParser(description="Chat with others using PubNub")
+parser.add_argument("-n", metavar="name", help="Your name", default=None, required=False)
+parser.add_argument("-c", metavar="channel", help="The channel you want to join", default=None, required=False)
+args = parser.parse_args()
+
+
+class ExampleCallback(SubscribeCallback):
+    def message(self, pubnub, message):
+        print(f"{message.publisher}> {message.message}\n")
+
+    def presence(self, pubnub, presence):
+        print(f"-- {presence.uuid} {'joined' if presence.event == 'join' else 'left'} \n")
+
+    def status(self, pubnub, status):
+        if status.is_error():
+            print(f"! Error: {status.error_data}")
+        else:
+            print(f"* Status: {status.category.name}")
+
+
+async def async_input():
+    print()
+    await asyncio.sleep(0.1)
+    return (await asyncio.get_event_loop().run_in_executor(None, input))
+
+
+async def main():
+    name = args.name if hasattr(args, "name") else input("Enter your name: ")
+    channel = args.channel if hasattr(args, "channel") else input("Enter the channel you want to join: ")
+
+    print("Welcome to the chat room. Type 'exit' to leave the chat.")
+
+    config = PNConfiguration()
+    config.subscribe_key = getenv("PN_KEY_SUBSCRIBE")
+    config.publish_key = getenv("PN_KEY_PUBLISH")
+    config.uuid = name
+
+    pubnub = PubNubAsyncio(config, subscription_manager=EventEngineSubscriptionManager)
+    pubnub.add_listener(ExampleCallback())
+
+    pubnub.subscribe().channels(channel).with_presence().execute()
+
+    while True:
+        message = await async_input()
+        print("\x1b[2K")
+        if message == "exit":
+            print("Goodbye!")
+            break
+
+        await pubnub.publish().channel(channel).message(message).future()
+
+
+if __name__ == '__main__':
+    loop = asyncio.get_event_loop()
+    loop.run_until_complete(main())

pubnub/dtos.py

@@ -10,6 +10,18 @@ def __init__(self, channels=None, channel_groups=None, presence_enabled=None, ti
         self.presence_enabled = presence_enabled
         self.timetoken = timetoken
 
+    @property
+    def channels_with_pressence(self):
+        if not self.presence_enabled:
+            return self.channels
+        return self.channels + [ch + '-pnpres' for ch in self.channels]
+
+    @property
+    def groups_with_pressence(self):
+        if not self.presence_enabled:
+            return self.channel_groups
+        return self.channel_groups + [ch + '-pnpres' for ch in self.channel_groups]
+
 
 class UnsubscribeOperation(object):
     def __init__(self, channels=None, channel_groups=None):
@@ -19,6 +31,18 @@ def __init__(self, channels=None, channel_groups=None):
         self.channels = channels
         self.channel_groups = channel_groups
 
+    def get_subscribed_channels(self, channels, with_presence=False) -> list:
+        result = [ch for ch in channels if ch not in self.channels and not ch.endswith('-pnpres')]
+        if not with_presence:
+            return result
+        return result + [ch + '-pnpres' for ch in result]
+
+    def get_subscribed_channel_groups(self, channel_groups, with_presence=False) -> list:
+        result = [grp for grp in channel_groups if grp not in self.channel_groups and not grp.endswith('-pnpres')]
+        if not with_presence:
+            return result
+        return result + [grp + '-pnpres' for grp in result]
+
 
 class StateOperation(object):
     def __init__(self, channels=None, channel_groups=None, state=None):

pubnub/endpoints/presence/heartbeat.py

@@ -52,6 +52,9 @@ def custom_params(self):
         if self._state is not None and len(self._state) > 0:
             params['state'] = utils.url_write(self._state)
 
+        if hasattr(self.pubnub, '_subscription_manager'):
+            params.update(self.pubnub._subscription_manager.get_custom_params())
+
         return params
 
     def create_response(self, envelope):

pubnub/endpoints/presence/leave.py

@@ -36,6 +36,9 @@ def custom_params(self):
         if len(self._groups) > 0:
             params['channel-group'] = utils.join_items(self._groups)
 
+        if hasattr(self.pubnub, '_subscription_manager'):
+            params.update(self.pubnub._subscription_manager.get_custom_params())
+
         return params
 
     def build_path(self):

pubnub/endpoints/pubsub/subscribe.py

@@ -18,6 +18,7 @@ def __init__(self, pubnub):
         self._filter_expression = None
         self._timetoken = None
         self._with_presence = None
+        self._state = None
 
     def channels(self, channels):
         utils.extend_list(self._channels, channels)
@@ -44,6 +45,10 @@ def region(self, region):
 
         return self
 
+    def state(self, state):
+        self._state = state
+        return self
+
     def http_method(self):
         return HttpMethod.GET
 
@@ -75,6 +80,12 @@ def custom_params(self):
         if not self.pubnub.config.heartbeat_default_values:
             params['heartbeat'] = self.pubnub.config.presence_timeout
 
+        if self._state is not None and len(self._state) > 0:
+            params['state'] = utils.url_write(self._state)
+
+        if hasattr(self.pubnub, '_subscription_manager'):
+            params.update(self.pubnub._subscription_manager.get_custom_params())
+
         return params
 
     def create_response(self, envelope):

pubnub/enums.py

@@ -19,7 +19,7 @@ def string(cls, method):
             return "PATCH"
 
 
-class PNStatusCategory(object):
+class PNStatusCategory(Enum):
     PNUnknownCategory = 1
     PNAcknowledgmentCategory = 2
     PNAccessDeniedCategory = 3

pubnub/event_engine/containers.py

@@ -0,0 +1,15 @@
+class PresenceStateContainer:
+    channel_states: dict
+
+    def __init__(self):
+        self.channel_states = {}
+
+    def register_state(self, state: dict, channels: list):
+        for channel in channels:
+            self.channel_states[channel] = state
+
+    def get_state(self, channels: list):
+        return {channel: self.channel_states[channel] for channel in channels if channel in self.channel_states}
+
+    def get_channels_states(self, channels: list):
+        return {channel: self.channel_states[channel] for channel in channels if channel in self.channel_states}