diff --git a/trustgraph-base/trustgraph/base/input_output.py b/trustgraph-base/trustgraph/base/input_output.py
index 91af74f7..6c08fa33 100644
--- a/trustgraph-base/trustgraph/base/input_output.py
+++ b/trustgraph-base/trustgraph/base/input_output.py
@@ -16,85 +16,101 @@ class InputOutputProcessor(AsyncProcessor):
         
         self.id = params.get("id")
         self.subscriber = params.get("subscriber")
-        self.input_schema = params.get("input_schema")
-        self.output_schema = params.get("output_schema")
 
         ProcessorMetrics(id=self.id).info(
             {
                 "subscriber": self.subscriber,
-                "input_schema": self.input_schema.__name__,
-                "output_schema": self.output_schema.__name__,
             }
         )
 
         super(InputOutputProcessor, self).__init__(
             **params | {
                 "id": self.id,
-                "input_schema": self.input_schema.__name__,
-                "output_schema": self.output_schema.__name__,
             }
         )
 
         self.on_config(self.on_configuration)
 
         self.consumers = {}
+        self.producers = {}
 
         # These can be overriden by a derived class
-        self.consumer_spec = ("input", self.input_schema)
-        self.producer_spec = [
-            ("output", self.output_schema)
-        ]
+        self.consumer_spec = []
+        self.producer_spec = []
+
+# "input", self.input_schema)
+#                 "input_schema": Document,
+#                 "output_schema": TextDocument,
 
         print("Service initialised.")
 
+    def register_consumer(self, name, schema, handler):
+        self.consumer_spec.append((name, schema, handler))
+
+    def register_producer(self, name, schema):
+        self.producer_spec.append((name, schema))
+
     async def start_flow(self, flow, defn):
 
-        consumer_tag = self.consumer_spec[0]
-        consumer_schema = self.consumer_spec[1]
-        consumer_metrics = ConsumerMetrics(self.id, f"{flow}-{consumer_tag}")
-
-        consumer = self.subscribe(
-            queue = defn[consumer_tag],
-            subscriber = self.subscriber,
-            schema = consumer_schema,
-            handler = self.on_message,
-            metrics = consumer_metrics,
-        )
-
-        class Queues:
-            pass
-
-        consumer.q = Queues()
-        consumer.id = self.id
-        consumer.flow = flow
-
         producers = {}
 
         for spec in self.producer_spec:
 
-            producer_tag = spec[0]
-            producer_schema = spec[1]
+            name, schema = spec
+
             producer_metrics = ProducerMetrics(
-                self.id, f"{flow}-{producer_tag}"
+                self.id, f"{flow}-{name}"
             )
 
             producer = self.publish(
-                queue = defn[producer_tag],
-                schema = spec[1],
+                queue = defn[name],
+                schema = schema,
                 metrics = producer_metrics,
             )
 
-            setattr(consumer.q, producer_tag, producer)
+            producers[name] = producer
 
-        self.consumers[flow] = consumer
+        consumers = {}
 
-        await consumer.start()
+        for spec in self.consumer_spec:
 
+            name, schema, handler = spec
+
+            consumer_metrics = ConsumerMetrics(
+                self.id, f"{flow}-{name}"
+            )
+
+            consumer = self.subscribe(
+                queue = defn[name],
+                subscriber = self.subscriber,
+                schema = schema,
+                handler = handler,
+                metrics = consumer_metrics,
+            )
+
+            # Consumer handle gets access to producers and other
+            # metadata
+            consumer.id = self.id
+            consumer.name = name
+            consumer.flow = flow
+            consumer.q = producers
+
+            consumers[name] = consumer
+
+            await consumer.start()
+
+        self.consumers[flow] = consumers
+        self.producers[flow] = producers
+            
         print("Started flow: ", flow)
 
     async def stop_flow(self, flow):
-        await self.consumers[flow].stop()
+
+        for c in self.consumers[flow]:
+            await c.stop()
+
         del self.consumers[flow]
+        del self.producers[flow]
 
         print("Stopped flow: ", flow, flush=True)
 
