Replace references to blueapi messaging package (#589)

Replace references to blueapi messaging package with references to the
bluesky stomp library

Fixes #588

---------

Co-authored-by: Zoheb Shaikh <zoheb.shaikh@diamond.ac.uk>
Co-authored-by: DiamondJoseph <53935796+DiamondJoseph@users.noreply.github.com>

.github/workflows/_test.yml

@@ -24,19 +24,7 @@ jobs:
   run:
     runs-on: ${{ inputs.runs-on }}
 
-    services:
-      activemq:
-        image: rmohr/activemq:5.14.5-alpine
-        ports:
-          - 61613:61613
-
     steps:
-      - name: Start RabbitMQ
-        uses: namoshek/rabbitmq-github-action@v1
-        with:
-          ports: "61614:61613"
-          plugins: rabbitmq_stomp
-
       - name: Checkout
         uses: actions/checkout@v4
         with:

dev-requirements.txt

@@ -17,6 +17,7 @@ bidict==0.23.1
 bluesky==1.13.0a4
 bluesky-kafka==0.10.0
 bluesky-live==0.0.8
+bluesky-stomp==0.1.0
 boltons==24.0.0
 cachetools==5.5.0
 caproto==1.1.1
@@ -212,10 +213,10 @@ tzlocal==5.2
 urllib3==2.2.2
 uvicorn==0.30.6
 virtualenv==20.26.3
-watchfiles==0.23.0
+watchfiles==0.24.0
 wcwidth==0.2.13
 websocket-client==1.8.0
-websockets==13.0
+websockets==13.0.1
 widgetsnbextension==4.0.13
 workflows==2.27
 xarray==2024.7.0

docs/reference/openapi.yaml

@@ -104,12 +104,10 @@ components:
       description: Request to change the state of the worker.
       properties:
         defer:
-          anyOf:
-          - type: boolean
-          - type: 'null'
           default: false
           description: Should worker defer Pausing until the next checkpoint
           title: Defer
+          type: boolean
         new_state:
           $ref: '#/components/schemas/WorkerState'
         reason:

pyproject.toml

@@ -31,6 +31,7 @@ dependencies = [
     "dls-dodal>=1.24.0",
     "super-state-machine", # See GH issue 553
     "GitPython",
+    "bluesky-stomp>=0.1.0"
 ]
 dynamic = ["version"]
 license.file = "LICENSE"
@@ -90,9 +91,6 @@ addopts = """
 filterwarnings = ["error", "ignore::DeprecationWarning"]
 # Doctest python code in docs, python code in src docstrings, test functions in tests
 testpaths = "docs src tests"
-markers = [
-    "handler: marks tests that interact with the global handler object in handler.py",
-]
 asyncio_mode = "auto"
 
 [tool.coverage.run]

src/blueapi/cli/cli.py

@@ -6,6 +6,8 @@
 
 import click
 from bluesky.callbacks.best_effort import BestEffortCallback
+from bluesky_stomp.messaging import MessageContext, MessagingTemplate
+from bluesky_stomp.models import Broker
 from pydantic import ValidationError
 from requests.exceptions import ConnectionError
 
@@ -16,8 +18,6 @@
 from blueapi.client.rest import BlueskyRemoteControlError
 from blueapi.config import ApplicationConfig, ConfigLoader
 from blueapi.core import DataEvent
-from blueapi.messaging import MessageContext
-from blueapi.messaging.stomptemplate import StompMessagingTemplate
 from blueapi.service.main import start
 from blueapi.service.openapi import (
     DOCS_SCHEMA_LOCATION,
@@ -147,14 +147,20 @@ def listen_to_events(obj: dict) -> None:
     config: ApplicationConfig = obj["config"]
     if config.stomp is not None:
         event_bus_client = EventBusClient(
-            StompMessagingTemplate.autoconfigured(config.stomp)
+            MessagingTemplate.for_broker(
+                broker=Broker(
+                    host=config.stomp.host,
+                    port=config.stomp.port,
+                    auth=config.stomp.auth,
+                )
+            )
         )
     else:
         raise RuntimeError("Message bus needs to be configured")
 
     def on_event(
-        context: MessageContext,
         event: WorkerEvent | ProgressEvent | DataEvent,
+        context: MessageContext,
     ) -> None:
         converted = json.dumps(event.dict(), indent=2)
         print(converted)

src/blueapi/client/client.py

@@ -1,9 +1,11 @@
 import time
 from concurrent.futures import Future
 
+from bluesky_stomp.messaging import MessageContext, MessagingTemplate
+from bluesky_stomp.models import Broker
+
 from blueapi.config import ApplicationConfig
 from blueapi.core.bluesky_types import DataEvent
-from blueapi.messaging import MessageContext, StompMessagingTemplate
 from blueapi.service.model import (
     DeviceModel,
     DeviceResponse,
@@ -38,7 +40,13 @@ def __init__(
     def from_config(cls, config: ApplicationConfig) -> "BlueapiClient":
         rest = BlueapiRestClient(config.api)
         if config.stomp is not None:
-            template = StompMessagingTemplate.autoconfigured(config.stomp)
+            template = MessagingTemplate.for_broker(
+                broker=Broker(
+                    host=config.stomp.host,
+                    port=config.stomp.port,
+                    auth=config.stomp.auth,
+                )
+            )
             events = EventBusClient(template)
         else:
             events = None
@@ -178,7 +186,7 @@ def run_task(
 
         complete: Future[WorkerEvent] = Future()
 
-        def inner_on_event(ctx: MessageContext, event: AnyEvent) -> None:
+        def inner_on_event(event: AnyEvent, ctx: MessageContext) -> None:
             match event:
                 case WorkerEvent(task_status=TaskStatus(task_id=test_id)):
                     relates_to_task = test_id == task_id

src/blueapi/client/event_bus.py

@@ -1,7 +1,9 @@
 from collections.abc import Callable
 
+from bluesky_stomp.messaging import MessageContext, MessagingTemplate
+from bluesky_stomp.models import MessageTopic
+
 from blueapi.core import DataEvent
-from blueapi.messaging import MessageContext, MessagingTemplate
 from blueapi.worker import ProgressEvent, WorkerEvent
 
 
@@ -28,11 +30,11 @@ def __exit__(self, exc_type, exc_value, exc_traceback) -> None:
 
     def subscribe_to_all_events(
         self,
-        on_event: Callable[[MessageContext, AnyEvent], None],
+        on_event: Callable[[AnyEvent, MessageContext], None],
     ) -> None:
         try:
             self.app.subscribe(
-                self.app.destinations.topic("public.worker.event"),
+                MessageTopic(name="public.worker.event"),
                 on_event,
             )
         except Exception as err: