From c3f1aacbfc92bb6e500edd9585057acaf479d12a Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Tue, 15 Apr 2025 16:18:03 +0100 Subject: [PATCH] Remove cruft --- .../trustgraph/base/async_processor.py | 17 +---------------- trustgraph-base/trustgraph/base/consumer.py | 14 +++----------- trustgraph-base/trustgraph/base/input_output.py | 14 ++++---------- .../trustgraph/decoding/pdf/pdf_decoder.py | 2 +- 4 files changed, 9 insertions(+), 38 deletions(-) diff --git a/trustgraph-base/trustgraph/base/async_processor.py b/trustgraph-base/trustgraph/base/async_processor.py index 94649d41..0aa2b97c 100644 --- a/trustgraph-base/trustgraph/base/async_processor.py +++ b/trustgraph-base/trustgraph/base/async_processor.py @@ -66,8 +66,6 @@ class AsyncProcessor: self.config_handlers.append(handler) async def start(self): - - print("STARTING CONFIG SUB", flush=True) await self.config_sub_task.start() async def on_config_change(self, message, consumer): @@ -79,18 +77,8 @@ class AsyncProcessor: print("Config change event", config, version, flush=True) for ch in self.config_handlers: - print("Invoke... handler...", flush=True) await ch(config, version) - - async def run_config_queue(self): - - if self.module == "config.service": - print("I am config-svc, not looking at config queue", flush=True) - return - - print("Config thread running", flush=True) - async def run(self): while self.running: await asyncio.sleep(2) @@ -100,7 +88,6 @@ class AsyncProcessor: start_of_messages=False ): - print("Processing subscription!!!!") return Consumer( taskgroup = self.taskgroup, client = self.client, @@ -148,7 +135,6 @@ class AsyncProcessor: async with asyncio.TaskGroup() as tg: try: - print("CREATING...", flush=True) p = cls(**args | { "taskgroup": tg }) @@ -156,13 +142,12 @@ class AsyncProcessor: p.module = ident p.config_ident = args.get("ident", "FIXME") - print("STARTING...", flush=True) await p.start() task2 = tg.create_task(p.run()) except Exception as e: - print("Exception, dropping out", flush=True) + print("Exception, closing taskgroup", flush=True) raise e @classmethod diff --git a/trustgraph-base/trustgraph/base/consumer.py b/trustgraph-base/trustgraph/base/consumer.py index 9a6a981e..14bd7c5d 100644 --- a/trustgraph-base/trustgraph/base/consumer.py +++ b/trustgraph-base/trustgraph/base/consumer.py @@ -28,8 +28,6 @@ class Consumer: self.metrics = metrics - print("Consumer: __init__d") - async def stop(self): self.running = False @@ -38,12 +36,11 @@ class Consumer: self.running = True - print("Subscribing...", flush=True) - self.consumer = self.client.subscribe( self.queue, self.subscriber, self.schema, start_of_messages = self.start_of_messages, ) + print("Subscribed.", flush=True) # Puts it in the stopped state, the run thread should set running @@ -57,17 +54,12 @@ class Consumer: async def run(self): if self.metrics: - print("RUNNING") self.metrics.state("running") while self.running: - print("Consumer: WAIT FOR MESSAGE", flush=True) - msg = await asyncio.to_thread(self.consumer.receive) - print("Consumer: MESSAGE RECEIVED", flush=True) - expiry = time.time() + self.rate_limit_timeout # This loop is for retry on rate-limit / resource limits @@ -89,7 +81,7 @@ class Consumer: try: - print("Handle...") + print("Handle...", flush=True) if self.metrics: @@ -99,7 +91,7 @@ class Consumer: else: await self.handler(msg, self.consumer) - print("Handled.") + print("Handled.", flush=True) # Acknowledge successful processing of the message self.consumer.acknowledge(msg) diff --git a/trustgraph-base/trustgraph/base/input_output.py b/trustgraph-base/trustgraph/base/input_output.py index 82a8c513..507f669c 100644 --- a/trustgraph-base/trustgraph/base/input_output.py +++ b/trustgraph-base/trustgraph/base/input_output.py @@ -19,8 +19,6 @@ class InputOutputProcessor(AsyncProcessor): self.input_schema = params.get("input_schema") self.output_schema = params.get("output_schema") - print("alsdkjasdlasdjkL") - ProcessorMetrics(id=self.id).info( { "subscriber": self.subscriber, @@ -29,7 +27,6 @@ class InputOutputProcessor(AsyncProcessor): } ) - print("ASD") super(InputOutputProcessor, self).__init__( **params | { "id": self.id, @@ -38,7 +35,6 @@ class InputOutputProcessor(AsyncProcessor): } ) - print("alsdkj") self.on_config(self.on_configuration) self.subs = {} @@ -67,18 +63,18 @@ class InputOutputProcessor(AsyncProcessor): await self.subs[flow].start() - print("Started flow for", flow) + print("Started flow: ", flow) async def stop_handler(self, flow): - print("Stopping ", flow, flush=True) await self.subs[flow].stop() del self.subs[flow] del self.pubs[flow] - + + print("Stopped flow: ", flow, flush=True) async def on_configuration(self, config, version): - print("Got config version", version) + print("Got config version", version, flush=True) if "flows" not in config: return @@ -100,8 +96,6 @@ class InputOutputProcessor(AsyncProcessor): print("Handled config update") async def start(self): - - print("INPUT OUTPU START") await super(InputOutputProcessor, self).start() @staticmethod diff --git a/trustgraph-flow/trustgraph/decoding/pdf/pdf_decoder.py b/trustgraph-flow/trustgraph/decoding/pdf/pdf_decoder.py index 3cdb8e3f..f03d2f75 100755 --- a/trustgraph-flow/trustgraph/decoding/pdf/pdf_decoder.py +++ b/trustgraph-flow/trustgraph/decoding/pdf/pdf_decoder.py @@ -37,7 +37,7 @@ class Processor(InputOutputProcessor): async def on_message(self, msg, consumer): - print("PDF message received") + print("PDF message received", flush=True) v = msg.value()