Update base structures for plugin architecture

trustgraph-base/trustgraph/base/async_processor.py

@@ -15,7 +15,7 @@ from prometheus_client import start_http_server, Info
 
 from .. schema import ConfigPush, config_push_queue
 from .. log_level import LogLevel
-from . pubsub import PulsarClient
+from . pubsub import PulsarClient, get_pubsub
 from . producer import Producer
 from . consumer import Consumer
 from . metrics import ProcessorMetrics, ConsumerMetrics
@@ -34,7 +34,10 @@ class AsyncProcessor:
         # Store the identity
         self.id = params.get("id")
 
-        # Register a pulsar client
+        # Create pub/sub backend via factory
+        self.pubsub_backend = get_pubsub(**params)
+
+        # Keep old PulsarClient for backward compatibility with pulsar_host property
         self.pulsar_client_object = PulsarClient(**params)
 
         # Initialise metrics, records the parameters
@@ -70,7 +73,7 @@ class AsyncProcessor:
         self.config_sub_task = Consumer(
 
             taskgroup = self.taskgroup,
-            client = self.pulsar_client,
+            backend = self.pubsub_backend,  # Changed from client to backend
             subscriber = config_subscriber_id,
             flow = None,
 
@@ -96,14 +99,19 @@ class AsyncProcessor:
     # This is called to stop all threads.  An over-ride point for extra
     # functionality
     def stop(self):
+        self.pubsub_backend.close()
         self.pulsar_client.close()
         self.running = False
 
-    # Returns the pulsar host
+    # Returns the pub/sub backend (new interface)
+    @property
+    def pubsub(self): return self.pubsub_backend
+
+    # Returns the pulsar host (backward compatibility)
     @property
     def pulsar_host(self): return self.pulsar_client_object.pulsar_host
 
-    # Returns the pulsar client
+    # Returns the pulsar client (backward compatibility)
     @property
     def pulsar_client(self): return self.pulsar_client_object.client
 
@@ -247,6 +255,14 @@ class AsyncProcessor:
     @staticmethod
     def add_args(parser):
 
+        # Pub/sub backend selection
+        parser.add_argument(
+            '--pubsub-backend',
+            default=os.getenv('PUBSUB_BACKEND', 'pulsar'),
+            choices=['pulsar', 'mqtt'],
+            help='Pub/sub backend (default: pulsar, env: PUBSUB_BACKEND)',
+        )
+
         PulsarClient.add_args(parser)
         add_logging_args(parser)
 

trustgraph-base/trustgraph/base/backend.py

@@ -0,0 +1,148 @@
+"""
+Backend abstraction interfaces for pub/sub systems.
+
+This module defines Protocol classes that all pub/sub backends must implement,
+allowing TrustGraph to work with different messaging systems (Pulsar, MQTT, Kafka, etc.)
+"""
+
+from typing import Protocol, Any, runtime_checkable
+
+
+@runtime_checkable
+class Message(Protocol):
+    """Protocol for a received message."""
+
+    def value(self) -> Any:
+        """
+        Get the deserialized message content.
+
+        Returns:
+            Dataclass instance representing the message
+        """
+        ...
+
+    def properties(self) -> dict:
+        """
+        Get message properties/metadata.
+
+        Returns:
+            Dictionary of message properties
+        """
+        ...
+
+
+@runtime_checkable
+class BackendProducer(Protocol):
+    """Protocol for backend-specific producer."""
+
+    def send(self, message: Any, properties: dict = {}) -> None:
+        """
+        Send a message (dataclass instance) with optional properties.
+
+        Args:
+            message: Dataclass instance to send
+            properties: Optional metadata properties
+        """
+        ...
+
+    def flush(self) -> None:
+        """Flush any buffered messages."""
+        ...
+
+    def close(self) -> None:
+        """Close the producer."""
+        ...
+
+
+@runtime_checkable
+class BackendConsumer(Protocol):
+    """Protocol for backend-specific consumer."""
+
+    def receive(self, timeout_millis: int = 2000) -> Message:
+        """
+        Receive a message from the topic.
+
+        Args:
+            timeout_millis: Timeout in milliseconds
+
+        Returns:
+            Message object
+
+        Raises:
+            TimeoutError: If no message received within timeout
+        """
+        ...
+
+    def acknowledge(self, message: Message) -> None:
+        """
+        Acknowledge successful processing of a message.
+
+        Args:
+            message: The message to acknowledge
+        """
+        ...
+
+    def negative_acknowledge(self, message: Message) -> None:
+        """
+        Negative acknowledge - triggers redelivery.
+
+        Args:
+            message: The message to negatively acknowledge
+        """
+        ...
+
+    def unsubscribe(self) -> None:
+        """Unsubscribe from the topic."""
+        ...
+
+    def close(self) -> None:
+        """Close the consumer."""
+        ...
+
+
+@runtime_checkable
+class PubSubBackend(Protocol):
+    """Protocol defining the interface all pub/sub backends must implement."""
+
+    def create_producer(self, topic: str, schema: type, **options) -> BackendProducer:
+        """
+        Create a producer for a topic.
+
+        Args:
+            topic: Generic topic format (qos/tenant/namespace/queue)
+            schema: Dataclass type for messages
+            **options: Backend-specific options (e.g., chunking_enabled)
+
+        Returns:
+            Backend-specific producer instance
+        """
+        ...
+
+    def create_consumer(
+        self,
+        topic: str,
+        subscription: str,
+        schema: type,
+        initial_position: str = 'latest',
+        consumer_type: str = 'shared',
+        **options
+    ) -> BackendConsumer:
+        """
+        Create a consumer for a topic.
+
+        Args:
+            topic: Generic topic format (qos/tenant/namespace/queue)
+            subscription: Subscription/consumer group name
+            schema: Dataclass type for messages
+            initial_position: 'earliest' or 'latest' (some backends may ignore)
+            consumer_type: 'shared', 'exclusive', 'failover' (some backends may ignore)
+            **options: Backend-specific options
+
+        Returns:
+            Backend-specific consumer instance
+        """
+        ...
+
+    def close(self) -> None:
+        """Close the backend connection."""
+        ...

trustgraph-base/trustgraph/base/consumer.py

@@ -9,9 +9,6 @@
 # one handler, and a single thread of concurrency, nothing too outrageous
 # will happen if synchronous / blocking code is used
 
-from pulsar.schema import JsonSchema
-import pulsar
-import _pulsar
 import asyncio
 import time
 import logging
@@ -21,11 +18,15 @@ from .. exceptions import TooManyRequests
 # Module logger
 logger = logging.getLogger(__name__)
 
+# Timeout exception - can come from different backends
+class TimeoutError(Exception):
+    pass
+
 class Consumer:
 
     def __init__(
-            self, taskgroup, flow, client, topic, subscriber, schema,
+            self, taskgroup, flow, backend, topic, subscriber, schema,
-            handler, 
+            handler,
             metrics = None,
             start_of_messages=False,
             rate_limit_retry_time = 10, rate_limit_timeout = 7200,
@@ -35,7 +36,7 @@ class Consumer:
 
         self.taskgroup = taskgroup
         self.flow = flow
-        self.client = client
+        self.backend = backend  # Changed from 'client' to 'backend'
         self.topic = topic
         self.subscriber = subscriber
         self.schema = schema
@@ -96,18 +97,20 @@ class Consumer:
 
                 logger.info(f"Subscribing to topic: {self.topic}")
 
+                # Determine initial position
                 if self.start_of_messages:
-                    pos = pulsar.InitialPosition.Earliest
+                    initial_pos = 'earliest'
                 else:
-                    pos = pulsar.InitialPosition.Latest
+                    initial_pos = 'latest'
 
+                # Create consumer via backend
                 self.consumer = await asyncio.to_thread(
-                    self.client.subscribe,
+                    self.backend.create_consumer,
                     topic = self.topic,
-                    subscription_name = self.subscriber,
+                    subscription = self.subscriber,
-                    schema = JsonSchema(self.schema),
+                    schema = self.schema,
-                    initial_position = pos,
+                    initial_position = initial_pos,
-                    consumer_type = pulsar.ConsumerType.Shared,
+                    consumer_type = 'shared',
                 )
 
             except Exception as e:
@@ -159,9 +162,10 @@ class Consumer:
                     self.consumer.receive,
                     timeout_millis=2000
                 )
-            except _pulsar.Timeout:
-                continue
             except Exception as e:
+                # Handle timeout from any backend
+                if 'timeout' in str(type(e)).lower() or 'timeout' in str(e).lower():
+                    continue
                 raise e
 
             await self.handle_one_from_queue(msg)

trustgraph-base/trustgraph/base/consumer_spec.py

@@ -19,7 +19,7 @@ class ConsumerSpec(Spec):
         consumer = Consumer(
             taskgroup = processor.taskgroup,
             flow = flow,
-            client = processor.pulsar_client,
+            backend = processor.pubsub,  # Changed from pulsar_client to pubsub
             topic = definition[self.name],
             subscriber = processor.id + "--" + flow.name + "--" + self.name,
             schema = self.schema,

trustgraph-base/trustgraph/base/producer.py

@@ -1,5 +1,4 @@
 
-from pulsar.schema import JsonSchema
 import asyncio
 import logging
 
@@ -8,10 +7,10 @@ logger = logging.getLogger(__name__)
 
 class Producer:
 
-    def __init__(self, client, topic, schema, metrics=None,
+    def __init__(self, backend, topic, schema, metrics=None,
                  chunking_enabled=True):
 
-        self.client = client
+        self.backend = backend  # Changed from 'client' to 'backend'
         self.topic = topic
         self.schema = schema
 
@@ -44,9 +43,9 @@ class Producer:
 
             try:
                 logger.info(f"Connecting publisher to {self.topic}...")
-                self.producer = self.client.create_producer(
+                self.producer = self.backend.create_producer(
                     topic = self.topic,
-                    schema = JsonSchema(self.schema),
+                    schema = self.schema,
                     chunking_enabled = self.chunking_enabled,
                 )
                 logger.info(f"Connected publisher to {self.topic}")

trustgraph-base/trustgraph/base/producer_spec.py

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

trustgraph-base/trustgraph/base/publisher.py

@@ -1,9 +1,6 @@
 
-from pulsar.schema import JsonSchema
-
 import asyncio
 import time
-import pulsar
 import logging
 
 # Module logger
@@ -11,9 +8,9 @@ logger = logging.getLogger(__name__)
 
 class Publisher:
 
-    def __init__(self, client, topic, schema=None, max_size=10,
+    def __init__(self, backend, topic, schema=None, max_size=10,
                  chunking_enabled=True, drain_timeout=5.0):
-        self.client = client
+        self.backend = backend  # Changed from 'client' to 'backend'
         self.topic = topic
         self.schema = schema
         self.q = asyncio.Queue(maxsize=max_size)
@@ -47,9 +44,9 @@ class Publisher:
 
             try:
 
-                producer = self.client.create_producer(
+                producer = self.backend.create_producer(
                     topic=self.topic,
-                    schema=JsonSchema(self.schema),
+                    schema=self.schema,
                     chunking_enabled=self.chunking_enabled,
                 )
 

trustgraph-base/trustgraph/base/pubsub.py

@@ -4,8 +4,45 @@ import pulsar
 import _pulsar
 import uuid
 from pulsar.schema import JsonSchema
+import logging
 
 from .. log_level import LogLevel
+from .pulsar_backend import PulsarBackend
+
+logger = logging.getLogger(__name__)
+
+
+def get_pubsub(**config):
+    """
+    Factory function to create a pub/sub backend based on configuration.
+
+    Args:
+        config: Configuration dictionary from command-line args
+                Must include 'pubsub_backend' key
+
+    Returns:
+        Backend instance (PulsarBackend, MQTTBackend, etc.)
+
+    Example:
+        backend = get_pubsub(
+            pubsub_backend='pulsar',
+            pulsar_host='pulsar://localhost:6650'
+        )
+    """
+    backend_type = config.get('pubsub_backend', 'pulsar')
+
+    if backend_type == 'pulsar':
+        return PulsarBackend(
+            host=config.get('pulsar_host', PulsarClient.default_pulsar_host),
+            api_key=config.get('pulsar_api_key', PulsarClient.default_pulsar_api_key),
+            listener=config.get('pulsar_listener'),
+        )
+    elif backend_type == 'mqtt':
+        # TODO: Implement MQTT backend
+        raise NotImplementedError("MQTT backend not yet implemented")
+    else:
+        raise ValueError(f"Unknown pub/sub backend: {backend_type}")
+
 
 class PulsarClient:
 

trustgraph-base/trustgraph/base/pulsar_backend.py

@@ -0,0 +1,315 @@
+"""
+Pulsar backend implementation for pub/sub abstraction.
+
+This module provides a Pulsar-specific implementation of the backend interfaces,
+handling topic mapping, serialization, and Pulsar client management.
+"""
+
+import pulsar
+import _pulsar
+import json
+import logging
+from dataclasses import asdict, is_dataclass
+from typing import Any
+
+from .backend import PubSubBackend, BackendProducer, BackendConsumer, Message
+
+logger = logging.getLogger(__name__)
+
+
+def dataclass_to_dict(obj: Any) -> dict:
+    """
+    Recursively convert a dataclass to a dictionary, handling None values.
+
+    None values are excluded from the dictionary (not serialized).
+    """
+    if obj is None:
+        return None
+
+    if is_dataclass(obj):
+        result = {}
+        for key, value in asdict(obj).items():
+            if value is not None:
+                if is_dataclass(value):
+                    result[key] = dataclass_to_dict(value)
+                elif isinstance(value, list):
+                    result[key] = [dataclass_to_dict(item) if is_dataclass(item) else item for item in value]
+                elif isinstance(value, dict):
+                    result[key] = {k: dataclass_to_dict(v) if is_dataclass(v) else v for k, v in value.items()}
+                else:
+                    result[key] = value
+        return result
+    return obj
+
+
+def dict_to_dataclass(data: dict, cls: type) -> Any:
+    """
+    Convert a dictionary back to a dataclass instance.
+
+    Handles nested dataclasses and missing fields.
+    """
+    if data is None:
+        return None
+
+    if not is_dataclass(cls):
+        return data
+
+    # Get field types from the dataclass
+    field_types = {f.name: f.type for f in cls.__dataclass_fields__.values()}
+    kwargs = {}
+
+    for key, value in data.items():
+        if key in field_types:
+            field_type = field_types[key]
+
+            # Handle Optional types (type | None)
+            if hasattr(field_type, '__args__'):
+                # Get the non-None type from Union
+                actual_type = next((t for t in field_type.__args__ if t is not type(None)), field_type)
+            else:
+                actual_type = field_type
+
+            # Recursively convert nested dataclasses
+            if is_dataclass(actual_type):
+                kwargs[key] = dict_to_dataclass(value, actual_type)
+            elif hasattr(actual_type, '__origin__'):
+                # Handle generic types like list[T] or dict[K, V]
+                if actual_type.__origin__ == list:
+                    item_type = actual_type.__args__[0] if actual_type.__args__ else None
+                    if item_type and is_dataclass(item_type):
+                        kwargs[key] = [dict_to_dataclass(item, item_type) for item in value]
+                    else:
+                        kwargs[key] = value
+                else:
+                    kwargs[key] = value
+            else:
+                kwargs[key] = value
+
+    return cls(**kwargs)
+
+
+class PulsarMessage:
+    """Wrapper for Pulsar messages to match Message protocol."""
+
+    def __init__(self, pulsar_msg, schema_cls):
+        self._msg = pulsar_msg
+        self._schema_cls = schema_cls
+        self._value = None
+
+    def value(self) -> Any:
+        """Deserialize and return the message value as a dataclass."""
+        if self._value is None:
+            # Get JSON string from Pulsar message
+            json_data = self._msg.data().decode('utf-8')
+            data_dict = json.loads(json_data)
+            # Convert to dataclass
+            self._value = dict_to_dataclass(data_dict, self._schema_cls)
+        return self._value
+
+    def properties(self) -> dict:
+        """Return message properties."""
+        return self._msg.properties()
+
+
+class PulsarBackendProducer:
+    """Pulsar-specific producer implementation."""
+
+    def __init__(self, pulsar_producer, schema_cls):
+        self._producer = pulsar_producer
+        self._schema_cls = schema_cls
+
+    def send(self, message: Any, properties: dict = {}) -> None:
+        """Send a dataclass message."""
+        # Convert dataclass to dict, excluding None values
+        data_dict = dataclass_to_dict(message)
+        # Serialize to JSON
+        json_data = json.dumps(data_dict)
+        # Send via Pulsar
+        self._producer.send(json_data.encode('utf-8'), properties=properties)
+
+    def flush(self) -> None:
+        """Flush buffered messages."""
+        self._producer.flush()
+
+    def close(self) -> None:
+        """Close the producer."""
+        self._producer.close()
+
+
+class PulsarBackendConsumer:
+    """Pulsar-specific consumer implementation."""
+
+    def __init__(self, pulsar_consumer, schema_cls):
+        self._consumer = pulsar_consumer
+        self._schema_cls = schema_cls
+
+    def receive(self, timeout_millis: int = 2000) -> Message:
+        """Receive a message."""
+        pulsar_msg = self._consumer.receive(timeout_millis=timeout_millis)
+        return PulsarMessage(pulsar_msg, self._schema_cls)
+
+    def acknowledge(self, message: Message) -> None:
+        """Acknowledge a message."""
+        if isinstance(message, PulsarMessage):
+            self._consumer.acknowledge(message._msg)
+
+    def negative_acknowledge(self, message: Message) -> None:
+        """Negative acknowledge a message."""
+        if isinstance(message, PulsarMessage):
+            self._consumer.negative_acknowledge(message._msg)
+
+    def unsubscribe(self) -> None:
+        """Unsubscribe from the topic."""
+        self._consumer.unsubscribe()
+
+    def close(self) -> None:
+        """Close the consumer."""
+        self._consumer.close()
+
+
+class PulsarBackend:
+    """
+    Pulsar backend implementation.
+
+    Handles topic mapping, client management, and creation of Pulsar-specific
+    producers and consumers.
+    """
+
+    def __init__(self, host: str, api_key: str = None, listener: str = None):
+        """
+        Initialize Pulsar backend.
+
+        Args:
+            host: Pulsar broker URL (e.g., pulsar://localhost:6650)
+            api_key: Optional API key for authentication
+            listener: Optional listener name for multi-homed setups
+        """
+        self.host = host
+        self.api_key = api_key
+        self.listener = listener
+
+        # Create Pulsar client
+        client_args = {'service_url': host}
+
+        if listener:
+            client_args['listener_name'] = listener
+
+        if api_key:
+            client_args['authentication'] = pulsar.AuthenticationToken(api_key)
+
+        self.client = pulsar.Client(**client_args)
+        logger.info(f"Pulsar client connected to {host}")
+
+    def map_topic(self, generic_topic: str) -> str:
+        """
+        Map generic topic format to Pulsar URI.
+
+        Format: qos/tenant/namespace/queue
+        Example: q1/tg/flow/my-queue -> persistent://tg/flow/my-queue
+
+        Args:
+            generic_topic: Generic topic string
+
+        Returns:
+            Pulsar topic URI
+        """
+        parts = generic_topic.split('/', 3)
+        if len(parts) != 4:
+            raise ValueError(f"Invalid topic format: {generic_topic}, expected qos/tenant/namespace/queue")
+
+        qos, tenant, namespace, queue = parts
+
+        # Map QoS to persistence
+        if qos == 'q0':
+            persistence = 'non-persistent'
+        elif qos in ['q1', 'q2']:
+            persistence = 'persistent'
+        else:
+            raise ValueError(f"Invalid QoS level: {qos}, expected q0, q1, or q2")
+
+        return f"{persistence}://{tenant}/{namespace}/{queue}"
+
+    def create_producer(self, topic: str, schema: type, **options) -> BackendProducer:
+        """
+        Create a Pulsar producer.
+
+        Args:
+            topic: Generic topic format (qos/tenant/namespace/queue)
+            schema: Dataclass type for messages
+            **options: Backend-specific options (e.g., chunking_enabled)
+
+        Returns:
+            PulsarBackendProducer instance
+        """
+        pulsar_topic = self.map_topic(topic)
+
+        producer_args = {
+            'topic': pulsar_topic,
+            'schema': pulsar.schema.BytesSchema(),  # We handle serialization ourselves
+        }
+
+        # Add optional parameters
+        if 'chunking_enabled' in options:
+            producer_args['chunking_enabled'] = options['chunking_enabled']
+
+        pulsar_producer = self.client.create_producer(**producer_args)
+        logger.debug(f"Created producer for topic: {pulsar_topic}")
+
+        return PulsarBackendProducer(pulsar_producer, schema)
+
+    def create_consumer(
+        self,
+        topic: str,
+        subscription: str,
+        schema: type,
+        initial_position: str = 'latest',
+        consumer_type: str = 'shared',
+        **options
+    ) -> BackendConsumer:
+        """
+        Create a Pulsar consumer.
+
+        Args:
+            topic: Generic topic format (qos/tenant/namespace/queue)
+            subscription: Subscription name
+            schema: Dataclass type for messages
+            initial_position: 'earliest' or 'latest'
+            consumer_type: 'shared', 'exclusive', or 'failover'
+            **options: Backend-specific options
+
+        Returns:
+            PulsarBackendConsumer instance
+        """
+        pulsar_topic = self.map_topic(topic)
+
+        # Map initial position
+        if initial_position == 'earliest':
+            pos = pulsar.InitialPosition.Earliest
+        else:
+            pos = pulsar.InitialPosition.Latest
+
+        # Map consumer type
+        if consumer_type == 'exclusive':
+            ctype = pulsar.ConsumerType.Exclusive
+        elif consumer_type == 'failover':
+            ctype = pulsar.ConsumerType.Failover
+        else:
+            ctype = pulsar.ConsumerType.Shared
+
+        consumer_args = {
+            'topic': pulsar_topic,
+            'subscription_name': subscription,
+            'schema': pulsar.schema.BytesSchema(),  # We handle deserialization ourselves
+            'initial_position': pos,
+            'consumer_type': ctype,
+        }
+
+        pulsar_consumer = self.client.subscribe(**consumer_args)
+        logger.debug(f"Created consumer for topic: {pulsar_topic}, subscription: {subscription}")
+
+        return PulsarBackendConsumer(pulsar_consumer, schema)
+
+    def close(self) -> None:
+        """Close the Pulsar client."""
+        self.client.close()
+        logger.info("Pulsar client closed")

trustgraph-base/trustgraph/base/subscriber.py

@@ -3,9 +3,7 @@
 # off of a queue and make it available using an internal broker system,
 # so suitable for when multiple recipients are reading from the same queue
 
-from pulsar.schema import JsonSchema
 import asyncio
-import _pulsar
 import time
 import logging
 import uuid
@@ -13,12 +11,16 @@ import uuid
 # Module logger
 logger = logging.getLogger(__name__)
 
+# Timeout exception - can come from different backends
+class TimeoutError(Exception):
+    pass
+
 class Subscriber:
 
-    def __init__(self, client, topic, subscription, consumer_name,
+    def __init__(self, backend, topic, subscription, consumer_name,
                  schema=None, max_size=100, metrics=None,
                  backpressure_strategy="block", drain_timeout=5.0):
-        self.client = client
+        self.backend = backend  # Changed from 'client' to 'backend'
         self.topic = topic
         self.subscription = subscription
         self.consumer_name = consumer_name
@@ -43,18 +45,14 @@ class Subscriber:
 
     async def start(self):
 
-        # Build subscribe arguments
+        # Create consumer via backend
-        subscribe_args = {
+        self.consumer = await asyncio.to_thread(
-            'topic': self.topic,
+            self.backend.create_consumer,
-            'subscription_name': self.subscription,
+            topic=self.topic,
-            'consumer_name': self.consumer_name,
+            subscription=self.subscription,
-        }
+            schema=self.schema,
-
+            consumer_type='shared',
-        # 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())
 
@@ -94,12 +92,13 @@ class Subscriber:
                         drain_end_time = time.time() + self.drain_timeout
                         logger.info(f"Subscriber entering drain mode, timeout={self.drain_timeout}s")
 
-                        # Stop accepting new messages from Pulsar during drain
+                        # Stop accepting new messages during drain
-                        if self.consumer:
+                        # Note: Not all backends support pausing message listeners
+                        if self.consumer and hasattr(self.consumer, 'pause_message_listener'):
                             try:
                                 self.consumer.pause_message_listener()
-                            except _pulsar.InvalidConfiguration:
+                            except Exception:
-                                # Not all consumers have message listeners (e.g., blocking receive mode)
+                                # Not all consumers support message listeners
                                 pass
                     
                     # Check drain timeout
@@ -133,9 +132,10 @@ class Subscriber:
                                 self.consumer.receive,
                                 timeout_millis=250
                             )
-                        except _pulsar.Timeout:
-                            continue
                         except Exception as e:
+                            # Handle timeout from any backend
+                            if 'timeout' in str(type(e)).lower() or 'timeout' in str(e).lower():
+                                continue
                             logger.error(f"Exception in subscriber receive: {e}", exc_info=True)
                             raise e
 
@@ -157,19 +157,20 @@ class Subscriber:
                 for msg in self.pending_acks.values():
                     try:
                         self.consumer.negative_acknowledge(msg)
-                    except _pulsar.AlreadyClosed:
+                    except Exception:
-                        pass  # Consumer already closed
+                        pass  # Consumer already closed or error
                 self.pending_acks.clear()
 
                 if self.consumer:
-                    try:
+                    if hasattr(self.consumer, 'unsubscribe'):
-                        self.consumer.unsubscribe()
+                        try:
-                    except _pulsar.AlreadyClosed:
+                            self.consumer.unsubscribe()
-                        pass  # Already closed
+                        except Exception:
+                            pass  # Already closed or error
                     try:
                         self.consumer.close()
-                    except _pulsar.AlreadyClosed:
+                    except Exception:
-                        pass  # Already closed
+                        pass  # Already closed or error
                     self.consumer = None