Primitives: Add Broker class.

Author: peterhinch

v3/docs/DRIVERS.md

@@ -26,11 +26,15 @@ MicroPython's `asyncio` when used in a microcontroller context.
   6.1 [Encoder class](./DRIVERS.md#61-encoder-class)  
  7. [Ringbuf Queue](./DRIVERS.md#7-ringbuf-queue) A MicroPython optimised queue primitive.  
  8. [Delay_ms class](./DRIVERS.md#8-delay_ms-class) A flexible retriggerable delay with callback or Event interface.  
- 9. [Additional functions](./DRIVERS.md#9-additional-functions)  
-  9.1 [launch](./DRIVERS.md#91-launch) Run a coro or callback interchangeably.  
-  9.2 [set_global_exception](./DRIVERS.md#92-set_global_exception) Simplify debugging with a global exception handler.  
+ 9. [Message Broker](./DRIVERS.md#9-message-broker) A flexible means of messaging between
+ tasks.  
+  9.1 [Further examples](./DRIVERS.md#91-further-examples)  
+  9.2 [User agents](./DRIVERS.md#92-user-agents)  
+ 10. [Additional functions](./DRIVERS.md#10-additional-functions)  
+  10.1 [launch](./DRIVERS.md#101-launch) Run a coro or callback interchangeably.  
+  10.2 [set_global_exception](./DRIVERS.md#102-set_global_exception) Simplify debugging with a global exception handler.  
 
-###### [Tutorial](./TUTORIAL.md#contents)
+###### [asyncio Tutorial](./TUTORIAL.md#contents)
 
 # 1. Introduction
 
@@ -1126,9 +1130,116 @@ finally:
 ```
 ###### [Contents](./DRIVERS.md#0-contents)
 
-# 9. Additional functions
+# 9. Message Broker
+
+This is under development: please check for updates.
+
+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
+message. This enables one to one, one to many or many to many messaging.
+
+A task subscribes to a topic with an `agent`. This is stored by the broker. When
+the broker publishes a message, the `agent` of each task subscribed to its topic
+will be triggered. In the simplest case the `agent` is a `Queue` instance: the
+broker puts the topic and message onto the subscriber's queue for retrieval.
+
+More advanced agents can perform actions in response to a message, such as
+calling a function or launching a `task`.
+
+Broker methods. All are synchronous, constructor has no args:
+* `subscribe(topic, agent)` Passed `agent` will be triggered by messages with a
+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, in which case data for that queue has been lost.
+
+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)`.
+* `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.
+* `coroutine` Task created when a message is received with 2 args, topic and
+message.
+* `bound coroutine` Task created when a message is received with 2 args, topic
+and message.
+* Instance of a user class. See user agents below.
+* `Event` Set when a message is received.
+
+Note that synchronous `agent` instances must run to completion quickly otherwise
+the `publish` method will be slowed.
+
+The following is a simple example:
+```py
+import asyncio
+from primitives import Broker, Queue
+
+broker = Broker()
+queue = Queue()
+async def sender(t):
+    for x in range(t):
+        await asyncio.sleep(1)
+        broker.publish("foo_topic", f"test {x}")
+
+async def main():
+    broker.subscribe("foo_topic", queue)
+    n = 10
+    asyncio.create_task(sender(n))
+    print("Letting queue part-fill")
+    await asyncio.sleep(5)
+    for _ in range(n):
+        topic, message = await queue.get()
+        print(topic, message)
+
+asyncio.run(main())
+```
+## 9.1 Further examples
+
+An interesting application is to extend MQTT into the Python code
+(see [mqtt_as](https://github.com/peterhinch/micropython-mqtt/tree/master)).
+This is as simple as:
+```py
+async def messages(client):
+    async for topic, msg, retained in client.queue:
+        broker.publish(topic.decode(), msg.decode())
+```
+Assuming the MQTT client is subscribed to multiple topics, message strings are
+directed to individual tasks each supporting one topic.
+
+## 9.2 User agents
+
+An `agent` can be an instance of a user class. The class must be a subclass of
+`Agent`, and it must support a synchronous `.put` method. The latter takes two
+args, being `topic` and `message`. It should run to completion quickly.
+
+```py
+import asyncio
+from primitives import Broker, Agent
+
+broker = Broker()
+class MyAgent(Agent):
+    def put(sef, topic, message):
+        print(f"User agent. Topic: {topic} Message: {message}")
+
+async def sender(t):
+    for x in range(t):
+        await asyncio.sleep(1)
+        broker.publish("foo_topic", f"test {x}")
+
+async def main():
+    broker.subscribe("foo_topic", MyAgent())
+    await sender(10)
+
+asyncio.run(main())
+```
+
+###### [Contents](./DRIVERS.md#0-contents)
+
+# 10. Additional functions
 
-## 9.1 Launch
+## 10.1 Launch
 
 Import as follows:
 ```python
@@ -1140,7 +1251,7 @@ runs it and returns the callback's return value. If a coro is passed, it is
 converted to a `task` and run asynchronously. The return value is the `task`
 instance. A usage example is in `primitives/switch.py`.
 
-## 9.2 set_global_exception
+## 10.2 set_global_exception
 
 Import as follows:
 ```python

v3/primitives/__init__.py

@@ -53,6 +53,8 @@ def _handle_exception(loop, context):
     "RingbufQueue": "ringbuf_queue",
     "Keyboard": "sw_array",
     "SwArray": "sw_array",
+    "Broker": "broker",
+    "Agent": "broker",
 }
 
 # Copied from uasyncio.__init__.py

v3/primitives/broker.py

@@ -0,0 +1,52 @@
+# broker.py A message broker for MicroPython
+
+# Copyright (c) 2024 Peter Hinch
+# Released under the MIT License (MIT) - see LICENSE file
+
+# Inspired by the following
+# https://www.joeltok.com/posts/2021-03-building-an-event-bus-in-python/
+
+import asyncio
+from primitives import Queue, type_coro
+
+
+class Agent:
+    pass
+
+
+class Broker(dict):
+    def subscribe(self, topic, agent):
+        if not self.get(topic, False):
+            self[topic] = {agent}
+        else:
+            self[topic].add(agent)
+
+    def unsubscribe(self, topic, agent):
+        try:
+            self[topic].remove(agent)
+            if len(self[topic]) == 0:
+                del self[topic]
+        except KeyError:
+            pass  # Topic already removed
+
+    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:
+                    agent.put_nowait((topic, message))
+                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/package.json

@@ -3,6 +3,7 @@
     ["primitives/__init__.py", "github:peterhinch/micropython-async/v3/primitives/__init__.py"],
     ["primitives/aadc.py", "github:peterhinch/micropython-async/v3/primitives/aadc.py"],
     ["primitives/barrier.py", "github:peterhinch/micropython-async/v3/primitives/barrier.py"],
+    ["primitives/broker.py", "github:peterhinch/micropython-async/v3/primitives/broker.py"],
     ["primitives/condition.py", "github:peterhinch/micropython-async/v3/primitives/condition.py"],
     ["primitives/delay_ms.py", "github:peterhinch/micropython-async/v3/primitives/delay_ms.py"],
     ["primitives/encoder.py", "github:peterhinch/micropython-async/v3/primitives/encoder.py"],

v3/primitives/tests/broker_test.py

@@ -0,0 +1,101 @@
+# broker_test.py Test various types of subscriber
+
+# import primitives.tests.broker_test
+
+import asyncio
+from primitives import Broker, Queue
+
+broker = Broker()
+
+# Periodically publish messages to two topics
+async def test(t):
+    for x in range(t):
+        await asyncio.sleep(1)
+        broker.publish("foo_topic", f"dogs {x}")
+        broker.publish("bar_topic", f"rats {x}")
+
+
+# Suscribe via coroutine
+async def subs(topic, message):
+    await asyncio.sleep_ms(100)
+    print("coroutine", topic, message)
+
+
+# Subscribe via function
+def func(topic, message):
+    print("function", topic, message)
+
+
+# Subscribe via Event
+
+event = asyncio.Event()
+
+
+async def event_test():
+    while True:
+        await event.wait()
+        event.clear()
+        print("Event triggered")
+
+
+class TestClass:
+    async def fetch_data(self, topic, message):
+        await asyncio.sleep_ms(100)
+        print("bound coro", topic, message)
+
+    def get_data(self, topic, message):
+        print("bound method", topic, message)
+
+
+async def print_queue(q):
+    while True:
+        topic, message = await q.get()
+        print(topic, message)
+
+
+async def main():
+    tc = TestClass()
+    q = Queue(10)
+    print("Subscribing Event, coroutine, Queue 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)
+
+    asyncio.create_task(test(30))  # Publish to topics for 30s
+    asyncio.create_task(event_test())
+    await asyncio.sleep(5)
+    print()
+    print("Unsubscribing coroutine")
+    broker.unsubscribe("bar_topic", subs)
+    await asyncio.sleep(5)
+    print()
+    print("Unsubscribing Event")
+    broker.unsubscribe("bar_topic", event)
+    print()
+    print("Subscribing function")
+    broker.subscribe("bar_topic", func)
+    await asyncio.sleep(5)
+    print()
+    print("Unsubscribing function")
+    broker.unsubscribe("bar_topic", func)
+    print()
+    print("Unsubscribing bound coroutine")
+    broker.unsubscribe("foo_topic", tc.fetch_data)  # Async method
+    print()
+    print("Subscribing method")
+    broker.subscribe("foo_topic", tc.get_data)  # Sync method
+    await asyncio.sleep(5)
+    print()
+    print("Unsubscribing method")
+    broker.unsubscribe("foo_topic", tc.get_data)  # Async method
+    print("Pause 5s")
+    await asyncio.sleep(5)
+    print("Retrieving foo_topic messages from queue")
+    try:
+        await asyncio.wait_for(print_queue(q), 5)
+    except asyncio.TimeoutError:
+        print("Done")
+
+
+asyncio.run(main())