Updated/rewrote toolkit listener implementation.

Author: chsou

samples/notification2_asynchronous.py

@@ -29,13 +29,17 @@ async def main():
     print(f"Subscription created: {sub_name}")
 
     # Create a listener for previously created subscription
-    listener = AsyncListener(c8y, sub.name)
+    listener = AsyncListener(
+        c8y,
+        subscription_name=sub.name,
+        auto_ack=True,
+        auto_unsubscribe=True,
+    )
 
     # Define callback function.
     # This function is invoked (asynchronously) for each received event.
     async def callback(msg:AsyncListener.Message):
         print(f"Received message, ID: {msg.id}, Source: {msg.source}, Action: {msg.action}, Body: {msg.json}")
-        await msg.ack()
 
     # Start listening
     listener.start(callback)
@@ -48,7 +52,7 @@ async def callback(msg:AsyncListener.Message):
     # The update event is now being processed
     await asyncio.sleep(5)
 
-    # close the listener and wait for it to end.
+    # stop the listener and wait for it to end.
     listener.stop()
     await listener.wait()
 

samples/notification2_synchronous.py

@@ -32,18 +32,21 @@
 print(f"Subscription created: {sub_name}")
 
 # Create a listener for previously created subscription
-listener = Listener(c8y, sub.name)
+listener = Listener(
+    c8y,
+    subscription_name=sub.name,
+    auto_ack=False,
+    auto_unsubscribe=False,
+)
 
 # Define callback function.
 # This function is invoked (synchronously) for each received event.
 def callback(msg: Listener.Message):
     print(f"Received message, ID: {msg.id}, Source: {msg.source}, Action: {msg.action}, Body: {msg.json}")
     msg.ack()
 
-# Wrap listen function into separate thread
-# and start listening
-listener_thread = threading.Thread(target=listener.listen, args=[callback])
-listener_thread.start()
+# Start listening, thread is handled automatically
+listener.start(callback)
 
 # Some action: Update the managed object
 time.sleep(5)
@@ -53,9 +56,9 @@ def callback(msg: Listener.Message):
 # The update event is now being processed
 time.sleep(5)
 
-# close the listener
+# stop the listener and wait for completion
 listener.stop()
-listener_thread.join()
+listener.wait()
 
 # cleanup subscription and managed object
 sub.delete()