diff --git a/pyproject.toml b/pyproject.toml index 284ad4abdc..d4d217a862 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -39,6 +39,8 @@ dependencies = [ "deepdiff", "matplotlib", "cachetools", + "bluesky-stomp >= 0.2.0", + # # These dependencies may be issued as pre-release versions and should have a pin constraint # as by default pip-install will not upgrade to a pre-release. diff --git a/src/mx_bluesky/hyperion/external_interaction/callbacks/__main__.py b/src/mx_bluesky/hyperion/external_interaction/callbacks/__main__.py index 26cd1ccd3b..a703061fa8 100644 --- a/src/mx_bluesky/hyperion/external_interaction/callbacks/__main__.py +++ b/src/mx_bluesky/hyperion/external_interaction/callbacks/__main__.py @@ -1,12 +1,17 @@ import logging -from collections.abc import Callable, Sequence +from abc import abstractmethod +from collections.abc import Callable +from contextlib import AbstractContextManager from threading import Thread from time import sleep # noqa from urllib import request from urllib.error import URLError +from blueapi.config import ApplicationConfig, ConfigLoader from bluesky.callbacks import CallbackBase from bluesky.callbacks.zmq import Proxy, RemoteDispatcher +from bluesky_stomp.messaging import StompClient +from bluesky_stomp.models import Broker from dodal.log import LOGGER as DODAL_LOGGER from dodal.log import set_up_all_logging_handlers @@ -52,6 +57,9 @@ from mx_bluesky.hyperion.external_interaction.callbacks.snapshot_callback import ( BeamDrawingCallback, ) +from mx_bluesky.hyperion.external_interaction.callbacks.stomp.dispatcher import ( + StompDispatcher, +) from mx_bluesky.hyperion.parameters.cli import CallbackArgs, parse_callback_args from mx_bluesky.hyperion.parameters.constants import CONST from mx_bluesky.hyperion.parameters.gridscan import ( @@ -143,19 +151,6 @@ def log_debug(msg, *args, **kwargs): NEXUS_LOGGER.debug(msg, *args, **kwargs) -def wait_for_threads_forever(threads: Sequence[Thread]): - alive = [t.is_alive() for t in threads] - try: - log_debug("Trying to wait forever on callback and dispatcher threads") - while all(alive): - sleep(LIVENESS_POLL_SECONDS) - alive = [t.is_alive() for t in threads] - except KeyboardInterrupt: - log_info("Main thread received interrupt - exiting.") - else: - log_info("Proxy or dispatcher thread ended - exiting.") - - class HyperionCallbackRunner: """Runs Nexus, ISPyB and Zocalo callbacks in their own process.""" @@ -166,44 +161,26 @@ def __init__(self, callback_args: CallbackArgs) -> None: self.callbacks = setup_callbacks() - self.proxy = Proxy(*CONST.CALLBACK_0MQ_PROXY_PORTS) - self.proxy_thread = Thread( - target=self.proxy.start, daemon=True, name="0MQ Proxy" - ) - - self.dispatcher = RemoteDispatcher( - f"localhost:{CONST.CALLBACK_0MQ_PROXY_PORTS[1]}" - ) - - def start_dispatcher(callbacks: list[Callable]): - for cb in callbacks: - self.dispatcher.subscribe(cb) - self.dispatcher.start() - - self.dispatcher_thread = Thread( - target=start_dispatcher, - args=[self.callbacks], - daemon=True, - name="0MQ Dispatcher", - ) - self.watchdog_thread = Thread( target=run_watchdog, daemon=True, name="Watchdog", args=[callback_args.watchdog_port], ) - log_info("Created 0MQ proxy and local RemoteDispatcher.") + + self._dispatcher_cm: DispatcherContextMgr + if callback_args.stomp_config: + self._dispatcher_cm = StompDispatcherContextMgr( + callback_args, self.callbacks + ) + else: + self._dispatcher_cm = RemoteDispatcherContextMgr(self.callbacks) def start(self): log_info(f"Launching threads, with callbacks: {self.callbacks}") - self.proxy_thread.start() - self.dispatcher_thread.start() self.watchdog_thread.start() - log_info("Proxy and dispatcher thread launched.") - wait_for_threads_forever( - [self.proxy_thread, self.dispatcher_thread, self.watchdog_thread] - ) + with self._dispatcher_cm: + ping_watchdog_while_alive(self._dispatcher_cm, self.watchdog_thread) def run_watchdog(watchdog_port: int): @@ -231,5 +208,96 @@ def main(dev_mode=False) -> None: runner.start() +class DispatcherContextMgr(AbstractContextManager): + @abstractmethod + def is_alive(self) -> bool: ... + + +class RemoteDispatcherContextMgr(DispatcherContextMgr): + def __init__(self, callbacks: list[CallbackBase]): + super().__init__() + + self.proxy = Proxy(*CONST.CALLBACK_0MQ_PROXY_PORTS) + self.proxy_thread = Thread( + target=self.proxy.start, daemon=True, name="0MQ Proxy" + ) + + self.dispatcher = RemoteDispatcher( + f"localhost:{CONST.CALLBACK_0MQ_PROXY_PORTS[1]}" + ) + + def start_dispatcher(callbacks: list[Callable]): + for cb in callbacks: + self.dispatcher.subscribe(cb) + self.dispatcher.start() + + self.dispatcher_thread = Thread( + target=start_dispatcher, + args=[callbacks], + daemon=True, + name="0MQ Dispatcher", + ) + log_info("Created 0MQ proxy and local RemoteDispatcher.") + + def __enter__(self): + log_info("Proxy and dispatcher thread launched.") + self.proxy_thread.start() + self.dispatcher_thread.start() + return self + + def __exit__(self, exc_type, exc_value, traceback, /): + self.dispatcher.stop() + # proxy has no way to stop + + def is_alive(self): + return self.proxy_thread.is_alive() and self.dispatcher_thread.is_alive() + + +class StompDispatcherContextMgr(DispatcherContextMgr): + def __init__(self, args: CallbackArgs, callbacks: list[CallbackBase]): + super().__init__() + loader = ConfigLoader(ApplicationConfig) + loader.use_values_from_yaml(args.stomp_config) + config = loader.load() + log_info( + f"Stomp client configured on {config.stomp.url.host}:{config.stomp.url.port}" + ) + self._stomp_client = StompClient.for_broker( + broker=Broker( + host=config.stomp.url.host, + port=config.stomp.url.port, + auth=config.stomp.auth, + ) + ) + self._dispatcher = StompDispatcher(self._stomp_client) + for cb in callbacks: + self._dispatcher.subscribe(cb) + + def is_alive(self) -> bool: + return self._stomp_client.is_connected() + + def __enter__(self): + self._dispatcher.__enter__() + return self + + def __exit__(self, exc_type, exc_value, traceback, /): + self._dispatcher.__exit__(exc_type, exc_value, traceback) + + +def ping_watchdog_while_alive( + dispatcher_cm: DispatcherContextMgr, watchdog_thread: Thread +): + alive = watchdog_thread.is_alive() and dispatcher_cm.is_alive() + try: + log_debug("Trying to wait forever on callback and dispatcher threads") + while alive: + sleep(LIVENESS_POLL_SECONDS) + alive = watchdog_thread.is_alive() and dispatcher_cm.is_alive() + except KeyboardInterrupt: + log_info("Main thread received interrupt - exiting.") + else: + log_info("Proxy or dispatcher thread ended - exiting.") + + if __name__ == "__main__": main() diff --git a/src/mx_bluesky/hyperion/external_interaction/callbacks/stomp/__init__.py b/src/mx_bluesky/hyperion/external_interaction/callbacks/stomp/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/src/mx_bluesky/hyperion/external_interaction/callbacks/stomp/dispatcher.py b/src/mx_bluesky/hyperion/external_interaction/callbacks/stomp/dispatcher.py new file mode 100644 index 0000000000..40a7075f99 --- /dev/null +++ b/src/mx_bluesky/hyperion/external_interaction/callbacks/stomp/dispatcher.py @@ -0,0 +1,33 @@ +from blueapi.client.event_bus import AnyEvent +from blueapi.core import DataEvent +from bluesky.run_engine import Dispatcher +from bluesky_stomp.messaging import MessageContext, StompClient +from bluesky_stomp.models import MessageTopic +from event_model import DocumentNames + +from mx_bluesky.common.utils.log import ISPYB_ZOCALO_CALLBACK_LOGGER as LOGGER + +BLUEAPI_EVENT_TOPIC = "public.worker.event" + + +class StompDispatcher(Dispatcher): + def __init__(self, stomp_client: StompClient): + super().__init__() + self._client = stomp_client + + def __enter__(self): + self._subscription_id = self._client.subscribe( + MessageTopic(name=BLUEAPI_EVENT_TOPIC), self._on_event + ) + LOGGER.info("Connecting to stomp broker...") + self._client.connect() + + def __exit__(self, exc_type, exc_val, exc_tb): + LOGGER.info("Disconnecting from stomp and unsubscribing...") + self._client.disconnect() + self._client.unsubscribe(self._subscription_id) + + def _on_event(self, event: AnyEvent, context: MessageContext): + match event: + case DataEvent(name=name, doc=doc): # type: ignore + self.process(DocumentNames[name], doc) diff --git a/src/mx_bluesky/hyperion/parameters/cli.py b/src/mx_bluesky/hyperion/parameters/cli.py index 661aa741f0..620a2222f2 100644 --- a/src/mx_bluesky/hyperion/parameters/cli.py +++ b/src/mx_bluesky/hyperion/parameters/cli.py @@ -1,5 +1,6 @@ import argparse from enum import StrEnum +from pathlib import Path from pydantic.dataclasses import dataclass @@ -25,6 +26,7 @@ class HyperionArgs: class CallbackArgs: dev_mode: bool = False watchdog_port: int = HyperionConstants.HYPERION_PORT + stomp_config: Path | None = None def _add_callback_relevant_args(parser: argparse.ArgumentParser) -> None: @@ -45,8 +47,18 @@ def parse_callback_args() -> CallbackArgs: type=int, help="Liveness port for callbacks to ping regularly", ) + parser.add_argument( + "--stomp-config", + type=Path, + default=None, + help="Specify config yaml for the STOMP backend (default is 0MQ)", + ) args = parser.parse_args() - return CallbackArgs(dev_mode=args.dev, watchdog_port=args.watchdog_port) + return CallbackArgs( + dev_mode=args.dev, + watchdog_port=args.watchdog_port, + stomp_config=args.stomp_config, + ) def parse_cli_args() -> HyperionArgs: diff --git a/tests/system_tests/hyperion/external_interaction/callbacks/stomp_callback_wrapper/__init__.py b/tests/system_tests/hyperion/external_interaction/callbacks/stomp_callback_wrapper/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/system_tests/hyperion/external_interaction/callbacks/stomp_callback_wrapper/__main__.py b/tests/system_tests/hyperion/external_interaction/callbacks/stomp_callback_wrapper/__main__.py new file mode 100644 index 0000000000..fa78880bcb --- /dev/null +++ b/tests/system_tests/hyperion/external_interaction/callbacks/stomp_callback_wrapper/__main__.py @@ -0,0 +1,70 @@ +from unittest.mock import patch + +from bluesky.callbacks import CallbackBase +from bluesky_stomp.models import MessageTopic +from event_model import Event, RunStart, RunStop + +from mx_bluesky.hyperion.external_interaction.callbacks.__main__ import ( + StompDispatcherContextMgr, + main, +) +from mx_bluesky.hyperion.parameters.cli import CallbackArgs + +CALLBACK_TOPIC = "callback_test_events" + + +class PatchedStompCallbackMgr(StompDispatcherContextMgr): + def __init__(self, args: CallbackArgs, callbacks: list[CallbackBase]) -> None: + super().__init__(args, callbacks) + callback = callbacks[0] + assert isinstance(callback, StompTestCallback) + self.callback = callback + + def __enter__(self): + super().__enter__() + self.callback.init(self._stomp_client) + return self + + +class StompTestCallback(CallbackBase): + def __init__(self) -> None: + super().__init__() + self.stomp_client = None + self.destination = MessageTopic(name=CALLBACK_TOPIC) + + def init(self, stomp_client): + self.stomp_client = stomp_client + self.fire_event_back("init") + + def start(self, doc: RunStart) -> RunStart | None: + self.fire_event_back(f"start: {doc['run_name']}") # type: ignore + return super().start(doc) + + def stop(self, doc: RunStop) -> RunStop | None: + self.fire_event_back("stop") + return super().stop(doc) + + def event(self, doc: Event) -> Event: + self.fire_event_back(f"event: {doc['data']['baton-requested_user']}") + return super().event(doc) + + def fire_event_back(self, msg: str): + self.stomp_client.send(destination=self.destination, obj=msg) # type: ignore + + +if __name__ == "__main__": + + def mock_setup_callbacks(): + return [StompTestCallback()] + + with ( + patch( + "mx_bluesky.hyperion.external_interaction.callbacks.__main__.setup_callbacks", + return_value=mock_setup_callbacks(), + ), + patch( + "mx_bluesky.hyperion.external_interaction.callbacks.__main__.StompDispatcherContextMgr", + PatchedStompCallbackMgr, + ), + ): + main() diff --git a/tests/system_tests/hyperion/external_interaction/callbacks/test_external_callbacks.py b/tests/system_tests/hyperion/external_interaction/callbacks/test_external_callbacks.py index 4cda1dbde3..905f226c0a 100644 --- a/tests/system_tests/hyperion/external_interaction/callbacks/test_external_callbacks.py +++ b/tests/system_tests/hyperion/external_interaction/callbacks/test_external_callbacks.py @@ -4,18 +4,26 @@ import re import signal import subprocess +import sys import threading +from collections.abc import Mapping from genericpath import isfile from http.server import BaseHTTPRequestHandler, HTTPServer +from pathlib import Path from time import sleep +from typing import Any from unittest.mock import MagicMock, patch import bluesky.plan_stubs as bps import pytest import zmq +from blueapi.config import ApplicationConfig, ConfigLoader +from blueapi.core import BlueskyContext, DataEvent, EventPublisher from bluesky.callbacks import CallbackBase from bluesky.callbacks.zmq import Publisher from bluesky.run_engine import RunEngine +from bluesky_stomp.messaging import MessageContext, StompClient +from bluesky_stomp.models import Broker, MessageTopic from dodal.devices.oav.oav_detector import OAV from dodal.devices.smargon import Smargon from zmq.utils.monitor import recv_monitor_message @@ -37,6 +45,9 @@ from mx_bluesky.hyperion.experiment_plans.rotation_scan_plan import ( rotation_scan, ) +from mx_bluesky.hyperion.external_interaction.callbacks.stomp.dispatcher import ( + BLUEAPI_EVENT_TOPIC, +) from mx_bluesky.hyperion.parameters.constants import CONST, HyperionConstants from mx_bluesky.hyperion.parameters.device_composites import ( HyperionFlyScanXRayCentreComposite, @@ -45,6 +56,7 @@ from .....conftest import fake_read from ..conftest import fetch_comment # noqa # type: ignore +from .stomp_callback_wrapper.__main__ import CALLBACK_TOPIC """ Note that because these tests use the external processes some of the errors coming from @@ -53,6 +65,9 @@ """ +TASK_ID = "TASK_ID" + + class DocumentCatcher(CallbackBase): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) @@ -80,18 +95,10 @@ def event_monitor(monitor: zmq.Socket, connection_active_lock: threading.Lock) - LOGGER.info("event monitor thread done!") -@pytest.fixture -def external_callbacks(): +def external_callbacks_with_args(*args): process_env = os.environ.copy() external_callbacks_process = subprocess.Popen( - [ - "python", - "src/mx_bluesky/hyperion/external_interaction/callbacks/__main__.py", - "--dev", - "--watchdog-port", - str(HyperionConstants.HYPERION_PORT), - ], - env=process_env, + [sys.executable, *args], env=process_env ) yield external_callbacks_process external_callbacks_process.send_signal(signal.SIGINT) @@ -100,6 +107,64 @@ def external_callbacks(): external_callbacks_process.wait(10) +@pytest.fixture +def external_callbacks(): + yield from external_callbacks_with_args( + "-m", + "mx_bluesky.hyperion.external_interaction.callbacks", + "--dev", + "--watchdog-port", + str(HyperionConstants.HYPERION_PORT), + ) + + +@pytest.fixture +def external_callbacks_with_stomp(): + yield from external_callbacks_with_args( + "tests/system_tests/hyperion/external_interaction/callbacks/stomp_callback_wrapper/__main__.py", + "--dev", + "--watchdog-port", + str(HyperionConstants.HYPERION_PORT), + "--stomp-config", + "tests/test_data/stomp_callback_test_config.yaml", + ) + + +@pytest.fixture +def bluesky_context_with_stomp(run_engine: RunEngine): + loader = ConfigLoader(ApplicationConfig) + loader.use_values_from_yaml(Path("tests/test_data/stomp_callback_test_config.yaml")) + config = loader.load() + context = BlueskyContext(configuration=config, run_engine=run_engine) + stomp_client = StompClient.for_broker( + broker=Broker( + host=config.stomp.url.host, + port=config.stomp.url.port, + auth=config.stomp.auth, + ) + ) + + event_publisher = EventPublisher() + + def fire_event(name: str, doc: Mapping[str, Any]): + event_publisher.publish(DataEvent(name=name, doc=doc, task_id=TASK_ID)) + + re_sub_id = context.run_engine.subscribe(fire_event) + stomp_client.connect() + + def send_message(event, correlation_id): + stomp_client.send(MessageTopic(name=BLUEAPI_EVENT_TOPIC), event, correlation_id) + + publisher_sub_id = event_publisher.subscribe(send_message) + + try: + yield context, stomp_client + finally: + event_publisher.unsubscribe(publisher_sub_id) + stomp_client.disconnect() + context.run_engine.unsubscribe(re_sub_id) + + @pytest.fixture def run_engine_with_external_callbacks( external_callbacks, @@ -275,3 +340,42 @@ def do_GET(self): # noqa: N802 finally: httpd.shutdown() server_thread.join() + + +@pytest.mark.system_test +def test_external_callbacks_with_stomp( + external_callbacks_with_stomp, bluesky_context_with_stomp +): + bluesky_context, stomp_client = bluesky_context_with_stomp + + def read_the_baton(): + yield from bps.open_run(md={"run_name": "READ BATON PLAN"}) + baton = bluesky_context.find_device("baton") + yield from bps.abs_set(baton.requested_user, "TEST_USER") + yield from bps.create(name="BATON_READ_EVENT") + yield from bps.read(baton) + yield from bps.save() + yield from bps.close_run() + + msgs = [] + + started = threading.Event() + stopped = threading.Event() + + def on_msg_from_callback(msg: str, context: MessageContext): + msgs.append(msg) + if msg == "stop": + stopped.set() + + sub_id = stomp_client.subscribe( + MessageTopic(name=CALLBACK_TOPIC), on_msg_from_callback + ) + + try: + started.wait(10) + bluesky_context.run_engine(read_the_baton()) + stopped.wait(10) + finally: + stomp_client.unsubscribe(sub_id) + + assert msgs == ["init", "start: READ BATON PLAN", "event: TEST_USER", "stop"] diff --git a/tests/test_data/stomp_callback_test_config.yaml b/tests/test_data/stomp_callback_test_config.yaml new file mode 100644 index 0000000000..7b69fbf3f9 --- /dev/null +++ b/tests/test_data/stomp_callback_test_config.yaml @@ -0,0 +1,11 @@ +# Configuration for the test harness to send stomp messages to the callbacks +env: + sources: + - kind: deviceManager + module: dodal.beamlines.i03_supervisor + mock: true + events: + broadcast_status_events: false +stomp: + enabled: true + url: tcp://localhost:61613 diff --git a/tests/unit_tests/hyperion/external_interaction/callbacks/test_external_callbacks.py b/tests/unit_tests/hyperion/external_interaction/callbacks/test_external_callbacks.py index 51259d7c7f..6371f553ed 100644 --- a/tests/unit_tests/hyperion/external_interaction/callbacks/test_external_callbacks.py +++ b/tests/unit_tests/hyperion/external_interaction/callbacks/test_external_callbacks.py @@ -3,9 +3,11 @@ from functools import partial from threading import Event from time import sleep -from unittest.mock import MagicMock, patch +from unittest.mock import MagicMock, Mock, call, patch import pytest +from bluesky.callbacks import CallbackBase +from bluesky_stomp.models import Broker from dodal.log import LOGGER as DODAL_LOGGER from mx_bluesky.common.external_interaction.alerting.log_based_service import ( @@ -15,10 +17,10 @@ from mx_bluesky.hyperion.external_interaction.callbacks.__main__ import ( PING_TIMEOUT_S, main, + ping_watchdog_while_alive, run_watchdog, setup_callbacks, setup_logging, - wait_for_threads_forever, ) from mx_bluesky.hyperion.parameters.cli import CallbackArgs from mx_bluesky.hyperion.parameters.constants import HyperionConstants @@ -94,9 +96,10 @@ def test_wait_for_threads_forever_calls_time_sleep(mock_sleep: MagicMock): thread_that_stops_after_one_call = MagicMock() thread_that_stops_after_one_call.is_alive.side_effect = [True, False] - mock_threads = [thread_that_stops_after_one_call, MagicMock()] + mock_context_manager = MagicMock() + mock_context_manager.is_alive.side_effect = [True, False] - wait_for_threads_forever(mock_threads) + ping_watchdog_while_alive(thread_that_stops_after_one_call, mock_context_manager) assert mock_sleep.call_count == 1 @@ -137,3 +140,52 @@ def test_launch_with_watchdog_port_arg_applies_port(mock_callback_runner: MagicM callback_args = mock_callback_runner.mock_calls[0].args[0] assert callback_args.dev_mode assert callback_args.watchdog_port == 1234 + + +@patch( + "sys.argv", + new=[ + "hyperion-callbacks", + "--watchdog-port", + "1234", + "--stomp-config", + "tests/test_data/stomp_callback_test_config.yaml", + ], +) +@patch("mx_bluesky.hyperion.external_interaction.callbacks.__main__.StompDispatcher") +@patch("mx_bluesky.hyperion.external_interaction.callbacks.__main__.StompClient") +@patch( + "mx_bluesky.hyperion.external_interaction.callbacks.__main__.setup_callbacks", + return_value=[Mock(spec=CallbackBase)], +) +@patch( + "mx_bluesky.hyperion.external_interaction.callbacks.__main__.LIVENESS_POLL_SECONDS", + 0.1, +) +def test_launch_with_stomp_launches_stomp_backend( + mock_setup_callbacks: MagicMock, + mock_client_cls: MagicMock, + mock_dispatcher_cls: MagicMock, +): + stomp_client = mock_client_cls.for_broker.return_value + dispatcher = mock_dispatcher_cls.return_value + stomp_client.is_connected.side_effect = [True, False] + + parent = MagicMock() + parent.attach_mock(stomp_client, "stomp_client") + parent.attach_mock(dispatcher, "dispatcher") + main(dev_mode=True) + + mock_client_cls.for_broker.assert_called_once_with( + broker=Broker(host="localhost", port=61613, auth=None) + ) + mock_dispatcher_cls.assert_called_once_with(stomp_client) + parent.assert_has_calls( + [ + call.dispatcher.subscribe(mock_setup_callbacks.return_value[0]), + call.dispatcher.__enter__(), + call.stomp_client.is_connected(), + call.stomp_client.is_connected(), + call.dispatcher.__exit__(None, None, None), + ] + ) diff --git a/tests/unit_tests/hyperion/external_interaction/stomp/__init__.py b/tests/unit_tests/hyperion/external_interaction/stomp/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/unit_tests/hyperion/external_interaction/stomp/test_dispatcher.py b/tests/unit_tests/hyperion/external_interaction/stomp/test_dispatcher.py new file mode 100644 index 0000000000..40e715f32d --- /dev/null +++ b/tests/unit_tests/hyperion/external_interaction/stomp/test_dispatcher.py @@ -0,0 +1,49 @@ +from unittest.mock import MagicMock, Mock + +import pytest +from blueapi.core import DataEvent +from bluesky_stomp.messaging import MessageContext, StompClient + +from mx_bluesky.hyperion.external_interaction.callbacks.stomp.dispatcher import ( + StompDispatcher, +) + +TEST_TASK_ID = "TEST_TASK_ID" + +TEST_SUB_ID = "SUB_ID" + + +@pytest.fixture +def mock_client(): + stomp_client = Mock(spec=StompClient) + stomp_client.subscribe.return_value = TEST_SUB_ID + return stomp_client + + +def test_dispatcher_connect_and_disconnects_on_enter_and_exit(mock_client): + dispatcher = StompDispatcher(mock_client) + with dispatcher: + mock_client.connect.assert_called_once() + mock_client.subscribe.assert_called_once() + mock_client.unsubscribe.assert_called_once_with(TEST_SUB_ID) + mock_client.disconnect.assert_called_once() + + +def test_dispatcher_forwards_data_event(mock_client, test_event_data): + dispatcher = StompDispatcher(mock_client) + callback = MagicMock() + + with dispatcher: + stomp_subs_callback = mock_client.subscribe.mock_calls[0].args[1] + dispatcher.subscribe(callback) + stomp_subs_callback( + DataEvent( + name="start", + doc=test_event_data.test_grid_detect_and_gridscan_start_document, + task_id=TEST_TASK_ID, + ), + Mock(spec=MessageContext), + ) + callback.assert_called_once_with( + "start", test_event_data.test_grid_detect_and_gridscan_start_document + ) diff --git a/uv.lock b/uv.lock index b1fabcea5f..a1f52a6fe0 100644 --- a/uv.lock +++ b/uv.lock @@ -1846,6 +1846,7 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/1f/cb/48e964c452ca2b92175a9b2dca037a553036cb053ba69e284650ce755f13/greenlet-3.3.0-cp311-cp311-macosx_11_0_universal2.whl", hash = "sha256:e29f3018580e8412d6aaf5641bb7745d38c85228dacf51a73bd4e26ddf2a6a8e", size = 274908, upload-time = "2025-12-04T14:23:26.435Z" }, { url = "https://files.pythonhosted.org/packages/28/da/38d7bff4d0277b594ec557f479d65272a893f1f2a716cad91efeb8680953/greenlet-3.3.0-cp311-cp311-manylinux_2_24_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:a687205fb22794e838f947e2194c0566d3812966b41c78709554aa883183fb62", size = 577113, upload-time = "2025-12-04T14:50:05.493Z" }, { url = "https://files.pythonhosted.org/packages/3c/f2/89c5eb0faddc3ff014f1c04467d67dee0d1d334ab81fadbf3744847f8a8a/greenlet-3.3.0-cp311-cp311-manylinux_2_24_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:4243050a88ba61842186cb9e63c7dfa677ec146160b0efd73b855a3d9c7fcf32", size = 590338, upload-time = "2025-12-04T14:57:41.136Z" }, + { url = "https://files.pythonhosted.org/packages/80/d7/db0a5085035d05134f8c089643da2b44cc9b80647c39e93129c5ef170d8f/greenlet-3.3.0-cp311-cp311-manylinux_2_24_s390x.manylinux_2_28_s390x.whl", hash = "sha256:670d0f94cd302d81796e37299bcd04b95d62403883b24225c6b5271466612f45", size = 601098, upload-time = "2025-12-04T15:07:11.898Z" }, { url = "https://files.pythonhosted.org/packages/dc/a6/e959a127b630a58e23529972dbc868c107f9d583b5a9f878fb858c46bc1a/greenlet-3.3.0-cp311-cp311-manylinux_2_24_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:6cb3a8ec3db4a3b0eb8a3c25436c2d49e3505821802074969db017b87bc6a948", size = 590206, upload-time = "2025-12-04T14:26:01.254Z" }, { url = "https://files.pythonhosted.org/packages/48/60/29035719feb91798693023608447283b266b12efc576ed013dd9442364bb/greenlet-3.3.0-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:2de5a0b09eab81fc6a382791b995b1ccf2b172a9fec934747a7a23d2ff291794", size = 1550668, upload-time = "2025-12-04T15:04:22.439Z" }, { url = "https://files.pythonhosted.org/packages/0a/5f/783a23754b691bfa86bd72c3033aa107490deac9b2ef190837b860996c9f/greenlet-3.3.0-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:4449a736606bd30f27f8e1ff4678ee193bc47f6ca810d705981cfffd6ce0d8c5", size = 1615483, upload-time = "2025-12-04T14:27:28.083Z" }, @@ -1853,6 +1854,7 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/f8/0a/a3871375c7b9727edaeeea994bfff7c63ff7804c9829c19309ba2e058807/greenlet-3.3.0-cp312-cp312-macosx_11_0_universal2.whl", hash = "sha256:b01548f6e0b9e9784a2c99c5651e5dc89ffcbe870bc5fb2e5ef864e9cc6b5dcb", size = 276379, upload-time = "2025-12-04T14:23:30.498Z" }, { url = "https://files.pythonhosted.org/packages/43/ab/7ebfe34dce8b87be0d11dae91acbf76f7b8246bf9d6b319c741f99fa59c6/greenlet-3.3.0-cp312-cp312-manylinux_2_24_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:349345b770dc88f81506c6861d22a6ccd422207829d2c854ae2af8025af303e3", size = 597294, upload-time = "2025-12-04T14:50:06.847Z" }, { url = "https://files.pythonhosted.org/packages/a4/39/f1c8da50024feecd0793dbd5e08f526809b8ab5609224a2da40aad3a7641/greenlet-3.3.0-cp312-cp312-manylinux_2_24_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:e8e18ed6995e9e2c0b4ed264d2cf89260ab3ac7e13555b8032b25a74c6d18655", size = 607742, upload-time = "2025-12-04T14:57:42.349Z" }, + { url = "https://files.pythonhosted.org/packages/77/cb/43692bcd5f7a0da6ec0ec6d58ee7cddb606d055ce94a62ac9b1aa481e969/greenlet-3.3.0-cp312-cp312-manylinux_2_24_s390x.manylinux_2_28_s390x.whl", hash = "sha256:c024b1e5696626890038e34f76140ed1daf858e37496d33f2af57f06189e70d7", size = 622297, upload-time = "2025-12-04T15:07:13.552Z" }, { url = "https://files.pythonhosted.org/packages/75/b0/6bde0b1011a60782108c01de5913c588cf51a839174538d266de15e4bf4d/greenlet-3.3.0-cp312-cp312-manylinux_2_24_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:047ab3df20ede6a57c35c14bf5200fcf04039d50f908270d3f9a7a82064f543b", size = 609885, upload-time = "2025-12-04T14:26:02.368Z" }, { url = "https://files.pythonhosted.org/packages/49/0e/49b46ac39f931f59f987b7cd9f34bfec8ef81d2a1e6e00682f55be5de9f4/greenlet-3.3.0-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:2d9ad37fc657b1102ec880e637cccf20191581f75c64087a549e66c57e1ceb53", size = 1567424, upload-time = "2025-12-04T15:04:23.757Z" }, { url = "https://files.pythonhosted.org/packages/05/f5/49a9ac2dff7f10091935def9165c90236d8f175afb27cbed38fb1d61ab6b/greenlet-3.3.0-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:83cd0e36932e0e7f36a64b732a6f60c2fc2df28c351bae79fbaf4f8092fe7614", size = 1636017, upload-time = "2025-12-04T14:27:29.688Z" }, @@ -1860,6 +1862,7 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/02/2f/28592176381b9ab2cafa12829ba7b472d177f3acc35d8fbcf3673d966fff/greenlet-3.3.0-cp313-cp313-macosx_11_0_universal2.whl", hash = "sha256:a1e41a81c7e2825822f4e068c48cb2196002362619e2d70b148f20a831c00739", size = 275140, upload-time = "2025-12-04T14:23:01.282Z" }, { url = "https://files.pythonhosted.org/packages/2c/80/fbe937bf81e9fca98c981fe499e59a3f45df2a04da0baa5c2be0dca0d329/greenlet-3.3.0-cp313-cp313-manylinux_2_24_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:9f515a47d02da4d30caaa85b69474cec77b7929b2e936ff7fb853d42f4bf8808", size = 599219, upload-time = "2025-12-04T14:50:08.309Z" }, { url = "https://files.pythonhosted.org/packages/c2/ff/7c985128f0514271b8268476af89aee6866df5eec04ac17dcfbc676213df/greenlet-3.3.0-cp313-cp313-manylinux_2_24_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:7d2d9fd66bfadf230b385fdc90426fcd6eb64db54b40c495b72ac0feb5766c54", size = 610211, upload-time = "2025-12-04T14:57:43.968Z" }, + { url = "https://files.pythonhosted.org/packages/79/07/c47a82d881319ec18a4510bb30463ed6891f2ad2c1901ed5ec23d3de351f/greenlet-3.3.0-cp313-cp313-manylinux_2_24_s390x.manylinux_2_28_s390x.whl", hash = "sha256:30a6e28487a790417d036088b3bcb3f3ac7d8babaa7d0139edbaddebf3af9492", size = 624311, upload-time = "2025-12-04T15:07:14.697Z" }, { url = "https://files.pythonhosted.org/packages/fd/8e/424b8c6e78bd9837d14ff7df01a9829fc883ba2ab4ea787d4f848435f23f/greenlet-3.3.0-cp313-cp313-manylinux_2_24_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:087ea5e004437321508a8d6f20efc4cfec5e3c30118e1417ea96ed1d93950527", size = 612833, upload-time = "2025-12-04T14:26:03.669Z" }, { url = "https://files.pythonhosted.org/packages/b5/ba/56699ff9b7c76ca12f1cdc27a886d0f81f2189c3455ff9f65246780f713d/greenlet-3.3.0-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:ab97cf74045343f6c60a39913fa59710e4bd26a536ce7ab2397adf8b27e67c39", size = 1567256, upload-time = "2025-12-04T15:04:25.276Z" }, { url = "https://files.pythonhosted.org/packages/1e/37/f31136132967982d698c71a281a8901daf1a8fbab935dce7c0cf15f942cc/greenlet-3.3.0-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:5375d2e23184629112ca1ea89a53389dddbffcf417dad40125713d88eb5f96e8", size = 1636483, upload-time = "2025-12-04T14:27:30.804Z" }, @@ -1867,6 +1870,7 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/d7/7c/f0a6d0ede2c7bf092d00bc83ad5bafb7e6ec9b4aab2fbdfa6f134dc73327/greenlet-3.3.0-cp314-cp314-macosx_11_0_universal2.whl", hash = "sha256:60c2ef0f578afb3c8d92ea07ad327f9a062547137afe91f38408f08aacab667f", size = 275671, upload-time = "2025-12-04T14:23:05.267Z" }, { url = "https://files.pythonhosted.org/packages/44/06/dac639ae1a50f5969d82d2e3dd9767d30d6dbdbab0e1a54010c8fe90263c/greenlet-3.3.0-cp314-cp314-manylinux_2_24_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:0a5d554d0712ba1de0a6c94c640f7aeba3f85b3a6e1f2899c11c2c0428da9365", size = 646360, upload-time = "2025-12-04T14:50:10.026Z" }, { url = "https://files.pythonhosted.org/packages/e0/94/0fb76fe6c5369fba9bf98529ada6f4c3a1adf19e406a47332245ef0eb357/greenlet-3.3.0-cp314-cp314-manylinux_2_24_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:3a898b1e9c5f7307ebbde4102908e6cbfcb9ea16284a3abe15cab996bee8b9b3", size = 658160, upload-time = "2025-12-04T14:57:45.41Z" }, + { url = "https://files.pythonhosted.org/packages/93/79/d2c70cae6e823fac36c3bbc9077962105052b7ef81db2f01ec3b9bf17e2b/greenlet-3.3.0-cp314-cp314-manylinux_2_24_s390x.manylinux_2_28_s390x.whl", hash = "sha256:dcd2bdbd444ff340e8d6bdf54d2f206ccddbb3ccfdcd3c25bf4afaa7b8f0cf45", size = 671388, upload-time = "2025-12-04T15:07:15.789Z" }, { url = "https://files.pythonhosted.org/packages/b8/14/bab308fc2c1b5228c3224ec2bf928ce2e4d21d8046c161e44a2012b5203e/greenlet-3.3.0-cp314-cp314-manylinux_2_24_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:5773edda4dc00e173820722711d043799d3adb4f01731f40619e07ea2750b955", size = 660166, upload-time = "2025-12-04T14:26:05.099Z" }, { url = "https://files.pythonhosted.org/packages/4b/d2/91465d39164eaa0085177f61983d80ffe746c5a1860f009811d498e7259c/greenlet-3.3.0-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:ac0549373982b36d5fd5d30beb8a7a33ee541ff98d2b502714a09f1169f31b55", size = 1615193, upload-time = "2025-12-04T15:04:27.041Z" }, { url = "https://files.pythonhosted.org/packages/42/1b/83d110a37044b92423084d52d5d5a3b3a73cafb51b547e6d7366ff62eff1/greenlet-3.3.0-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:d198d2d977460358c3b3a4dc844f875d1adb33817f0613f663a656f463764ccc", size = 1683653, upload-time = "2025-12-04T14:27:32.366Z" }, @@ -1874,6 +1878,7 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/a0/66/bd6317bc5932accf351fc19f177ffba53712a202f9df10587da8df257c7e/greenlet-3.3.0-cp314-cp314t-macosx_11_0_universal2.whl", hash = "sha256:d6ed6f85fae6cdfdb9ce04c9bf7a08d666cfcfb914e7d006f44f840b46741931", size = 282638, upload-time = "2025-12-04T14:25:20.941Z" }, { url = "https://files.pythonhosted.org/packages/30/cf/cc81cb030b40e738d6e69502ccbd0dd1bced0588e958f9e757945de24404/greenlet-3.3.0-cp314-cp314t-manylinux_2_24_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:d9125050fcf24554e69c4cacb086b87b3b55dc395a8b3ebe6487b045b2614388", size = 651145, upload-time = "2025-12-04T14:50:11.039Z" }, { url = "https://files.pythonhosted.org/packages/9c/ea/1020037b5ecfe95ca7df8d8549959baceb8186031da83d5ecceff8b08cd2/greenlet-3.3.0-cp314-cp314t-manylinux_2_24_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:87e63ccfa13c0a0f6234ed0add552af24cc67dd886731f2261e46e241608bee3", size = 654236, upload-time = "2025-12-04T14:57:47.007Z" }, + { url = "https://files.pythonhosted.org/packages/69/cc/1e4bae2e45ca2fa55299f4e85854606a78ecc37fead20d69322f96000504/greenlet-3.3.0-cp314-cp314t-manylinux_2_24_s390x.manylinux_2_28_s390x.whl", hash = "sha256:2662433acbca297c9153a4023fe2161c8dcfdcc91f10433171cf7e7d94ba2221", size = 662506, upload-time = "2025-12-04T15:07:16.906Z" }, { url = "https://files.pythonhosted.org/packages/57/b9/f8025d71a6085c441a7eaff0fd928bbb275a6633773667023d19179fe815/greenlet-3.3.0-cp314-cp314t-manylinux_2_24_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:3c6e9b9c1527a78520357de498b0e709fb9e2f49c3a513afd5a249007261911b", size = 653783, upload-time = "2025-12-04T14:26:06.225Z" }, { url = "https://files.pythonhosted.org/packages/f6/c7/876a8c7a7485d5d6b5c6821201d542ef28be645aa024cfe1145b35c120c1/greenlet-3.3.0-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:286d093f95ec98fdd92fcb955003b8a3d054b4e2cab3e2707a5039e7b50520fd", size = 1614857, upload-time = "2025-12-04T15:04:28.484Z" }, { url = "https://files.pythonhosted.org/packages/4f/dc/041be1dff9f23dac5f48a43323cd0789cb798342011c19a248d9c9335536/greenlet-3.3.0-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:6c10513330af5b8ae16f023e8ddbfb486ab355d04467c4679c5cfe4659975dd9", size = 1676034, upload-time = "2025-12-04T14:27:33.531Z" }, @@ -3305,6 +3310,7 @@ dependencies = [ { name = "annotated-types" }, { name = "blueapi" }, { name = "bluesky" }, + { name = "bluesky-stomp" }, { name = "cachetools" }, { name = "caproto" }, { name = "daq-config-server" }, @@ -3366,6 +3372,7 @@ requires-dist = [ { name = "annotated-types" }, { name = "blueapi", specifier = ">=1.8.0" }, { name = "bluesky", specifier = ">=1.14.6" }, + { name = "bluesky-stomp", specifier = ">=0.2.0" }, { name = "cachetools" }, { name = "caproto" }, { name = "daq-config-server", specifier = ">=1.0.0rc2" },