PDF done but not working

trustgraph-base/trustgraph/base/__init__.py

@@ -6,4 +6,5 @@ from . producer import Producer
 from . publisher import Publisher
 from . subscriber import Subscriber
 from . metrics import ProcessorMetrics, ConsumerMetrics, ProducerMetrics
+from . input_output import InputOutputProcessor
 

trustgraph-base/trustgraph/base/async_processor.py

@@ -50,6 +50,9 @@ class AsyncProcessor:
     @property
     def pulsar_host(self): return self.client.pulsar_host
 
+    def on_config(self, handler):
+        self.config_handlers.append(handler)
+
     async def start(self):
 
         self.config_sub_task = self.subscribe(

trustgraph-base/trustgraph/base/input_output.py

@@ -0,0 +1,117 @@
+
+from pulsar.schema import JsonSchema
+
+from .. schema import Error
+from .. schema import config_request_queue, config_response_queue
+from .. schema import config_push_queue
+from .. log_level import LogLevel
+from .. base import AsyncProcessor, Consumer, Producer
+
+from .. base import ProcessorMetrics, ConsumerMetrics, ProducerMetrics
+
+class InputOutputProcessor(AsyncProcessor):
+
+    def __init__(self, **params):
+        
+        self.id = params.get("id")
+        self.subscriber = params.get("subscriber")
+        self.input_schema = params.get("input_schema")
+        self.output_schema = params.get("output_schema")
+
+        ProcessorMetrics(id=self.id).info(
+            {
+                "subscriber": self.subscriber,
+                "input_schema": self.input_schema.__name__,
+                "output_schema": self.output_schema.__name__,
+            }
+        )
+
+        super(InputOutputProcessor, self).__init__(
+            **params | {
+                "id": self.id,
+                "input_schema": self.input_schema.__name__,
+                "output_schema": self.output_schema.__name__,
+            }
+        )
+
+        self.on_config(self.on_configuration)
+
+        self.subs = {}
+        self.pubs = {}
+
+        print("Service initialised.")
+
+    async def start_handler(self, flow, defn):
+
+        input_metrics = ConsumerMetrics(self.id, flow)
+        output_metrics = ProducerMetrics(self.id, flow)
+
+        self.subs[flow] = self.subscribe(
+            queue = defn["input"],
+            subscriber = subscriber,
+            schema = self.input_schema,
+            handler = self.on_message,
+            metrics = input_metrics,
+        )
+        
+        self.pubs[flow] = self.publish(
+            queue = defn["output"],
+            schema = self.output_schema,
+            metrics = output_metrics,
+        )
+
+        self.subs[flow].start()
+
+        print("Started flow for", flow)
+
+    async def on_configuration(self, config, version):
+
+        if "flows" not in config: return
+
+        if self.id in config["flows"]:
+
+            wanted_keys = config["flows"][self.id].keys()
+            current_keys = self.subs.keys()
+
+            for key in wanted_keys:
+                if key not in current_keys:
+                    self.start_handler(key, config["flows"][key])
+
+            for key in current_keys:
+                if key not in wanted_keys:
+                    self.stop_handler(key, config["flows"][key])
+
+            print("Handled config update")
+
+    async def start(self):
+        pass
+
+    @staticmethod
+    def add_args(parser, default_subscriber):
+
+        AsyncProcessor.add_args(parser)
+
+        parser.add_argument(
+            '-s', '--subscriber',
+            default=default_subscriber,
+            help=f'Queue subscriber name (default: {default_subscriber})'
+        )
+
+        # parser.add_argument(
+        #     '--rate-limit-retry',
+        #     type=int,
+        #     default=default_rate_limit_retry,
+        #     help=f'Rate limit retry (default: {default_rate_limit_retry})'
+        # )
+
+        # parser.add_argument(
+        #     '--rate-limit-timeout',
+        #     type=int,
+        #     default=default_rate_limit_timeout,
+        #     help=f'Rate limit timeout (default: {default_rate_limit_timeout})'
+        # )
+
+def run():
+
+    Processor.launch(module, __doc__)
+

trustgraph-base/trustgraph/base/request_response.py

@@ -0,0 +1,118 @@
+
+from pulsar.schema import JsonSchema
+
+from .. schema import Error
+from .. schema import config_request_queue, config_response_queue
+from .. schema import config_push_queue
+from .. log_level import LogLevel
+from .. base import AsyncProcessor, Consumer, Producer
+
+from .. base import ProcessorMetrics, ConsumerMetrics, ProducerMetrics
+
+class RequestResponseProcessor(AsyncProcessor):
+
+    def __init__(self, **params):
+        
+        id = params.get("id")
+        self.subscriber = params.get("subscriber")
+        self.request_schema = params.get("request_schema")
+        self.response_schema = params.get("response_schema")
+
+        ProcessorMetrics(id=id).info(
+            {
+                "subscriber": self.subscriber,
+                "request_schema": self.request_schema.__name__,
+                "response_schema": self.response_schema.__name__,
+            }
+        )
+
+        super(Processor, self).__init__(
+            **params | {
+                "request_schema": self.request_schema.__name__,
+                "response_schema": self.response_schema.__name__,
+            }
+        )
+
+        self.on_config(self.on_configuration)
+
+        self.subs = {}
+        self.pubs = {}
+
+        print("Service initialised.")
+
+    async def start_handler(self, flow, defn):
+
+        request_metrics = ConsumerMetrics(self.id, flow)
+        response_metrics = ProducerMetrics(self.id, flow)
+
+        self.subs[flow] = self.subscribe(
+            queue = defn["input"]
+            subscriber = subscriber,
+            schema = self.request_schema,
+            handler = self.on_message,
+            metrics = request_metrics,
+        )
+        
+        self.pubs[flow] = self.publish(
+            queue = defn["output"],
+            schema = self.response_schema,
+            metrics = response_metrics,
+        )
+
+        self.subs[flow].start()
+
+        print("Started flow for", flow)
+
+    async def on_configuration(self, config, version):
+
+        if "flows" not in config: return
+
+        if self.id in config["flows"]:
+
+            wanted_keys = config["flows"][self.id].keys()
+            current_keys = self.subs.keys()
+
+            for key in wanted_keys:
+                if key not in current_keys:
+                    self.start_handler(key, config["flows"][key])
+
+            for key in current_keys:
+                if key not in wanted_keys:
+                    self.stop_handler(key, config["flows"][key])
+
+            print("Handled config update")
+
+    async def start(self):
+
+        await self.request_sub.start()
+        
+    async def on_message(self, msg, consumer):
+
+        try:
+
+            v = msg.value()
+
+            # Sender-produced ID
+            id = msg.properties()["id"]
+
+            print(f"Handling {id}...", flush=True)
+
+            consumer.acknowledge(msg)
+
+        except Exception as e:
+            
+            resp = "error"
+
+            await self.response_pub.send(resp, properties={"id": id})
+
+            consumer.acknowledge(msg)
+
+    @staticmethod
+    def add_args(parser):
+
+        AsyncProcessor.add_args(parser)
+
+def run():
+
+    Processor.launch(module, __doc__)
+