diff --git a/src/dlstbx/cli/service.py b/src/dlstbx/cli/service.py
index fe06d76e3..c8ba8aac4 100644
--- a/src/dlstbx/cli/service.py
+++ b/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):
diff --git a/src/dlstbx/services/archiver.py b/src/dlstbx/services/archiver.py
index cc989cfbc..5832b2308 100644
--- a/src/dlstbx/services/archiver.py
+++ b/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):
diff --git a/src/dlstbx/services/cloudwatcher.py b/src/dlstbx/services/cloudwatcher.py
index 5b309ddba..0a61295ce 100644
--- a/src/dlstbx/services/cloudwatcher.py
+++ b/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):
         """
diff --git a/src/dlstbx/services/cluster.py b/src/dlstbx/services/cluster.py
index 7f5591015..091e6e86f 100644
--- a/src/dlstbx/services/cluster.py
+++ b/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
diff --git a/src/dlstbx/services/filewatcher.py b/src/dlstbx/services/filewatcher.py
index cd6950035..182623ddb 100644
--- a/src/dlstbx/services/filewatcher.py
+++ b/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."""
diff --git a/src/dlstbx/services/images.py b/src/dlstbx/services/images.py
index 282e01bd1..99fbb2e51 100644
--- a/src/dlstbx/services/images.py
+++ b/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."""
diff --git a/src/dlstbx/services/indexer.py b/src/dlstbx/services/indexer.py
index 82d21cdd4..c97afc2ab 100644
--- a/src/dlstbx/services/indexer.py
+++ b/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})
diff --git a/src/dlstbx/services/ispybsvc.py b/src/dlstbx/services/ispybsvc.py
index d85891369..d2ed86fcb 100644
--- a/src/dlstbx/services/ispybsvc.py
+++ b/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,
         )
 
diff --git a/src/dlstbx/services/ispybsvc_pia.py b/src/dlstbx/services/ispybsvc_pia.py
index cbdfad963..9aac1a63f 100644
--- a/src/dlstbx/services/ispybsvc_pia.py
+++ b/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)
diff --git a/src/dlstbx/services/mimas.py b/src/dlstbx/services/mimas.py
index e99c562da..dc3e2664e 100644
--- a/src/dlstbx/services/mimas.py
+++ b/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
diff --git a/src/dlstbx/services/mimas_backlog.py b/src/dlstbx/services/mimas_backlog.py
index dac4c715a..0dea391d4 100644
--- a/src/dlstbx/services/mimas_backlog.py
+++ b/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
diff --git a/src/dlstbx/services/nexusparser.py b/src/dlstbx/services/nexusparser.py
index 9fa2746a0..4f589bfee 100644
--- a/src/dlstbx/services/nexusparser.py
+++ b/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."""
diff --git a/src/dlstbx/services/notifygda.py b/src/dlstbx/services/notifygda.py
index 6bc270c10..402321208 100644
--- a/src/dlstbx/services/notifygda.py
+++ b/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):
diff --git a/src/dlstbx/services/per_image_analysis.py b/src/dlstbx/services/per_image_analysis.py
index 4886e1dec..e1661c1f6 100644
--- a/src/dlstbx/services/per_image_analysis.py
+++ b/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(
diff --git a/src/dlstbx/services/s3echo_collector.py b/src/dlstbx/services/s3echo_collector.py
index 5b74b6a54..e1d542382 100644
--- a/src/dlstbx/services/s3echo_collector.py
+++ b/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):
         """
diff --git a/src/dlstbx/services/s3echo_uploader.py b/src/dlstbx/services/s3echo_uploader.py
index 94ed2d32b..6f0ecedaf 100644
--- a/src/dlstbx/services/s3echo_uploader.py
+++ b/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"):
diff --git a/src/dlstbx/services/s3echo_watcher.py b/src/dlstbx/services/s3echo_watcher.py
index bfbaa5624..6e136652d 100644
--- a/src/dlstbx/services/s3echo_watcher.py
+++ b/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):
         """
diff --git a/src/dlstbx/services/ssx_plotter.py b/src/dlstbx/services/ssx_plotter.py
index def9234a2..0b29ab389 100644
--- a/src/dlstbx/services/ssx_plotter.py
+++ b/src/dlstbx/services/ssx_plotter.py
@@ -74,13 +74,7 @@ class SSXPlotter(CommonService):
     _logger_name = "dlstbx.services.ssx_plotter"
 
     def initializing(self):
