genichinandClaude Opus 4.8
d182464b7a
발행 전용 연결의 idle heartbeat 사망 후 첫 발행 유실 결함 제거. - REQ-091 발행 재시도 + 강제 재연결(_force_reconnect, is_open 불신) - REQ-092 idle keepalive(execution-loop 단일스레드 틱, process_data_events time_limit=0) - REQ-093 DLQ 가시성 + emit 실패 ERROR 경보(구조화 로그, dcy-029 재사용) - REQ-094 baseline 문서 정합(runtime-flows/operations/project-structure) 독립 opus QC PASS-WITH-NOTES → 시정 2건(ruff F401, AC-3 프로덕션 경로) 완료. 554 passed/47 skipped/회귀 0/ruff clean. NFR-1 wire 무회귀(interface-contract diff 0). BAT-047 release-candidate, REQ-091~094 Implemented. dcy-031. M2 트랙B. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VHtM5Ws1HKCKQ4JGi6GRb5
131 lines
3.9 KiB
Python
131 lines
3.9 KiB
Python
"""Messaging transport port + DTOs (REQ-021, REQ-022, dcy-006).
|
|
|
|
Application-layer contract for trex-core ↔ 전략 모듈 통신. Domain stays
|
|
free of AMQP / pika dependencies (NFR-1); concrete adapters live in
|
|
trex_core.adapters.messaging.
|
|
|
|
Three patterns (dcy-006 OQ-2):
|
|
1. snapshot RPC (request-reply) — read-only, transient.
|
|
2. intent admission RPC (request-reply) — durable, correlation_id dedup.
|
|
3. event pub/sub — topic exchange, per-strategy queue, last-wins.
|
|
|
|
Request payloads carry view/strategy_id in the message body; events carry
|
|
strategy_id/symbol in the routing key (REQ-021 mapping principle).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum
|
|
from typing import Any, Callable, Optional, Protocol
|
|
|
|
|
|
class TransportHealth(str, Enum):
|
|
"""Connection state exposed to application for degrade branches (REQ-024)."""
|
|
CONNECTED = "connected"
|
|
DISCONNECTED = "disconnected"
|
|
RECONNECTING = "reconnecting"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RpcRequest:
|
|
"""Inbound RPC message handed to a consumer handler."""
|
|
payload: dict
|
|
correlation_id: str
|
|
reply_to: Optional[str] = None
|
|
properties: dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RpcResponse:
|
|
"""Outbound RPC response returned from a consumer handler."""
|
|
payload: dict
|
|
|
|
|
|
class TransportError(Exception):
|
|
"""Base class for messaging transport failures."""
|
|
|
|
|
|
class TransportUnavailableError(TransportError):
|
|
"""Raised when transport.health() is not CONNECTED and the caller
|
|
cannot proceed (REQ-024: no local buffering, conservative halt)."""
|
|
|
|
|
|
class RpcTimeoutError(TransportError):
|
|
"""RPC reply not received within the configured timeout."""
|
|
|
|
|
|
# Type aliases for handlers
|
|
RpcHandler = Callable[[RpcRequest], RpcResponse]
|
|
EventHandler = Callable[[str, dict], None] # (routing_key, payload)
|
|
|
|
|
|
class MessagingTransport(Protocol):
|
|
"""Transport boundary for messaging-based communication (REQ-022)."""
|
|
|
|
def declare_topology(self) -> None:
|
|
"""Declare exchanges, queues, bindings (REQ-021)."""
|
|
...
|
|
|
|
def publish_event(
|
|
self,
|
|
routing_key: str,
|
|
payload: dict,
|
|
*,
|
|
mandatory: bool = True,
|
|
) -> None:
|
|
"""Publish to the events topic exchange. Uses publisher confirms;
|
|
unroutable or NACKed messages are routed to the DLQ (REQ-023).
|
|
Raises TransportUnavailableError if not connected."""
|
|
...
|
|
|
|
def call_rpc(
|
|
self,
|
|
queue: str,
|
|
payload: dict,
|
|
*,
|
|
correlation_id: str,
|
|
timeout: float,
|
|
) -> dict:
|
|
"""Client-side RPC: publish to `queue`, wait for reply.
|
|
Raises RpcTimeoutError on timeout, TransportUnavailableError on outage."""
|
|
...
|
|
|
|
def consume_rpc_forever(
|
|
self,
|
|
queue: str,
|
|
handler: RpcHandler,
|
|
stop: threading.Event,
|
|
) -> None:
|
|
"""Server-side RPC consumption. Blocks until `stop` is set.
|
|
Auto-reconnects on connection drop (REQ-022 AC-4)."""
|
|
...
|
|
|
|
def consume_events_forever(
|
|
self,
|
|
queue: str,
|
|
binding_pattern: str,
|
|
handler: EventHandler,
|
|
stop: threading.Event,
|
|
) -> None:
|
|
"""Subscribe to events matching `binding_pattern` on the events
|
|
exchange, dispatch to handler. Blocks until `stop` is set."""
|
|
...
|
|
|
|
def keepalive_tick(self) -> None:
|
|
"""Process pending I/O frames to keep the connection alive (REQ-092).
|
|
|
|
MUST be called from the thread that owns this transport instance.
|
|
Non-blocking (time_limit=0 equivalent). Implementations that do not
|
|
need keepalive (e.g. InMemoryTransport) should be a no-op.
|
|
"""
|
|
...
|
|
|
|
def health(self) -> TransportHealth:
|
|
"""Current connection state (REQ-024)."""
|
|
...
|
|
|
|
def close(self) -> None:
|
|
"""Release connection resources."""
|
|
...
|