Consumer 소유 스레드 기반 타이머 등록·취소 지원

docs/README.en.md

@@ -35,5 +35,5 @@ Messages published by a producer pass through the broker to a Worker Consumer, a
 - [Lifecycle hooks](./LifecycleHooks.en.md): hook order and resource management across Workers, pools, and tasks
 - [Request/reply contracts and validation](./Contracts.en.md): schemas, validators, and task contracts
 - [Operations and monitoring](./Operations.en.md): Inspector, logging, and operational broker considerations
-- [RabbitMQ](./RabbitMQ.en.md): queues, exchanges, streams, RPC, and RabbitMQ workers
+- [RabbitMQ](./RabbitMQ.en.md): queues, exchanges, streams, RPC, connection timers, and RabbitMQ workers
 - [Redis Streams](./Redis.en.md): messages, consumer groups, acknowledgements, and Redis workers

docs/README.md

@@ -35,5 +35,5 @@ Producer가 발행한 메시지는 브로커를 거쳐 Worker의 Consumer로 전
 - [생명주기 훅](./LifecycleHooks.md): Worker·풀·작업 훅의 실행 순서와 자원 관리
 - [요청·응답 계약과 검증](./Contracts.md): 스키마, validator와 작업 계약
 - [운영과 모니터링](./Operations.md): Inspector, 로깅, 브로커 선택과 운영 고려사항
-- [RabbitMQ](./RabbitMQ.md): 큐, Exchange, 스트림, RPC와 RabbitMQ Worker
+- [RabbitMQ](./RabbitMQ.md): 큐, Exchange, 스트림, RPC, 연결 타이머와 RabbitMQ Worker
 - [Redis Streams](./Redis.md): 메시지, consumer group, ACK와 Redis Worker

docs/RabbitMQ.en.md

@@ -88,6 +88,116 @@ Call `ack()` on success, `nack()` for redelivery, or `reject()` to reject.
 Rejected messages go to the broker-configured dead-letter destination or are
 discarded when none is configured.
 
+### Connection timers
+
+RabbitMQ connections support the optional `SetTimer` protocol.
+`set_timer(cb, seconds)` schedules a one-shot callback using integer seconds
+from registration and returns a cancelable `Timer`. The callback takes no
+arguments and returns no value. Code using a connection can check support
+with `isinstance(connection, SetTimer)`. Timer support is not required by
+the base `Connection` contract or Redis connections.
+
+This example consumes an empty queue and stops consumption from a callback
+after five seconds. It requires local RabbitMQ and leaves
+`defstream.example.timer` in place. Messages sent to this queue by other
+producers are printed and acknowledged.
+
+```python
+from DefStream.Broker import Event, SetTimer
+from DefStream.Broker.Implements.RabbitMQ import Configuration, Connection
+
+
+def receive(event: Event) -> None:
+  print(event.body)
+  event.ack()
+
+
+connection = Connection(Configuration(
+  host='127.0.0.1',
+  port=5672,
+  username='guest',
+  password='guest',
+))
+try:
+  connection.connect()
+  if not isinstance(connection, SetTimer):
+    raise TypeError('Connection does not support timers')
+  connection.create_queue('defstream.example.timer')
+  consumer = connection.consumer()
+  consumer.subs('defstream.example.timer', receive)
+
+  def expired() -> None:
+    print('Timer fired')
+    consumer.stop()
+
+  timer = connection.set_timer(cb=expired, seconds=5)
+  try:
+    consumer.run()
+  finally:
+    timer.remove()
+    consumer.close()
+finally:
+  connection.close()
+```
+
+Call `timer.remove()` before execution to cancel the scheduled callback.
+The handle belongs to the connection that registered it; reopening a connection
+does not redirect cancellation to the new connection.
+Already-fired or canceled timers, closed connections, and negative or zero
+delays retain Pika's behavior and errors. `remove()` does not interrupt a
+callback that is already running.
+
+For repetition, register the next timer inside the callback.
+The following excerpt replaces the `expired` definition and the one-shot
+registration, execution, and consumer cleanup above, after preparing the
+subscription through `consumer.subs(...)`. Keep the outer `finally` that
+closes the connection. It fires three times at five-second intervals, then
+stops, retaining the latest handle for cancellation.
+
+```python
+from DefStream.Broker import Consumer, SetTimer, Timer
+
+
+def run_repeated(connection: SetTimer, consumer: Consumer) -> None:
+  remaining = 3
+  timer: Timer
+
+  def tick() -> None:
+    nonlocal remaining, timer
+    print('Timer fired')
+    remaining -= 1
+    if remaining:
+      timer = connection.set_timer(tick, 5)
+    else:
+      consumer.stop()
+
+  timer = connection.set_timer(tick, 5)
+  try:
+    consumer.run()
+  finally:
+    timer.remove()
+
+
+try:
+  run_repeated(connection, consumer)
+finally:
+  consumer.close()
+```
+
+The implementation delegates directly to Pika's `call_later` and
+`remove_timeout`. Callbacks run only while Pika processes events, such as
+during `consumer.run()`. Registering a timer and calling `sleep()` alone
+does not dispatch it. Long message handlers and other event-loop work can
+delay execution. Callbacks run on the connection owner thread, and their
+exceptions propagate through the existing event-processing path.
+
+Registration and cancellation follow Pika's owner-thread restrictions;
+no additional thread safety is provided. Consumer exit does not automatically
+cancel connection timers, so callers must cancel reservations when appropriate.
+The examples cancel before closing the connection; cancellation on a closed
+connection is not guaranteed to succeed. No separate timer thread,
+general-purpose scheduler, or automatic cleanup layer is provided.
+
 ### Logging
 
 `Connection`, `Consumer`, `Queue`, `Exchange`, `Stream`, and `Event` expose a class
