apunkt/trustgraph
1
0
Fork 0
mirror of https://github.com/trustgraph-ai/trustgraph.git synced 2026-05-22 22:05:13 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Processors use shared queues, means there can be more than process on a queue to share load (#265)

Browse source
This commit is contained in:
cybermaggedon 2025-01-11 18:10:04 +00:00 • committed by GitHub
parent c603caa3cc
commit cd9a208432
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 4 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -1,5 +1,6 @@
from pulsar.schema import JsonSchema from pulsar.schema import JsonSchema
import pulsar
from prometheus_client import Histogram, Info, Counter, Enum from prometheus_client import Histogram, Info, Counter, Enum
import time import time
@ -51,6 +52,7 @@ class Consumer(BaseProcessor):
self.consumer = self.client.subscribe( self.consumer = self.client.subscribe(
input_queue, subscriber, input_queue, subscriber,
consumer_type=pulsar.ConsumerType.Shared,
schema=JsonSchema(input_schema), schema=JsonSchema(input_schema),
) )

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

@ -1,5 +1,6 @@
from pulsar.schema import JsonSchema from pulsar.schema import JsonSchema
import pulsar
from prometheus_client import Histogram, Info, Counter, Enum from prometheus_client import Histogram, Info, Counter, Enum
import time import time
@ -71,6 +72,7 @@ class ConsumerProducer(BaseProcessor):
self.consumer = self.client.subscribe( self.consumer = self.client.subscribe(
input_queue, subscriber, input_queue, subscriber,
consumer_type=pulsar.ConsumerType.Shared,
schema=JsonSchema(input_schema), schema=JsonSchema(input_schema),
) )