1.4.2 - Allow an array of applications

README.md

@@ -2,11 +2,11 @@
     =====================================
     generator=datazen
     version=3.1.2
-    hash=1b0b047c6997f2915a3cf8fe7fc78ee4
+    hash=36f9aec8c2e9ae7d8015f17f7164d382
     =====================================
 -->
 
-# runtimepy ([1.4.1](https://pypi.org/project/runtimepy/))
+# runtimepy ([1.4.2](https://pypi.org/project/runtimepy/))
 
 [![python](https://img.shields.io/pypi/pyversions/runtimepy.svg)](https://pypi.org/project/runtimepy/)
 ![Build Status](https://github.com/vkottler/runtimepy/workflows/Python%20Package/badge.svg)

local/variables/package.yaml

@@ -1,5 +1,5 @@
 ---
 major: 1
 minor: 4
-patch: 1
+patch: 2
 entry: runtimepy

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta:__legacy__"
 
 [project]
 name = "runtimepy"
-version = "1.4.1"
+version = "1.4.2"
 description = "A framework for implementing Python services."
 readme = "README.md"
 requires-python = ">=3.7"

runtimepy/__init__.py

@@ -1,7 +1,7 @@
 # =====================================
 # generator=datazen
 # version=3.1.2
-# hash=e59954f20776156e96b052340653bca0
+# hash=513fdff327693a37914605dff5e6f0eb
 # =====================================
 
 """
@@ -10,4 +10,4 @@
 
 DESCRIPTION = "A framework for implementing Python services."
 PKG_NAME = "runtimepy"
-VERSION = "1.4.1"
+VERSION = "1.4.2"

runtimepy/data/schemas/ConnectionArbiterConfig.yaml

@@ -13,10 +13,14 @@ properties:
     items:
       $ref: package://runtimepy/schemas/ServerConnectionConfig.yaml
 
+  # Runtime application or applications.
+  # defaults to: "runtimepy.net.apps.init_only"
   app:
-    # This is the default application.
-    # default: "runtimepy.net.apps.init_only"
-    type: string
+    oneOf:
+      - type: string
+      - type: array
+        items:
+          type: string
 
   # Application configuration data.
   config:

runtimepy/net/arbiter/base.py

@@ -39,6 +39,7 @@ class AppInfo(NamedTuple):
 
 
 NetworkApplication = _Callable[[AppInfo], _Awaitable[int]]
+NetworkApplicationlike = _Union[NetworkApplication, _List[NetworkApplication]]
 ServerTask = _Awaitable[None]
 
 
@@ -49,6 +50,20 @@ async def init_only(app: AppInfo) -> int:
     return 0
 
 
+def normalize_app(
+    app: NetworkApplicationlike = None,
+) -> _List[NetworkApplication]:
+    """
+    Normalize some application parameter into a list of network applications.
+    """
+
+    if app is None:
+        app = [init_only]
+    elif not isinstance(app, list):
+        app = [app]
+    return app
+
+
 class BaseConnectionArbiter(_NamespaceMixin, _LoggerMixin):
     """
     A class implementing a base connection-manager for a broader application.
@@ -60,7 +75,7 @@ def __init__(
         stop_sig: _asyncio.Event = None,
         namespace: _Namespace = None,
         logger: _LoggerType = None,
-        app: NetworkApplication = init_only,
+        app: NetworkApplicationlike = None,
         config: _JsonObject = None,
     ) -> None:
         """Initialize this connection arbiter."""
@@ -79,7 +94,7 @@ def __init__(
 
         # A fallback application. Set a class attribute so this can be more
         # easily externally updated.
-        self._app = app
+        self._apps: _List[NetworkApplication] = normalize_app(app)
 
         # Application configuration data.
         if config is None:
@@ -136,7 +151,7 @@ def register_connection(
 
     async def _entry(
         self,
-        app: NetworkApplication = None,
+        app: NetworkApplicationlike = None,
         check_connections: bool = True,
         config: _JsonObject = None,
     ) -> int:
@@ -174,18 +189,28 @@ async def _entry(
                 async with _AsyncExitStack() as stack:
                     self.logger.info("Application starting.")
 
-                    if app is None:
-                        app = self._app
-
-                    result = await app(
-                        AppInfo(
-                            stack,
-                            self._connections,
-                            self.stop_sig,
-                            config if config is not None else self._config,
-                        )
+                    info = AppInfo(
+                        stack,
+                        self._connections,
+                        self.stop_sig,
+                        config if config is not None else self._config,
                     )
-                    self.logger.info("Application returned %d.", result)
+
+                    # Get application methods.
+                    apps = self._apps
+                    if app is not None:
+                        apps = normalize_app(app)
+
+                    result = 0
+                    for curr_app in apps:
+                        if result == 0:
+                            result = await curr_app(info)
+
+                            self.logger.info(
+                                "Application '%s' returned %d.",
+                                curr_app.__name__,
+                                result,
+                            )
 
         finally:
             for conn in self._connections.values():
@@ -196,7 +221,7 @@ async def _entry(
 
     async def app(
         self,
-        app: NetworkApplication = None,
+        app: NetworkApplicationlike = None,
         check_connections: bool = True,
         config: _JsonObject = None,
     ) -> int:
@@ -206,7 +231,7 @@ async def app(
 
         result = await _asyncio.gather(
             self._entry(
-                app, check_connections=check_connections, config=config
+                app=app, check_connections=check_connections, config=config
             ),
             self.manager.manage(self.stop_sig),
             *self._servers,
@@ -215,7 +240,7 @@ async def app(
 
     def run(
         self,
-        app: NetworkApplication = None,
+        app: NetworkApplicationlike = None,
         eloop: _asyncio.AbstractEventLoop = None,
         signals: _Iterable[int] = None,
         check_connections: bool = True,

runtimepy/net/arbiter/imports.py

@@ -5,7 +5,9 @@
 
 # built-in
 from importlib import import_module as _import_module
+from typing import List as _List
 from typing import Tuple as _Tuple
+from typing import Union as _Union
 
 # internal
 from runtimepy.net.arbiter.factory import (
@@ -32,13 +34,21 @@ class ImportConnectionArbiter(_FactoryConnectionArbiter):
     arbitrary Python modules.
     """
 
-    def set_app(self, module_path: str) -> None:
+    def set_app(self, module_path: _Union[str, _List[str]]) -> None:
         """
         Attempt to update the application method from the provided string.
         """
 
-        module, app = import_str_and_item(module_path)
-        self._app = getattr(_import_module(module), app)
+        if isinstance(module_path, str):
+            module_path = [module_path]
+
+        # Load all application methods.
+        apps = []
+        for path in module_path:
+            module, app = import_str_and_item(path)
+            apps.append(getattr(_import_module(module), app))
+
+        self._apps = apps
 
     def register_module_factory(
         self, module_path: str, *namespaces: str, **kwargs

runtimepy/net/websocket/connection.py

@@ -7,6 +7,7 @@
 # built-in
 import asyncio as _asyncio
 from contextlib import asynccontextmanager as _asynccontextmanager
+from contextlib import suppress as _suppress
 from logging import getLogger as _getLogger
 from typing import AsyncIterator as _AsyncIterator
 from typing import Awaitable as _Awaitable
@@ -18,6 +19,7 @@
 from typing import Union as _Union
 
 # third-party
+from vcorelib.asyncio import log_exceptions as _log_exceptions
 import websockets
 from websockets.client import (
     WebSocketClientProtocol as _WebSocketClientProtocol,
@@ -109,6 +111,7 @@ def server_handler(
 
         async def _handler(protocol: _WebSocketServerProtocol) -> None:
             """A handler that runs the callers initialization function."""
+
             conn = cls(protocol)
             if init is None or await init(conn):
                 if manager is not None:
@@ -127,9 +130,13 @@ async def _handler(protocol: _WebSocketServerProtocol) -> None:
                         tasks,
                         return_when=_asyncio.FIRST_COMPLETED,
                     )
+
+                    # Cleaning up tasks is always a nightmare.
                     for task in pending:
                         task.cancel()
-                        await task
+                        with _suppress(_asyncio.CancelledError):
+                            await task
+                    _log_exceptions(pending, logger=conn.logger)
 
                 # If there's no connection manager, just process the
                 # connection here.

tests/data/valid/connection_arbiter/basic.yaml

@@ -3,7 +3,9 @@ includes:
   - ports.yaml
   - basic_factories.yaml
 
-app: "runtimepy.net.apps.init_only"
+app:
+  - runtimepy.net.apps.init_only
+  - runtimepy.net.apps.init_only
 
 clients:
   - factory: sample_tcp_conn

tests/net/arbiter/test_arbiter.py

@@ -10,6 +10,7 @@
 
 # module under test
 from runtimepy.net import get_free_socket_name
+from runtimepy.net.apps import init_only
 from runtimepy.net.arbiter import ConnectionArbiter
 
 # internal
@@ -21,7 +22,7 @@ def test_connection_arbiter_run():
     """Test the synchronous 'run' entry."""
 
     arbiter = ConnectionArbiter()
-    assert arbiter.run() == 0
+    assert arbiter.run(app=init_only) == 0
 
 
 @mark.asyncio