@@ -383,3 +493,18 @@ publishing the reply but before ACK, the task may be redelivered and run again.
 - [Sample.py](../samples/Broker/RabbitMQ/Sample.py)
 - [Sample.Stream.py](../samples/Broker/RabbitMQ/Sample.Stream.py)
 - [Sample.RPC.py](../samples/Broker/RabbitMQ/Sample.RPC.py)
+- [Sample.Timer.py](../samples/Broker/RabbitMQ/Sample.Timer.py): one-shot cancellation and repetition by re-registering callbacks
+
+The timer sample uses local RabbitMQ at `127.0.0.1:5672` with the `guest` account.
+Enable the RabbitMQ implementation in your development environment and run it
+from the repository root.
+
+```shell
+python samples/Broker/RabbitMQ/Sample.Timer.py
+```
+
+Each run creates a uniquely named empty queue, cancels one callback before
+execution, then logs three timer callbacks at one-second intervals and stops
+consumption. It cleans up the latest timer, consumer channel, sample queue,
+and connection. The Stream sample also uses the connection's `set_timer`
+instead of a separate timer thread to stop consumption.

docs/RabbitMQ.md

@@ -103,6 +103,109 @@ finally:
 일반 연결의 heartbeat는 애플리케이션 작업·유휴 시간에 맞게 설정하고 처리되도록 관리하세요.
 연결과 그 리소스는 **소유 스레드에서만 사용**해야 하며, 스레드·프로세스 사이에 공유하지 마세요.
 
+### 연결 타이머
+
+RabbitMQ 연결은 선택적 `SetTimer` 프로토콜을 지원합니다.
+`set_timer(cb, seconds)`는 등록 시점부터 정수 초 단위로 일회성 콜백을 예약하고,
+취소할 수 있는 `Timer`를 반환합니다. 콜백은 인자 없이 호출되며 반환값이 없습니다.
+연결을 사용하는 코드에서 `isinstance(connection, SetTimer)`로 지원 여부를 확인할 수 있습니다.
+기본 `Connection` 계약과 Redis 연결에는 타이머 지원이 강제되지 않습니다.
+
+다음 예제는 메시지가 없는 큐를 소비하면서 5초 뒤 콜백으로 소비를 중단합니다.
+로컬 RabbitMQ가 필요하며 `defstream.example.timer` 큐는 실행 후에도 유지합니다.
+다른 생산자가 이 큐에 보낸 메시지는 출력한 뒤 ACK합니다.
+
+```python
+from DefStream.Broker import Event, SetTimer
+from DefStream.Broker.Implements.RabbitMQ import Configuration, Connection
+
+
+def receive(event: Event) -> None:
+  print(event.body)
+  event.ack()
+
+
+connection = Connection(Configuration(
+  host='127.0.0.1',
+  port=5672,
+  username='guest',
+  password='guest',
+))
+try:
+  connection.connect()
+  if not isinstance(connection, SetTimer):
+    raise TypeError('Connection does not support timers')
+  connection.create_queue('defstream.example.timer')
+  consumer = connection.consumer()
+  consumer.subs('defstream.example.timer', receive)
+
+  def expired() -> None:
+    print('Timer fired')
+    consumer.stop()
+
+  timer = connection.set_timer(cb=expired, seconds=5)
+  try:
+    consumer.run()
+  finally:
+    timer.remove()
+    consumer.close()
+finally:
+  connection.close()
+```
+
+실행 전에 `timer.remove()`를 호출하면 해당 예약을 취소합니다.
+핸들은 등록 당시 연결에 연결되며, 연결을 다시 열어도 새 연결의 타이머를 취소하지 않습니다.
+이미 실행·취소된 예약, 닫힌 연결, 음수·0초 지연의 동작과 오류는 Pika API를 그대로 따릅니다.
+`remove()`는 실행 중인 콜백을 강제로 중단하지 않습니다.
+
+반복 실행은 콜백 안에서 다음 타이머를 등록하면 됩니다.
+아래는 반복 실행용 발췌입니다. 위 예제에서 `consumer.subs(...)`까지 준비한 뒤,
+`expired` 정의와 일회성 타이머 등록·실행·소비자 정리 부분을 아래 코드로 교체합니다.
+바깥쪽 `finally`의 연결 종료는 유지합니다.
+5초 간격으로 세 번 실행하고 중단하며, 취소할 때 사용할 최신 핸들을 보관합니다.
+
+```python
+from DefStream.Broker import Consumer, SetTimer, Timer
+
+
+def run_repeated(connection: SetTimer, consumer: Consumer) -> None:
+  remaining = 3
+  timer: Timer
+
+  def tick() -> None:
+    nonlocal remaining, timer
+    print('Timer fired')
+    remaining -= 1
+    if remaining:
+      timer = connection.set_timer(tick, 5)
+    else:
+      consumer.stop()
+
+  timer = connection.set_timer(tick, 5)
+  try:
+    consumer.run()
+  finally:
+    timer.remove()
+
+
+try:
+  run_repeated(connection, consumer)
+finally:
+  consumer.close()
+```
+
+RabbitMQ 구현은 Pika의 `call_later`·`remove_timeout`에 직접 위임합니다.
+`consumer.run()`과 같은 Pika 이벤트 처리가 진행되어야 콜백이 실행됩니다.
+단순히 등록 후 `sleep()`하는 것만으로는 실행되지 않으며,
+긴 메시지 처리나 이벤트 루프 작업 때문에 실행이 늦어질 수 있습니다.
+콜백은 연결 소유 스레드에서 실행되고, 예외는 기존 이벤트 처리 경로로 전파됩니다.
+
+등록·취소는 Pika의 소유 스레드 제약을 따르며 추가 스레드 안전성을 제공하지 않습니다.
+Consumer 종료 시 연결 타이머를 자동으로 취소하지 않으므로,
+필요한 예약은 호출자가 적절한 시점에 취소해야 합니다.
+위 예제는 연결을 닫기 전에 취소하며, 닫힌 연결에서 취소가 항상 성공한다고 보장하지 않습니다.
+별도 타이머 스레드·범용 스케줄러·자동 정리 계층은 제공하지 않습니다.
+
 ### 로깅
 
 여섯 클래스는 `Broker.RabbitMQ.ClassName` 이름의 클래스 멤버
@@ -370,3 +473,16 @@ sequenceDiagram
 - [Sample.py](../samples/Broker/RabbitMQ/Sample.py)
 - [Sample.Stream.py](../samples/Broker/RabbitMQ/Sample.Stream.py)
 - [Sample.RPC.py](../samples/Broker/RabbitMQ/Sample.RPC.py)
+- [Sample.Timer.py](../samples/Broker/RabbitMQ/Sample.Timer.py): 일회성 예약 취소와 콜백 재등록에 의한 반복 실행
+
+타이머 샘플은 로컬 RabbitMQ의 `127.0.0.1:5672`와 `guest` 계정을 사용합니다.
+개발 환경에서 RabbitMQ 구현체를 사용할 수 있어야 하며, 저장소 루트에서 실행하세요.
+
+```shell
+python samples/Broker/RabbitMQ/Sample.Timer.py
+```
+
+실행마다 고유한 빈 큐를 만들고, 취소한 콜백은 실행하지 않으며,
+1초 간격으로 세 번 출력한 뒤 소비를 중단합니다.
+종료 시 최신 타이머·소비 채널·샘플 큐·연결을 정리합니다.
+Stream 샘플의 소비 중단도 별도 스레드 타이머 대신 연결의 `set_timer`를 사용합니다.

issues/feature/62/CHANGELOG.md

@@ -0,0 +1,26 @@
+# 연결 타이머 등록과 취소
+
+## 추가된 기능
+
+RabbitMQ 연결에서 `set_timer(cb, seconds)`로 일회성 콜백을 예약하고,
+반환된 `Timer`의 `remove()`로 취소할 수 있습니다.
+호출자는 Pika 연결이나 내부 타이머 ID를 직접 다루지 않아도 됩니다.
+
+`SetTimer`는 연결의 선택적 기능이며, `Timer`, `TimerCallback`과 함께
+`DefStream.Broker` 및 `DefStream.Broker.Connection`에서 import할 수 있습니다.
+반복 실행은 콜백에서 다음 타이머를 등록하는 방식으로 구성합니다.
+
+[실행 가능한 타이머 샘플](../../../samples/Broker/RabbitMQ/Sample.Timer.py)에서
+취소와 재등록을 확인할 수 있습니다. 기존 Stream 샘플의 소비 중단도
+외부 스레드 타이머 대신 연결 타이머를 사용합니다.
+
+## 사용 조건
+
+- 지연 시간은 등록 시점부터 정수 초 단위로 지정합니다.
+- RabbitMQ 소비 루프 등 Pika의 이벤트 처리가 진행되어야 콜백이 실행됩니다.
+- 등록·취소는 연결 소유 스레드에서 수행하며, 긴 이벤트 처리로 실행이 지연될 수 있습니다.
+- Consumer 종료 시 예약을 자동 취소하지 않습니다. 취소 시점과 연결의 사용 순서는 호출자가 관리합니다.
+- 중복 취소·종료된 연결·음수·0초 등의 동작과 오류는 기반 Pika API를 따릅니다.
+- Redis 타이머 지원이나 기존 Consumer·Worker API 변경은 포함하지 않습니다.
+
+자세한 사용법은 [RabbitMQ 연결 타이머 안내](../../../docs/RabbitMQ.md#연결-타이머)를 참고하세요.

issues/feature/62/IMPLEMENTS.md

@@ -0,0 +1,26 @@
+# 연결 타이머 구현 결과
+
+## 구현 내용
+
+- [공통 연결 계약](../../../src/Broker/Connection.py)에 `Timer`, `TimerCallback`, `SetTimer`를 추가하고 [Broker 공개 진입점](../../../src/Broker/__init__.py)에서도 노출했다.
+- `Timer`, `SetTimer`의 런타임 검사·추상 메서드 선언과 `set_timer(cb: TimerCallback, seconds: int) -> Timer` 시그니처를 확정 명세대로 구현했다.
+- [RabbitMQ 연결](../../../src/Broker/Implements/RabbitMQ/Connection.py)은 `SetTimer`를 명시적으로 상속한다. 기존 연결 확인 후 즉시 `call_later(seconds, cb)`로 일회성 예약을 등록하므로 지연 시간은 등록 시점부터 계산된다.
+- 반환하는 구체 `Timer`는 등록 당시 실제 연결과 불투명 핸들만 보관한다. `remove()`는 해당 연결의 `remove_timeout`에 직접 위임하여 상위 연결 객체가 재연결되어도 취소 대상이 바뀌지 않는다.
+- 콜백 래퍼·스레드·추가 동기화·상태 검사·자동 정리 계층은 추가하지 않았다. 스레드 제약·취소·입력값·오류 동작은 Pika API를 따른다.
+- 한국어·영어 [RabbitMQ 사용 안내](../../../docs/RabbitMQ.md)에 일회성·취소·재등록 예제와 사용 제약을 반영했다.
+- 실행 가능한 [타이머 샘플](../../../samples/Broker/RabbitMQ/Sample.Timer.py)을 추가하고, [Stream 샘플](../../../samples/Broker/RabbitMQ/Sample.Stream.py)의 별도 스레드 타이머와 중계 함수를 연결의 `set_timer(consumer.stop, 1)`로 교체했다.
+
+## 유지한 범위
+
+Consumer·Worker·Ping·Pool·Executor·Event·Redis 구현과 연결·소비 종료 처리는 변경하지 않았다. 기본 Connection 계약에 타이머를 강제하지 않으며, 별도 이슈 #63·#64의 리팩터링을 포함하지 않았다.
+
+## 검증과 제한
+
+샘플 검증 추가 후 단위·회귀 45개가 통과했으며 실제 브로커 테스트 4개는 기본 실행에서 건너뛰었다. 별도 유휴 실행 테스트는 로컬 RabbitMQ 연결 실패로 동작 검증까지 진행하지 못했다. 타입·구문 진단과 사용 예제 검증 결과는 [작업 종합](./TASKS.md)에 구분하여 기록했다.
+
+구현은 반영했지만 유휴 실행·실제 소유 스레드·취소·재등록·콜백 예외의 실제 브로커 인수 확인은 남아 있다.
+
+## 구현 기록
+
+- [연결 타이머 계약과 RabbitMQ 등록·취소 구현](./implements/1793777682327.md)
+- [실행 가능한 연결 타이머 샘플 추가](./implements/1793778339166.md)