diff --git a/trustgraph-base/trustgraph/base/async_processor.py b/trustgraph-base/trustgraph/base/async_processor.py
index 1203aa09..94649d41 100644
--- a/trustgraph-base/trustgraph/base/async_processor.py
+++ b/trustgraph-base/trustgraph/base/async_processor.py
@@ -47,6 +47,18 @@ class AsyncProcessor:
 
         self.config_handlers = []
 
+        self.config_sub_task = self.subscribe(
+            self.config_push_queue, config_subscriber_id,
+            schema=ConfigPush, handler=self.on_config_change,
+            start_of_messages=True
+        )
+
+        self.running = True
+
+    def stop(self):
+        self.client.close()
+        self.running = False
+
     @property
     def pulsar_host(self): return self.client.pulsar_host
 
@@ -55,14 +67,21 @@ class AsyncProcessor:
 
     async def start(self):
 
-        self.config_sub_task = self.subscribe(
-            self.config_push_queue, config_subscriber_id,
-            schema=ConfigPush, handler=self.on_config_change
-        )
+        print("STARTING CONFIG SUB", flush=True)
+        await self.config_sub_task.start()
 
-    async def on_config_change(self, config, version):
+    async def on_config_change(self, message, consumer):
+
+        config = message.value().config
+        version = message.value().version
+
+        consumer.acknowledge(message)
+
+        print("Config change event", config, version, flush=True)
         for ch in self.config_handlers:
-            ch(config, version)
+            print("Invoke... handler...", flush=True)
+            await ch(config, version)
+
 
     async def run_config_queue(self):
 
@@ -73,11 +92,15 @@ class AsyncProcessor:
         print("Config thread running", flush=True)
 
     async def run(self):
-        while True:
+        while self.running:
             await asyncio.sleep(2)
 
-    def subscribe(self, queue, subscriber, schema, handler, metrics=None):
+    def subscribe(
+            self, queue, subscriber, schema, handler, metrics=None,
+            start_of_messages=False
+    ):
 
+        print("Processing subscription!!!!")
         return Consumer(
             taskgroup = self.taskgroup,
             client = self.client,
@@ -86,6 +109,7 @@ class AsyncProcessor:
             schema = schema,
             handler = handler,
             metrics = metrics,
+            start_of_messages=start_of_messages,
         )
 
     def publish(self, queue, schema, metrics=None):
@@ -123,15 +147,23 @@ class AsyncProcessor:
 
         async with asyncio.TaskGroup() as tg:
 
-            p = cls(**args | { "taskgroup": tg })
+            try:
+                print("CREATING...", flush=True)
 
-            # FIXME: Two sort of 'ident' things going on here?
-            p.module = ident
-            p.config_ident = args.get("ident", "FIXME")
+                p = cls(**args | { "taskgroup": tg })
 
-            await p.start()
+                # FIXME: Two sort of 'ident' things going on here?
+                p.module = ident
+                p.config_ident = args.get("ident", "FIXME")
 
-            task2 = tg.create_task(p.run())
+                print("STARTING...", flush=True)
+                await p.start()
+
+                task2 = tg.create_task(p.run())
+
+            except Exception as e:
+                print("Exception, dropping out", flush=True)
+                raise e
 
     @classmethod
     def launch(cls, ident, doc):
@@ -152,13 +184,15 @@ class AsyncProcessor:
         args = parser.parse_args()
         args = vars(args)
 
-        print(args)
+        print(args, flush=True)
 
         if args["metrics"]:
             start_http_server(args["metrics_port"])
 
         while True:
 
+            print("Starting...", flush=True)
+
             try:
 
                 asyncio.run(cls.launch_async(
@@ -166,23 +200,23 @@ class AsyncProcessor:
                 ))
 
             except KeyboardInterrupt:
-                print("Keyboard interrupt.")
+                print("Keyboard interrupt.", flush=True)
                 return
 
             except _pulsar.Interrupted:
-                print("Pulsar Interrupted.")
+                print("Pulsar Interrupted.", flush=True)
                 return
 
             except ExceptionGroup as e:
 
-                print("Exception group:")
+                print("Exception group:", flush=True)
 
                 for se in e.exceptions:
-                    print("  Type:", type(se))
-                    print(f"  Exception: {se}")
+                    print("  Type:", type(se), flush=True)
+                    print(f"  Exception: {se}", flush=True)
 
             except Exception as e:
-                print("Type:", type(e))
+                print("Type:", type(e), flush=True)
                 print("Exception:", e, flush=True)
 
             print("Will retry...", flush=True)
diff --git a/trustgraph-base/trustgraph/base/consumer.py b/trustgraph-base/trustgraph/base/consumer.py
index 80fb25a5..9a6a981e 100644
--- a/trustgraph-base/trustgraph/base/consumer.py
+++ b/trustgraph-base/trustgraph/base/consumer.py
@@ -9,7 +9,8 @@ class Consumer:
     def __init__(
             self, taskgroup, client, queue, subscriber, schema,
             handler, rate_limit_retry_time = 10,
-            rate_limit_timeout = 7200, metrics = None
+            rate_limit_timeout = 7200, metrics = None,
+            start_of_messages=False,
     ):
 
         self.taskgroup = taskgroup
@@ -20,19 +21,30 @@ class Consumer:
         self.handler = handler
         self.rate_limit_retry_time = rate_limit_retry_time
         self.rate_limit_timeout = rate_limit_timeout
+        self.start_of_messages = start_of_messages
 
         self.running = True
         self.task = None
 
         self.metrics = metrics
 
+        print("Consumer: __init__d")
+
+    async def stop(self):
+
+        self.running = False
+
     async def start(self):
 
         self.running = True
 
+        print("Subscribing...", flush=True)
+
         self.consumer = self.client.subscribe(
-            self.queue, self.subscriber, self.schema
+            self.queue, self.subscriber, self.schema,
+            start_of_messages = self.start_of_messages,
         )
+        print("Subscribed.", flush=True)
 
         # Puts it in the stopped state, the run thread should set running
         if self.metrics:
@@ -40,6 +52,8 @@ class Consumer:
 
         self.task = self.taskgroup.create_task(self.run())
 
+        print("Subscriber started", flush=True)
+
     async def run(self):
 
         if self.metrics:
@@ -48,8 +62,12 @@ class Consumer:
 
         while self.running:
 
+            print("Consumer: WAIT FOR MESSAGE", flush=True)
+
             msg = await asyncio.to_thread(self.consumer.receive)
 
+            print("Consumer: MESSAGE RECEIVED", flush=True)
+
             expiry = time.time() + self.rate_limit_timeout
 
             # This loop is for retry on rate-limit / resource limits
diff --git a/trustgraph-base/trustgraph/base/input_output.py b/trustgraph-base/trustgraph/base/input_output.py
index 1e72c444..82a8c513 100644
--- a/trustgraph-base/trustgraph/base/input_output.py
+++ b/trustgraph-base/trustgraph/base/input_output.py
@@ -1,4 +1,5 @@
 
+import json
 from pulsar.schema import JsonSchema
 
 from .. schema import Error
@@ -18,6 +19,8 @@ class InputOutputProcessor(AsyncProcessor):
         self.input_schema = params.get("input_schema")
         self.output_schema = params.get("output_schema")
 
+        print("alsdkjasdlasdjkL")
+
         ProcessorMetrics(id=self.id).info(
             {
                 "subscriber": self.subscriber,
@@ -26,6 +29,7 @@ class InputOutputProcessor(AsyncProcessor):
             }
         )
 
+        print("ASD")
         super(InputOutputProcessor, self).__init__(
             **params | {
                 "id": self.id,
@@ -34,6 +38,7 @@ class InputOutputProcessor(AsyncProcessor):
             }
         )
 
+        print("alsdkj")
         self.on_config(self.on_configuration)
 
         self.subs = {}
@@ -48,7 +53,7 @@ class InputOutputProcessor(AsyncProcessor):
 
         self.subs[flow] = self.subscribe(
             queue = defn["input"],
-            subscriber = subscriber,
+            subscriber = self.subscriber,
             schema = self.input_schema,
             handler = self.on_message,
             metrics = input_metrics,
@@ -60,31 +65,44 @@ class InputOutputProcessor(AsyncProcessor):
             metrics = output_metrics,
         )
 
-        self.subs[flow].start()
+        await self.subs[flow].start()
 
         print("Started flow for", flow)
 
+    async def stop_handler(self, flow):
+        print("Stopping ", flow, flush=True)
+        await self.subs[flow].stop()
+        del self.subs[flow]
+        del self.pubs[flow]
+        
+
     async def on_configuration(self, config, version):
 
+        print("Got config version", version)
+
         if "flows" not in config: return
 
         if self.id in config["flows"]:
 
-            wanted_keys = config["flows"][self.id].keys()
+            flow_config = json.loads(config["flows"][self.id])
+
+            wanted_keys = flow_config.keys()
             current_keys = self.subs.keys()
 
             for key in wanted_keys:
                 if key not in current_keys:
-                    self.start_handler(key, config["flows"][key])
+                    await self.start_handler(key, flow_config[key])
 
             for key in current_keys:
                 if key not in wanted_keys:
-                    self.stop_handler(key, config["flows"][key])
+                    await self.stop_handler(key)
 
             print("Handled config update")
 
     async def start(self):
