Use new workflows self.subscribe and queue override

src/dlstbx/cli/service.py

@@ -110,13 +110,6 @@ def on_parser_preparation(self, parser):
             default=False,
             help="Restart service on failure",
         )
-        parser.add_option(
-            "-q",
-            "--queue",
-            dest="queue",
-            default=None,
-            help="Consume messages from this queue",
-        )
         self._zc.add_command_line_options(parser)
         self.log.debug("Launching %r", sys.argv)
 
@@ -137,8 +130,6 @@ def before_frontend_construction(self, kwargs):
         kwargs["environment"] = kwargs.get("environment", {})
         kwargs["environment"]["live"] = self.use_live_infrastructure  # XXX deprecated
         kwargs["environment"]["config"] = self._zc
-        if self.options.queue:
-            kwargs["environment"]["queue"] = self.options.queue
         return kwargs
 
     def on_frontend_preparation(self, frontend):

src/dlstbx/services/archiver.py

@@ -8,7 +8,6 @@
 from datetime import datetime
 from pathlib import Path
 
-import workflows.recipe
 from workflows.services.common_service import CommonService
 
 from dlstbx.util import INDUSTRIAL_CODES
@@ -112,20 +111,8 @@ def initializing(self):
         """Subscribe to the archiver queue. Received messages must be
         acknowledged."""
         self.log.info("Archiver starting")
-        workflows.recipe.wrap_subscribe(
-            self._transport,
-            "archive.pattern",
-            self.archive_dcid,
-            acknowledgement=True,
-            log_extender=self.extend_log,
-        )
-        workflows.recipe.wrap_subscribe(
-            self._transport,
-            "archive.filelist",
-            self.archive_filelist,
-            acknowledgement=True,
-            log_extender=self.extend_log,
-        )
+        self.subscribe("archive.pattern", self.archive_dcid)
+        self.subscribe("archive.filelist", self.archive_filelist)
 
     @staticmethod
     def rangifier(numbers):

src/dlstbx/services/cloudwatcher.py

@@ -4,7 +4,6 @@
 from math import log2
 from pprint import pformat
 
-import workflows.recipe
 from requests.exceptions import HTTPError
 from workflows.services.common_service import CommonService
 from zocalo.util import slurm
@@ -43,13 +42,7 @@ def initializing(self):
             "iris": self.iris_api,
         }
 
-        workflows.recipe.wrap_subscribe(
-            self._transport,
-            "cloudwatcher",
-            self.watch_jobs,
-            acknowledgement=True,
-            log_extender=self.extend_log,
-        )
+        self.subscribe("cloudwatcher", self.watch_jobs)
 
     def watch_jobs(self, rw, header, message):
         """

src/dlstbx/services/cluster.py

@@ -13,7 +13,6 @@
 
 import pydantic
 import requests
-import workflows.recipe
 from workflows.services.common_service import CommonService
 from zocalo.configuration import Configuration
 from zocalo.util import slurm
@@ -179,13 +178,7 @@ def initializing(self):
             )
         }
         self.log.debug(f"Supported schedulers: {', '.join(self.schedulers.keys())}")
-        workflows.recipe.wrap_subscribe(
-            self._transport,
-            "cluster.submission",
-            self.run_submit_job,
-            acknowledgement=True,
-            log_extender=self.extend_log,
-        )
+        self.subscribe("cluster.submission", self.run_submit_job)
 
     def _write_job_file(
         self, header, path: str, description: str, content: str

src/dlstbx/services/filewatcher.py

@@ -6,7 +6,6 @@
 from pathlib import Path
 
 import h5py
-import workflows.recipe
 from workflows.recipe import RecipeWrapper
 from workflows.services.common_service import CommonService
 
@@ -98,13 +97,7 @@ def initializing(self):
         acknowledged.
         """
         self.log.info("Filewatcher starting")
-        workflows.recipe.wrap_subscribe(
-            self._transport,
-            self._environment.get("queue") or "filewatcher",
-            self.watch_files,
-            acknowledgement=True,
-            log_extender=self.extend_log,
-        )
+        self.subscribe("filewatcher", self.watch_files)
 
     def watch_files(self, rw, header, message):
         """Check for presence of files."""

src/dlstbx/services/images.py

@@ -56,13 +56,7 @@ def initializing(self):
                 group="zocalo.services.images.plugins"
             )
         }
-        workflows.recipe.wrap_subscribe(
-            self._transport,
-            "images",
-            self.image_call,
-            acknowledgement=True,
-            log_extender=self.extend_log,
-        )
+        self.subscribe("images", self.image_call)
 
     def image_call(self, rw, header, message):
         """Pass incoming message to the relevant plugin function."""

src/dlstbx/services/indexer.py

@@ -81,13 +81,8 @@ class DLSIndexer(CommonService):
 
     def initializing(self):
         logging.getLogger("dials").setLevel(logging.WARNING)
-        workflows.recipe.wrap_subscribe(
-            self._transport,
-            "index",
-            self.index,
-            acknowledgement=True,
-            log_extender=self.extend_log,
-            mangle_for_receiving=msgpack_mangle_for_receiving,
+        self.subscribe(
+            "index", self.index, mangle_for_receiving=msgpack_mangle_for_receiving
         )
 
     @pydantic.validate_call(config={"arbitrary_types_allowed": True})

src/dlstbx/services/ispybsvc.py

@@ -11,7 +11,6 @@
 import mysql.connector
 import pydantic
 import sqlalchemy.orm
-import workflows.recipe
 from ispyb.sqlalchemy import PDB, ProteinHasPDB
 from workflows.services.common_service import CommonService
 
@@ -68,12 +67,9 @@ def initializing(self):
                 exc_info=True,
             )
         self.log.info("ISPyB service ready")
-        workflows.recipe.wrap_subscribe(
-            self._transport,
+        self.subscribe(
             "ispyb_connector",  # will become 'ispyb' in far future
             self.receive_msg,
-            acknowledgement=True,
-            log_extender=self.extend_log,
             allow_non_recipe_messages=True,
         )
 

src/dlstbx/services/ispybsvc_pia.py

@@ -2,7 +2,6 @@
 
 import ispyb
 import sqlalchemy
-import workflows.recipe
 
 from dlstbx.services.ispybsvc import DLSISPyB
 
@@ -33,11 +32,4 @@ def initializing(self):
             )
         )
         self.log.debug("ISPyB PIA inserter starting")
-        workflows.recipe.wrap_subscribe(
-            self._transport,
-            "ispyb_pia",
-            self.receive_msg,
-            acknowledgement=True,
-            log_extender=self.extend_log,
-            allow_non_recipe_messages=True,
-        )
+        self.subscribe("ispyb_pia", self.receive_msg, allow_non_recipe_messages=True)

src/dlstbx/services/mimas.py

@@ -4,7 +4,6 @@
 from pprint import pformat
 from typing import Any
 
-import workflows.recipe
 from workflows.services.common_service import CommonService
 
 from dlstbx import mimas
@@ -40,13 +39,7 @@ def initializing(self):
             "s3echo": {"total": 0.0, "last_cluster_update": time.time()},
         }
 
-        workflows.recipe.wrap_subscribe(
-            self._transport,
-            "mimas",
-            self.process,
-            acknowledgement=True,
-            log_extender=self.extend_log,
-        )
+        self.subscribe("mimas", self.process)
 
         # Subscribe to the transient.statistics.cluster topic, which we will
         # examine to determine the number of waiting jobs

src/dlstbx/services/mimas_backlog.py

@@ -3,7 +3,6 @@
 import logging
 import time
 
-import workflows.recipe
 from workflows.services.common_service import CommonService
 
 
@@ -27,14 +26,7 @@ def initializing(self):
 
         # Subscribe to the mimas.held queue, which contains the held mimas
         # recipes we would like to drip-feed to the dispatcher
-        workflows.recipe.wrap_subscribe(
-            self._transport,
-            "mimas.held",
-            self.on_mimas_held,
-            acknowledgement=True,
-            exclusive=True,
-            log_extender=self.extend_log,
-        )
+        self.subscribe("mimas.held", self.on_mimas_held, exclusive=True)
 
         # Subscribe to the transient.statistics.cluster topic, which we will
         # examine to determine the number of waiting jobs

src/dlstbx/services/nexusparser.py

@@ -2,7 +2,6 @@
 
 import os.path
 
-import workflows.recipe
 from workflows.services.common_service import CommonService
 
 import dlstbx.util.hdf5
@@ -20,13 +19,7 @@ class DLSNexusParser(CommonService):
     def initializing(self):
         """Subscribe to a queue. Received messages must be acknowledged."""
         self.log.info("Nexusparser starting")
