vs-queue

Pluggable async message queue library for Viveka Sutra — Redis Streams, RabbitMQ, retry policies, dead letter queues, and annotation-based consumers.

---

Overview

vs-queue is the messaging backbone for all Viveka Sutra services. It provides a unified API for publishing and consuming messages across multiple queue backends. Services publish a VsMessage to a named queue and consumers process it — the queue backend is swappable via config without changing application code.

Key features:

• Unified VsBaseQueue abstraction — swap Redis for RabbitMQ without changing consumer code
• @queue_listener decorator — register consumers declaratively, like Spring's @RabbitListener
• VsQueueManager — lifecycle management for all consumers: start, stop, health check
• Built-in retry with exponential backoff
• Dead letter queue (DLQ) support
• Registry pattern — register custom queue providers by name
• Fully async — built on asyncio, compatible with FastAPI and any async framework

---

The Problem It Solves

Services that call each other directly over HTTP are tightly coupled — a slow or unavailable downstream service blocks the caller.

Without vs-queue

# Caller blocks waiting for the downstream service
response = await http_client.post("/v1/process", json=payload)
response.raise_for_status()
return response.json()

If the downstream service is slow or down, every caller fails. There is no retry, no buffering, and no way to recover without restarting.

With vs-queue

# Caller publishes and returns immediately
await queue.publish("orchestrator:tasks", VsMessage(payload={"node_id": node_id}))

The consumer processes at its own pace. If processing fails it retries automatically. If it exceeds max retries it goes to the dead letter queue for inspection. The caller never blocks.

---

Installation

pip install vs-queue

With Redis Streams support:

pip install vs-queue[redis]

With RabbitMQ support:

pip install vs-queue[rabbitmq]

Both backends:

pip install vs-queue[redis,rabbitmq]

---

Dependencies

Package	Version	Required	Purpose
pydantic	>=2.0	Yes	VsMessage schema validation
redis	>=5.0	No — install with [redis] extra	Redis Streams backend
hiredis	>=2.0	No — install with [redis] extra	Redis protocol parser (performance)
aio-pika	>=9.0	No — install with [rabbitmq] extra	RabbitMQ backend via AMQP

---

Quick Start

Publisher

import asyncio
from vs_queue import VsQueueRegistry  # auto-registers redis and rabbitmq
from vs_queue.factory.vs_queue_factory import VsQueueFactory
from vs_queue.schema.vs_message import VsMessage

async def main():
    queue = VsQueueFactory.create(
        provider="redis",
        host="localhost",
        port=6379,
    )
    await queue.connect()

    message = VsMessage(
        headers={"type": "orchestrator.task", "trace_id": "abc123"},
        payload={"node_id": "xyz", "user_message": "hello"},
    )
    await queue.publish("orchestrator:tasks", message)
    await queue.disconnect()

asyncio.run(main())

Consumer (manual)

from vs_queue.base.vs_base_consumer import VsBaseConsumer
from vs_queue.schema.vs_message import VsMessage

class OrchestratorConsumer(VsBaseConsumer):

    async def handle(self, message: VsMessage) -> None:
        node_id = message.payload["node_id"]
        print(f"Processing node: {node_id}")

    async def start(self) -> None:
        print("OrchestratorConsumer started")

    async def stop(self) -> None:
        print("OrchestratorConsumer stopped")

Consumer (annotation-based)

from vs_queue.decorator.vs_queue_listener import queue_listener
from vs_queue.base.vs_base_consumer import VsBaseConsumer
from vs_queue.schema.vs_message import VsMessage

@queue_listener(
    queue="orchestrator:tasks",
    concurrency=2,
    max_retries=3,
    retry_backoff_seconds=2.0,
    dead_letter_queue="orchestrator:dlq",
)
class OrchestratorConsumer(VsBaseConsumer):

    async def handle(self, message: VsMessage) -> None:
        node_id = message.payload["node_id"]
        # process the message
        print(f"Processing node: {node_id}")

    async def start(self) -> None:
        pass

    async def stop(self) -> None:
        pass