-        pass
+
+        print("INPUT OUTPU START")
+        await super(InputOutputProcessor, self).start()
 
     @staticmethod
     def add_args(parser, default_subscriber):
diff --git a/trustgraph-base/trustgraph/base/pubsub.py b/trustgraph-base/trustgraph/base/pubsub.py
index c9c4ed92..3c727a67 100644
--- a/trustgraph-base/trustgraph/base/pubsub.py
+++ b/trustgraph-base/trustgraph/base/pubsub.py
@@ -51,12 +51,18 @@ class PulsarClient:
             if self.client:
                 self.client.close()
 
-    def subscribe(self, queue, subscriber, schema):
+    def subscribe(self, queue, subscriber, schema, start_of_messages=False):
+
+        if start_of_messages:
+            pos = pulsar.InitialPosition.Earliest
+        else:
+            pos =  pulsar.InitialPosition.Latest
 
         return self.client.subscribe(
             queue, subscriber,
             consumer_type=pulsar.ConsumerType.Shared,
             schema=JsonSchema(schema),
+            initial_position=pos,
         )
 
     def publish(self, queue, schema):
diff --git a/trustgraph-base/trustgraph/base/subscriber.py b/trustgraph-base/trustgraph/base/subscriber.py
index 30ade3ee..3b1f158d 100644
--- a/trustgraph-base/trustgraph/base/subscriber.py
+++ b/trustgraph-base/trustgraph/base/subscriber.py
@@ -31,6 +31,8 @@ class Subscriber:
 
     def run(self):
 
+        print("SUBSCRIBER LOOP KICKS IN", flush=True)
+
         while self.running:
 
             try:
@@ -42,10 +44,14 @@ class Subscriber:
                     schema=self.schema,
                 )
 
+                print("SUBSCRIBER RUNNING...", flush=True)
+
                 while self.running:
 
                     msg = consumer.receive()
 
+                    print("GOT MESSAGE...", flush=True)
+
                     # Acknowledge successful reception of the message
                     consumer.acknowledge(msg)
 
@@ -73,7 +79,7 @@ class Subscriber:
                                 pass
 
             except Exception as e:
-                print("Exception:", e, flush=True)
+                print("Subscriber exception:", e, flush=True)
          
             # If handler drops out, sleep a retry
             time.sleep(2)
diff --git a/trustgraph-flow/trustgraph/config/service/config.py b/trustgraph-flow/trustgraph/config/service/config.py
index f96f3ea5..597fb443 100644
--- a/trustgraph-flow/trustgraph/config/service/config.py
+++ b/trustgraph-flow/trustgraph/config/service/config.py
@@ -11,11 +11,14 @@ class Configuration(dict):
     # one config service.  Should be more than one, and use a
     # back-end state store.
 
-    def __init__(self):
+    def __init__(self, push):
 
         # Version counter
         self.version = 0
 
+        # External function to respond to update
+        self.push = push
+
     def __getitem__(self, key):
         if key not in self:
             self[key] = ConfigurationItems()
@@ -169,6 +172,8 @@ class Configuration(dict):
 
     async def handle(self, msg):
 
+        print("Handle message ", msg.operation)
+
         if msg.operation == "get":
 
             resp = await self.handle_get(msg)
diff --git a/trustgraph-flow/trustgraph/config/service/service.py b/trustgraph-flow/trustgraph/config/service/service.py
index 93f4101b..ac9ca053 100644
--- a/trustgraph-flow/trustgraph/config/service/service.py
+++ b/trustgraph-flow/trustgraph/config/service/service.py
@@ -81,7 +81,7 @@ class Processor(AsyncProcessor):
             metrics = self.request_metrics,
         )
 
-        self.config = Configuration()
+        self.config = Configuration(self.push)
 
         print("Service initialised.")
 
@@ -103,7 +103,7 @@ class Processor(AsyncProcessor):
 
         await self.push_pub.send(resp)
 
-        print("Pushed.")
+        print("Pushed version ", self.config.version)
         
     async def on_message(self, msg, consumer):
 
diff --git a/trustgraph-flow/trustgraph/decoding/pdf/pdf_decoder.py b/trustgraph-flow/trustgraph/decoding/pdf/pdf_decoder.py
index 0fabe0aa..3cdb8e3f 100755
--- a/trustgraph-flow/trustgraph/decoding/pdf/pdf_decoder.py
+++ b/trustgraph-flow/trustgraph/decoding/pdf/pdf_decoder.py
@@ -33,7 +33,7 @@ class Processor(InputOutputProcessor):
             }
         )
 
-        print("PDF inited")
+        print("PDF inited", flush=True)
 
     async def on_message(self, msg, consumer):