Roughly working on config svc

Makefile

@@ -51,14 +51,14 @@ container: update-package-versions
 	    -t ${CONTAINER_BASE}/trustgraph-base:${VERSION} .
 	${DOCKER} build -f containers/Containerfile.flow \
 	    -t ${CONTAINER_BASE}/trustgraph-flow:${VERSION} .
-	${DOCKER} build -f containers/Containerfile.bedrock \
+#	${DOCKER} build -f containers/Containerfile.bedrock \
-	    -t ${CONTAINER_BASE}/trustgraph-bedrock:${VERSION} .
+#	    -t ${CONTAINER_BASE}/trustgraph-bedrock:${VERSION} .
-	${DOCKER} build -f containers/Containerfile.vertexai \
+#	${DOCKER} build -f containers/Containerfile.vertexai \
-	    -t ${CONTAINER_BASE}/trustgraph-vertexai:${VERSION} .
+#	    -t ${CONTAINER_BASE}/trustgraph-vertexai:${VERSION} .
-	${DOCKER} build -f containers/Containerfile.hf \
+#	${DOCKER} build -f containers/Containerfile.hf \
-	    -t ${CONTAINER_BASE}/trustgraph-hf:${VERSION} .
+#	    -t ${CONTAINER_BASE}/trustgraph-hf:${VERSION} .
-	${DOCKER} build -f containers/Containerfile.ocr \
+#	${DOCKER} build -f containers/Containerfile.ocr \
-	    -t ${CONTAINER_BASE}/trustgraph-ocr:${VERSION} .
+#	    -t ${CONTAINER_BASE}/trustgraph-ocr:${VERSION} .
 
 basic-containers: update-package-versions
 	${DOCKER} build -f containers/Containerfile.base \

trustgraph-base/trustgraph/base/__init__.py

@@ -3,4 +3,6 @@ from . pubsub import PulsarClient
 from . async_processor import AsyncProcessor
 from . consumer import Consumer
 from . producer import Producer
+from . publisher import Publisher
+from . subscriber import Subscriber
 

trustgraph-base/trustgraph/base/async_processor.py

@@ -162,17 +162,20 @@ class AsyncProcessor:
                 print("Pulsar Interrupted.")
                 return
 
+            except ExceptionGroup as e:
+
+                print("Exception group:")
+
+                for se in e.exceptions:
+                    print("   Type:", type(se))
+                    print(f"  Exception: {se}")
+
             except Exception as e:
-
+                print("Type:", type(e))
-                print(type(e))
-
-#                print(e.message)
-#                print(e.exceptions)
-
                 print("Exception:", e, flush=True)
-                print("Will retry...", flush=True)
 
-                time.sleep(4)
+            print("Will retry...", flush=True)
+            time.sleep(4)
 
     @staticmethod
     def add_args(parser):

trustgraph-flow/trustgraph/config/service/service.py

@@ -12,7 +12,7 @@ from trustgraph.schema import ConfigValue, Error
 from trustgraph.schema import config_request_queue, config_response_queue
 from trustgraph.schema import config_push_queue
 from trustgraph.log_level import LogLevel
-from trustgraph.base import ConsumerProducer, BaseProcessor
+from trustgraph.base import AsyncProcessor, Consumer, Producer
 
 module = "config-svc"
 
@@ -34,7 +34,7 @@ class Configuration(dict):
             self[key] = ConfigurationItems()
         return dict.__getitem__(self, key)
         
-class Processor(BaseProcessor):
+class Processor(AsyncProcessor):
 
     def __init__(self, **params):
         
@@ -69,10 +69,10 @@ class Processor(BaseProcessor):
         )
 
         self.subs = self.subscribe(
-            input_queue=input_queue,
+            queue = input_queue,
-            subscriber=subscriber,
+            subscriber = subscriber,
-            schema=input_schema,
+            schema = input_schema,
-            handler=self.on_message,
+            handler = self.on_message,
         )
 
         if not hasattr(__class__, "request_metric"):
@@ -87,12 +87,12 @@ class Processor(BaseProcessor):
             )
 
         self.push_pub = self.publish(
-            output_queue = push_queue,
+            queue = push_queue,
             schema = ConfigPush
         )
 
         self.out_pub = self.publish(
-            output_queue = output_queue,
+            queue = output_queue,
             schema = ConfigResponse
         )
 
@@ -107,6 +107,7 @@ class Processor(BaseProcessor):
         print("Service initialised.")
 
     async def start(self):
+
         await self.push()
         await self.subs.start()
         
@@ -337,10 +338,7 @@ class Processor(BaseProcessor):
     @staticmethod
     def add_args(parser):
 
-        ConsumerProducer.add_args(
+        AsyncProcessor.add_args(parser)
-            parser, default_input_queue, default_subscriber,
-            default_output_queue,
-        )
 
         parser.add_argument(
             '-i', '--input-queue',