Starting all listeners

from vs_queue.factory.vs_queue_factory import VsQueueFactory
from vs_queue.manager.vs_queue_manager import VsQueueManager

import my_app.consumers  # noqa — import modules that contain @queue_listener classes

queue = VsQueueFactory.create(provider="redis", host="localhost", port=6379)
manager = VsQueueManager(queue)
await manager.register_listeners()  # discovers all @queue_listener classes, connects & starts immediately

# ... app runs ...
await manager.stop_all()

---

Configuration

Redis

[queue]
provider = redis
host     = localhost
port     = 6379
password = your_redis_password

RabbitMQ

[queue]
provider  = rabbitmq
host      = localhost
port      = 5672
username  = guest
password  = guest
vhost     = /

---

VsMessage

Every message published and consumed by vs-queue is a VsMessage. It is backend-agnostic — the same schema works for Redis Streams and RabbitMQ.

from vs_queue.schema.vs_message import VsMessage

message = VsMessage(
    headers={"type": "orchestrator.task", "trace_id": "abc123", "source_queue": "orchestrator:tasks"},
    payload={"node_id": "xyz", "user_id": "u1"},
)

Fields:

Field	Type	Default	Description
id	str	uuid4().hex	Unique message ID, auto-generated
timestamp	datetime	now(UTC)	Message creation time, auto-set
retry_count	int	0	Number of processing attempts. Managed by the library — do not set manually.
headers	Dict[str, str]	{}	Caller-defined metadata: type, trace_id, routing info, etc.
payload	Dict[str, Any]	None	Message body. Structure is defined by the caller.

Notes:

• id and timestamp are auto-generated — you rarely need to set them.
• retry_count is managed by the retry policy. Do not set it manually.
• headers is a flat Dict[str, str]. Nest complex metadata in payload.
• Include "source_queue" in headers if you use retry — the retry policy uses it to re-publish.

message = VsMessage(
    headers={
        "type": "task.process",
        "trace_id": "abc123",
        "source_queue": "orchestrator:tasks",
    },
    payload={
        "node_id": "n1",
        "conversation_id": "c1",
        "user_message": "what is the weather today?",
    },
)

print(message.id)           # e.g. "3f2a1b4c..."
print(message.timestamp)    # e.g. 2026-08-03T10:00:00+00:00
print(message.retry_count)  # 0

---

VsBaseQueue

Abstract base class for all queue backends. Extend this to implement a custom provider.

Constructor parameters:

Parameter	Type	Required	Description
host	str	Yes	Broker host
port	int	Yes	Broker port
credentials	Dict[str, Any]	No	Auth credentials. Keys depend on the backend.

Abstract methods:

Method	Description
connect()	Establish connection to the broker
disconnect()	Close connection cleanly
health_check()	Returns True if the broker is reachable
publish(queue, message)	Publish a VsMessage to a named queue
subscribe(queue, consumer, retry_policy)	Begin consuming from a queue using a VsBaseConsumer

Concrete method:

Method	Description
set_retention(queue, VsRetention)	Register a per-queue retention policy. Call before connect(). See Retention.

---

VsBaseConsumer

Abstract base class for all consumers. Extend this and implement handle to process messages.

Abstract methods:

Method	Description
handle(message)	Called for every message received. Raise any exception to trigger retry.

Optional overrides (concrete no-ops by default — override only if you need them):

Method	Description
start()	Called once when the consumer begins. Use for setup.
stop()	Called once on graceful shutdown. Use for cleanup.
on_error(message, error)	Called when handle raises an exception, before retry logic runs. Use for logging or alerting.

Example with error hook:

from vs_queue.base.vs_base_consumer import VsBaseConsumer
from vs_queue.schema.vs_message import VsMessage
import logging

_logger = logging.getLogger(__name__)

class TaskConsumer(VsBaseConsumer):

    async def handle(self, message: VsMessage) -> None:
        result = await process(message.payload)
        if not result:
            raise ValueError(f"Processing failed for node {message.payload['node_id']}")

    async def start(self) -> None:
        _logger.info("TaskConsumer starting")

    async def stop(self) -> None:
        _logger.info("TaskConsumer stopping")

    async def on_error(self, message: VsMessage, error: Exception) -> None:
        _logger.error(f"Message failed | id={message.id} error={error}")
        # send alert, update DB status, etc.

---

@queue_listener

Decorator that registers a VsBaseConsumer subclass as a listener. VsQueueManager.register_listeners() discovers all decorated classes and starts them automatically.

Parameters:

Parameter	Type	Required	Default	Description
queue	str	Yes	—	Queue name to consume from
concurrency	int	No	1	Number of parallel consumer instances to start
max_retries	int	No	3	Max retry attempts before sending to DLQ
retry_backoff_seconds	float	No	1.0	Base delay for exponential backoff between retries
dead_letter_queue	str	No	None	Queue name for messages that exceed max retries

Example — minimal:

@queue_listener(queue="tasks")
class SimpleConsumer(VsBaseConsumer):
    async def handle(self, message: VsMessage) -> None:
        print(message.payload)

Example — full configuration:

@queue_listener(
    queue="orchestrator:tasks",
    concurrency=3,
    max_retries=5,
    retry_backoff_seconds=2.0,
    dead_letter_queue="orchestrator:dlq",
)
class OrchestratorConsumer(VsBaseConsumer):

    async def handle(self, message: VsMessage) -> None:
        trace_id = message.headers.get("trace_id")
        node_id = message.payload["node_id"]
        await run_graph(trace_id, node_id)

    async def start(self) -> None:
        await db.connect()

    async def stop(self) -> None:
        await db.disconnect()

    async def on_error(self, message: VsMessage, error: Exception) -> None:
        await notify_ops(f"Failed: {message.id} — {error}")

Important: The module containing @queue_listener classes must be imported before calling register_listeners(), otherwise the decorator never runs and the class is never registered. register_listeners() is async and safe to call only once.

import my_app.consumers  # noqa — triggers @queue_listener registration
await manager.register_listeners()

---

VsQueueManager

Manages the lifecycle of all consumers. Connects to the broker, starts consumer tasks, handles graceful shutdown.

Constructor:

Parameter	Type	Required	Default	Description
queue	VsBaseQueue	Yes	—	The queue instance
max_consumer_restarts	int	No	3	Max times a crashed consumer task is restarted before giving up

Methods:

Method	Description
register(queue_name, consumer)	Register a consumer, auto-connect if needed, start immediately. Returns a UUID.
register_listeners()	Auto-discover all @queue_listener classes and register them. Safe to call only once.
get_listeners()	Returns Dict[str, List[VsBaseConsumer]] — queue name → running consumer instances
stop_listener(queue_name, uuid=None)	Stop all consumers on a queue, or one specific consumer by UUID
stop_all()	Stop all consumers and disconnect from broker
health_check()	Returns True if connected and broker is healthy. Returns False if not yet connected.

Example — with vs-server lifecycle:

from vs_queue.factory.vs_queue_factory import VsQueueFactory
from vs_queue.manager.vs_queue_manager import VsQueueManager
from vs_server.lifecycle.vs_lifecycle import startup, shutdown
import app.consumers  # noqa

queue = VsQueueFactory.create(provider="redis", host="localhost", port=6379)
manager = VsQueueManager(queue, max_consumer_restarts=5)

@startup
async def start_consumers():
    await manager.register_listeners()

@shutdown
async def stop_consumers():
    await manager.stop_all()

Example — dynamic registration:

manager = VsQueueManager(queue)

# auto-connects on first register(), starts task immediately
for tenant_id in ["tenant_a", "tenant_b"]:
    uuid = await manager.register(f"tasks:{tenant_id}", TenantConsumer())

# stop one specific consumer
await manager.stop_listener("tasks:tenant_a", uuid)

# stop all consumers on a queue
await manager.stop_listener("tasks:tenant_b")

# inspect running consumers
listeners = manager.get_listeners()
# {"tasks:tenant_a": [<TenantConsumer uuid=...>]}

---

Retry Policy

VsRetryPolicy defines what happens when handle() raises an exception. Pass it to subscribe() on the queue directly, or let @queue_listener configure it declaratively.

Constructor:

Parameter	Type	Default	Description
max_retries	int	3	Max attempts before giving up
backoff_seconds	float	1.0	Base delay. Doubles on each retry (exponential backoff).
dead_letter_queue	str	None	Queue for messages that exceed max retries

Retry timing:

Attempt	Delay
1st retry	1s
2nd retry	2s
3rd retry	4s
Beyond max	→ DLQ

Example — manual:

from vs_queue.retry.vs_retry_policy import VsRetryPolicy

retry_policy = VsRetryPolicy(
    max_retries=5,
    backoff_seconds=2.0,
    dead_letter_queue="orchestrator:dlq",
)

await queue.subscribe("orchestrator:tasks", consumer, retry_policy=retry_policy)

Important: Include "source_queue" in the message headers so the retry policy knows where to re-publish:

message = VsMessage(
    headers={
        "source_queue": "orchestrator:tasks",
        "trace_id": "abc123",
    },
    payload={"node_id": "n1"},
)

---

Dead Letter Queue

Messages that fail after all retry attempts are published to the dead letter queue (DLQ). The DLQ is just another queue — you can attach a consumer to it for inspection, alerting, or manual replay.

Example — DLQ consumer:

@queue_listener(queue="orchestrator:dlq")
class DlqConsumer(VsBaseConsumer):

    async def handle(self, message: VsMessage) -> None:
        await alert_ops(
            f"Message permanently failed | id={message.id} retries={message.retry_count}",
            payload=message.payload,
        )

Example — replaying a DLQ message:

dlq_message = ...  # fetch from DLQ consumer

# Reset retry count and re-publish to the original queue
dlq_message.retry_count = 0
await queue.publish("orchestrator:tasks", dlq_message)

---

Retention

Retention is a property of the queue, not the message — so you register it per queue name, once, before connect(). VsMessage is unchanged.

from vs_queue.schema.vs_retention import VsRetention

# via the factory
queue = VsQueueFactory.create(
    provider="redis", host="localhost", port=6379,
    retention={
        "vs_trace:logs": VsRetention(ttl_seconds=604800, max_len=100_000),
    },
)

# or directly, any time before the first publish/subscribe to that queue
queue.set_retention("vs_trace:logs", VsRetention(ttl_seconds=604800))

Field	Meaning
ttl_seconds	Drop messages older than this
max_len	Keep at most this many messages

Applied the first time the queue is published to or subscribed. Queues with no policy are unbounded, same as before.

Redis Streams — no native per-message TTL, so:

• max_len → XADD ... MAXLEN ~ n on every publish
• ttl_seconds → throttled XTRIM <queue> MINID ~ <now - ttl> (stream IDs are ms timestamps); runs at most once per ttl/10 seconds per queue, not on every publish

RabbitMQ — set as queue arguments on declare:

• ttl_seconds → x-message-ttl
• max_len → x-max-length

RabbitMQ rejects redeclaring a live queue with different arguments — the first declare wins. To change a policy, use a new queue name or a broker policy.

---

Logging

Both backends log consumer lifecycle and per-message activity on the provider logger (vs_queue.provider.vs_redis_queue / vs_queue.provider.vs_rabbitmq_queue). Every line uses a Label | key=value format.

Level	Line	When
INFO	Subscribed | queue=<q> consumer=<ClassName>	consumer.start() has run and the queue / consumer group is ready — the loop is now listening
INFO	Message received | queue=<q> id=<msg_id> retry_count=<n>	Each message pulled off the queue, before handle() is called
DEBUG	Message processed | queue=<q> id=<msg_id>	handle() returned and the message was acked
INFO	Unsubscribed | queue=<q> consumer=<ClassName>	The subscribe loop exited (graceful shutdown or permanent failure)
ERROR	Message failed, no retry policy | id=<msg_id> error=<e>	handle() raised and no VsRetryPolicy was supplied

