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

Unsubscribe when subscriber/consumer closes (#374)

Browse source
This commit is contained in:
cybermaggedon 2025-05-08 12:57:33 +01:00 • committed by GitHub
parent 31b7ade44d
commit 80ec00f212
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 24 additions and 16 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -45,7 +45,9 @@ class Consumer:
if hasattr(self, "consumer"): if hasattr(self, "consumer"):
if self.consumer: if self.consumer:
self.consumer.unsubscribe()
self.consumer.close() self.consumer.close()
self.consumer = None
async def stop(self): async def stop(self):
@ -108,11 +110,16 @@ class Consumer:
except Exception as e: except Exception as e:
print("consumer loop exception:", e, flush=True) print("consumer loop exception:", e, flush=True)
self.consumer.unsubscribe()
self.consumer.close() self.consumer.close()
self.consumer = None self.consumer = None
await asyncio.sleep(self.reconnect_time) await asyncio.sleep(self.reconnect_time)
continue continue
if self.consumer:
self.consumer.unsubscribe()
self.consumer.close()
async def consume(self): async def consume(self):
while self.running: while self.running:

33
trustgraph-base/trustgraph/base/subscriber.py
View file

@ -21,10 +21,21 @@ class Subscriber:
self.metrics = metrics self.metrics = metrics
self.task = None self.task = None
self.consumer = None
def __del__(self): def __del__(self):
self.running = False self.running = False
async def start(self): async def start(self):
self.consumer = self.client.subscribe(
topic = self.topic,
subscription_name = self.subscription,
consumer_name = self.consumer_name,
schema = JsonSchema(self.schema),
)
self.task = asyncio.create_task(self.run()) self.task = asyncio.create_task(self.run())
async def stop(self): async def stop(self):
@ -41,8 +52,6 @@ class Subscriber:
async def run(self): async def run(self):
consumer = None
while self.running: while self.running:
if self.metrics: if self.metrics:
@ -50,15 +59,6 @@ class Subscriber:
try: try:
# FIXME: Create consumer in start method so we know
# it is definitely running when start completes
consumer = self.client.subscribe(
topic = self.topic,
subscription_name = self.subscription,
consumer_name = self.consumer_name,
schema = JsonSchema(self.schema),
)
if self.metrics: if self.metrics:
self.metrics.state("running") self.metrics.state("running")
@ -68,7 +68,7 @@ class Subscriber:
try: try:
msg = await asyncio.to_thread( msg = await asyncio.to_thread(
consumer.receive, self.consumer.receive,
timeout_millis=250 timeout_millis=250
) )
except _pulsar.Timeout: except _pulsar.Timeout:
@ -82,7 +82,7 @@ class Subscriber:
self.metrics.received() self.metrics.received()
# Acknowledge successful reception of the message # Acknowledge successful reception of the message
consumer.acknowledge(msg) self.consumer.acknowledge(msg)
try: try:
id = msg.properties()["id"] id = msg.properties()["id"]
@ -124,9 +124,10 @@ class Subscriber:
finally: finally:
if consumer: if self.consumer:
consumer.close() self.consumer.unsubscribe()
consumer = None self.consumer.close()
self.consumer = None
if self.metrics: if self.metrics: