diff --git a/trustgraph-base/trustgraph/base/subscriber.py b/trustgraph-base/trustgraph/base/subscriber.py index 24b7a45c..35ac4d17 100644 --- a/trustgraph-base/trustgraph/base/subscriber.py +++ b/trustgraph-base/trustgraph/base/subscriber.py @@ -43,12 +43,18 @@ class Subscriber: 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), - ) + # Build subscribe arguments + subscribe_args = { + 'topic': self.topic, + 'subscription_name': self.subscription, + 'consumer_name': self.consumer_name, + } + + # Only add schema if provided (omit if None) + if self.schema is not None: + subscribe_args['schema'] = JsonSchema(self.schema) + + self.consumer = self.client.subscribe(**subscribe_args) self.task = asyncio.create_task(self.run()) diff --git a/trustgraph-cli/trustgraph/cli/dump_queues.py b/trustgraph-cli/trustgraph/cli/dump_queues.py index ce057fed..2cbe8c21 100644 --- a/trustgraph-cli/trustgraph/cli/dump_queues.py +++ b/trustgraph-cli/trustgraph/cli/dump_queues.py @@ -9,6 +9,7 @@ Uses TrustGraph's Subscriber abstraction for future-proof pub/sub compatibility. """ import pulsar +from pulsar.schema import BytesSchema import sys import json import asyncio @@ -161,7 +162,7 @@ async def async_main(queues, output_file, pulsar_host, listener_name, subscriber topic=queue_name, subscription=subscriber_name, consumer_name=f"{subscriber_name}-{queue_name}", - schema=None, # Generic - no schema validation + schema=None, # No schema - accept any message type ) await sub.start() subscribers.append((queue_name, sub))