A global config and setup method

README.md

@@ -1,15 +1,18 @@
 <!-- License Badge -->
 [![License](https://img.shields.io/badge/License-Apache_2.0-blue.svg)](https://github.com/ansible/dispatcher/blob/main/LICENSE)
 
-Working space for dispatcher prototyping
-
-This is firstly intended to be a code split of:
+This is intended to be a working space for prototyping a code split of:
 
 <https://github.com/ansible/awx/tree/devel/awx/main/dispatch>
 
 As a part of doing the split, we also want to resolve a number of
 long-standing design and sustainability issues, thus, asyncio.
 
+The philosophy of the dispatcher is to have a limited scope
+as a "local" runner of background tasks, but to be composable
+so that it can be "wrapped" easily to enable clustering and
+distributed task management by apps using it.
+
 Licensed under [Apache Software License 2.0](LICENSE)
 
 ### Usage
@@ -37,20 +40,11 @@ def print_hello():
     print('hello world!!')
 ```
 
-#### Dispatcher service
-
-The dispatcher service needs to be running before you submit tasks.
-This does not make any attempts at message durability or confirmation.
-If you submit a task in an outage of the service, it will be dropped.
-
-There are 2 ways to run the dispatcher service:
-
-- Importing and running (code snippet below)
-- A CLI entrypoint `dispatcher-standalone` for demo purposes
+Additionally, you need to configure dispatcher somewhere in your import path.
+This tells dispatcher how to submit tasks to be ran.
 
 ```python
-from dispatcher.main import DispatcherMain
-import asyncio
+from dispatcher.config import setup
 
 config = {
     "producers": {
@@ -63,13 +57,29 @@ config = {
     },
     "pool": {"max_workers": 4},
 }
-loop = asyncio.get_event_loop()
-dispatcher = DispatcherMain(config)
+setup(config)
+```
+
+For more on how to set up and the allowed options in the config,
+see the section [config](docs/config.md) docs.
+
+#### Dispatcher service
+
+The dispatcher service needs to be running before you submit tasks.
+This does not make any attempts at message durability or confirmation.
+If you submit a task in an outage of the service, it will be dropped.
+
+There are 2 ways to run the dispatcher service:
 
-try:
-    loop.run_until_complete(dispatcher.main())
-finally:
-    loop.close()
+- Importing and running (code snippet below)
+- A CLI entrypoint `dispatcher-standalone` for demo purposes
+
+```python
+from dispatcher import run_service
+
+# After the setup() method has been called
+
+run_service()
 ```
 
 Configuration tells how to connect to postgres, and what channel(s) to listen to.
@@ -88,6 +98,8 @@ The following code will submit `print_hello` to run in the background dispatcher
 ```python
 from test_methods import print_hello
 
+# After the setup() method has been called
+
 print_hello.delay()
 ```
 
@@ -96,6 +108,8 @@ Also valid:
 ```python
 from test_methods import print_hello
 
+# After the setup() method has been called
+
 print_hello.apply_async(args=[], kwargs={})
 ```
 

dispatcher.yml

@@ -1,20 +1,25 @@
 # Demo config
 ---
-pool:
-  max_workers: 3
-producers:
-  brokers:
-    # List of channels to listen on
+version: 2
+service:
+  pool_kwargs:
+    max_workers: 4
+brokers:
+  pg_notify:
+    config:
+      conninfo: dbname=dispatch_db user=dispatch password=dispatching host=localhost port=55777
+    sync_connection_factory: dispatcher.brokers.pg_notify.connection_saver
     channels:
     - test_channel
     - test_channel2
     - test_channel3
-    pg_notify:
-      # Database connection details
-      conninfo: dbname=dispatch_db user=dispatch password=dispatching host=localhost
-        port=55777
-  scheduled:
-    'lambda: __import__("time").sleep(1)':
-      schedule: 3
-    'lambda: __import__("time").sleep(2)':
-      schedule: 3
+    default_publish_channel: test_channel
+producers:
+  ScheduledProducer:
+    task_schedule:
+      'lambda: __import__("time").sleep(1)':
+        schedule: 3
+      'lambda: __import__("time").sleep(2)':
+        schedule: 3
+publish:
+  default_broker: pg_notify

dispatcher/__init__.py

@@ -0,0 +1,21 @@
+import asyncio
+import logging
+
+from dispatcher.factories import from_settings
+
+logger = logging.getLogger(__name__)
+
+
+def run_service() -> None:
+    """
+    Runs dispatcher task service (runs tasks due to messages from brokers and other local producers)
+    Before calling this you need to configure by calling dispatcher.config.setup
+    """
+    loop = asyncio.get_event_loop()
+    dispatcher = from_settings()
+    try:
+        loop.run_until_complete(dispatcher.main())
+    except KeyboardInterrupt:
+        logger.info('Dispatcher stopped by KeyboardInterrupt')
+    finally:
+        loop.close()

dispatcher/brokers/__init__.py

@@ -0,0 +1,20 @@
+import importlib
+from types import ModuleType
+
+from .base import BaseBroker
+
+
+def get_broker_module(broker_name) -> ModuleType:
+    "Static method to alias import_module so we use a consistent import path"
+    return importlib.import_module(f'dispatcher.brokers.{broker_name}')
+
+
+def get_broker(broker_name: str, broker_config: dict, **overrides) -> BaseBroker:
+    """
+    Given the name of the broker in the settings, and the data under that entry in settings,
+    return the broker object.
+    """
+    broker_module = get_broker_module(broker_name)
+    kwargs = broker_config.copy()
+    kwargs.update(overrides)
+    return broker_module.Broker(**kwargs)

dispatcher/brokers/base.py

@@ -0,0 +1,14 @@
+from typing import AsyncGenerator, Optional, Protocol
+
+
+class BaseBroker(Protocol):
+    async def aprocess_notify(self, connected_callback=None) -> AsyncGenerator[tuple[str, str], None]:
+        yield ('', '')  # yield affects CPython type https://github.com/python/mypy/pull/18422
+
+    async def apublish_message(self, channel: Optional[str] = None, message: str = '') -> None: ...
+
+    async def aclose(self) -> None: ...
+
+    def publish_message(self, channel=None, message=None): ...
+
+    def close(self): ...