diff --git a/Makefile b/Makefile index 4f4de9d2..78e0ff4d 100644 --- a/Makefile +++ b/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 \ - -t ${CONTAINER_BASE}/trustgraph-bedrock:${VERSION} . - ${DOCKER} build -f containers/Containerfile.vertexai \ - -t ${CONTAINER_BASE}/trustgraph-vertexai:${VERSION} . - ${DOCKER} build -f containers/Containerfile.hf \ - -t ${CONTAINER_BASE}/trustgraph-hf:${VERSION} . - ${DOCKER} build -f containers/Containerfile.ocr \ - -t ${CONTAINER_BASE}/trustgraph-ocr:${VERSION} . +# ${DOCKER} build -f containers/Containerfile.bedrock \ +# -t ${CONTAINER_BASE}/trustgraph-bedrock:${VERSION} . +# ${DOCKER} build -f containers/Containerfile.vertexai \ +# -t ${CONTAINER_BASE}/trustgraph-vertexai:${VERSION} . +# ${DOCKER} build -f containers/Containerfile.hf \ +# -t ${CONTAINER_BASE}/trustgraph-hf:${VERSION} . +# ${DOCKER} build -f containers/Containerfile.ocr \ +# -t ${CONTAINER_BASE}/trustgraph-ocr:${VERSION} . basic-containers: update-package-versions ${DOCKER} build -f containers/Containerfile.base \ diff --git a/trustgraph-base/trustgraph/base/__init__.py b/trustgraph-base/trustgraph/base/__init__.py index 074bc286..02548d66 100644 --- a/trustgraph-base/trustgraph/base/__init__.py +++ b/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 diff --git a/trustgraph-base/trustgraph/base/async_processor.py b/trustgraph-base/trustgraph/base/async_processor.py index b355d3a2..f56393de 100644 --- a/trustgraph-base/trustgraph/base/async_processor.py +++ b/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(e)) - -# print(e.message) -# print(e.exceptions) - + print("Type:", type(e)) 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): diff --git a/trustgraph-flow/trustgraph/config/service/service.py b/trustgraph-flow/trustgraph/config/service/service.py index b05ae446..cf798680 100644 --- a/trustgraph-flow/trustgraph/config/service/service.py +++ b/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, - subscriber=subscriber, - schema=input_schema, - handler=self.on_message, + queue = input_queue, + subscriber = subscriber, + schema = input_schema, + 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( - parser, default_input_queue, default_subscriber, - default_output_queue, - ) + AsyncProcessor.add_args(parser) parser.add_argument( '-i', '--input-queue',