-        workflows.recipe.wrap_subscribe(
-            self._transport,
-            "ssx.plot",
-            self.receive_msg,
-            acknowledgement=True,
-            log_extender=self.extend_log,
-        )
+        self.subscribe("ssx.plot", self.receive_msg)
 
     @pydantic.validate_call(config={"arbitrary_types_allowed": True})
     def receive_msg(
diff --git a/src/dlstbx/services/strategy.py b/src/dlstbx/services/strategy.py
index aa759f66e..c773ed5db 100644
--- a/src/dlstbx/services/strategy.py
+++ b/src/dlstbx/services/strategy.py
@@ -174,13 +174,7 @@ class DLSStrategy(CommonService):
     def initializing(self):
         """Subscribe to channel."""
         self.log.info("Strategy service starting")
-        workflows.recipe.wrap_subscribe(
-            self._transport,
-            "strategy",
-            self.generate_strategy,
-            acknowledgement=True,
-            log_extender=self.extend_log,
-        )
+        self.subscribe("strategy", self.generate_strategy)
 
     def generate_strategy(
         self, rw: workflows.recipe.RecipeWrapper, header: dict, message: dict
diff --git a/src/dlstbx/services/trigger.py b/src/dlstbx/services/trigger.py
index 1840e2a76..622ec02fe 100644
--- a/src/dlstbx/services/trigger.py
+++ b/src/dlstbx/services/trigger.py
@@ -328,13 +328,7 @@ def initializing(self):
                 ispyb.sqlalchemy.url(), connect_args={"use_pure": True}
             )
         )
-        workflows.recipe.wrap_subscribe(
-            self._transport,
-            "trigger",
-            self.trigger,
-            acknowledgement=True,
-            log_extender=self.extend_log,
-        )
+        self.subscribe("trigger", self.trigger)
         self.ispyb = ispyb.open()
 
         # Initialise metrics if requested
diff --git a/src/dlstbx/services/trigger_xchem.py b/src/dlstbx/services/trigger_xchem.py
index c0d15146b..8c7194a9f 100644
--- a/src/dlstbx/services/trigger_xchem.py
+++ b/src/dlstbx/services/trigger_xchem.py
@@ -125,13 +125,7 @@ def initializing(self):
                 ispyb.sqlalchemy.url(), connect_args={"use_pure": True}
             )
         )
-        workflows.recipe.wrap_subscribe(
-            self._transport,
-            "trigger_xchem",
-            self.trigger,
-            acknowledgement=True,
-            log_extender=self.extend_log,
-        )
+        self.subscribe("trigger_xchem", self.trigger)
         self.ispyb = ispyb.open()
 
         # Initialise metrics if requested
diff --git a/src/dlstbx/services/validation.py b/src/dlstbx/services/validation.py
index 78f92f964..0bc0dd9be 100644
--- a/src/dlstbx/services/validation.py
+++ b/src/dlstbx/services/validation.py
@@ -8,7 +8,6 @@
 import h5py
 import numpy as np
 import pytest
-import workflows.recipe
 
 try:
     from dxtbx.format.FormatNXmxDLS import find_meta_filename
@@ -35,13 +34,7 @@ class DLSValidation(CommonService):
     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,
-            "validation",
-            self.validate,
-            acknowledgement=True,
-            log_extender=self.extend_log,
-        )
+        self.subscribe("validation", self.validate)
 
     def fail_validation(self, rw, header, output, reason):
         output["reason"] = reason
diff --git a/src/dlstbx/services/xray_centering.py b/src/dlstbx/services/xray_centering.py
index d8ec59d67..35ade9c72 100644
--- a/src/dlstbx/services/xray_centering.py
+++ b/src/dlstbx/services/xray_centering.py
@@ -124,13 +124,10 @@ def initializing(self):
 
         self._next_garbage_collection = time.time() + 60
         self._register_idle(60, self.garbage_collect)
-        workflows.recipe.wrap_subscribe(
-            self._transport,
-            self._environment.get("queue") or "reduce.xray_centering",
+        self.subscribe(
+            "reduce.xray_centering",
             self.add_pia_result,
-            acknowledgement=True,
             exclusive=True,
-            log_extender=self.extend_log,
             prefetch_count=65535,
         )