apunkt/trustgraph
1
0
Fork 0
mirror of https://github.com/trustgraph-ai/trustgraph.git synced 2026-07-19 10:11:01 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

Tidying

Browse source
This commit is contained in:
Cyber MacGeddon 2025-04-14 20:44:22 +01:00
parent 74fb52dd3b
commit 726baa9932
2 changed files with 34 additions and 28 deletions
Download patch file Download diff file Expand all files Collapse all files

1
trustgraph-base/trustgraph/base/async_processor.py
View file

@ -119,7 +119,6 @@ class AsyncProcessor:
await p.start() await p.start()
# task1 = tg.create_task(p.run_config_queue())
task2 = tg.create_task(p.run()) task2 = tg.create_task(p.run())
@classmethod @classmethod

61
trustgraph-flow/trustgraph/config/service/service.py
View file

@ -18,8 +18,8 @@ from . config import Configuration
module = "config-svc" module = "config-svc"
default_input_queue = config_request_queue default_request_queue = config_request_queue
default_output_queue = config_response_queue default_response_queue = config_response_queue
default_push_queue = config_push_queue default_push_queue = config_push_queue
default_subscriber = module default_subscriber = module
@ -30,23 +30,27 @@ class Processor(AsyncProcessor):
def __init__(self, **params): def __init__(self, **params):
input_queue = params.get("input_queue", default_input_queue) request_queue = params.get("request_queue", default_request_queue)
output_queue = params.get("output_queue", default_output_queue) response_queue = params.get("response_queue", default_response_queue)
push_queue = params.get("push_queue", default_push_queue) push_queue = params.get("push_queue", default_push_queue)
subscriber = params.get("subscriber", default_subscriber) subscriber = params.get("subscriber", default_subscriber)
id = params.get("id")
input_schema = ConfigRequest request_schema = ConfigRequest
output_schema = ConfigResponse response_schema = ConfigResponse
push_schema = ConfigResponse push_schema = ConfigResponse
self.set_processor_state("FIXME", "starting") self.set_processor_state(id, "starting")
self.set_pubsub_info( self.set_pubsub_info(
"FIXME", id
{ {
"input_queue": input_queue, "id": id,
"subscriber": subscriber, "subscriber": subscriber,
"input_schema": input_schema.__name__, "request_queue": request_queue,
"request_schema": request_schema.__name__,
"response_queue": response_queue,
"response_schema": request_schema.__name__,
# "rate_limit_retry": str(self.rate_limit_retry), # "rate_limit_retry": str(self.rate_limit_retry),
# "rate_limit_timeout": str(self.rate_limit_timeout), # "rate_limit_timeout": str(self.rate_limit_timeout),
} }
@ -54,19 +58,12 @@ class Processor(AsyncProcessor):
super(Processor, self).__init__( super(Processor, self).__init__(
**params | { **params | {
"input_schema": input_schema.__name__, "request_schema": request_schema.__name__,
"output_schema": output_schema.__name__, "response_schema": response_schema.__name__,
"push_schema": push_schema.__name__, "push_schema": push_schema.__name__,
} }
) )
self.subs = self.subscribe(
queue = input_queue,
subscriber = subscriber,
schema = input_schema,
handler = self.on_message,
)
if not hasattr(__class__, "request_metric"): if not hasattr(__class__, "request_metric"):
__class__.request_metric = Histogram( __class__.request_metric = Histogram(
'request_latency', 'Request latency (seconds)' 'request_latency', 'Request latency (seconds)'
@ -84,14 +81,18 @@ class Processor(AsyncProcessor):
) )
self.out_pub = self.publish( self.out_pub = self.publish(
queue = output_queue, queue = response_queue,
schema = ConfigResponse schema = ConfigResponse
) )
self.config = Configuration() self.subs = self.subscribe(
queue = request_queue,
subscriber = subscriber,
schema = request_schema,
handler = self.on_message,
)
# Version counter self.config = Configuration()
self.version = 0
print("Service initialised.") print("Service initialised.")
@ -153,13 +154,19 @@ class Processor(AsyncProcessor):
AsyncProcessor.add_args(parser) AsyncProcessor.add_args(parser)
parser.add_argument( parser.add_argument(
'-i', '--input-queue', '-q', '--request-queue',
default=default_input_queue, default=default_request_queue,
help=f'Input queue (default: {default_input_queue})' help=f'Request queue (default: {default_request_queue})'
) )
parser.add_argument( parser.add_argument(
'-q', '--push-queue', '-r', '--response-queue',
default=default_response_queue,
help=f'Response queue {default_response_queue}',
)
parser.add_argument(
'-P', '--push-queue',
default=default_push_queue, default=default_push_queue,
help=f'Config push queue (default: {default_push_queue})' help=f'Config push queue (default: {default_push_queue})'
) )