Added async/sync queue listener implementations; improved tests; minor code tweaks.

Author: chsou

CHANGELOG.md

@@ -1,10 +1,13 @@
 # Changelog
 
- * Updated and rewrote Notification 2.0 listener implementation. Added additional parameters for more control: 
-   `consumer_name`, `shared`, `auto_ack` and `auto_unsubscribe`. Added `unsubscribe` function for removing
-   subscribers on demand. Both `AsyncListener` and `Listener` now provide consistent `start`/`stop` functions
-   which take care of coroutine and thread creation. The `listen ` function can still be invoked directly if
-   necessary. 
+* Added `QueueListener` and `AsyncQueueListener` classes to the Notification 2.0 toolkit. These pre-defined
+  listener implementation append new notifications to standard queues that can be monitored/listened to which 
+  makes Notification 2.0 solutions even simpler to implement.
+* Updated and rewrote Notification 2.0 listener implementation. Added additional parameters for more control: 
+  `consumer_name`, `shared`, `auto_ack` and `auto_unsubscribe`. Added `unsubscribe` function for removing
+  subscribers on demand. Both `AsyncListener` and `Listener` now provide consistent `start`/`stop` functions
+  which take care of coroutine and thread creation. The `listen ` function can still be invoked directly if
+  necessary. 
 
 ## Version 3.3.0
 

c8y_tk/notification2/__init__.py

@@ -2,4 +2,4 @@
 
 from c8y_tk.notification2.listener import *
 
-__all__ = ['Listener', 'AsyncListener']
+__all__ = ['Listener', 'AsyncListener', 'AsyncQueueListener', 'QueueListener']

c8y_tk/notification2/listener.py

@@ -8,6 +8,7 @@
 import threading
 import uuid
 from itertools import count
+import queue as sync_queue
 from typing import Callable, Awaitable
 
 import certifi
@@ -231,7 +232,7 @@ async def _callback(msg):
                 self._is_connected.clear()
                 if self._connection:
                     with contextlib.suppress(Exception):
-                        await self._connection.close()   # TODO: add code and reason
+                        await self._connection.close()
                 self._connection = None
 
         self._is_running.clear()
@@ -586,7 +587,7 @@ def wait(self, timeout=None) -> bool:
             timeout (float):  Timeout in seconds.
 
         Returns:
-            True if the listener has stopped (before timeout), False otherwise.
+            Whether the listener has stopped (before timeout).
         """
         self._thread.join(timeout=timeout)
         return not self._thread.is_alive()
@@ -621,3 +622,101 @@ def ack(self, msg_id: str = None, payload: str = None) -> None:
         """
         # assuming that we are already listening ...
         asyncio.run_coroutine_threadsafe(self._listener.ack(msg_id, payload), self._event_loop)
+
+
+class AsyncQueueListener(object):
+    """Special listener implementation which pushes notification messages
+    into a standard (async) queue which can be monitored and read."""
+
+    def __init__(
+            self,
+            c8y: CumulocityApi,
+            subscription_name: str,
+            subscriber_name: str = None,
+            consumer_name: str = None,
+            shared: bool = False,
+            auto_unsubscribe: bool = True,
+            queue: asyncio.Queue = None
+    ):
+        self.queue = queue or asyncio.Queue()
+        self.listener = AsyncListener(
+            c8y=c8y,
+            subscription_name=subscription_name,
+            subscriber_name=subscriber_name,
+            consumer_name=consumer_name,
+            shared=shared,
+            auto_ack=True,
+            auto_unsubscribe=auto_unsubscribe,
+        )
+
+    def start(self):
+        """Start the listener."""
+        async def push_message(msg: AsyncListener.Message):
+            self.queue.put_nowait(msg)
+
+        self.listener.start(push_message)
+
+    def stop(self):
+        """Stop the listener."""
+        self.listener.stop()
+
+    async def wait(self, timeout=None):
+        """Wait for the listener task to finish.
+
+        Args:
+            timeout (int): The number of seconds to wait for the listener
+                to finish. The listener will be cancelled if the timeout
+                occurs.
+        """
+        await self.listener.wait(timeout=timeout)
+
+
+class QueueListener(object):
+    """Special listener implementation which pushes notification messages
+    into a standard (sync) queue which can be monitored and read."""
+
+    def __init__(
+            self,
+            c8y: CumulocityApi,
+            subscription_name: str,
+            subscriber_name: str = None,
+            consumer_name: str = None,
+            shared: bool = False,
+            auto_unsubscribe: bool = True,
+            queue: sync_queue.Queue = None
+    ):
+        self.queue = queue or sync_queue.Queue()
+        self.listener = Listener(
+            c8y=c8y,
+            subscription_name=subscription_name,
+            subscriber_name=subscriber_name,
+            consumer_name=consumer_name,
+            shared=shared,
+            auto_ack=True,
+            auto_unsubscribe=auto_unsubscribe,
+        )
+
+    def start(self):
+        """Start the listener."""
+
+        def push_message(msg: AsyncListener.Message):
+            self.queue.put(msg)
+
+        self.listener.start(push_message)
+
+    def stop(self):
+        """Stop the listener."""
+        self.listener.stop()
+
+    def wait(self, timeout=None) -> bool:
+        """Wait for the listener task to finish.
+
+        Args:
+            timeout (int): The number of seconds to wait for the listener
+                to finish. The listener will be cancelled if the timeout
+                occurs.
+
+        Returns:
+            Whether the listener has stopped (before timeout).
+        """
+        return self.listener.wait(timeout=timeout)

