Broker: add wildcard subscriptions.

Author: peterhinch

v3/docs/DRIVERS.md

@@ -26,11 +26,11 @@ 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. [Message Broker](./DRIVERS.md#9-message-broker) A flexible means of messaging between
- tasks.  
+ 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) User defined Agent classes.  
-  9.3 [Notes](./DRIVERS.md#93-notes)  
+  9.3 [Wildcard subscriptions](./DRIVERS.md#93-wildcard-subscriptions)  
+  9.4 [Notes](./DRIVERS.md#9-notes)  
  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.  
@@ -1332,7 +1332,21 @@ async def main():
 
 asyncio.run(main())
 ```
-## 9.3 Notes
+## 9.3 Wildcard subscriptions
+
+In the case of publications whose topics are strings, a single call to
+`.subscribe` can subscribe an `agent` to multiple topics. This is by wildcard
+matching. By default exact matching is used, however this can be changed to use
+regular expressions as in this code fragment:
+```py
+from primitives import Broker, RegExp
+broker.subscribe(RegExp(".*_topic"), some_agent)
+```
+In this case `some_agent` would be triggered by publications to `foo_topic` or
+`bar_topic` because the string `".*_topic"` matches these by the rules of
+regular expressions.
+
+## 9.4 Notes
 
 #### The publish/subscribe model
 

v3/primitives/__init__.py

@@ -55,6 +55,7 @@ def _handle_exception(loop, context):
     "SwArray": "sw_array",
     "Broker": "broker",
     "Agent": "broker",
+    "RegExp": "broker",
 }
 
 # Copied from asyncio.__init__.py

v3/primitives/broker.py

@@ -8,12 +8,21 @@
 
 import asyncio
 from primitives import Queue, RingbufQueue, type_coro
+import re
 
 
 class Agent:
     pass
 
 
+class RegExp:
+    def __init__(self, re_str):
+        self.re = re.compile(re_str)
+
+    def matching(self, topic):
+        return re.match(self.re, topic) is not None
+
+
 def _validate(a):
     return (
         isinstance(a, asyncio.Event)
@@ -50,7 +59,13 @@ def unsubscribe(self, topic, agent, *args):
             print(f"Unsubscribe topic {topic} fail: topic not subscribed.")
 
     def publish(self, topic, message):
-        agents = self.get(topic, [])
+        agents = set()  # Agents which are triggered by this topic
+        if isinstance(topic, str):  # Check regexps
+            # Are any keys RegExp instances?
+            for regexp in [k for k in self.keys() if isinstance(k, RegExp)]:
+                if regexp.matching(topic):
+                    agents.update(self[regexp])  # Append matching agents
+        agents.update(self.get(topic, []))  # Exact match
         for agent, args in agents:
             if isinstance(agent, asyncio.Event):
                 agent.set()

v3/primitives/tests/broker_test.py

@@ -3,7 +3,7 @@
 # import primitives.tests.broker_test
 
 import asyncio
-from primitives import Broker, Queue, RingbufQueue
+from primitives import Broker, Queue, RingbufQueue, RegExp
 
 broker = Broker()
 
@@ -49,12 +49,13 @@ def get_data(self, topic, message):
 
 async def print_queue(q):
     while True:
-        topic, message = await q.get()
+        topic, message = await asyncio.wait_for(q.get(), 2)
         print(topic, message)
 
 
 async def print_ringbuf_q(q):
-    async for topic, message, args in q:
+    while True:
+        topic, message, args = await asyncio.wait_for(q.get(), 2)
         print(topic, message, args)
 
 
@@ -98,20 +99,19 @@ async def main():
     print("Unsubscribing method")
     broker.unsubscribe("foo_topic", tc.get_data)  # Async method
     print("Retrieving foo_topic messages from Queue")
-    try:
-        await asyncio.wait_for(print_queue(q), 5)
-    except asyncio.TimeoutError:
-        print("Timeout")
     print("Retrieving bar_topic messages from RingbufQueue")
-    try:
-        await asyncio.wait_for(print_ringbuf_q(rq), 5)
-    except asyncio.TimeoutError:
-        print("Timeout")
+    await asyncio.gather(print_queue(q), print_ringbuf_q(rq), return_exceptions=True)
+    # Queues are now empty
     print()
+    print("*** Unsubscribing queues ***")
+    broker.unsubscribe("foo_topic", q)
+    broker.unsubscribe("bar_topic", rq, "args", "added")
+    print()
+
     print("*** Testing error reports and exception ***")
     print()
     Broker.Verbose = True
-    print("*** Check error on invalid unsubscribe ***")
+    print("*** Produce warning messages on invalid unsubscribe ***")
     broker.unsubscribe("rats", "more rats")  # Invalid topic
     broker.unsubscribe("foo_topic", "rats")  # Invalid agent
     print("*** Check exception on invalid subscribe ***")
@@ -120,6 +120,12 @@ async def main():
         print("Test FAIL")
     except ValueError:
         print("Test PASS")
+    print()
+    print("*** Test wildcard subscribe ***")
+    broker.subscribe(RegExp(".*_topic"), func)
+    broker.publish("FAIL", func)  # No match
+    asyncio.create_task(test(5))
+    await asyncio.sleep(10)
 
 
 asyncio.run(main())