VsQueueManager logs lifecycle events on its own logger (vs_queue.manager.vs_queue_manager):

Level	Line
INFO	Listener registered | queue=<q> uuid=<uid>
INFO	Listeners registered | queue=<q> concurrency=<n>
INFO	Listener stopped | queue=<q> uuid=<uid>
ERROR	Consumer crashed, restarting in <n>s | queue=<q> attempt=<n> error=<e>
CRITICAL	Consumer permanently failed after <n> restarts | queue=<q> error=<e>

Subscribed / Message received land only once the loop is actually consuming, so the absence of Subscribed after Listener registered means subscribe() is failing before it starts.

Turn on the debug lines:

import logging
logging.getLogger("vs_queue").setLevel(logging.DEBUG)

---

Redis Streams Backend

VsRedisQueue implements the queue using Redis Streams (XADD / XREADGROUP). Redis Streams provide persistent, ordered, consumer-group-aware message delivery.

How it works:

• publish → XADD queue * field value ...
• subscribe → XREADGROUP GROUP group consumer STREAMS queue >
• On success → XACK removes the message from the pending entries list
• On failure → retry policy runs; if max retries exceeded → DLQ

Credentials:

Key	Description
password	Redis AUTH password

Example:

from vs_queue.provider.vs_redis_queue import VsRedisQueue

queue = VsRedisQueue(
    host="localhost",
    port=6379,
    credentials={"password": "secret"},
)
await queue.connect()

healthy = await queue.health_check()  # True if Redis responds to PING

Consumer group naming:

• Group name: {queue}:group
• Consumer name: {queue}:consumer

These are created automatically on first subscribe(). If the stream does not exist yet, it is created via mkstream=True.

---

RabbitMQ Backend

VsRabbitMQQueue implements the queue using AMQP via aio-pika. Messages are published to the default exchange with the queue name as the routing key. Queues are declared as durable — messages survive broker restarts.

Credentials:

Key	Default	Description
username	guest	RabbitMQ username
password	guest	RabbitMQ password
vhost	/	RabbitMQ virtual host

Example:

from vs_queue.provider.vs_rabbitmq_queue import VsRabbitMQQueue

queue = VsRabbitMQQueue(
    host="localhost",
    port=5672,
    credentials={"username": "admin", "password": "secret", "vhost": "/prod"},
)
await queue.connect()

Acknowledgement behaviour:

• On success → ack() removes the message
• On failure with retry → ack() the original, re-publish with incremented retry_count
• On failure beyond max retries → nack(requeue=False) + publish to DLQ

---

VsQueueRegistry

Stores registered queue provider classes by name. Pre-registered providers: redis, rabbitmq.

Methods:

Method	Description
register(name, queue_class)	Register a VsBaseQueue subclass under a name
get(name)	Return the class for a given name. Raises KeyError if not found.
available()	Return list of all registered provider names

Example — registering a custom provider:

from vs_queue.registry.vs_queue_registry import VsQueueRegistry
from vs_queue.base.vs_base_queue import VsBaseQueue

class VsKafkaQueue(VsBaseQueue):
    async def connect(self): ...
    async def disconnect(self): ...
    async def health_check(self): ...
    async def publish(self, queue, message): ...
    async def subscribe(self, queue, consumer, retry_policy=None): ...

VsQueueRegistry.register("kafka", VsKafkaQueue)

Once registered, the factory can create it:

queue = VsQueueFactory.create(provider="kafka", host="localhost", port=9092)

---

VsQueueFactory

Creates queue instances from a registered provider name.

Methods:

Method	Signature	Description
create	create(provider, host, port, credentials, retention) -> VsBaseQueue	Create and return a queue instance. Does not call connect(). retention is an optional {queue_name: VsRetention} map — see Retention.