-        workflows.recipe.wrap_subscribe(
-            self._transport,
-            "nexusparser.find_related_files",
-            self.find_related_files,
-            acknowledgement=True,
-            log_extender=self.extend_log,
-        )
+        self.subscribe("nexusparser.find_related_files", self.find_related_files)
 
     def find_related_files(self, rw, header, message):
         """Take a single file and recursively find all referenced files."""

src/dlstbx/services/notifygda.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import workflows.recipe
 from workflows.services.common_service import CommonService
 
 import dlstbx.util.gda
@@ -21,12 +20,9 @@ def initializing(self):
         Received messages must be acknowledged.
         """
         self.log.debug("GDA Bridge starting")
-        workflows.recipe.wrap_subscribe(
-            self._transport,
+        self.subscribe(
             "notify_gda",  # consider transient queue
             self.notify_gda,
-            acknowledgement=True,
-            log_extender=self.extend_log,
         )
 
     def notify_gda(self, rw, header, message):

src/dlstbx/services/per_image_analysis.py

@@ -90,13 +90,7 @@ def initializing(self):
 
         # The main per_image_analysis queue.
         # For every received message a single frame will be analysed.
-        workflows.recipe.wrap_subscribe(
-            self._transport,
-            self._environment.get("queue") or "per_image_analysis",
-            self.per_image_analysis,
-            acknowledgement=True,
-            log_extender=self.extend_log,
-        )
+        self.subscribe("per_image_analysis", self.per_image_analysis)
 
     # @pydantic.validate_call(config=dict(arbitrary_types_allowed=True))
     def per_image_analysis(

src/dlstbx/services/s3echo_collector.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import minio
-import workflows.recipe
 from workflows.services.common_service import CommonService
 
 from dlstbx.util import iris
@@ -34,21 +33,9 @@ def initializing(self):
 
         self._message_delay = 5
 
-        workflows.recipe.wrap_subscribe(
-            self._transport,
-            "s3echo.start",
-            self.on_start,
-            acknowledgement=True,
-            log_extender=self.extend_log,
-        )
+        self.subscribe("s3echo.start", self.on_start)
 
-        workflows.recipe.wrap_subscribe(
-            self._transport,
-            "s3echo.end",
-            self.on_end,
-            acknowledgement=True,
-            log_extender=self.extend_log,
-        )
+        self.subscribe("s3echo.end", self.on_end)
 
     def on_start(self, rw, header, message):
         """

src/dlstbx/services/s3echo_uploader.py

@@ -4,7 +4,6 @@
 from pprint import pformat
 
 import prometheus_client
-import workflows.recipe
 from minio.error import S3Error
 from workflows.services.common_service import CommonService
 
@@ -83,29 +82,11 @@ def initializing(self):
 
         self._message_delay = 5
 
-        workflows.recipe.wrap_subscribe(
-            self._transport,
-            "s3echo.upload.rpid",
-            self.on_upload_rpid,
-            acknowledgement=True,
-            log_extender=self.extend_log,
-        )
+        self.subscribe("s3echo.upload.rpid", self.on_upload_rpid)
 
-        workflows.recipe.wrap_subscribe(
-            self._transport,
-            "s3echo.upload.dcid",
-            self.on_upload_dcid,
-            acknowledgement=True,
-            log_extender=self.extend_log,
-        )
+        self.subscribe("s3echo.upload.dcid", self.on_upload_dcid)
 
-        workflows.recipe.wrap_subscribe(
-            self._transport,
-            "s3echo.download",
-            self.on_download,
-            acknowledgement=True,
-            log_extender=self.extend_log,
-        )
+        self.subscribe("s3echo.download", self.on_download)
 
         # Initialise metrics if requested
         if self._environment.get("metrics"):

src/dlstbx/services/s3echo_watcher.py

@@ -4,7 +4,6 @@
 from pprint import pformat
 
 import minio
-import workflows.recipe
 from minio.error import S3Error
 from workflows.services.common_service import CommonService
 
@@ -41,13 +40,7 @@ def initializing(self):
 
         self._message_delay = 5
 
-        workflows.recipe.wrap_subscribe(
-            self._transport,
-            "s3echo.watch",
-            self.watch,
-            acknowledgement=True,
-            log_extender=self.extend_log,
-        )
+        self.subscribe("s3echo.watch", self.watch)
 
     def watch(self, rw, header, message):
         """