broker: add topic exchange

Author: scailer

examples/push_agent.py

@@ -0,0 +1,59 @@
+# mypy: ignore-errors
+import asyncio
+import logging
+import os
+import sys
+
+from microagent import MicroAgent, consumer, load_stuff, on, periodic
+from microagent.tools import amqp, redis
+
+logging.basicConfig(format=(
+    '%(levelname)-8s [pid#%(process)d] %(asctime)s %(name)s '
+    '%(filename)s:%(lineno)d %(message)s'
+), stream=sys.stdout, level=logging.INFO)
+
+cur_dir = os.path.dirname(os.path.realpath(__file__))
+signals, queues = load_stuff('file://' + os.path.join(cur_dir, 'signals.json'))
+
+
+class UserAgent(MicroAgent):
+    @on('pre_start')
+    def setup(self):
+        self.log.info('Run ...\n %s', self.info())
+
+    @periodic(period=15, timeout=10, start_after=3)
+    async def example_periodic_task_send_push(self):
+        self.log.info('Run periodic task')
+        await self.broker.push3.send({'text': 'informer text'}, topic='msg.ios')
+        await self.broker.push.send({'text': 'informer 1'})
+
+    @consumer(queues.android)
+    async def example_read_queue_android(self, **kwargs):
+        self.log.info('Catch android %s', kwargs)
+
+    @consumer(queues.ios)
+    async def example_read_queue_ios(self, **kwargs):
+        self.log.info('Catch ios %s', kwargs)
+
+    @consumer(queues.android_a)
+    async def example_read_queue_android_a(self, **kwargs):
+        self.log.info('Catch android_a %s', kwargs)
+
+    @consumer(queues.ios_a)
+    async def example_read_queue_ios_a(self, **kwargs):
+        self.log.info('Catch ios_a %s', kwargs)
+
+
+async def main():
+    bus = redis.RedisSignalBus('redis://localhost/7')
+    broker = amqp.AMQPBroker('amqp://guest:guest@localhost:5672/')
+    await broker.push3.send({'q':1}, topic='msg.android')
+
+    agent = UserAgent(bus=bus, broker=broker)
+    await agent.start()
+
+    while True:
+        await asyncio.sleep(1)
+
+
+asyncio.run(main())

examples/signals.json

@@ -11,6 +11,10 @@
         }}
     ],
     "queues": [
-        {"name": "mailer"}
+        {"name": "mailer"},
+        {"name": "android", "exchange": "push3", "topics": ["*.android"]},
+        {"name": "ios", "exchange": "push3", "topics": ["*.ios"]},
+        {"name": "android_a", "exchange": "push"},
+        {"name": "ios_a", "exchange": "push"}
     ]
 }

microagent/__init__.py

@@ -3,7 +3,6 @@
 import importlib
 import json
 import urllib.request
-
 from collections import abc
 from typing import Any, NamedTuple
 
@@ -16,7 +15,6 @@
 from .timer import CRONArgs, CRONTask, PeriodicArgs, PeriodicTask, cron_parser
 from .utils import make_bound_key
 
-
 __all__ = ['Signal', 'Queue', 'MicroAgent', 'ServerInterrupt', 'receiver', 'consumer',
            'periodic', 'cron', 'on', 'load_stuff', 'load_signals', 'load_queues']
 
@@ -58,10 +56,18 @@ def load_stuff(source: str) -> tuple[Any, Any]:
             type_map = {name: get_types(_type) for name, _type in providing_args.items()}
             providing_args = list(providing_args)
 
-        Signal(name=_data['name'], providing_args=providing_args, type_map=type_map)
+        Signal(
+            name=_data['name'],
+            providing_args=providing_args,
+            type_map=type_map
+        )
 
     for _data in data.get('queues', []):
-        Queue(name=_data['name'], exchange=_data.get('exchange', ''))
+        Queue(
+            name=_data['name'],
+            exchange=_data.get('exchange', ''),
+            topics=_data.get('topics', [])
+        )
 
     if data.get('jsonlib'):
         jsonlib = importlib.import_module(data['jsonlib'])  # type: ignore

microagent/abc.py

@@ -1,9 +1,7 @@
 import logging
-
 from collections.abc import Awaitable, Callable
 from typing import Any, Literal, Protocol
 
-
 HookLabel = Literal['server', 'pre_start', 'post_start', 'pre_stop']
 HookFunc = Callable[[Any], Awaitable[None]]
 PeriodicFunc = Callable[[Any], Awaitable[None]]

microagent/agent.py

@@ -77,7 +77,6 @@ async def setup(self):
 
 import asyncio
 import logging
-
 from collections.abc import Callable, Iterable
 from dataclasses import dataclass, field
 from datetime import datetime, timedelta, timezone
@@ -90,7 +89,6 @@ async def setup(self):
 from .signal import Receiver
 from .timer import CRONTask, PeriodicTask
 
-
 HandlerTypes = TypeVar('HandlerTypes', Receiver, Consumer, PeriodicTask, CRONTask, Hook)
 
 

microagent/broker.py

@@ -47,7 +47,6 @@ async def example_read_queue(self, **kwargs):
 '''
 import logging
 import uuid
-
 from abc import abstractmethod
 from dataclasses import dataclass, field
 from typing import Any

microagent/bus.py

@@ -49,7 +49,6 @@ async def example(self, user_id, **kwargs):
 import logging
 import time
 import uuid
-
 from abc import abstractmethod
 from collections import abc, defaultdict
 from dataclasses import dataclass, field

microagent/hooks.py

@@ -24,14 +24,12 @@
 '''
 
 import inspect
-
 from collections import abc, defaultdict
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, ClassVar, TypedDict
 
 from .abc import BoundKey, HookFunc
 
-
 if TYPE_CHECKING:
     from .agent import MicroAgent
 

microagent/launcher.py

@@ -23,13 +23,11 @@
 import os
 import signal
 import time
-
 from collections.abc import Iterator
 from functools import partial
 from itertools import chain
 from typing import TYPE_CHECKING, Any
 
-
 if TYPE_CHECKING:
     from .agent import MicroAgent
     from .broker import AbstractQueueBroker

microagent/queue.py

@@ -1,12 +1,10 @@
 import json
-
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from types import ModuleType
 from typing import TYPE_CHECKING, ClassVar, TypedDict
 
 from .abc import BoundKey, ConsumerFunc
 
-
 if TYPE_CHECKING:
     from .agent import MicroAgent
 
@@ -41,9 +39,10 @@ class Queue:
 
             {
                 "queues": [
-                    {"name": "mailer"},
-                    {"name": "pusher", "exchange": "events"},
+                    {"name": "mailer"},  # simple
+                    {"name": "pusher", "exchange": "events"},  # fanout
                     {"name": "logger", "exchange": "events"},
+                    {"name": "save", "exchange": "msg", "topics": "m.*"},  # topics
                 ]
             }
 
@@ -57,6 +56,7 @@ class Queue:
     '''
     name: str
     exchange: str = ''
+    topics: list[str] = field(default_factory=list)
 
     _queues: ClassVar[dict[str, 'Queue']] = {}
     _exchanges: ClassVar[dict[str, 'Queue']] = {}