Remove cruft

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

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)

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

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()