Messaging Queues: RabbitMQ, SQS & PubSub

Relevant source files

The following files were used as context for generating this wiki page:

• skills/forze-messaging-streaming/SKILL.md
• skills/forze-specs-infrastructure/SKILL.md
• skills/forze-framework-usage/SKILL.md
• skills/forze-wiring/SKILL.md
• pyproject.toml

The Forze framework provides a robust set of messaging and streaming contracts designed to facilitate asynchronous communication within a hexagonal architecture. By utilizing logical specifications (Specs) and decoupled adapters, the system supports point-to-point queuing via RabbitMQ and SQS, as well as broadcast-style event distribution through Pub/Sub and append-only logs via Streams.

These messaging components allow the application layer to remain agnostic of the underlying transport provider. Handlers interact with high-level ports (Query and Command ports) resolved from the ExecutionContext, while infrastructure-specific details like queue namespaces, durable settings, and connection pooling are handled in dedicated dependency modules.

Architecture and Contracts

The messaging system is built upon a separation between logical routing and physical infrastructure. Application code defines QueueSpec, PubSubSpec, or StreamSpec objects that contain the message schema and a logical name.

Component Interaction Flow

The following diagram illustrates how a message flows from a producer to a consumer through the Forze abstraction layer:

The producer resolves a command port via the context, which uses the DepsRegistry to find the appropriate adapter (e.g., RabbitMQ or SQS) based on the spec name. Sources: skills/forze-messaging-streaming/SKILL.md:1-15, skills/forze-framework-usage/SKILL.md:25-45

Core Messaging Types

Contract Type	Key Purpose	Project Contract Reference
Queue	Point-to-point async task processing.	QueueSpec
Pub/Sub	Broadcast events to multiple subscribers by topic.	PubSubSpec
Stream	Append-only logs for consumer-group reads.	StreamSpec

Sources: skills/forze-messaging-streaming/SKILL.md:54-70, skills/forze-framework-usage/SKILL.md:112-118

Queue Operations: RabbitMQ & SQS

Forze integrates with both RabbitMQ (via aio-pika) and AWS SQS (via aioboto3). Both integrations share a similar wiring pattern using QueueQueryDepKey and QueueCommandDepKey.

SQS Specifics

SQS integration supports long polling and FIFO queues. Wiring is performed through the SQSDepsModule, which maps logical names to physical namespaces.
Sources: skills/forze-messaging-streaming/SKILL.md:43-52, pyproject.toml:79-85

RabbitMQ Specifics

RabbitMQ uses durable queues and persistent messages by default. It supports manual acknowledgement (ack) and negative acknowledgement (nack), often with publisher confirms enabled.
Sources: skills/forze-messaging-streaming/SKILL.md:43-52, pyproject.toml:104-104

Implementation Example

# Resolving a Queue Writer
writer = ctx.deps.resolve_configurable(
    ctx, QueueCommandDepKey, order_queue, route=order_queue.name
)

# Enqueueing with a delay
await writer.enqueue(
    "reminders",
    ReminderPayload(user_id="u-1"),
    delay=timedelta(minutes=10),
)

Sources: skills/forze-messaging-streaming/SKILL.md:31-41

Pub/Sub and Streaming

Pub/Sub is used for event-driven architectures where the producer does not know who the consumers are.

Pub/Sub Flow

Producers publish to a specific topic, and the underlying adapter handles the distribution.

Streams

Streams model append-only logs. They support append operations for producers and read, tail, or consumer-group operations for consumers. Consumer groups are handled via StreamGroupQueryDepKey.
Sources: skills/forze-messaging-streaming/SKILL.md:67-78

Infrastructure Wiring

Wiring messaging requires mapping the logical StrEnum names used in Specs to physical configuration in the DepsModule.

Module	Configuration Map	Purpose
RabbitMQDepsModule	queue_readers, queue_writers	Maps specs to RabbitMQ queues/exchanges.
SQSDepsModule	queue_readers, queue_writers	Maps specs to SQS URLs/namespaces.
MockDepsModule	Internal Registry	Used for unit testing messaging without a broker.

Sources: skills/forze-specs-infrastructure/SKILL.md:45-60, skills/forze-wiring/SKILL.md:215-225

Processing Rules

1. At-Least-Once Delivery: Systems should prefer idempotent consumers as brokers can redeliver messages.
2. Acknowledgement Timing: Only acknowledge a message (ack) after the business operation has successfully finished.
3. Transient Failures: Use nack(..., requeue=True) for temporary issues (e.g., network blips) and requeue=False to move messages to a Dead Letter Queue (DLQ).
4. Outbox Pattern: Wrap enqueueing in transactions using defer_after_commit to prevent premature event publication if a database transaction fails.

Sources: skills/forze-messaging-streaming/SKILL.md:80-87, skills/forze-framework-usage/SKILL.md:55-65

Anti-Patterns to Avoid

• Missing Route Parameter: Resolving queue ports without route=spec.name when using routed modules (SQS/RabbitMQ) will cause resolution errors.
• Premature Ack: Acknowledging a message before the handler successfully completes results in data loss if the process crashes mid-execution.
• Direct Imports: Importing SQS or RabbitMQ adapter classes directly into handlers violates hexagonal architecture boundaries; use contracts and dependency keys.
• Confusing Spec vs Provider Names: Spec names are for dependency routing; queue or topic names are provider-level strings.

Sources: skills/forze-messaging-streaming/SKILL.md:89-98, skills/forze-framework-usage/SKILL.md:11-20

Summary

Messaging in Forze is a high-level abstraction that enables scalable, asynchronous workflows. By standardizing on QueueSpec, PubSubSpec, and StreamSpec, developers can switch between local testing with MockDepsModule and production environments using RabbitMQ or SQS with minimal code changes. The system emphasizes reliability through explicit acknowledgement protocols and transactional integration via the outbox pattern.