53 lines
2.2 KiB
Python
53 lines
2.2 KiB
Python
"""Messaging topology constants + builders (REQ-021, dcy-006 OQ-2).
|
|
|
|
Three patterns:
|
|
1. snapshot RPC — direct queue `trex.snapshot.req`, reply via amq.rabbitmq.reply-to
|
|
2. intent RPC — direct queue `trex.intent.req`, reply via amq.rabbitmq.reply-to
|
|
3. events — topic exchange `trex.events`, routing key {event_type}.{strategy_id}.{symbol}
|
|
|
|
Mapping principle: requests carry view/strategy_id in *payload*; events carry
|
|
strategy_id/symbol in the *routing key* so brokers can filter per-strategy
|
|
queues with patterns like `*.{strategy_id}.#`.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
# --- RPC queues (durable for intent; transient-ish for snapshot but kept
|
|
# durable for simplicity since the queue itself is shared) ---
|
|
SNAPSHOT_REQ_QUEUE = "trex.snapshot.req"
|
|
INTENT_REQ_QUEUE = "trex.intent.req"
|
|
PAUSE_REQ_QUEUE = "trex.pause.req" # BAT-008 / REQ-032 control RPC
|
|
|
|
# --- Events ---
|
|
EVENTS_EXCHANGE = "trex.events" # topic, durable
|
|
EVENTS_DLQ_EXCHANGE = "trex.events.dlq" # fanout, durable; receives unroutable/NACKed
|
|
EVENTS_DLQ_QUEUE = "trex.events.dlq" # one collector queue bound to DLQ exchange
|
|
|
|
|
|
def event_routing_key(event_type: str, strategy_id: str, symbol: str) -> str:
|
|
"""Build the routing key for an event (dcy-006 OQ-2 mapping)."""
|
|
# Sanitize tokens — AMQP routing keys cannot contain '.', '*', '#'.
|
|
safe = lambda s: str(s or "_").replace(".", "_").replace("*", "_").replace("#", "_") # noqa: E731
|
|
return f"{safe(event_type)}.{safe(strategy_id)}.{safe(symbol)}"
|
|
|
|
|
|
def strategy_event_queue_name(strategy_id: str) -> str:
|
|
"""Per-strategy durable queue name (catch-up + last-wins)."""
|
|
return f"trex.events.strategy.{strategy_id}"
|
|
|
|
|
|
def strategy_event_binding_pattern(strategy_id: str) -> str:
|
|
"""Bind pattern for a strategy queue — all event types, all symbols.
|
|
|
|
Wildcards: `*` matches one word, `#` matches zero-or-more words.
|
|
"""
|
|
return f"*.{strategy_id}.#"
|
|
|
|
|
|
# --- per-queue arguments (REQ-023 TTL / max-length / drop-oldest) ---
|
|
STRATEGY_QUEUE_ARGS: dict = {
|
|
"x-message-ttl": 3_600_000, # 1h
|
|
"x-max-length": 10_000,
|
|
"x-overflow": "drop-head", # drop-oldest, then accept new (REQ-023 AC-7)
|
|
"x-dead-letter-exchange": EVENTS_DLQ_EXCHANGE,
|
|
}
|