diff --git a/trustgraph-base/trustgraph/base/publisher.py b/trustgraph-base/trustgraph/base/publisher.py
index ce9e364e..bc302599 100644
--- a/trustgraph-base/trustgraph/base/publisher.py
+++ b/trustgraph-base/trustgraph/base/publisher.py
@@ -21,6 +21,7 @@ class Publisher:
 
     async def stop(self):
         self.running = False
+        await self.task
 
     async def join(self):
         await self.stop()
@@ -42,7 +43,7 @@ class Publisher:
                     try:
                         id, item = await asyncio.wait_for(
                             self.q.get(),
-                            timeout=0.5
+                            timeout=0.25
                         )
                     except asyncio.TimeoutError:
                         continue
@@ -57,8 +58,11 @@ class Publisher:
             except Exception as e:
                 print("Exception:", e, flush=True)
 
+            if not self.running:
+                return
+
             # If handler drops out, sleep a retry
-            time.sleep(2)
+            await asyncio.sleep(1)
 
     async def send(self, id, item):
         await self.q.put((id, item))
diff --git a/trustgraph-base/trustgraph/base/subscriber.py b/trustgraph-base/trustgraph/base/subscriber.py
index 127d2add..4f5d1455 100644
--- a/trustgraph-base/trustgraph/base/subscriber.py
+++ b/trustgraph-base/trustgraph/base/subscriber.py
@@ -28,6 +28,7 @@ class Subscriber:
 
     async def stop(self):
         self.running = False
+        await self.task
 
     async def join(self):
         await self.stop()
@@ -35,6 +36,8 @@ class Subscriber:
 
     async def run(self):
 
+        consumer = None
+
         while self.running:
 
             if self.metrics:
@@ -59,7 +62,7 @@ class Subscriber:
                     try:
                         msg = await asyncio.to_thread(
                             consumer.receive,
-                            timeout_millis=2000
+                            timeout_millis=250
                         )
                     except _pulsar.Timeout:
                         continue
@@ -91,7 +94,7 @@ class Subscriber:
                                 # FIXME: Timeout means data goes missing
                                 await asyncio.wait_for(
                                     self.q[id].put(value),
-                                    timeout=2
+                                    timeout=1
                                 )
 
                             except Exception as e:
@@ -103,7 +106,7 @@ class Subscriber:
                                 # FIXME: Timeout means data goes missing
                                 await asyncio.wait_for(
                                     q.put(value),
-                                    timeout=2
+                                    timeout=1
                                 )
                             except Exception as e:
                                 self.metrics.dropped()
@@ -112,13 +115,21 @@ class Subscriber:
             except Exception as e:
                 print("Subscriber exception:", e, flush=True)
 
-            consumer.close()
+            finally:
+
+                if consumer:
+                    consumer.close()
+                    consumer = None
+                
          
             if self.metrics:
                 self.metrics.state("stopped")
 
+            if not self.running:
+                return
+            
             # If handler drops out, sleep a retry
-            time.sleep(2)
+            await asyncio.sleep(1)
 
     async def subscribe(self, id):