Fix/get multiple flows working (#355)

* Reduce log output

* Fix problems

trustgraph-base/trustgraph/base/async_processor.py

@@ -114,7 +114,7 @@ class AsyncProcessor:
         version = message.value().version
 
         # Invoke message handlers
-        print("Config change event", config, version, flush=True)
+        print("Config change event", version, flush=True)
         for ch in self.config_handlers:
             await ch(config, version)
 

trustgraph-base/trustgraph/base/consumer_spec.py

@@ -20,7 +20,7 @@ class ConsumerSpec(Spec):
             flow = flow,
             client = processor.pulsar_client,
             topic = definition[self.name],
-            subscriber = processor.id + "--" + self.name,
+            subscriber = processor.id + "--" + flow.name + "--" + self.name,
             schema = self.schema,
             handler = self.handler,
             metrics = consumer_metrics,

trustgraph-base/trustgraph/base/flow_processor.py

@@ -75,7 +75,8 @@ class FlowProcessor(AsyncProcessor):
         # Get list of flows which should be running and are currently
         # running
         wanted_flows = flow_config.keys()
-        current_flows = self.flows.keys()
+        # This takes a copy, needed because dict gets modified by stop_flow
+        current_flows = list(self.flows.keys())
 
         # Start all the flows which arent currently running
         for flow in wanted_flows:

trustgraph-base/trustgraph/base/request_response_spec.py

@@ -127,7 +127,9 @@ class RequestResponseSpec(Spec):
 
         rr = self.impl(
             client = processor.pulsar_client,
-            subscription = flow.id,
+            subscription = (
+                processor.id + "--" + flow.name + "--" + self.request_name
+            ),
             consumer_name = flow.id,
             request_topic = definition[self.request_name],
             request_schema = self.request_schema,

trustgraph-base/trustgraph/base/subscriber.py

@@ -20,7 +20,7 @@ class Subscriber:
         self.running = True
         self.metrics = metrics
 
-    async def __del__(self):
+    def __del__(self):
         self.running = False
 
     async def start(self):