broker: Prior to adding args.

Author: peterhinch

v3/docs/DRIVERS.md

@@ -1134,7 +1134,9 @@ finally:
 
 This is under development: please check for updates. See
 [code](https://github.com/peterhinch/micropython-async/blob/master/v3/primitives/broker.py).
-
+```py
+from primitives import Broker
+```
 The `Broker` class provides a flexible means of messaging between running tasks.
 It uses a publish-subscribe model (akin to MQTT) whereby the transmitting task
 publishes to a topic. Any tasks subscribed to that topic will receive the
@@ -1153,13 +1155,13 @@ Broker methods. All are synchronous, constructor has no args:
 matching `topic`.
 * `unsubscribe(topic, agent)` The `agent` will stop being triggered.
 * `publish(topic, message)` All `agent` instances subscribed to `topic` will be
-triggered, receiving `topic` and `message` args. Returns `True` unless a `Queue`
-agent has become full. A `False` value indicates that at least one message has
-been lost.
+triggered, receiving `topic` and `message` args. The method is not threadsafe;
+it should not be called from a hard ISR or from another thread.
 
 The `topic` arg is typically a string but may be any hashable object. A
 `message` is an arbitrary Python object. An `agent` may be any of the following:
 * `Queue` When a message is received receives 2-tuple `(topic, message)`.
+* `RingbufQueue` When a message is received receives 2-tuple `(topic, message)`.
 * `function` Called when a message is received. Gets 2 args, topic and message.
 * `bound method` Called when a message is received. Gets 2 args, topic and
 message.
@@ -1179,7 +1181,7 @@ import asyncio
 from primitives import Broker, Queue
 
 broker = Broker()
-queue = Queue()
+queue = Queue()  # Or (e.g. RingbufQueue(20))
 async def sender(t):
     for x in range(t):
         await asyncio.sleep(1)

v3/primitives/broker.py

@@ -7,7 +7,7 @@
 # https://www.joeltok.com/posts/2021-03-building-an-event-bus-in-python/
 
 import asyncio
-from primitives import Queue, type_coro
+from primitives import Queue, RingbufQueue, type_coro
 
 
 class Agent:
@@ -31,22 +31,20 @@ def unsubscribe(self, topic, agent):
 
     def publish(self, topic, message):
         agents = self.get(topic, [])
-        result = True
         for agent in agents:
             if isinstance(agent, asyncio.Event):
                 agent.set()
                 continue
             if isinstance(agent, Agent):  # User class
                 agent.put(topic, message)  # Must support .put
                 continue
-            if isinstance(agent, Queue):
-                if agent.full():
-                    result = False
-                else:
+            if isinstance(agent, Queue) or isinstance(agent, RingbufQueue):
+                try:
                     agent.put_nowait((topic, message))
+                except Exception:  # TODO
+                    pass
                 continue
             # agent is function, method, coroutine or bound coroutine
             res = agent(topic, message)
             if isinstance(res, type_coro):
                 asyncio.create_task(res)
-        return result

v3/primitives/tests/broker_test.py

@@ -3,7 +3,7 @@
 # import primitives.tests.broker_test
 
 import asyncio
-from primitives import Broker, Queue
+from primitives import Broker, Queue, RingbufQueue
 
 broker = Broker()
 
@@ -56,11 +56,13 @@ async def print_queue(q):
 async def main():
     tc = TestClass()
     q = Queue(10)
-    print("Subscribing Event, coroutine, Queue and bound coroutine.")
+    rq = RingbufQueue(10)
+    print("Subscribing Event, coroutine, Queue, RingbufQueue and bound coroutine.")
     broker.subscribe("foo_topic", tc.fetch_data)  # Bound coroutine
     broker.subscribe("bar_topic", subs)  # Coroutine
     broker.subscribe("bar_topic", event)
     broker.subscribe("foo_topic", q)
+    broker.subscribe("bar_topic", rq)
 
     asyncio.create_task(test(30))  # Publish to topics for 30s
     asyncio.create_task(event_test())
@@ -91,11 +93,14 @@ async def main():
     broker.unsubscribe("foo_topic", tc.get_data)  # Async method
     print("Pause 5s")
     await asyncio.sleep(5)
-    print("Retrieving foo_topic messages from queue")
+    print("Retrieving foo_topic messages from Queue")
     try:
         await asyncio.wait_for(print_queue(q), 5)
     except asyncio.TimeoutError:
-        print("Done")
+        print("Timeout")
+    print("Retrieving bar_topic messages from RingbufQueue")
+    async for topic, message in rq:
+        print(topic, message)
 
 
 asyncio.run(main())