apunkt/trustgraph
1
0
Fork 0
mirror of https://github.com/trustgraph-ai/trustgraph.git synced 2026-07-20 18:51:03 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

- Rename 'client' to 'pulsar_client' in async_processor. Less likely to

Browse source 
be a name clash.
- Turned on Metrics API in api-gateway.
This commit is contained in:
Cyber MacGeddon 2025-04-23 11:28:19 +01:00
parent 93a50ab218
commit 950c624013
7 changed files with 19 additions and 19 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -30,7 +30,7 @@ class AsyncProcessor:
self.id = params.get("id")
# Register a pulsar client
self.pulsar_client = PulsarClient(**params)
self.pulsar_client_object = PulsarClient(**params)
# Initialise metrics, records the parameters
ProcessorMetrics(processor = self.id).info({
@ -61,7 +61,7 @@ class AsyncProcessor:
self.config_sub_task = Consumer(
taskgroup = self.taskgroup,
client = self.client,
client = self.pulsar_client,
subscriber = config_subscriber_id,
flow = None,
@ -85,16 +85,16 @@ class AsyncProcessor:
# This is called to stop all threads. An over-ride point for extra
# functionality
def stop(self):
self.client.close()
self.pulsar_client.close()
self.running = False
# Returns the pulsar host
@property
def pulsar_host(self): return self.client.pulsar_host
def pulsar_host(self): return self.pulsar_client_object.pulsar_host
# Returns the pulsar client
@property
def client(self): return self.pulsar_client.client
def pulsar_client(self): return self.pulsar_client_object.client
# Register a new event handler for configuration change
def register_config_handler(self, handler):

2
trustgraph-base/trustgraph/base/consumer_spec.py
View file

@ -18,7 +18,7 @@ class ConsumerSpec(Spec):
consumer = Consumer(
taskgroup = processor.taskgroup,
flow = flow,
client = processor.client,
client = processor.pulsar_client,
topic = definition[self.name],
subscriber = processor.id + "--" + self.name,
schema = self.schema,

2
trustgraph-base/trustgraph/base/producer_spec.py
View file

@ -15,7 +15,7 @@ class ProducerSpec(Spec):
)
producer = Producer(
client = processor.client,
client = processor.pulsar_client,
topic = definition[self.name],
schema = self.schema,
metrics = producer_metrics,

2
trustgraph-base/trustgraph/base/request_response_spec.py
View file

@ -121,7 +121,7 @@ class RequestResponseSpec(Spec):
)
rr = self.impl(
client = processor.client,
client = processor.pulsar_client,
subscription = flow.id,
consumer_name = flow.id,
request_topic = definition[self.request_name],

2
trustgraph-base/trustgraph/base/subscriber_spec.py
View file

@ -17,7 +17,7 @@ class SubscriberSpec(Spec):
)
subscriber = Subscriber(
client = processor.client,
client = processor.pulsar_client,
topic = definition[self.name],
subscription = flow.id,
consumer_name = flow.id,

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

@ -87,7 +87,7 @@ class Processor(AsyncProcessor):
self.config_request_consumer = Consumer(
taskgroup = self.taskgroup,
client = self.client,
client = self.pulsar_client,
flow = None,
topic = config_request_queue,
subscriber = id,
@ -97,14 +97,14 @@ class Processor(AsyncProcessor):
)
self.config_response_producer = Producer(
client = self.client,
client = self.pulsar_client,
topic = config_response_queue,
schema = ConfigResponse,
metrics = config_response_metrics,
)
self.config_push_producer = Producer(
client = self.client,
client = self.pulsar_client,
topic = config_push_queue,
schema = ConfigPush,
metrics = config_push_metrics,
@ -112,7 +112,7 @@ class Processor(AsyncProcessor):
self.flow_request_consumer = Consumer(
taskgroup = self.taskgroup,
client = self.client,
client = self.pulsar_client,
flow = None,
topic = flow_request_queue,
subscriber = id,
@ -122,7 +122,7 @@ class Processor(AsyncProcessor):
)
self.flow_response_producer = Producer(
client = self.client,
client = self.pulsar_client,
topic = flow_response_queue,
schema = FlowResponse,
metrics = flow_response_metrics,

10
trustgraph-flow/trustgraph/gateway/service.py
View file

@ -266,11 +266,11 @@ class Api:
# auth = self.auth,
# services = self.services,
# ),
# MetricsEndpoint(
# endpoint_path = "/api/v1/metrics",
# prometheus_url = self.prometheus_url,
# auth = self.auth,
# ),
MetricsEndpoint(
endpoint_path = "/api/v1/metrics",
prometheus_url = self.prometheus_url,
auth = self.auth,
),
]
for ep in self.endpoints: