Fix OpalServer-EvenBroadcaster integration & other OpalServer refactors

documentation/docs/getting-started/configuration.mdx

@@ -74,7 +74,6 @@ Please use this table as a reference.
 | POLICY_BUNDLE_TMP_PATH                        | Path for temp policy file. It needs to be writable.                                                                                                                                                                                                                                                                         |                                          |
 | POLICY_BUNDLE_GIT_ADD_PATTERN                 | File pattern to add files to all the git default files.                                                                                                                                                                                                                                                                     |                                          |
 | REPO_WATCHER_ENABLED                          |                                                                                                                                                                                                                                                                                                                             |                                          |
-| PUBLISHER_ENABLED                             |                                                                                                                                                                                                                                                                                                                             |                                          |
 | BROADCAST_KEEPALIVE_INTERVAL                  | The time to wait between sending two consecutive broadcaster keepalive messages.                                                                                                                                                                                                                                            |                                          |
 | BROADCAST_KEEPALIVE_TOPIC                     | The topic on which we should send broadcaster keepalive messages.                                                                                                                                                                                                                                                           |                                          |
 | MAX_CHANNELS_PER_CLIENT                       | Max number of records per client, after this number it will not be added to statistics, relevant only if `STATISTICS_ENABLED`.                                                                                                                                                                                              |                                          |

packages/opal-client/opal_client/tests/server_to_client_intergation_test.py

@@ -52,7 +52,6 @@ def setup_server(event):
     # Server without git watcher and with a test specific url for data, and without broadcasting
     server = OpalServer(
         init_policy_watcher=False,
-        init_publisher=False,
         data_sources_config=DATA_SOURCES_CONFIG,
         broadcaster_uri=None,
         enable_jwks_endpoint=False,

packages/opal-common/opal_common/async_utils.py

@@ -97,13 +97,24 @@ def __init__(self):
         self._tasks: List[asyncio.Task] = []
 
     def _cleanup_task(self, done_task):
-        self._tasks.remove(done_task)
+        try:
+            self._tasks.remove(done_task)
+        except KeyError:
+            ...
 
     def add_task(self, f):
         t = asyncio.create_task(f)
         self._tasks.append(t)
         t.add_done_callback(self._cleanup_task)
 
+    async def join(self, cancel=False):
+        if cancel:
+            for t in self._tasks:
+                if not t.done():
+                    t.cancel()
+        await asyncio.gather(*self._tasks, return_exceptions=True)
+        self._tasks.clear()
+
 
 async def repeated_call(
     func: Coroutine,

packages/opal-common/opal_common/confi/confi.py

@@ -75,7 +75,7 @@ def wrapped_cast(value, *args, **kwargs):
     return wrapped_cast
 
 
-def load_conf_if_none(variable, conf):
+def load_conf_if_none(variable: Any, conf: Any):
     if variable is None:
         return conf
     else:

packages/opal-server/opal_server/config.py

@@ -146,9 +146,6 @@ class OpalServerConfig(Confi):
 
     REPO_WATCHER_ENABLED = confi.bool("REPO_WATCHER_ENABLED", True)
 
-    # publisher
-    PUBLISHER_ENABLED = confi.bool("PUBLISHER_ENABLED", True)
-
     # broadcaster keepalive
     BROADCAST_KEEPALIVE_INTERVAL = confi.int(
         "BROADCAST_KEEPALIVE_INTERVAL",

packages/opal-server/opal_server/data/data_update_publisher.py

@@ -1,23 +1,18 @@
-import asyncio
 import os
-from typing import List
+from typing import List, Union
 
-from fastapi_utils.tasks import repeat_every
 from opal_common.logger import logger
-from opal_common.schemas.data import (
-    DataSourceEntryWithPollingInterval,
-    DataUpdate,
-    ServerDataSourceConfig,
-)
-from opal_common.topics.publisher import TopicPublisher
+from opal_common.schemas.data import DataUpdate
+from opal_server.pubsub import PubSub
+from opal_server.scopes.scoped_pubsub import ScopedPubSub
 
 TOPIC_DELIMITER = "/"
 PREFIX_DELIMITER = ":"
 
 
 class DataUpdatePublisher:
-    def __init__(self, publisher: TopicPublisher) -> None:
-        self._publisher = publisher
+    def __init__(self, pubsub: Union[PubSub, ScopedPubSub]) -> None:
+        self._pubsub = pubsub
 
     @staticmethod
     def get_topic_combos(topic: str) -> List[str]:
@@ -108,6 +103,4 @@ async def publish_data_updates(self, update: DataUpdate):
             entries=logged_entries,
         )
 
-        await self._publisher.publish(
-            list(all_topic_combos), update.dict(by_alias=True)
-        )
+        await self._pubsub.publish(list(all_topic_combos), update.dict(by_alias=True))

packages/opal-server/opal_server/policy/watcher/callbacks.py

@@ -16,8 +16,8 @@
     PolicyUpdateMessage,
     PolicyUpdateMessageNotification,
 )
-from opal_common.topics.publisher import TopicPublisher
 from opal_common.topics.utils import policy_topics
+from opal_server.pubsub import PubSub
 
 
 async def create_update_all_directories_in_repo(
@@ -104,7 +104,7 @@ def is_path_affected(path: Path) -> bool:
 async def publish_changed_directories(
     old_commit: Commit,
     new_commit: Commit,
-    publisher: TopicPublisher,
+    pubsub: PubSub,
     file_extensions: Optional[List[str]] = None,
     bundle_ignore: Optional[List[str]] = None,
 ):
@@ -116,7 +116,4 @@ async def publish_changed_directories(
     )
 
     if notification:
-        async with publisher:
-            await publisher.publish(
-                topics=notification.topics, data=notification.update.dict()
-            )
+        await pubsub.publish_sync(notification.topics, notification.update.dict())