integration_tests/test_notification2.py

@@ -13,9 +13,9 @@
 import pytest
 
 from c8y_api import CumulocityApi
-from c8y_api.model import Device, ManagedObject, Subscription, Measurement, Value, Event, Alarm
-from c8y_tk.notification2 import AsyncListener, Listener
-from tests.utils import assert_in_any, assert_no_failures, assert_all_in
+from c8y_api.model import Device, ManagedObject, Subscription, Measurement, Value, Event, Alarm, Operation
+from c8y_tk.notification2 import AsyncListener, Listener, AsyncQueueListener, QueueListener
+from tests.utils import assert_in_any, assert_no_failures
 
 from util.testing_util import RandomNameGenerator
 
@@ -86,9 +86,8 @@ def build():
     return build
 
 
-# TODO: Add Operation
 @pytest.mark.parametrize("api_filters, expected", [
-    ('*', 'M,E,EwC,A,AwC,MO'),
+    ('*', 'M,E,EwC,A,AwC,MO,O'),
     ('M', 'M'),
     ('E', 'E'),
     ('EwC', 'E,EwC'),
@@ -126,6 +125,7 @@ def test_api_filters(live_c8y: CumulocityApi, sample_object, api_filters, expect
 
     mo = sample_object
     mo['c8y_IsDevice'] = {}
+    mo['com_cumulocity_model_Agent'] = {}
     mo.update()
     sub = Subscription(
         live_c8y,
@@ -153,7 +153,7 @@ def receive_notification(m:Listener.Message):
         e_id = Event(live_c8y, source=mo.id, type="c8y_TestEvent", time='now', text='text').create().id
         a_id = Alarm(live_c8y, source=mo.id, type="c8y_TestAlarm", time='now', text='text',
                      severity=Alarm.Severity.WARNING).create().id
-        # o_id = Operation(live_c8y, device_id=mo.id, c8y_Operation={}).create().id
+        o_id = Operation(live_c8y, device_id=mo.id, c8y_Operation={}).create().id
         mo.apply({'some_tag': {}})
 
         time.sleep(1)
@@ -169,6 +169,8 @@ def receive_notification(m:Listener.Message):
             assert e_id in ids
         if 'alarms' in expected:
             assert a_id in ids
+        if 'operations' in expected:
+            assert o_id in ids
     finally:
         # (99) cleanup
         listener.stop()
@@ -666,3 +668,51 @@ async def receive_notification(m:AsyncListener.Message):
             == (n_listeners-1 if shared else 0))
     assert sum("cancelled" in x for x in log_messages) == n_listeners
     assert_no_failures(caplog)
+
+
+@pytest.mark.asyncio(loop_scope='function')
+async def test_asyncio_queue_listener(live_c8y: CumulocityApi, sample_object):
+    """Verify that the queue listener works as expected."""
+    mo = sample_object
+    sub = create_managed_object_subscription(live_c8y, mo)
+
+    q = asyncio.Queue()
+    listener = AsyncQueueListener(
+        c8y=live_c8y,
+        subscription_name=sub.name,
+        queue=q,
+    )
+
+    listener.start()
+    try:
+        await asyncio.sleep(5)  # ensure creation
+        mo.apply({'test_CustomFragment': {'num': 42}})
+        msg = await q.get()
+        assert msg.json['test_CustomFragment']['num'] == 42
+    finally:
+        listener.stop()
+        await listener.wait()
+
+
+@pytest.mark.asyncio(loop_scope='function')
+async def test_queue_listener(live_c8y: CumulocityApi, sample_object):
+    """Verify that the queue listener works as expected."""
+    mo = sample_object
+    sub = create_managed_object_subscription(live_c8y, mo)
+
+    q = queue.Queue()
+    listener = QueueListener(
+        c8y=live_c8y,
+        subscription_name=sub.name,
+        queue=q,
+    )
+
+    listener.start()
+    try:
+        time.sleep(5)  # ensure creation
+        mo.apply({'test_CustomFragment': {'num': 42}})
+        msg = q.get()
+        assert msg.json['test_CustomFragment']['num'] == 42
+    finally:
+        listener.stop()
+        listener.wait()

samples/notification2_synchronous.py

@@ -2,7 +2,6 @@
 
 # pylint: disable=missing-function-docstring
 
-import threading
 import time
 
 from c8y_api.app import SimpleCumulocityApp