diff --git a/trustgraph-base/trustgraph/base/producer.py b/trustgraph-base/trustgraph/base/producer.py
index dbbf0237..9cf62e0b 100644
--- a/trustgraph-base/trustgraph/base/producer.py
+++ b/trustgraph-base/trustgraph/base/producer.py
@@ -10,6 +10,10 @@ class Producer:
 
         self.metrics = metrics
 
+    def __del__(self):
+
+        self.producer.close()
+
     async def send(self, msg, properties={}):
 
         if self.metrics:
@@ -17,6 +21,4 @@ class Producer:
 
         self.producer.send(msg, properties)
 
-        # FIXME
-#         __class__.output_metric.inc()
 
diff --git a/trustgraph-flow/trustgraph/chunking/recursive/chunker.py b/trustgraph-flow/trustgraph/chunking/recursive/chunker.py
index 348707f0..fce1b61f 100755
--- a/trustgraph-flow/trustgraph/chunking/recursive/chunker.py
+++ b/trustgraph-flow/trustgraph/chunking/recursive/chunker.py
@@ -28,8 +28,6 @@ class Processor(InputOutputProcessor):
         super(Processor, self).__init__(
             **params | {
                 "subscriber": subscriber,
-                "input_schema": TextDocument,
-                "output_schema": Chunk,
             }
         )
 
@@ -48,6 +46,17 @@ class Processor(InputOutputProcessor):
             is_separator_regex=False,
         )
 
+        self.register_consumer(
+            name = "input",
+            schema = TextDocument,
+            handler = self.on_message,
+        )
+
+        self.register_producer(
+            name = "output",
+            schema = Chunk,
+        )
+
         print("Chunker initialised", flush=True)
 
     async def on_message(self, msg, consumer):
@@ -72,7 +81,7 @@ class Processor(InputOutputProcessor):
                 id=consumer.id, flow=consumer.flow
             ).observe(len(chunk.page_content))
 
-            await consumer.q.output.send(r)
+            await consumer.q["output"].send(r)
 
         print("Done.", flush=True)
 
diff --git a/trustgraph-flow/trustgraph/chunking/token/chunker.py b/trustgraph-flow/trustgraph/chunking/token/chunker.py
index 497333b3..eee16324 100755
--- a/trustgraph-flow/trustgraph/chunking/token/chunker.py
+++ b/trustgraph-flow/trustgraph/chunking/token/chunker.py
@@ -28,8 +28,6 @@ class Processor(InputOutputProcessor):
         super(Processor, self).__init__(
             **params | {
                 "subscriber": subscriber,
-                "input_schema": TextDocument,
-                "output_schema": Chunk,
             }
         )
 
@@ -47,6 +45,17 @@ class Processor(InputOutputProcessor):
             chunk_overlap=chunk_overlap,
         )
 
+        self.register_consumer(
+            name = "input",
+            schema = TextDocument,
+            handler = self.on_message,
+        )
+
+        self.register_producer(
+            name = "output",
+            schema = Chunk,
+        )
+
         print("Chunker initialised", flush=True)
 
     async def on_message(self, msg, consumer):
@@ -71,7 +80,7 @@ class Processor(InputOutputProcessor):
                 id=consumer.id, flow=consumer.flow
             ).observe(len(chunk.page_content))
 
-            await consumer.q.output.send(r)
+            await consumer.q["output"].send(r)
 
         print("Done.", flush=True)
 
diff --git a/trustgraph-flow/trustgraph/decoding/mistral_ocr/processor.py b/trustgraph-flow/trustgraph/decoding/mistral_ocr/processor.py
index 669ec1b5..e42d1601 100755
--- a/trustgraph-flow/trustgraph/decoding/mistral_ocr/processor.py
+++ b/trustgraph-flow/trustgraph/decoding/mistral_ocr/processor.py
@@ -17,12 +17,10 @@ from mistralai.models import OCRResponse
 from ... schema import Document, TextDocument, Metadata
 from ... schema import document_ingest_queue, text_ingest_queue
 from ... log_level import LogLevel
-from ... base import ConsumerProducer
+from ... base import InputOutputProcessor
 
 module = "ocr"
 
-default_input_queue = document_ingest_queue
-default_output_queue = text_ingest_queue
 default_subscriber = module
 default_api_key = os.getenv("MISTRAL_TOKEN")
 
@@ -71,19 +69,17 @@ def get_combined_markdown(ocr_response: OCRResponse) -> str:
 
     return "\n\n".join(markdowns)
 
-class Processor(ConsumerProducer):
+class Processor(InputOutputProcessor):
 
     def __init__(self, **params):
 
-        input_queue = params.get("input_queue", default_input_queue)
-        output_queue = params.get("output_queue", default_output_queue)
+        id = params.get("id")
         subscriber = params.get("subscriber", default_subscriber)
         api_key = params.get("api_key", default_api_key)
 
         super(Processor, self).__init__(
             **params | {
-                "input_queue": input_queue,
-                "output_queue": output_queue,
+                "id": id,
                 "subscriber": subscriber,
                 "input_schema": Document,
                 "output_schema": TextDocument,
@@ -151,7 +147,7 @@ class Processor(ConsumerProducer):
 
             return markdown
 
-    async def handle(self, msg):
+    async def on_message(self, msg, consumer):
 
         print("PDF message received")
 
@@ -166,17 +162,14 @@ class Processor(ConsumerProducer):
             text=markdown.encode("utf-8"),
         )
 
-        await self.send(r)
+        await consumer.q.output.send(r)
 
         print("Done.", flush=True)
 
     @staticmethod
     def add_args(parser):
 
-        ConsumerProducer.add_args(
-            parser, default_input_queue, default_subscriber,
-            default_output_queue,
-        )
+        InputOutputProcessor.add_args(parser, default_subscriber)
 
         parser.add_argument(
             '-k', '--api-key',
diff --git a/trustgraph-flow/trustgraph/decoding/pdf/pdf_decoder.py b/trustgraph-flow/trustgraph/decoding/pdf/pdf_decoder.py
index 689df9c4..08f427aa 100755
--- a/trustgraph-flow/trustgraph/decoding/pdf/pdf_decoder.py
+++ b/trustgraph-flow/trustgraph/decoding/pdf/pdf_decoder.py
@@ -22,17 +22,26 @@ class Processor(InputOutputProcessor):
     def __init__(self, **params):
 
         id = params.get("id")
-        subscriber = params.get("subscriber")
+        subscriber = params.get("subscriber", default_subscriber)
 
         super(Processor, self).__init__(
             **params | {
                 "id": id,
                 "subscriber": subscriber,
-                "input_schema": Document,
-                "output_schema": TextDocument,
             }
         )
 
+        self.register_consumer(
+            name = "input",
+            schema = Document,
+            handler = self.on_message,
+        )
+
+        self.register_producer(
+            name = "output",
+            schema = TextDocument,
+        )
+
         print("PDF inited", flush=True)
 
     async def on_message(self, msg, consumer):
@@ -62,7 +71,7 @@ class Processor(InputOutputProcessor):
                         text=page.page_content.encode("utf-8"),
                     )
 
-                    await consumer.q.output.send(r)
+                    await consumer.q["output"].send(r)
 
         print("Done.", flush=True)
 
diff --git a/trustgraph-flow/trustgraph/embeddings/document_embeddings/embeddings.py b/trustgraph-flow/trustgraph/embeddings/document_embeddings/embeddings.py
index 1a51d559..55074634 100755
--- a/trustgraph-flow/trustgraph/embeddings/document_embeddings/embeddings.py
+++ b/trustgraph-flow/trustgraph/embeddings/document_embeddings/embeddings.py
@@ -11,20 +11,17 @@ from ... schema import document_embeddings_store_queue
 from ... schema import embeddings_request_queue, embeddings_response_queue
 from ... clients.embeddings_client import EmbeddingsClient
 from ... log_level import LogLevel
-from ... base import ConsumerProducer
+from ... base import InputOutputProcessor
 
 module = "document-embeddings"
 
-default_input_queue = chunk_ingest_queue
-default_output_queue = document_embeddings_store_queue
 default_subscriber = module
 
-class Processor(ConsumerProducer):
+class Processor(InputOutputProcessor):
 
     def __init__(self, **params):
 
-        input_queue = params.get("input_queue", default_input_queue)
-        output_queue = params.get("output_queue", default_output_queue)
+        id = params.get("id")
         subscriber = params.get("subscriber", default_subscriber)
         emb_request_queue = params.get(
             "embeddings_request_queue", embeddings_request_queue