Example:

from vs_queue.factory.vs_queue_factory import VsQueueFactory

queue = VsQueueFactory.create(
    provider="redis",
    host="localhost",
    port=6379,
    credentials={"password": "secret"},
)
await queue.connect()

---

Extending vs-queue

Custom Queue Provider

Implement VsBaseQueue and register it:

from vs_queue.base.vs_base_queue import VsBaseQueue
from vs_queue.base.vs_base_consumer import VsBaseConsumer
from vs_queue.retry.vs_retry_policy import VsRetryPolicy
from vs_queue.schema.vs_message import VsMessage
from vs_queue.registry.vs_queue_registry import VsQueueRegistry
from typing import Optional

class VsSqsQueue(VsBaseQueue):

    def __init__(self, host: str, port: int, credentials=None):
        super().__init__(host, port, credentials)
        self._client = None

    async def connect(self) -> None:
        import aiobotocore.session
        session = aiobotocore.session.get_session()
        self._client = await session.create_client(
            "sqs",
            region_name=self._credentials.get("region", "us-east-1"),
            aws_access_key_id=self._credentials.get("access_key"),
            aws_secret_access_key=self._credentials.get("secret_key"),
        ).__aenter__()

    async def disconnect(self) -> None:
        if self._client:
            await self._client.__aexit__(None, None, None)

    async def health_check(self) -> bool:
        try:
            await self._client.list_queues(MaxResults=1)
            return True
        except Exception:
            return False

    async def publish(self, queue: str, message: VsMessage) -> None:
        import json
        await self._client.send_message(
            QueueUrl=queue,
            MessageBody=json.dumps({"id": message.id, "payload": message.payload, "headers": message.headers}),
        )

    async def subscribe(self, queue: str, consumer: VsBaseConsumer, retry_policy: Optional[VsRetryPolicy] = None) -> None:
        await consumer.start()
        try:
            while True:
                response = await self._client.receive_message(QueueUrl=queue, WaitTimeSeconds=20)
                for msg in response.get("Messages", []):
                    # parse and call consumer.handle(message)
                    ...
        finally:
            await consumer.stop()

VsQueueRegistry.register("sqs", VsSqsQueue)

---

Full Example — Orchestrator Integration

Publisher (chat controller):

from vs_queue.factory.vs_queue_factory import VsQueueFactory
from vs_queue.schema.vs_message import VsMessage

queue = VsQueueFactory.create(provider="redis", host="localhost", port=6379)
await queue.connect()

await queue.publish(
    "orchestrator:tasks",
    VsMessage(
        headers={
            "type": "orchestrator.task",
            "trace_id": trace_id,
            "source_queue": "orchestrator:tasks",
        },
        payload={
            "node_id": node["node"]["id"],
            "conversation_id": node["conversation_id"],
            "user_message": request.message,
            "user_id": user_id,
        },
    ),
)

Consumer (orchestrator worker):

from vs_queue.decorator.vs_queue_listener import queue_listener
from vs_queue.base.vs_base_consumer import VsBaseConsumer
from vs_queue.schema.vs_message import VsMessage
from orchestrator_agent.graph.orchestrator_graph import OrchestratorGraph
from orchestrator_agent.graph.orchestrator_state import OrchestratorState

@queue_listener(
    queue="orchestrator:tasks",
    concurrency=2,
    max_retries=3,
    retry_backoff_seconds=2.0,
    dead_letter_queue="orchestrator:dlq",
)
class OrchestratorTaskConsumer(VsBaseConsumer):

    def __init__(self):
        super().__init__()
        self._graph = OrchestratorGraph()

    async def handle(self, message: VsMessage) -> None:
        payload = message.payload
        result = await self._graph.invoke(OrchestratorState(
            trace_id=message.headers.get("trace_id", ""),
            error=None,
            current_message=payload["user_message"],
            user_id=payload["user_id"],
            message=payload["user_message"],
            resolved_context=None,
            response=None,
        ))

    async def start(self) -> None:
        pass

    async def stop(self) -> None:
        pass

    async def on_error(self, message: VsMessage, error: Exception) -> None:
        import logging
        logging.getLogger(__name__).error(
            f"Graph execution failed | node_id={message.payload.get('node_id')} error={error}"
        )

