Processors use shared queues, means there can be more than process on…

… a queue to share load (#265)
trustgraph-ai · Jan 11, 2025 · cd9a208 · cd9a208
1 parent c603caa
commit cd9a208
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 0 deletions.
diff --git a/trustgraph-base/trustgraph/base/consumer.py b/trustgraph-base/trustgraph/base/consumer.py
@@ -1,5 +1,6 @@
 
 from pulsar.schema import JsonSchema
+import pulsar
 from prometheus_client import Histogram, Info, Counter, Enum
 import time
 
@@ -51,6 +52,7 @@ def __init__(self, **params):
 
         self.consumer = self.client.subscribe(
             input_queue, subscriber,
+            consumer_type=pulsar.ConsumerType.Shared,
             schema=JsonSchema(input_schema),
         )
 

diff --git a/trustgraph-base/trustgraph/base/consumer_producer.py b/trustgraph-base/trustgraph/base/consumer_producer.py
@@ -1,5 +1,6 @@
 
 from pulsar.schema import JsonSchema
+import pulsar
 from prometheus_client import Histogram, Info, Counter, Enum
 import time
 
@@ -71,6 +72,7 @@ def __init__(self, **params):
 
         self.consumer = self.client.subscribe(
             input_queue, subscriber,
+            consumer_type=pulsar.ConsumerType.Shared,
             schema=JsonSchema(input_schema),
         )