App startup:

from vs_queue.factory.vs_queue_factory import VsQueueFactory
from vs_queue.manager.vs_queue_manager import VsQueueManager
from vs_server.lifecycle.vs_lifecycle import startup, shutdown
import orchestrator_agent.consumers  # noqa — registers @queue_listener classes

queue = VsQueueFactory.create(provider="redis", host="localhost", port=6379)
manager = VsQueueManager(queue, max_consumer_restarts=5)

@startup
async def start_consumers():
    await manager.register_listeners()

@shutdown
async def stop_consumers():
    await manager.stop_all()

---

Class Reference

---

VsMessage

Standard message envelope. Every message published and consumed by vs-queue is a VsMessage.

Field	Type	Default	Description
id	str	uuid4().hex	Auto-generated unique ID
timestamp	datetime	now(UTC)	Auto-set creation time
retry_count	int	0	Managed by retry policy
headers	Dict[str, str]	{}	Caller metadata
payload	Dict[str, Any]	None	Message body

---

VsBaseQueue

Abstract base. Extend to implement a custom queue provider.

Method	Description
connect()	Establish broker connection
disconnect()	Close connection cleanly
health_check()	Returns True if broker is reachable
publish(queue, message)	Publish a message
subscribe(queue, consumer, retry_policy)	Start consuming
set_retention(queue, VsRetention)	Register a retention policy for a queue (before connect())

---

VsRetention

Per-queue retention policy. See Retention.

Field	Type	Default	Description
ttl_seconds	Optional[int]	None	Drop messages older than this
max_len	Optional[int]	None	Keep at most this many messages

---

VsBaseConsumer

Abstract base. Extend and implement handle to process messages.

Attribute:

Attribute	Type	Description
uuid	Optional[str]	Set automatically at registration time. Use it with stop_listener() to stop a specific instance.

Methods:

Method	Required	Description
handle(message)	Yes	Process a message. Raise to trigger retry.
start()	No — no-op by default	Setup before consuming begins
stop()	No — no-op by default	Cleanup on shutdown — called on graceful cancel
on_error(message, error)	No — no-op by default	Called on handle failure before retry

---

VsQueueManager

Lifecycle manager for all consumers.

Method	Description
register(queue_name, consumer)	Register a consumer, auto-connect, start immediately. Returns UUID.
register_listeners()	Auto-discover @queue_listener classes. Safe to call only once.
get_listeners()	Returns Dict[str, List[VsBaseConsumer]]
stop_listener(queue_name, uuid=None)	Stop one or all consumers on a queue
stop_all()	Stop all consumers and disconnect
health_check()	Returns True if connected and broker is healthy

---

VsRetryPolicy

Defines retry and DLQ behaviour on consumer failure.

Parameter	Default	Description
max_retries	3	Max attempts
backoff_seconds	1.0	Base exponential backoff delay
dead_letter_queue	None	DLQ name

---

VsQueueRegistry

Registry of queue provider classes.

Method	Description
register(name, queue_class)	Register a provider
get(name)	Get a provider class by name
available()	List all registered provider names

---

VsQueueFactory

Creates queue instances from the registry.

Method	Description
create(provider, host, port, credentials)	Instantiate a queue. Call connect() before use.

---

@queue_listener

Decorator. Registers a VsBaseConsumer subclass for auto-discovery by VsQueueManager.

Parameter	Default	Description
queue	—	Queue name
concurrency	1	Parallel consumer instances
max_retries	3	Max retries on failure
retry_backoff_seconds	1.0	Exponential backoff base delay
dead_letter_queue	None	DLQ name

---

Running Tests

./run_tests.sh