From 9a39b04096fa32c9afd2186868d40384403150b3 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=ED=97=88=EC=9A=A9=EC=84=A0?= <yongseon.heo@gmail.com>
Date: Wed, 4 Nov 2026 14:12:06 +0900
Subject: [PATCH 1/4] =?UTF-8?q?Consumer=20=EC=86=8C=EC=9C=A0=20=EC=8A=A4?=
 =?UTF-8?q?=EB=A0=88=EB=93=9C=20=EA=B8=B0=EB=B0=98=20=ED=83=80=EC=9D=B4?=
 =?UTF-8?q?=EB=A8=B8=20=EB=93=B1=EB=A1=9D=C2=B7=EC=B7=A8=EC=86=8C=20?=
 =?UTF-8?q?=EC=A7=80=EC=9B=90=20=EB=AC=B8=EC=84=9C=20=EC=B6=94=EA=B0=80?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- RabbitMQ Consumer의 메시지 소비 중 타이머 등록·취소 기능 설명
- 지연 시간과 콜백을 지정해 일회성 타이머 등록 및 취소 방법 제안
- 소비 종료 시 타이머 수명주기 정리 및 콜백 오류 처리 방식 설명
- 관련 이슈: #62
- 이슈 URL: https://github.com/yong5eon/DefStream/issues/62
---
 issues/feature/62/README.md | 59 +++++++++++++++++++++++++++++++++++++
 1 file changed, 59 insertions(+)
 create mode 100644 issues/feature/62/README.md

diff --git a/issues/feature/62/README.md b/issues/feature/62/README.md
new file mode 100644
index 0000000..7ef33d2
--- /dev/null
+++ b/issues/feature/62/README.md
@@ -0,0 +1,59 @@
+# Consumer 소유 스레드 기반 타이머 등록·취소 지원
+
+## 이슈와 작업 상태
+
+- 원본: GitHub Issues [#62](https://github.com/yong5eon/DefStream/issues/62)
+- 제목: Consumer 소유 스레드 기반 타이머 등록·취소 지원
+- 상태: Open
+- 유형·레이블: 기능 (`feature`), 원본 이슈에 적용됨
+- 원본 레이블: `feature`, `rabbitmq`
+- 작업 대상: `yong5eon/DefStream`, 작업 브랜치 `feature/62`
+- 분기 기준: `origin/master`와 동일한 로컬 `master`의 `f1bf897`
+- 병합 대상: `master`
+- 로컬 진행 상태: 브랜치 생성과 이슈 원문 기록 완료, 요구사항 분석 전
+- 댓글: 없음
+
+## 이슈 원문
+
+## 해결하려는 문제
+
+RabbitMQ Consumer로 메시지를 소비하면서, 메시지가 없는 동안에도 연결 유지용 콜백 등을 일정 시간 뒤 실행할 필요가 있습니다.
+
+현재 `Consumer.call(callback)`은 연결 소유 스레드로 콜백을 전달하지만 지연 타이머의 등록·취소 API는 제공하지 않습니다. 따라서 호출자가 Pika의 `BlockingConnection.call_later()`·`remove_timeout()`에 직접 의존하거나 별도 스레드 타이머의 수명주기를 관리해야 합니다.
+
+`threading.Timer`에서 직접 출력·소비 중단을 처리하면 메시지 처리와 동시 실행될 수 있고, 타이머 스레드에서 발생한 예외도 소비 루프로 자동 전파되지 않습니다. `Consumer.call()`로 다시 전달할 수는 있지만 타이머 생성·재등록·취소·소비 종료 후 실행 방지는 호출자가 계속 관리해야 합니다.
+
+## 제안하는 해결 방법
+
+DefStream의 소비자 API를 통해 소비자 소유 스레드에서 실행되는 타이머를 등록·취소할 수 있게 지원해 주세요. 우선 현재 사용 사례인 RabbitMQ Consumer를 대상으로 합니다.
+
+- 지연 시간과 콜백을 지정해 일회성 타이머를 등록하고 취소 가능한 핸들을 반환합니다.
+- 메시지가 없는 소비 대기 중에도 콜백이 소비자 소유 스레드에서 실행되어야 합니다.
+- 반복 실행은 콜백에서 다음 타이머를 재등록하는 방식으로도 충분합니다. 별도 범용 스케줄러나 전용 타이머 스레드는 요구하지 않습니다.
+- 소비 종료·close 시 남은 타이머가 이후에 실행되지 않도록 수명주기를 정리합니다.
+- 콜백 오류는 기존 소비 루프의 오류 처리·전파 방식과 일관되게 드러내며 별도 스레드에서 조용히 유실하지 않습니다.
+- 호출자는 Pika 연결 객체·구현체 타이머 ID를 직접 다루지 않아도 되어야 합니다.
+
+구체적인 메서드 이름과 핸들 타입은 구현 시 기존 인터페이스와 대조해 결정하면 됩니다. 예를 들어 지연 콜백 등록과 해당 핸들 취소 형태를 고려할 수 있으며, 이 예시를 확정 API로 지정하는 것은 아닙니다.
+
+### 확인 기준
+
+- 메시지가 없는 동안에도 예약 콜백이 실행되고 메시지 처리와 같은 소유 스레드에서 실행되는지 확인합니다.
+- 취소한 타이머와 소비 종료 후의 타이머가 실행되지 않는지 확인합니다.
+- 재등록을 통한 주기 실행과 정상·오류 종료의 자원 정리를 검증합니다.
+- 콜백 예외가 기존 소비 오류 처리 경계로 전파되는지 확인합니다.
+- 기존 구독·ACK·소비 순서와 `Consumer.call()` 동작을 유지합니다.
+- 지원 범위·시간 단위·실행 지연 가능성·등록 및 취소 사용법을 관련 매뉴얼에 반영합니다.
+
+## 고려한 대안
+
+- Pika 연결의 타이머를 직접 사용: 같은 스레드에서 실행할 수 있지만 호출자가 브로커 구현체에 결합됩니다.
+- `threading.Timer`와 `Consumer.call()` 조합: 실행 스레드 전달은 가능하지만 별도 스레드·타이머 수명주기를 호출자가 관리해야 합니다.
+- 메시지를 받을 때만 경과 시간 검사: 유휴 중 실행이 필요한 요구를 충족하지 못합니다.
+
+## 추가 정보
+
+연결 유지용 출력은 사용 예시일 뿐이며 DefStream이 HTTP·SSE 프로토콜이나 해당 출력 내용을 구현해 달라는 요청은 아닙니다. 우선 타이머 등록·취소와 소비자 수명주기 통합만 지원하면 됩니다.
+
+- 현재 구현: https://github.com/yong5eon/DefStream/blob/master/src/Broker/Implements/RabbitMQ/Consumer.py
+- Pika BlockingConnection API: https://pika.readthedocs.io/en/stable/modules/adapters/blocking.html

From f6ee976ac63e37169d2df8506739517c297f5358 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=ED=97=88=EC=9A=A9=EC=84=A0?= <yongseon.heo@gmail.com>
Date: Wed, 4 Nov 2026 16:23:25 +0900
Subject: [PATCH 2/4] =?UTF-8?q?Connection=20=EA=B8=B0=EB=B0=98=20=ED=83=80?=
 =?UTF-8?q?=EC=9D=B4=EB=A8=B8=20=EC=9A=94=EA=B5=AC=EC=82=AC=ED=95=AD=20?=
 =?UTF-8?q?=EB=AC=B8=EC=84=9C=20=EC=B6=94=EA=B0=80=20=EB=B0=8F=20=EC=9E=91?=
 =?UTF-8?q?=EC=97=85=20=EA=B8=B0=EB=A1=9D=20=EC=A0=95=EB=A6=AC?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- Connection 기반 타이머 계약 명세와 관련 문서 추가
- Consumer 메서드 제거 및 close 추상 메서드 선언 철회
- 요구사항 분석 결과 및 검증 기록 정리

- 관련 이슈: #62
- 이슈 URL: https://github.com/yong5eon/DefStream/issues/62
---
 issues/feature/62/README.md              |   8 +-
 issues/feature/62/REQUIREMENTS.md        | 121 +++++++++++++++++++++++
 issues/feature/62/TASKS.md               |  25 +++++
 issues/feature/62/tasks/1793772101827.md |  17 ++++
 issues/feature/62/tasks/1793774701691.md |  13 +++
 issues/feature/62/tasks/1793775544541.md |  13 +++
 issues/feature/62/tasks/1793775771772.md |  13 +++
 7 files changed, 209 insertions(+), 1 deletion(-)
 create mode 100644 issues/feature/62/REQUIREMENTS.md
 create mode 100644 issues/feature/62/TASKS.md
 create mode 100644 issues/feature/62/tasks/1793772101827.md
 create mode 100644 issues/feature/62/tasks/1793774701691.md
 create mode 100644 issues/feature/62/tasks/1793775544541.md
 create mode 100644 issues/feature/62/tasks/1793775771772.md

diff --git a/issues/feature/62/README.md b/issues/feature/62/README.md
index 7ef33d2..8ce5ec4 100644
--- a/issues/feature/62/README.md
+++ b/issues/feature/62/README.md
@@ -10,9 +10,15 @@
 - 작업 대상: `yong5eon/DefStream`, 작업 브랜치 `feature/62`
 - 분기 기준: `origin/master`와 동일한 로컬 `master`의 `f1bf897`
 - 병합 대상: `master`
-- 로컬 진행 상태: 브랜치 생성과 이슈 원문 기록 완료, 요구사항 분석 전
+- 로컬 진행 상태: Connection 기반 타이머 계약 명세, 부가 동작은 기반 구현에 위임, 구현 전
 - 댓글: 없음
 
+## 작업 문서
+
+- [요구사항 명세](./REQUIREMENTS.md)
+- [작업 종합](./TASKS.md)
+- [최신 작업 기록](./tasks/1793775771772.md)
+
 ## 이슈 원문
 
 ## 해결하려는 문제
diff --git a/issues/feature/62/REQUIREMENTS.md b/issues/feature/62/REQUIREMENTS.md
new file mode 100644
index 0000000..b8fc1fa
--- /dev/null
+++ b/issues/feature/62/REQUIREMENTS.md
@@ -0,0 +1,121 @@
+# Connection 기반 선택적 타이머 요구사항
+
+## 목적과 범위
+
+[원본 이슈와 작업 정보](./README.md)를 기준으로, 연결 구현체가 제공하는 타이머를 공통 계약으로 이용한다. 이번 범위는 Connection을 이용하는 `SetTimer`, `Timer`, `TimerCallback`과 이를 지원하는 RabbitMQ 연결 구현, 직접 관련된 테스트·문서로 한정한다.
+
+Consumer의 인터페이스·구현과 Worker·Ping·Executor 등 다른 구성 요소는 변경하지 않는다. 앞서 논의한 리팩터링은 후속 작업으로 분리한다.
+
+이번 문서는 요구사항 분석과 명세이다. 제품 코드 변경과 구현 계획 작성은 아직 수행하지 않았다.
+
+## 분석 근거와 현재 동작
+
+- [Connection](../../../src/Broker/Connection.py)은 연결의 기본 계약과 `GetExchange`, `GetQueue`, `GetStream`, `GetConsumer` 등 선택적 기능의 프로토콜을 구분한다. 타이머도 모든 연결에 강제하지 않고 같은 방식으로 분리한다.
+- [RabbitMQ Consumer](../../../src/Broker/Implements/RabbitMQ/Consumer.py)는 이미 Pika의 `call_later`와 `remove_timeout`으로 내부 확인용 타이머를 관리한다. 소비 루프의 `finally`에서 해당 타이머를 취소한다.
+- 기존 Consumer의 타이머와 종료 처리는 변경하지 않는다. 새 예약의 등록·취소는 연결 객체를 통해 제공하며 소비자별 예약 관리 기능은 추가하지 않는다.
+
+## 원문과 대화에서 변경한 사항
+
+- 원문은 소비자 API를 통한 타이머 지원을 제안했지만, 대화에서는 연결 객체의 선택적 `SetTimer` 프로토콜로 제공하기로 확정했다.
+- 프로토콜과 반환 핸들·콜백의 명칭은 `SetTimer`, `Timer`, `TimerCallback`으로 확정했다.
+- 앞서 논의한 Consumer의 `complete`, `fail`, `call` 제거와 `close` 추상 메서드 선언은 이번 요구사항에서 철회했다. 기존 선언·구현·호출 흐름을 그대로 유지한다.
+- 원문의 소비 종료 시 예약 정리는 이번 범위에서 제외한다. 타이머 수명주기와 적절한 취소 시점은 호출자와 기반 구현이 관리하며, 별도 자동 정리 계층을 추가하지 않는다.
+
+## 확정 요구사항
+
+### R1. 선택적 타이머 등록
+
+- [연결 계약 모듈](../../../src/Broker/Connection.py)에 별도 `SetTimer` 프로토콜을 정의한다. 기본 `Connection`과 Consumer에는 타이머 등록을 필수 계약으로 추가하지 않는다.
+- 지원하는 연결 구현체가 지연 시간과 콜백을 받아 일회성 예약을 등록하고 `Timer` 핸들을 반환한다. 타이머의 구현 방식은 각 연결 구현체에 맡긴다.
+- 우선 RabbitMQ 연결에서 구현한다. Redis 등 다른 연결에는 타이머 구현이나 미지원 예외 메서드를 강제하지 않는다.
+- 이번 RabbitMQ 구현은 Pika의 내부 타이머를 사용한다. 별도 타이머 스레드나 범용 스케줄러를 추가하지 않으며, 호출자가 Pika 연결 객체나 타이머 ID를 직접 다루지 않게 한다.
+
+### R2. 타이머 핸들과 콜백
+
+- `Timer`는 내부 구현을 숨기는 반환 핸들의 프로토콜이며, 공개 기능은 `remove() -> None`으로 한정한다.
+- `remove`는 아직 실행되지 않은 예약을 취소한다. 실행 중인 콜백을 강제로 중단하는 기능은 포함하지 않는다.
+- 실행 완료·중복 취소 여부를 이유로 자체 오류나 별도 상태 관리 계층을 추가하지 않는다. 취소 동작과 오류는 기반 타이머 API를 따른다.
+- `TimerCallback`은 인자 없이 실행되며 반환값이 없는 `__call__() -> None` 계약이다.
+- 반복 실행은 콜백에서 다음 타이머를 등록하는 방식으로 지원한다. 별도 반복 타이머 API, 재설정, 상태 조회 기능은 요구하지 않는다.
+
+등록과 반환 타입의 계약은 다음과 같이 확정한다. 아래 코드는 구현할 계약의 명세이며, 현재 제품에 추가된 코드는 아니다.
+
+```python
+from abc import abstractmethod
+from typing import Protocol, runtime_checkable
+
+
+@runtime_checkable
+class Timer(Protocol):
+  @abstractmethod
+  def remove(self) -> None: ...
+
+
+class TimerCallback(Protocol):
+  def __call__(self) -> None: ...
+
+
+@runtime_checkable
+class SetTimer(Protocol):
+  @abstractmethod
+  def set_timer(self, cb: TimerCallback, seconds: int) -> Timer: ...
+```
+
+- 등록 메서드는 `set_timer`이며, 인자 순서는 `cb`, `seconds`이다.
+- 두 인자는 기본값 없이 필수로 받으며, 위치 인자 또는 `cb=`, `seconds=` 키워드 인자로 전달할 수 있다.
+- `seconds`의 타입은 `int`, 단위는 초이며 반환 타입은 `Timer`이다.
+- `Timer`, `SetTimer`에는 `@runtime_checkable`을 적용하고, 각각 `remove`, `set_timer`에 `@abstractmethod`를 적용한다. `TimerCallback`에는 이 두 데코레이터를 추가하지 않는다.
+
+### R3. 기반 구현의 동작 이용
+
+- 지연 시간은 등록 시점부터 초 단위로 계산한다. 소비 루프 시작에 맞춰 예약을 연기하거나 재등록하지 않는다.
+- RabbitMQ는 Pika의 타이머 등록·취소 API에 위임한다. 이벤트 처리가 진행되면 메시지가 없어도 연결 소유 스레드에서 콜백이 실행된다. 긴 처리 등에 따른 실행 지연은 기반 구현의 동작을 따른다.
+- 등록·취소 가능한 시점, 스레드 제약, 연결 종료 후 동작은 기반 API의 계약을 따르며 호출자가 이에 맞춰 사용한다. 완전한 스레드 안전성을 독자적으로 보장하는 동기화·전달 계층을 추가하지 않는다.
+- 타이머 목록·종료 상태를 별도로 추적하여 자동 취소하는 기능을 요구하지 않는다. 새 타이머를 위해 Consumer나 연결의 종료 처리를 변경하지 않는다.
+- 이미 실행·취소된 핸들, 종료된 연결의 취소, 음수·0초 지연의 처리는 기반 API에 맡긴다. 자체 입력 검증·기본값·폴백을 추가하지 않으며 발생한 오류를 숨기지 않는다.
+- 콜백 예외는 기반 타이머의 이벤트 처리 경로를 통해 드러나게 한다. 별도로 잡아서 무시하거나 정상 종료로 바꾸지 않는다.
+
+### R4. 기존 코드와의 경계
+
+- Consumer의 `complete`, `fail`, `call`, `close`, `run` 등 인터페이스와 구현은 변경하지 않는다.
+- Worker·Ping·Pool·Executor·Event의 구현 및 호출 흐름을 변경하지 않는다.
+- 타이머 추가를 이유로 ACK·NACK·RPC·중단·복구 처리나 프로세스 시작 방식을 리팩터링하지 않는다.
+- 기존 구독·소비 순서·소유 스레드 전달·연결 소유권을 유지한다.
+
+### R5. 사용 문서
+
+- 관련 한국어·영어 매뉴얼에 지원 범위, 등록 시점 기준의 초 단위 시간, 실행 지연 가능성, 등록·취소, 재등록 예시와 기반 API의 사용 제약을 반영한다.
+- Connection을 이용하는 선택적 기능으로 설명하고 Consumer 종료 시 자동 취소를 보장하지 않음을 명시한다. 아직 구현하지 않은 API를 사용 가능한 기능으로 안내하지 않는다.
+
+## 계약 확정 결과
+
+1. 등록 계약은 위 R2의 `set_timer(self, cb: TimerCallback, seconds: int) -> Timer`로 확정했다.
+2. 지연 시간은 등록 시점부터 계산한다. 등록 가능한 시점은 기반 구현을 따른다.
+3. 타이머의 수명주기와 취소 시점은 호출자와 기반 구현의 책임이다. 별도 연결 종료·오류 정리 기능을 추가하지 않는다.
+4. 스레드 제약은 기반 구현을 따른다. 공통 프로토콜이나 래퍼에서 추가 스레드 안전성을 보장하지 않는다.
+5. 중복 취소·종료된 연결·음수·0초 등의 동작은 기반 API에 맡긴다. 이 차이를 통일하기 위한 자체 상태 관리·검증을 추가하지 않는다.
+
+## 인수 기준과 검증 범위
+
+| 기준 | 검증 내용 |
+| --- | --- |
+| 선택적 기능 | RabbitMQ 연결의 타이머 지원과 기본 Connection·Consumer 및 Redis에 타이머 의무가 없음을 확인 |
+| API 계약 | 확정한 시그니처·데코레이터, 위치·키워드 인자 호출과 반환 핸들의 계약 확인 |
+| 유휴 실행 | 메시지가 없는 실제 소비 대기 중 콜백 실행 및 메시지 처리와 소유 스레드 일치 확인 |
+| 등록·취소 위임 | 등록 시점의 지연 시간이 기반 타이머에 전달되고 remove가 기반 취소 API에 위임되는지 확인 |
+| 반복·예외 | 콜백 재등록에 의한 반복 실행 및 콜백 예외의 소비 루프 전파 확인 |
+| 변경 경계 | Consumer·Worker·Ping·Pool·Executor·Event 및 Redis 구현이 수정되지 않았음을 확인 |
+| 회귀 | 연결 타이머 관련 테스트와 직접 영향받는 기존 검증으로 소비·연결 동작 유지 확인 |
+| 문서 | 확정 API와 한국어·영어 예제, 지원 범위·제약 및 호환성 안내의 일치 확인 |
+
+위 항목은 향후 구현 검증 기준이며, 현재 제품 테스트를 실행하거나 기능 구현을 완료한 것은 아니다.
+
+## 제외 범위
+
+- HTTP·SSE 프로토콜 및 연결 유지용 출력 자체의 구현
+- Redis 타이머 지원 추가와 모든 브로커에 대한 스케줄러 제공
+- 별도 타이머 스레드, 범용 작업 스케줄러, 실시간 실행 보장
+- Consumer 인터페이스 정리, `complete`·`fail`·`call` 제거 및 `close` 추상 메서드 선언
+- Worker·Ping·Pool·Executor·Event 리팩터링, 실행기 교체와 프로세스 시작 방식 변경
+- 소비자별 타이머 소유권 관리와 소비 종료 시 새 예약의 자동 취소
+- 타이머 수명주기의 별도 추적·자동 정리, 추가 스레드 동기화 및 기반 API와 중복되는 입력·상태 검증
diff --git a/issues/feature/62/TASKS.md b/issues/feature/62/TASKS.md
new file mode 100644
index 0000000..6a713ea
--- /dev/null
+++ b/issues/feature/62/TASKS.md
@@ -0,0 +1,25 @@
+# 요구사항 분석 결과
+
+[요구사항 명세](./REQUIREMENTS.md)의 범위를 Connection 기반 `SetTimer`, `Timer`, `TimerCallback`과 RabbitMQ 연결 지원 및 직접 관련된 테스트·문서로 한정했다.
+
+앞서 명세한 Consumer의 메서드 제거·close 추상 메서드 선언은 이번 범위에서 철회했다. Consumer·Worker·Ping·Pool·Executor·Event는 변경하지 않으며, 리팩터링은 후속 작업으로 분리한다. 타이머 수명주기는 호출자와 기반 구현이 관리하며, 별도 추적·자동 정리 계층이나 소비 종료 연동은 추가하지 않는다.
+
+등록 API는 `set_timer(self, cb: TimerCallback, seconds: int) -> Timer`로 확정했으며, `Timer`·`SetTimer`의 런타임 검사 및 추상 메서드 선언도 명세했다.
+
+지연 시간은 등록 시점부터 계산한다. 스레드 제약과 중복 취소·종료된 연결·음수·0초 등의 동작은 기반 API를 따른다. 이를 공통화하기 위한 추가 동기화·입력 검증·상태 관리는 요구하지 않는다.
+
+현재는 요구사항 명세 단계이며, 구현 계획과 제품 구현은 진행하지 않았다.
+
+제품 동작은 변경하지 않아 현재 고객 영향은 없다. 향후 구현 시 타이머 지원에 대한 한국어·영어 매뉴얼 안내가 필요하며, 현 단계에서는 미구현 기능을 매뉴얼에 추가하지 않았다.
+
+## 작업 기록
+
+- [타이머 및 Consumer 계약 요구사항 분석](./tasks/1793772101827.md)
+- [Connection 기반 타이머로 범위 한정](./tasks/1793774701691.md)
+- [타이머 등록 시그니처 확정](./tasks/1793775544541.md)
+- [기반 타이머 위임과 부가 요구 제거](./tasks/1793775771772.md)
+
+## 검증
+
+- 제품 코드 변경이 없는 요구사항 문서 작업으로 제품 테스트는 실행하지 않았다.
+- 최초 문서 검증과 이번 범위 변경의 검증 결과는 각 작업 기록에 구분하여 남긴다. 제품 코드 변경은 없다.
diff --git a/issues/feature/62/tasks/1793772101827.md b/issues/feature/62/tasks/1793772101827.md
new file mode 100644
index 0000000..024166a
--- /dev/null
+++ b/issues/feature/62/tasks/1793772101827.md
@@ -0,0 +1,17 @@
+# 타이머 및 Consumer 계약 요구사항 분석
+
+- 이슈 원문과 대화에서 확정한 내용을 Connection·Consumer·RabbitMQ·Redis 구현, Worker 결과 처리·중단, Inspector Ping의 실제 호출 흐름에 대조했다.
+- `SetTimer`, `Timer`, `TimerCallback`과 연결 객체의 선택적 타이머 기능을 명세했다. 외부 스레드 타이머나 Redis 타이머 구현은 범위에 추가하지 않았다.
+- Consumer는 ABC를 유지하는 것으로 사용자 확인을 받았다. `complete`, `fail`, `call` 제거와 `close`의 `@abstractmethod` 선언을 명세했다.
+- Redis Consumer의 close는 중단·닫힘 상태 설정만, RabbitMQ Consumer의 close는 소비자 측 채널 정리만 수행하며 연결 자체는 닫지 않음을 확인했다.
+- Redis가 상속하는 기본 ACK와 Worker·Ping의 직접 호출 의존성을 확인하여 회귀 검증 범위에 포함했다.
+- 연결 단위 타이머와 소비 종료 정리의 관계, 시간 계산 시작점·스레드 계약·등록 시그니처 등 미확정 사항을 [요구사항](../REQUIREMENTS.md)에 분리했다. 이를 확정한 뒤 구현 계획을 작성해야 한다.
+- 적용 지침: `instructions/`의 범위를 확인했으며, 요구사항 문서는 공통 Markdown·이슈 기록 지침을 적용했다. `instructions/README.instructions.md`는 루트 README 전용이고 `instructions/DOCS.instructions.md`는 향후 제품 매뉴얼 갱신에 적용한다.
+- 제품 코드·테스트·매뉴얼은 변경하지 않았다. 제품 테스트는 실행하지 않았다. 문서 검증 결과는 아래에 기록한다.
+
+## 문서 검증 결과
+
+- PowerShell 검사로 이슈 디렉터리의 Markdown 문서 4개에 대한 최상위 제목, 상대 링크 대상 존재, 로컬 절대 경로 미포함을 확인했다.
+- `gh issue view 62 -R yong5eon/DefStream --json body`로 원문을 다시 조회하여 README의 원문 구역과 줄바꿈·끝 공백을 정규화한 내용이 일치함을 확인했다.
+- 최초 절대 경로 검사식이 HTTPS URL을 드라이브 경로로 오인하여 실패했다. 검사식을 수정한 후 전체 문서 검사를 통과했으며 문서 내용의 문제는 아니었다.
+- 저장소 루트에서 `git diff --check`가 통과했다. `git status --short`에서 변경이 이 이슈의 문서에만 있음을 확인했으며 실제 브랜치는 `feature/62`이다.
diff --git a/issues/feature/62/tasks/1793774701691.md b/issues/feature/62/tasks/1793774701691.md
new file mode 100644
index 0000000..ef1f576
--- /dev/null
+++ b/issues/feature/62/tasks/1793774701691.md
@@ -0,0 +1,13 @@
+# Connection 기반 타이머로 범위 한정
+
+- 사용자 지시에 따라 이번 범위를 Connection을 이용하는 `SetTimer`, `Timer`, `TimerCallback`과 RabbitMQ 연결 지원 및 직접 관련된 테스트·문서로 한정했다.
+- [앞선 분석 기록](./1793772101827.md)에서 확정했던 Consumer의 메서드 제거·close 추상 메서드 선언은 후속 리팩터링으로 분리했다. 이전 기록은 당시 결정으로 보존하며 현재 범위는 [요구사항 명세](../REQUIREMENTS.md)를 따른다.
+- Consumer·Worker·Ping·Pool·Executor·Event 및 Redis 구현을 변경 대상으로 삼지 않는다. 관련 호출 경계 재설계도 이번 선행 조건에서 제거했다.
+- 연결 단위 예약에 Consumer 종료 연동을 추가하지 않음을 명시했다. 원문의 소비 종료 정리와 차이를 구분하고 연결 종료·오류 시 핸들 처리는 구체화할 사항으로 남겼다.
+- 제품 코드는 변경하지 않았으며 제품 테스트도 실행하지 않았다. 구현 전 미확정 사항 확인과 계획 작성이 남아 있다.
+
+## 검증
+
+- PowerShell 검사로 이슈 문서 5개의 최상위 제목·상대 링크·로컬 절대 경로 미포함을 확인했다.
+- `gh issue view 62 -R yong5eon/DefStream --json body`로 조회한 원문과 README의 원문 구역이 줄바꿈·끝 공백 정규화 후 일치했다.
+- 저장소 루트에서 `git diff --check`가 통과했다. `git status --short`에서 변경이 이 이슈 문서에만 있음을 확인했고, 실제 브랜치는 `feature/62`이다.
diff --git a/issues/feature/62/tasks/1793775544541.md b/issues/feature/62/tasks/1793775544541.md
new file mode 100644
index 0000000..59a34e2
--- /dev/null
+++ b/issues/feature/62/tasks/1793775544541.md
@@ -0,0 +1,13 @@
+# 타이머 등록 시그니처 확정
+
+- 사용자 제시 계약을 [요구사항 R2](../REQUIREMENTS.md#r2-타이머-핸들과-콜백)에 반영했다.
+- `set_timer(self, cb: TimerCallback, seconds: int) -> Timer`로 이름·인자 순서·타입·반환 타입을 확정했다. 두 인자는 기본값 없이 위치 또는 키워드로 전달한다.
+- `Timer`, `SetTimer`에 `@runtime_checkable`, 각각 `remove`, `set_timer`에 `@abstractmethod`를 적용하는 계약을 명세했다. `TimerCallback.__call__() -> None`도 그대로 유지했다.
+- 제시된 코드의 혼합 들여쓰기만 문서 지침의 공백으로 통일했다. 제품 코드는 변경하지 않았고 Consumer 등 제외 범위도 유지했다.
+- 시간 계산 시작점·등록 및 취소의 스레드 허용 범위·핸들 정리 등 나머지 확인 사항은 확정하지 않았다. 제품 테스트는 실행하지 않았다.
+
+## 검증
+
+- PowerShell 검사로 이슈 문서 6개의 최상위 제목·코드 펜스·상대 링크 대상 파일·로컬 절대 경로 미포함을 확인했다.
+- 요구사항의 등록 시그니처가 제시된 계약과 같고 이전의 등록 계약 미확정 문구가 제거됐음을 확인했다.
+- 저장소 루트에서 `git diff --check`가 통과했으며 `feature/62`에서 변경이 이슈 문서에만 있음을 확인했다.
diff --git a/issues/feature/62/tasks/1793775771772.md b/issues/feature/62/tasks/1793775771772.md
new file mode 100644
index 0000000..69da1d0
--- /dev/null
+++ b/issues/feature/62/tasks/1793775771772.md
@@ -0,0 +1,13 @@
+# 기반 타이머 위임과 부가 요구 제거
+
+- 사용자 확인에 따라 지연 시간을 등록 시점부터 계산하도록 [요구사항](../REQUIREMENTS.md)에 반영했다.
+- 앞선 기록에서 미확정으로 남긴 수명주기·스레드·상태·입력 처리를 별도 설계 과제로 확대하지 않고 기반 API의 계약에 맡겼다.
+- 별도 타이머 목록 추적, 연결 종료 시 자동 정리, 추가 동기화·스레드 전달, 중복 취소·음수·0초의 자체 검증 요구를 제거했다. 타이머 사용·취소 시점은 호출자가 관리하며 기반 오류는 숨기지 않는다.
+- 이번 RabbitMQ 구현은 Pika 타이머를 이용한다. 다른 연결 구현체의 타이머 구현 방식은 제한하지 않지만, 다른 브로커 지원이나 외부 스레드 타이머를 이번에 추가하지는 않는다.
+- 이전 기록은 당시 분석으로 보존했다. 현재 범위는 정정한 요구사항과 작업 종합을 따른다.
+- 제품 코드는 변경하지 않았으며 제품 테스트도 실행하지 않았다. 다음 단계는 확정된 계약에 대한 구현 계획 작성이다.
+
+## 검증
+
+- PowerShell 검사로 이슈 문서 7개의 제목·코드 펜스·상대 링크 대상 파일·로컬 절대 경로 미포함을 확인했다. 등록 시그니처 유지와 과도한 미확정·자동 정리 요구 문구 제거도 확인했다.
+- 저장소 루트의 `git diff --check`가 통과했으며 `feature/62`의 변경은 이슈 문서에만 한정된다.

From 20ebc479b9b35a047a5e3afc9078cbc9ea1e3a88 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=ED=97=88=EC=9A=A9=EC=84=A0?= <yongseon.heo@gmail.com>
Date: Wed, 4 Nov 2026 16:31:45 +0900
Subject: [PATCH 3/4] =?UTF-8?q?Connection=20=EA=B8=B0=EB=B0=98=20=ED=83=80?=
 =?UTF-8?q?=EC=9D=B4=EB=A8=B8=20=EC=9E=91=EC=97=85=20=EA=B3=84=ED=9A=8D=20?=
 =?UTF-8?q?=EB=AC=B8=EC=84=9C=20=EC=B6=94=EA=B0=80?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- Connection 기반 타이머 요구사항 및 구현 방향 정리
- 작업 문서에 작업 계획 링크 추가
- 요구사항 문서에 작업 계획 링크 추가
- 작업 기록 문서에 계획 작성 내용 추가
---
 issues/feature/62/PLAN.md                | 170 +++++++++++++++++++++++
 issues/feature/62/README.md              |   5 +-
 issues/feature/62/REQUIREMENTS.md        |   2 +-
 issues/feature/62/TASKS.md               |   3 +-
 issues/feature/62/tasks/1793777337621.md |  15 ++
 5 files changed, 191 insertions(+), 4 deletions(-)
 create mode 100644 issues/feature/62/PLAN.md
 create mode 100644 issues/feature/62/tasks/1793777337621.md

diff --git a/issues/feature/62/PLAN.md b/issues/feature/62/PLAN.md
new file mode 100644
index 0000000..7d7368a
--- /dev/null
+++ b/issues/feature/62/PLAN.md
@@ -0,0 +1,170 @@
+# Connection 기반 타이머 작업 계획
+
+## 기준과 현재 상태
+
+- 범위와 공개 계약은 [요구사항](./REQUIREMENTS.md)을 따른다. 작업 브랜치·분기 기준·병합 대상은 [이슈 정보](./README.md)에 모은다.
+- 현재는 계획 작성 단계이며 제품 코드·테스트·매뉴얼은 아직 변경하지 않았다. 아래 체크 항목은 수행할 작업이지 완료 결과가 아니다.
+- [워커 리팩토링 #63](https://github.com/yong5eon/DefStream/issues/63)과 [Recoverable 명시적 상속 #64](https://github.com/yong5eon/DefStream/issues/64)는 별도 작업이며 이번 선행 조건이 아니다.
+
+## 구현 방향
+
+추가할 동작은 다음 두 경로로 한정한다.
+
+```text
+connection.set_timer(cb, seconds)
+  → 현재 Pika 연결의 call_later(seconds, cb)
+  → 취소 핸들을 보관하는 Timer 반환
+
+timer.remove()
+  → 등록 당시 Pika 연결의 remove_timeout(등록 핸들)
+```
+
+- 등록 시 콜백과 초 단위 지연 시간을 그대로 전달한다. 콜백을 다른 함수로 감싸거나 소비 시작까지 등록을 미루지 않는다.
+- 반환 객체는 기반 연결과 등록 핸들만 보관하는 작은 구현체로 둔다. 호출자는 내부 Pika 객체나 타이머 ID를 다루지 않는다.
+- 핸들은 등록 당시의 실제 연결을 참조한다. 취소 시 상위 연결 객체의 현재 연결을 다시 조회하지 않아, 재연결 후 다른 연결에 이전 핸들을 전달하지 않는다.
+- `remove`는 기반 취소 API를 호출한다. 실행·취소 상태 플래그, 타이머 목록, 중복 취소 방지, 별도 입력 검증, 자동 정리 계층은 추가하지 않는다.
+- 기존 `connected()`와 오류 관찰 관례를 재사용한다. 새 폴백이나 오류를 정상 반환으로 바꾸는 처리는 추가하지 않는다.
+- 기존 Consumer의 확인용 타이머, `run`·`stop`·`close`, 연결 종료 처리에는 손대지 않는다.
+
+## 변경 대상
+
+| 대상 | 변경 내용 |
+| --- | --- |
+| [공통 연결 계약](../../../src/Broker/Connection.py) | `Timer`, `TimerCallback`, `SetTimer`와 `__all__` 등록 |
+| [Broker 공개 진입점](../../../src/Broker/__init__.py) | 세 계약의 import와 `__all__` 재노출 |
+| [RabbitMQ 연결](../../../src/Broker/Implements/RabbitMQ/Connection.py) | `SetTimer` 명시적 상속, `set_timer`, 취소 핸들 구현 |
+| [프로토콜 계약 테스트](../../../tests/test_protocol_contracts.py) | 공개 타입·시그니처·데코레이터·선택적 지원 검증 |
+| `tests/rabbitmq/test_rabbitmq_timers.py` 신규 | 등록·취소 위임 단위 테스트와 opt-in 실제 타이머 테스트 |
+| [RabbitMQ 한국어 안내](../../../docs/RabbitMQ.md)·[영어 안내](../../../docs/RabbitMQ.en.md) | 지원 범위와 등록·취소·재등록 사용법 |
+| [매뉴얼 한국어 진입점](../../../docs/README.md)·[영어 진입점](../../../docs/README.en.md) | 기존 RabbitMQ 안내 항목에 연결 타이머 주제 반영 |
+| 이 이슈의 기록 문서 | 진행 상태·검증 결과·최종 변경 로그 기록 |
+
+새 제품 소스 모듈은 만들지 않는 방향으로 진행한다. 구체 타이머는 RabbitMQ 연결 모듈 안에 `Timer` 클래스로 두고 공통 계약을 `BaseTimer`로 import하여 명시적으로 구현한다. 구체 핸들 클래스를 RabbitMQ 패키지의 새 공개 진입점으로 노출할 필요는 없다.
+
+## 수행 순서와 완료 조건
+
+### 1. 공통 계약과 공개 import 추가
+
+선행 조건: 확정 요구사항 확인.
+
+1. 공통 연결 계약 모듈에 다음 선언을 추가한다.
+   - `Timer`: `@runtime_checkable`, `remove(self) -> None`에 `@abstractmethod`
+   - `TimerCallback`: `__call__(self) -> None`, 추가 데코레이터 없음
+   - `SetTimer`: `@runtime_checkable`, `set_timer(self, cb: TimerCallback, seconds: int) -> Timer`에 `@abstractmethod`
+2. `cb`, `seconds`는 기본값 없는 위치 또는 키워드 인자로 유지한다. 위치 전용·키워드 전용 제한을 추가하지 않는다.
+3. 모듈과 Broker 패키지의 `__all__`·import를 함께 갱신한다. 사용자는 `DefStream.Broker` 및 `DefStream.Broker.Connection`에서 같은 계약을 import할 수 있어야 한다.
+4. 기본 `Connection`과 Consumer의 필수 메서드는 변경하지 않는다.
+5. 기존 계약 테스트에 공개 경로, 타입 힌트와 시그니처, 런타임 판별, 명시적 상속 시 필수 메서드 미구현 여부를 검증하는 항목을 추가한다.
+
+완료 조건: 요구사항 R1·R2의 선언과 공개 import가 일치하며 기존 계약은 유지된다.
+
+### 2. RabbitMQ 등록·취소 구현
+
+선행 조건: 1단계의 공통 계약 추가.
+
+1. 실제 개발 환경의 Pika `call_later`·`remove_timeout` 선언과 반환 핸들 타입을 확인하고 타입 힌트에 반영한다. 반환 핸들을 임의로 정수 ID로 가정하거나 근거 없는 캐스팅으로 숨기지 않는다.
+2. RabbitMQ `Connection`의 기존 상속 목록에 `SetTimer`를 추가한다.
+3. `set_timer`는 기존 `connected()`로 연결을 얻고 즉시 `call_later(seconds, cb)`를 호출한다. 설정값 변환·보정·추가 검증은 하지 않는다.
+4. 등록 당시 Pika 연결과 반환 핸들을 보관하는 구체 `Timer`를 반환한다.
+5. 구체 `Timer.remove()`는 보관한 연결의 `remove_timeout`에 보관한 핸들을 전달한다. 중복 호출과 종료된 연결의 결과도 기반 API에 맡긴다.
+6. 등록·취소의 기반 오류는 숨기지 않는다. 콜백은 그대로 전달하여 예외가 기존 Pika 이벤트 처리 경로에서 드러나게 한다.
+
+완료 조건: 요구사항 R1~R3을 만족하며 추가 스레드·이벤트 큐·상태 관리 계층이 없다.
+
+### 3. 직접 영향 범위 검증
+
+선행 조건: 1·2단계 구현.
+
+기존 `unittest`와 `unittest.mock` 관례를 사용한다. 새 검증 라이브러리는 추가하지 않는다.
+
+#### 단위 테스트
+
+| 항목 | 검증 방법과 기대 결과 | 요구사항 |
+| --- | --- | --- |
+| 공개 계약 | 두 공개 import 경로가 같은 타입을 가리키고 `__all__`에 포함되는지 확인 | R1·R2 |
+| 선택적 기능 | RabbitMQ 연결은 `SetTimer`로 판별되고 기본 Connection·Consumer·Redis에는 타이머 의무가 추가되지 않았는지 확인 | R1·R4 |
+| 정확한 등록 | 위치·키워드 호출 모두 `call_later(seconds, cb)`에 동일한 값과 같은 콜백 객체를 한 번 전달하는지 확인 | R2·R3 |
+| 콜백 계약 | 일반 함수와 `__call__` 객체를 `TimerCallback`으로 사용할 수 있고 등록 자체가 콜백을 호출하지 않는지 확인 | R2 |
+| 반환·취소 | 반환 핸들이 `Timer` 계약을 충족하고 `remove_timeout`에 기반 핸들을 그대로 전달하는지 확인 | R2·R3 |
+| 연결 대응 | 상위 연결 객체의 연결 참조가 바뀌어도 기존 핸들은 등록 당시 연결에 취소를 요청하는지 확인 | R3 |
+| 오류 전파 | 기반 등록·취소 API가 발생시킨 예외를 그대로 관찰할 수 있는지 확인 | R3 |
+| 얇은 위임 | 중복 취소를 자체 차단하지 않고 0·음수도 기반 API에 전달하는지 확인. 기반 동작을 새 정책으로 재정의하지 않음 | R3 |
+
+#### 실제 RabbitMQ 테스트
+
+- 신규 타이머 테스트 모듈의 `RabbitMQTimerIntegrationTests`에 기존 `RABBITMQ_TESTS=1` opt-in 방식을 적용한다. 포트는 기존 `rabbitmq_test_port()`를 사용한다.
+- 연결·소비·타이머 등록은 같은 소유 스레드에서 수행한다. 다른 스레드의 안전성을 검증한다는 명목으로 제품에 동기화 기능을 추가하지 않는다.
+- 테스트 전용 이름의 빈 큐를 구독하여 메시지 없는 소비 대기 중 콜백 실행과 소유 스레드 일치를 확인한다.
+- 개별 취소 후 미실행, 콜백 재등록에 의한 반복 실행, 콜백 예외가 `consumer.run()` 호출자에게 전달되는 동작을 확인한다.
+- 테스트에 유한 실행 제한을 두고, 테스트가 만든 큐·채널·연결은 테스트의 정리 코드에서 정리한다. 이를 제품의 자동 정리 요구로 옮기지 않는다.
+- 실제 서버를 사용할 수 없으면 단위 테스트로 유휴 실행까지 검증했다고 쓰지 않고 통합 검증 미실행으로 기록한다.
+
+완료 조건: 아래 명령과 진단 결과를 실제 수행 기록에 남긴다. 실패가 이번 변경에서 발생했으면 수정하고 관련 없는 기존 실패는 근거와 함께 구분한다.
+
+### 4. 한국어·영어 사용 안내
+
+선행 조건: 3단계에서 공개 동작 확인.
+
+1. 두 RabbitMQ 매뉴얼에 같은 범위의 연결 타이머 절을 추가한다.
+2. `isinstance(connection, SetTimer)`로 지원 여부를 확인하고 연결 후 `set_timer`로 등록하는 방법을 안내한다.
+3. 빈 큐 소비 중 실행되는 일회성 콜백과 콜백 안의 재등록, `Timer.remove()`를 설명한다. 기존 Consumer API를 그대로 사용하며 제품에 새로운 실행 API를 만들지 않는다.
+4. 초 단위·등록 시점 기준·Pika 이벤트 처리 필요·소유 스레드 제약·실행 지연 가능성·호출자의 취소 책임을 명시한다.
+5. 예제는 필요한 import·연결·구독·실행·테스트 자원 정리까지 확인 가능하게 작성한다. 종료된 연결의 핸들을 무조건 안전하게 취소할 수 있다고 보장하지 않는다.
+6. 두 매뉴얼 진입점의 기존 RabbitMQ 항목에 타이머 주제를 반영한다. 별도 타이머 매뉴얼이나 다이어그램은 단순 등록·취소 설명에 필요하지 않으면 만들지 않는다.
+7. 루트 README 요약에 타이머 안내가 필요한지만 확인한다. 필요할 때에만 한국어·영어 요약을 함께 갱신하며, 상세 사용법은 매뉴얼에 둔다.
+
+완료 조건: 요구사항 R5와 실제 구현이 일치하며 미지원 브로커나 추가 스레드 안전성을 보장하는 표현이 없다.
+
+### 5. 결과 정리와 범위 확인
+
+선행 조건: 3·4단계 완료.
+
+1. 변경 파일 목록을 확인하여 Consumer·Worker·Ping·Pool·Executor·Event·Redis 구현이 섞이지 않았는지 확인한다. #63·#64의 작업은 하지 않는다.
+2. 구현 단계는 `implements/`, 검증 등 구현 외 작업은 `tasks/`에 실제 수행 시점의 기록을 남긴다.
+3. 구현 결과를 `IMPLEMENTS.md`, 작업 결과를 `TASKS.md`에 종합한다. 이 단계에서 아직 없는 구현 문서를 빈 파일로 미리 만들지 않는다.
+4. 실제 제공된 타이머 등록·취소 기능과 사용 제약만 이슈의 `CHANGELOG.md`에 작성한다. 내부 계획·문서 작성·테스트 통과를 고객 기능으로 나열하지 않는다.
+5. README와 이 계획의 진행 상태를 갱신한다. 필요한 검증이 남았다면 전체 구현 완료로 표시하지 않는다.
+
+## 검증 명령과 환경
+
+명령은 저장소 루트에서 개발 환경에 선택된 `python`으로 실행한다. 아래 신규 테스트 모듈·클래스는 구현 단계에서 추가할 대상이며 현재 존재하는 것으로 간주하지 않는다.
+
+### 먼저 실행할 단위·회귀 검증
+
+실제 RabbitMQ opt-in 환경변수가 꺼진 상태에서 다음 모듈을 한 번의 실행으로 검사한다.
+
+```shell
+python -m unittest tests.test_protocol_contracts tests.rabbitmq.test_rabbitmq_timers tests.rabbitmq.test_rabbitmq_broker_methods tests.rabbitmq.test_rabbitmq_logging tests.rabbitmq.test_rabbitmq_timeouts -v
+```
+
+- 등록·취소 단위 테스트와 공통 계약, 기존 연결 API·로깅·시간 설정 검증을 포함한다.
+- 변경 Python 파일의 구문·타입 진단을 확인한다. 새로운 타입 검사 도구를 설치하거나 저장소 전체의 기존 진단을 정리하지 않는다.
+- 이 범위의 결과에서 더 넓은 영향이 확인될 때만 관련 검증으로 확대한다. 전체 Worker·프로세스 테스트를 이번 선행 조건으로 삼지 않는다.
+- 설치된 코드가 현재 소스를 가리키는지 먼저 확인한다. 의존성·선택 구현체가 없어서 검증이 실패할 때에만 [기존 개발용 설치 절차](../../../docs/Packaging.md)를 따른다. 이 경우 공통 계약 테스트가 RabbitMQ·Redis를 모두 import한다는 점을 고려하고 기존 선택을 임의로 축소하지 않는다.
+
+### 실제 RabbitMQ 검증
+
+테스트 사용이 허용된 서버에만 접속한다. 기존 `RABBITMQ_PORT` 설정을 이용하며, 테스트를 위해 운영 서버에 큐를 생성하거나 별도 서버를 자동 설치하지 않는다.
+
+다음은 PowerShell에서 opt-in하는 실행 예시이다. 환경변수는 해당 검증 프로세스에서만 설정한다.
+
+```powershell
+$env:RABBITMQ_TESTS = '1'
+python -m unittest tests.rabbitmq.test_rabbitmq_timers.RabbitMQTimerIntegrationTests -v
+```
+
+### 문서·변경 범위 검증
+
+- 한국어·영어 예제와 공개 import·시그니처를 대조하고 상대 링크·코드 펜스를 검사한다.
+- 이슈 문서에는 로컬 절대 경로를 남기지 않는다.
+- `git diff --check`와 변경 파일 목록을 확인한다. 현재 단계에서는 계획 문서만 검증하고 제품 테스트 결과와 구분한다.
+
+## 진행 상태
+
+- [x] 확정 요구사항과 현재 구현·공개 경로·기존 테스트·문서 위치 확인
+- [x] 범위에 맞는 구현·검증 계획 작성
+- [ ] 공통 계약·공개 import와 계약 테스트 추가
+- [ ] RabbitMQ 등록·취소와 단위 테스트 구현
+- [ ] 단위·회귀·실제 RabbitMQ 검증
+- [ ] 한국어·영어 매뉴얼 갱신
+- [ ] 종합 결과·고객 변경 로그 작성 및 최종 범위 확인
diff --git a/issues/feature/62/README.md b/issues/feature/62/README.md
index 8ce5ec4..7868c9a 100644
--- a/issues/feature/62/README.md
+++ b/issues/feature/62/README.md
@@ -10,14 +10,15 @@
 - 작업 대상: `yong5eon/DefStream`, 작업 브랜치 `feature/62`
 - 분기 기준: `origin/master`와 동일한 로컬 `master`의 `f1bf897`
 - 병합 대상: `master`
-- 로컬 진행 상태: Connection 기반 타이머 계약 명세, 부가 동작은 기반 구현에 위임, 구현 전
+- 로컬 진행 상태: Connection 기반 타이머 요구사항·구체 작업 계획 작성, 구현 전
 - 댓글: 없음
 
 ## 작업 문서
 
 - [요구사항 명세](./REQUIREMENTS.md)
+- [작업 계획](./PLAN.md)
 - [작업 종합](./TASKS.md)
-- [최신 작업 기록](./tasks/1793775771772.md)
+- [최신 작업 기록](./tasks/1793777337621.md)
 
 ## 이슈 원문
 
diff --git a/issues/feature/62/REQUIREMENTS.md b/issues/feature/62/REQUIREMENTS.md
index b8fc1fa..a971aae 100644
--- a/issues/feature/62/REQUIREMENTS.md
+++ b/issues/feature/62/REQUIREMENTS.md
@@ -6,7 +6,7 @@
 
 Consumer의 인터페이스·구현과 Worker·Ping·Executor 등 다른 구성 요소는 변경하지 않는다. 앞서 논의한 리팩터링은 후속 작업으로 분리한다.
 
-이번 문서는 요구사항 분석과 명세이다. 제품 코드 변경과 구현 계획 작성은 아직 수행하지 않았다.
+이번 문서는 요구사항 분석과 명세이다. 구현 순서와 검증 방법은 [작업 계획](./PLAN.md)에 정리하며, 제품 코드는 아직 변경하지 않았다.
 
 ## 분석 근거와 현재 동작
 
diff --git a/issues/feature/62/TASKS.md b/issues/feature/62/TASKS.md
index 6a713ea..1dc78d9 100644
--- a/issues/feature/62/TASKS.md
+++ b/issues/feature/62/TASKS.md
@@ -8,7 +8,7 @@
 
 지연 시간은 등록 시점부터 계산한다. 스레드 제약과 중복 취소·종료된 연결·음수·0초 등의 동작은 기반 API를 따른다. 이를 공통화하기 위한 추가 동기화·입력 검증·상태 관리는 요구하지 않는다.
 
-현재는 요구사항 명세 단계이며, 구현 계획과 제품 구현은 진행하지 않았다.
+[작업 계획](./PLAN.md)에 변경 파일·구현 순서·단위 및 실제 RabbitMQ 검증·한국어와 영어 문서 갱신·완료 조건을 구체화했다. 현재는 계획 작성까지 수행했으며 제품 구현은 진행하지 않았다.
 
 제품 동작은 변경하지 않아 현재 고객 영향은 없다. 향후 구현 시 타이머 지원에 대한 한국어·영어 매뉴얼 안내가 필요하며, 현 단계에서는 미구현 기능을 매뉴얼에 추가하지 않았다.
 
@@ -18,6 +18,7 @@
 - [Connection 기반 타이머로 범위 한정](./tasks/1793774701691.md)
 - [타이머 등록 시그니처 확정](./tasks/1793775544541.md)
 - [기반 타이머 위임과 부가 요구 제거](./tasks/1793775771772.md)
+- [Connection 타이머 구현 계획 작성](./tasks/1793777337621.md)
 
 ## 검증
 
diff --git a/issues/feature/62/tasks/1793777337621.md b/issues/feature/62/tasks/1793777337621.md
new file mode 100644
index 0000000..0d478d9
--- /dev/null
+++ b/issues/feature/62/tasks/1793777337621.md
@@ -0,0 +1,15 @@
+# Connection 타이머 구현 계획 작성
+
+- 확정 요구사항과 현재 연결 계약·공개 import·RabbitMQ의 `connected`·종료 처리·기존 테스트 및 매뉴얼을 대조하여 [작업 계획](../PLAN.md)을 작성했다.
+- 공통 계약과 공개 import 추가, 기존 RabbitMQ 연결 모듈 안의 등록·취소 핸들 구현, 직접 관련 검증, 한국어·영어 사용 안내, 결과 기록 순서와 각 완료 조건을 명시했다.
+- 등록 당시 Pika 연결과 핸들만 보관하여 직접 취소 API에 전달하는 방향을 계획했다. 별도 상태 관리·타이머 목록·자동 정리·동기화·콜백 래퍼는 추가하지 않는다.
+- 기존 unittest·Mock, `RABBITMQ_TESTS` opt-in, `rabbitmq_test_port`를 재사용하는 검증을 계획했다. 신규 테스트 모듈·클래스는 아직 없는 예정 대상으로 명시했으며, 실제 RabbitMQ 검증을 단위 테스트로 대체 완료하지 않도록 구분했다.
+- 적용 지침은 `instructions/CODE.instructions.md`, `instructions/DOCS.instructions.md`, `instructions/README.instructions.md`의 해당 범위와 공통 Python·Markdown 지침이다. 향후 Python 코드 내 설명·진단은 프로젝트의 영어 정책을 적용하고, 계획·진행 설명은 한국어로 작성한다.
+- Consumer 등 제외 범위와 별도 이슈 #63·#64를 변경하지 않는다. 계획 작성 전 실제 브랜치는 `feature/62`였고 작업 트리는 깨끗했다.
+- 이번에는 계획과 이슈 기록만 작성했으며 제품 코드·테스트·매뉴얼은 변경하지 않았다. 제품 테스트도 실행하지 않았다. 다음 단계는 계획의 공통 계약·공개 import 구현이다.
+
+## 검증
+
+- PowerShell 검사로 이슈 문서 9개의 제목·코드 펜스·상대 링크 대상 파일·로컬 절대 경로 미포함을 확인했다.
+- 계획에 공개 계약·등록·취소 위임·기존 테스트 실행 명령·실제 RabbitMQ opt-in·진행 상태가 포함됐음을 확인했다.
+- 저장소 루트의 `git diff --check`가 통과했으며 `git status --short`에서 변경이 이 이슈의 문서 5개에만 있음을 확인했다.

From 8b5ee0a7055534819d59a48f6caed5c3ceddbbd7 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=ED=97=88=EC=9A=A9=EC=84=A0?= <yongseon.heo@gmail.com>
Date: Wed, 4 Nov 2026 16:53:47 +0900
Subject: [PATCH 4/4] =?UTF-8?q?=EC=97=B0=EA=B2=B0=20=ED=83=80=EC=9D=B4?=
 =?UTF-8?q?=EB=A8=B8=20=EA=B8=B0=EB=8A=A5=20=EA=B5=AC=ED=98=84=20=EB=B0=8F?=
 =?UTF-8?q?=20=EC=83=98=ED=94=8C=20=EC=B6=94=EA=B0=80?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- RabbitMQ 연결 타이머 기능을 구현하고 관련 계약을 정의
- 타이머 샘플을 추가하여 예약 및 취소 기능을 시연
- 기존 Stream 샘플을 연결 타이머로 전환
- 문서 및 매뉴얼에 변경 사항 반영
- 관련 테스트를 추가하고 검증 결과 기록

- 관련 이슈: feature/62
---
 docs/README.en.md                             |   2 +-
 docs/README.md                                |   2 +-
 docs/RabbitMQ.en.md                           | 125 ++++++++
 docs/RabbitMQ.md                              | 116 +++++++
 issues/feature/62/CHANGELOG.md                |  26 ++
 issues/feature/62/IMPLEMENTS.md               |  26 ++
 issues/feature/62/PLAN.md                     |  23 +-
 issues/feature/62/README.md                   |   7 +-
 issues/feature/62/REQUIREMENTS.md             |   7 +-
 issues/feature/62/TASKS.md                    |  17 +-
 issues/feature/62/implements/1793777682327.md |   8 +
 issues/feature/62/implements/1793778339166.md |   7 +
 issues/feature/62/tasks/1793777847993.md      |  11 +
 issues/feature/62/tasks/1793778003070.md      |  21 ++
 issues/feature/62/tasks/1793778440735.md      |   9 +
 samples/Broker/RabbitMQ/Sample.Stream.py      |  18 +-
 samples/Broker/RabbitMQ/Sample.Timer.py       |  90 ++++++
 src/Broker/Connection.py                      |  35 +++
 src/Broker/Implements/RabbitMQ/Connection.py  |  37 +++
 src/Broker/Implements/RabbitMQ/__init__.py    |   2 +
 src/Broker/__init__.py                        |   6 +
 tests/rabbitmq/test_rabbitmq_timers.py        | 288 ++++++++++++++++++
 tests/test_protocol_contracts.py              | 101 +++++-
 23 files changed, 951 insertions(+), 33 deletions(-)
 create mode 100644 issues/feature/62/CHANGELOG.md
 create mode 100644 issues/feature/62/IMPLEMENTS.md
 create mode 100644 issues/feature/62/implements/1793777682327.md
 create mode 100644 issues/feature/62/implements/1793778339166.md
 create mode 100644 issues/feature/62/tasks/1793777847993.md
 create mode 100644 issues/feature/62/tasks/1793778003070.md
 create mode 100644 issues/feature/62/tasks/1793778440735.md
 create mode 100644 samples/Broker/RabbitMQ/Sample.Timer.py
 create mode 100644 tests/rabbitmq/test_rabbitmq_timers.py

diff --git a/docs/README.en.md b/docs/README.en.md
index c3656bb..e51b173 100644
--- a/docs/README.en.md
+++ b/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
diff --git a/docs/README.md b/docs/README.md
index d4072a0..b9202fb 100644
--- a/docs/README.md
+++ b/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
diff --git a/docs/RabbitMQ.en.md b/docs/RabbitMQ.en.md
index 036bfcf..4d23701 100644
--- a/docs/RabbitMQ.en.md
+++ b/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.
diff --git a/docs/RabbitMQ.md b/docs/RabbitMQ.md
index 51647e4..41fff51 100644
--- a/docs/RabbitMQ.md
+++ b/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`를 사용합니다.
diff --git a/issues/feature/62/CHANGELOG.md b/issues/feature/62/CHANGELOG.md
new file mode 100644
index 0000000..2a8fea7
--- /dev/null
+++ b/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#연결-타이머)를 참고하세요.
diff --git a/issues/feature/62/IMPLEMENTS.md b/issues/feature/62/IMPLEMENTS.md
new file mode 100644
index 0000000..86ca98a
--- /dev/null
+++ b/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)
diff --git a/issues/feature/62/PLAN.md b/issues/feature/62/PLAN.md
index 7d7368a..18641a6 100644
--- a/issues/feature/62/PLAN.md
+++ b/issues/feature/62/PLAN.md
@@ -3,7 +3,7 @@
 ## 기준과 현재 상태
 
 - 범위와 공개 계약은 [요구사항](./REQUIREMENTS.md)을 따른다. 작업 브랜치·분기 기준·병합 대상은 [이슈 정보](./README.md)에 모은다.
-- 현재는 계획 작성 단계이며 제품 코드·테스트·매뉴얼은 아직 변경하지 않았다. 아래 체크 항목은 수행할 작업이지 완료 결과가 아니다.
+- 계약·RabbitMQ 구현·테스트·한국어와 영어 매뉴얼을 반영했다. 단위·회귀 검증은 통과했으며 실제 RabbitMQ 검증은 서버 연결 불가로 남아 있다. 세부 결과는 [구현 종합](./IMPLEMENTS.md)과 [작업 종합](./TASKS.md)을 따른다.
 - [워커 리팩토링 #63](https://github.com/yong5eon/DefStream/issues/63)과 [Recoverable 명시적 상속 #64](https://github.com/yong5eon/DefStream/issues/64)는 별도 작업이며 이번 선행 조건이 아니다.
 
 ## 구현 방향
@@ -34,7 +34,8 @@ timer.remove()
 | [Broker 공개 진입점](../../../src/Broker/__init__.py) | 세 계약의 import와 `__all__` 재노출 |
 | [RabbitMQ 연결](../../../src/Broker/Implements/RabbitMQ/Connection.py) | `SetTimer` 명시적 상속, `set_timer`, 취소 핸들 구현 |
 | [프로토콜 계약 테스트](../../../tests/test_protocol_contracts.py) | 공개 타입·시그니처·데코레이터·선택적 지원 검증 |
-| `tests/rabbitmq/test_rabbitmq_timers.py` 신규 | 등록·취소 위임 단위 테스트와 opt-in 실제 타이머 테스트 |
+| [타이머 테스트](../../../tests/rabbitmq/test_rabbitmq_timers.py) 신규 | 등록·취소 위임 단위 테스트와 opt-in 실제 타이머 테스트 |
+| [타이머 샘플](../../../samples/Broker/RabbitMQ/Sample.Timer.py) 신규·[Stream 샘플](../../../samples/Broker/RabbitMQ/Sample.Stream.py) | 실행 가능한 취소·재등록 예제와 기존 외부 타이머 교체 |
 | [RabbitMQ 한국어 안내](../../../docs/RabbitMQ.md)·[영어 안내](../../../docs/RabbitMQ.en.md) | 지원 범위와 등록·취소·재등록 사용법 |
 | [매뉴얼 한국어 진입점](../../../docs/README.md)·[영어 진입점](../../../docs/README.en.md) | 기존 RabbitMQ 안내 항목에 연결 타이머 주제 반영 |
 | 이 이슈의 기록 문서 | 진행 상태·검증 결과·최종 변경 로그 기록 |
@@ -112,6 +113,7 @@ timer.remove()
 5. 예제는 필요한 import·연결·구독·실행·테스트 자원 정리까지 확인 가능하게 작성한다. 종료된 연결의 핸들을 무조건 안전하게 취소할 수 있다고 보장하지 않는다.
 6. 두 매뉴얼 진입점의 기존 RabbitMQ 항목에 타이머 주제를 반영한다. 별도 타이머 매뉴얼이나 다이어그램은 단순 등록·취소 설명에 필요하지 않으면 만들지 않는다.
 7. 루트 README 요약에 타이머 안내가 필요한지만 확인한다. 필요할 때에만 한국어·영어 요약을 함께 갱신하며, 상세 사용법은 매뉴얼에 둔다.
+8. 사용자 요청으로 실행 가능한 타이머 샘플을 추가하고 기존 Stream 샘플도 연결 타이머를 이용하도록 갱신한다. 샘플의 등록·취소·재등록·자원 정리를 단위 검증하며, import 시 연결과 로깅 설정이 실행되지 않는지 기존 검증에 포함한다.
 
 완료 조건: 요구사항 R5와 실제 구현이 일치하며 미지원 브로커나 추가 스레드 안전성을 보장하는 표현이 없다.
 
@@ -127,7 +129,7 @@ timer.remove()
 
 ## 검증 명령과 환경
 
-명령은 저장소 루트에서 개발 환경에 선택된 `python`으로 실행한다. 아래 신규 테스트 모듈·클래스는 구현 단계에서 추가할 대상이며 현재 존재하는 것으로 간주하지 않는다.
+명령은 저장소 루트에서 개발 환경에 선택된 `python`으로 실행한다. 아래 타이머 테스트 모듈·클래스는 구현 단계에서 추가했다.
 
 ### 먼저 실행할 단위·회귀 검증
 
@@ -157,14 +159,17 @@ python -m unittest tests.rabbitmq.test_rabbitmq_timers.RabbitMQTimerIntegrationT
 
 - 한국어·영어 예제와 공개 import·시그니처를 대조하고 상대 링크·코드 펜스를 검사한다.
 - 이슈 문서에는 로컬 절대 경로를 남기지 않는다.
-- `git diff --check`와 변경 파일 목록을 확인한다. 현재 단계에서는 계획 문서만 검증하고 제품 테스트 결과와 구분한다.
+- `git diff --check`와 변경 파일 목록을 확인한다. 문서 검증과 단위·회귀·실제 브로커 검증 결과는 구분하여 기록한다.
 
 ## 진행 상태
 
 - [x] 확정 요구사항과 현재 구현·공개 경로·기존 테스트·문서 위치 확인
 - [x] 범위에 맞는 구현·검증 계획 작성
-- [ ] 공통 계약·공개 import와 계약 테스트 추가
-- [ ] RabbitMQ 등록·취소와 단위 테스트 구현
-- [ ] 단위·회귀·실제 RabbitMQ 검증
-- [ ] 한국어·영어 매뉴얼 갱신
-- [ ] 종합 결과·고객 변경 로그 작성 및 최종 범위 확인
+- [x] 공통 계약·공개 import와 계약 테스트 추가
+- [x] RabbitMQ 등록·취소와 단위 테스트 구현
+- [x] 단위·회귀 검증
+- [ ] 실제 RabbitMQ 검증: 로컬 서버 연결 단계에서 `AMQPConnectionError`
+- [x] 한국어·영어 매뉴얼 갱신
+- [x] 실행 가능한 타이머 샘플 추가와 Stream 샘플 전환·단위 검증
+- [x] 종합 결과·고객 변경 로그 작성
+- [ ] 최종 인수 확인: 실제 RabbitMQ 검증 완료 후 판단
diff --git a/issues/feature/62/README.md b/issues/feature/62/README.md
index 7868c9a..34c8314 100644
--- a/issues/feature/62/README.md
+++ b/issues/feature/62/README.md
@@ -10,15 +10,18 @@
 - 작업 대상: `yong5eon/DefStream`, 작업 브랜치 `feature/62`
 - 분기 기준: `origin/master`와 동일한 로컬 `master`의 `f1bf897`
 - 병합 대상: `master`
-- 로컬 진행 상태: Connection 기반 타이머 요구사항·구체 작업 계획 작성, 구현 전
+- 로컬 진행 상태: 타이머 구현·실행 샘플·사용 문서 반영, 단위·회귀 45개 통과, 실제 RabbitMQ 검증은 서버 연결 불가
 - 댓글: 없음
 
 ## 작업 문서
 
 - [요구사항 명세](./REQUIREMENTS.md)
 - [작업 계획](./PLAN.md)
+- [구현 종합](./IMPLEMENTS.md)
 - [작업 종합](./TASKS.md)
-- [최신 작업 기록](./tasks/1793777337621.md)
+- [변경 로그](./CHANGELOG.md)
+- [최신 구현 기록](./implements/1793778339166.md)
+- [최신 작업 기록](./tasks/1793778440735.md)
 
 ## 이슈 원문
 
diff --git a/issues/feature/62/REQUIREMENTS.md b/issues/feature/62/REQUIREMENTS.md
index a971aae..8eb3cb8 100644
--- a/issues/feature/62/REQUIREMENTS.md
+++ b/issues/feature/62/REQUIREMENTS.md
@@ -6,7 +6,7 @@
 
 Consumer의 인터페이스·구현과 Worker·Ping·Executor 등 다른 구성 요소는 변경하지 않는다. 앞서 논의한 리팩터링은 후속 작업으로 분리한다.
 
-이번 문서는 요구사항 분석과 명세이다. 구현 순서와 검증 방법은 [작업 계획](./PLAN.md)에 정리하며, 제품 코드는 아직 변경하지 않았다.
+이번 문서는 확정 요구사항을 담는다. 구현 순서와 검증 방법은 [작업 계획](./PLAN.md), 실제 결과와 남은 검증은 [구현 종합](./IMPLEMENTS.md)에 구분하여 기록한다.
 
 ## 분석 근거와 현재 동작
 
@@ -38,7 +38,7 @@ Consumer의 인터페이스·구현과 Worker·Ping·Executor 등 다른 구성
 - `TimerCallback`은 인자 없이 실행되며 반환값이 없는 `__call__() -> None` 계약이다.
 - 반복 실행은 콜백에서 다음 타이머를 등록하는 방식으로 지원한다. 별도 반복 타이머 API, 재설정, 상태 조회 기능은 요구하지 않는다.
 
-등록과 반환 타입의 계약은 다음과 같이 확정한다. 아래 코드는 구현할 계약의 명세이며, 현재 제품에 추가된 코드는 아니다.
+등록과 반환 타입의 계약은 다음과 같다. 아래 코드는 공개 계약의 명세이며, 실제 선언은 [연결 계약 모듈](../../../src/Broker/Connection.py)에 있다.
 
 ```python
 from abc import abstractmethod
@@ -86,6 +86,7 @@ class SetTimer(Protocol):
 
 - 관련 한국어·영어 매뉴얼에 지원 범위, 등록 시점 기준의 초 단위 시간, 실행 지연 가능성, 등록·취소, 재등록 예시와 기반 API의 사용 제약을 반영한다.
 - Connection을 이용하는 선택적 기능으로 설명하고 Consumer 종료 시 자동 취소를 보장하지 않음을 명시한다. 아직 구현하지 않은 API를 사용 가능한 기능으로 안내하지 않는다.
+- 실행 가능한 RabbitMQ 타이머 샘플을 제공한다. 기존 Stream 샘플의 소비 중단도 별도 스레드 타이머 대신 연결 타이머를 이용한다.
 
 ## 계약 확정 결과
 
@@ -108,7 +109,7 @@ class SetTimer(Protocol):
 | 회귀 | 연결 타이머 관련 테스트와 직접 영향받는 기존 검증으로 소비·연결 동작 유지 확인 |
 | 문서 | 확정 API와 한국어·영어 예제, 지원 범위·제약 및 호환성 안내의 일치 확인 |
 
-위 항목은 향후 구현 검증 기준이며, 현재 제품 테스트를 실행하거나 기능 구현을 완료한 것은 아니다.
+위 항목은 인수 기준이다. 실제 검증 결과와 실행하지 못한 항목은 [작업 종합](./TASKS.md)에 기록하며, 실제 RabbitMQ 검증은 서버 연결 불가로 남아 있다.
 
 ## 제외 범위
 
diff --git a/issues/feature/62/TASKS.md b/issues/feature/62/TASKS.md
index 1dc78d9..157aa83 100644
--- a/issues/feature/62/TASKS.md
+++ b/issues/feature/62/TASKS.md
@@ -8,9 +8,9 @@
 
 지연 시간은 등록 시점부터 계산한다. 스레드 제약과 중복 취소·종료된 연결·음수·0초 등의 동작은 기반 API를 따른다. 이를 공통화하기 위한 추가 동기화·입력 검증·상태 관리는 요구하지 않는다.
 
-[작업 계획](./PLAN.md)에 변경 파일·구현 순서·단위 및 실제 RabbitMQ 검증·한국어와 영어 문서 갱신·완료 조건을 구체화했다. 현재는 계획 작성까지 수행했으며 제품 구현은 진행하지 않았다.
+[작업 계획](./PLAN.md)에 따라 계약·RabbitMQ 구현과 테스트·한국어 및 영어 매뉴얼을 반영했다. 제품 구현의 상세 결과는 [구현 종합](./IMPLEMENTS.md)에 기록했다.
 
-제품 동작은 변경하지 않아 현재 고객 영향은 없다. 향후 구현 시 타이머 지원에 대한 한국어·영어 매뉴얼 안내가 필요하며, 현 단계에서는 미구현 기능을 매뉴얼에 추가하지 않았다.
+선택된 개발 환경에서 샘플 검증 추가 후 관련 49개 테스트 중 45개가 통과했으며, opt-in 실제 RabbitMQ 테스트 4개는 기본 실행에서 건너뛰었다. 별도로 유휴 실행 통합 테스트를 시도했으나 연결 단계에서 `AMQPConnectionError`로 막혔다. 전체 인수 완료로 판단하지 않는다.
 
 ## 작업 기록
 
@@ -19,8 +19,17 @@
 - [타이머 등록 시그니처 확정](./tasks/1793775544541.md)
 - [기반 타이머 위임과 부가 요구 제거](./tasks/1793775771772.md)
 - [Connection 타이머 구현 계획 작성](./tasks/1793777337621.md)
+- [연결 타이머 단위 검증과 사용 안내 반영](./tasks/1793777847993.md)
+- [최종 검증과 남은 통합 검증](./tasks/1793778003070.md)
+- [타이머·Stream 샘플 검증](./tasks/1793778440735.md)
 
 ## 검증
 
-- 제품 코드 변경이 없는 요구사항 문서 작업으로 제품 테스트는 실행하지 않았다.
-- 최초 문서 검증과 이번 범위 변경의 검증 결과는 각 작업 기록에 구분하여 남긴다. 제품 코드 변경은 없다.
+- 구현 전 기준선: 관련 4개 모듈 35개 통과.
+- 구현 후 단위·회귀: 샘플 검증을 포함한 관련 5개 모듈 45개 통과, 실제 브로커 테스트 4개 skip.
+- 실제 브로커: 유휴 타이머 테스트 1개를 시도했으나 서버 연결 오류로 동작 검증 전에 실패했다. 연결 가능한 테스트 서버에서 통합 클래스 4개 테스트를 다시 실행해야 한다.
+- 정적 검증: 변경 Python 파일 5개에 타입 오류가 없으며, 테스트 콜백의 미사용 인자에 대한 정보성 진단만 있다. 신규 테스트의 구문 검사도 통과했다.
+- 한국어·영어 타이머 예제 두 개가 동일하며 Python 3.9 문법으로 파싱·컴파일되는지 확인했다.
+- 기본 터미널 환경의 기존 editable 설치 문제와 cp949 콘솔 출력 문제는 제품 변경과 구분하여 기록했다. 설치나 파일 인코딩을 변경하지 않았다.
+- 최종 문서 18개의 구조·링크, 이슈 원문 보존, 제외 코드 미변경과 `git diff --check`를 확인했다. 변경 Python 파일 5개는 Python 3.9 문법 파싱도 통과했으며 실제 실행 환경은 Python 3.12.10이다.
+- 샘플 추가 후 Mock 기반 등록·취소·재등록·정리 흐름과 기존 샘플 import·로깅 경계 검증이 통과했다. 두 샘플에 Pylance 진단은 없으며 Python 3.9 문법 파싱도 통과했다. 실제 브로커에서 샘플을 실행한 것은 아니다.
diff --git a/issues/feature/62/implements/1793777682327.md b/issues/feature/62/implements/1793777682327.md
new file mode 100644
index 0000000..8b35e1e
--- /dev/null
+++ b/issues/feature/62/implements/1793777682327.md
@@ -0,0 +1,8 @@
+# 연결 타이머 계약과 RabbitMQ 등록·취소 구현
+
+- 공통 연결 계약에 확정한 `Timer`, `TimerCallback`, `SetTimer`와 공개 import를 추가했다. `Timer`, `SetTimer`는 런타임 검사와 명시적 상속을 지원한다.
+- RabbitMQ 연결은 `SetTimer`를 명시적으로 구현한다. `set_timer`는 기존 `connected()` 확인 후 `call_later(seconds, cb)`에 그대로 위임한다.
+- 구체 `Timer`는 등록 당시 Pika 연결과 불투명 핸들만 보관하며, `remove`는 그 연결의 `remove_timeout`에 직접 위임한다. 취소 시 상위 연결을 다시 조회하지 않는다.
+- Pika의 공개 메서드 선언은 반환·취소 핸들을 `int`로 표기하므로 해당 선언을 타입 힌트에 반영했다. 실제 핸들은 해석하거나 변환하지 않는다.
+- Consumer·Worker·Executor와 종료 처리에는 손대지 않았다. 추가 스레드·콜백 래퍼·상태 관리·자동 정리 계층도 없다.
+- 계약·위임 단위 테스트와 기존 opt-in 방식의 실제 RabbitMQ 테스트를 추가했다. 검증 결과는 후속 작업 기록에 남긴다.
diff --git a/issues/feature/62/implements/1793778339166.md b/issues/feature/62/implements/1793778339166.md
new file mode 100644
index 0000000..63cacbd
--- /dev/null
+++ b/issues/feature/62/implements/1793778339166.md
@@ -0,0 +1,7 @@
+# 실행 가능한 연결 타이머 샘플 추가
+
+- 사용자 요청에 따라 `samples/Broker/RabbitMQ/Sample.Timer.py`를 추가했다. 로컬 RabbitMQ에서 고유한 빈 큐를 구독하고 예약 취소, 1초 간격 세 번의 콜백 재등록과 소비 중단을 보여준다.
+- 샘플은 기반 연결 타이머만 사용하며 현재 예약 핸들을 보관한다. 정상·오류 종료 시 타이머·소비 채널·샘플 큐·연결을 정리한다.
+- 기존 Stream 샘플의 `threading.Timer`와 중계 함수를 제거하고 `con.set_timer(consumer.stop, 1)`로 교체했다. Consumer 구현은 변경하지 않았다.
+- 기존 샘플의 실행 진입점 로깅 패턴을 유지하고 import 시 연결이나 루트 로깅 설정이 실행되지 않게 했다.
+- 한국어·영어 매뉴얼에 샘플 링크·실행 방법·자원 정리 범위를 추가했다. 실제 브로커 실행 검증은 서버 연결 제한과 구분하여 후속 작업 기록에 남긴다.
diff --git a/issues/feature/62/tasks/1793777847993.md b/issues/feature/62/tasks/1793777847993.md
new file mode 100644
index 0000000..be6223d
--- /dev/null
+++ b/issues/feature/62/tasks/1793777847993.md
@@ -0,0 +1,11 @@
+# 연결 타이머 단위 검증과 사용 안내 반영
+
+- VS Code에서 선택된 개발 가상환경의 Python 3.12.10과 Pika 1.4.4를 확인했다. editable import 내용이 현재 저장소 소스와 일치함도 확인했다.
+- 기본 터미널 Python 3.14의 기존 editable 설치는 다른 드라이브 간 링크 생성 실패로 모듈을 불러오지 못했다. 해당 환경이나 설치를 수정하지 않고 선택된 가상환경에서 검증했다.
+- 선택된 환경에서도 기본 콘솔 출력이 cp949여서 기존 Worker 배너 출력이 실패했다. 파일 인코딩을 변경하지 않고 검증 프로세스의 `PYTHONIOENCODING=utf-8`을 지정했다.
+- 구현 전 관련 4개 모듈의 기준선 35개 테스트가 통과했다. 구현 후 계획의 5개 모듈을 한 번에 실행하여 48개 중 44개가 통과하고 opt-in 통합 테스트 4개는 건너뛰었다.
+- 변경 Python 파일 5개의 Pylance 진단에서 타입 오류는 없었다. 테스트용 콜백의 미사용 인자에 대한 정보성 진단만 있었으며, 신규 테스트 파일의 구문 검사도 통과했다.
+- 로컬 RabbitMQ 테스트 포트의 연결 확인이 시간 초과됐다. 실제 브로커의 유휴 실행·재등록·예외 전파까지 통과한 것으로 보고하지 않는다.
+- 한국어·영어 RabbitMQ 매뉴얼에 등록·취소·제한된 반복 실행 예제와 기반 API의 이벤트 처리·소유 스레드 제약을 반영했다. 매뉴얼 진입점의 기존 RabbitMQ 항목도 함께 갱신했다.
+- 상세 사용법은 기존 RabbitMQ 매뉴얼에 집중하며 프로젝트 소개·최소 Worker 예제를 바꾸지 않으므로 루트 README 요약은 변경하지 않는다.
+- 사용한 단위·회귀 명령은 저장소 루트의 `python -m unittest tests.test_protocol_contracts tests.rabbitmq.test_rabbitmq_timers tests.rabbitmq.test_rabbitmq_broker_methods tests.rabbitmq.test_rabbitmq_logging tests.rabbitmq.test_rabbitmq_timeouts -q`이다. 실제 호출 시 선택된 개발 가상환경의 Python을 사용했다.
diff --git a/issues/feature/62/tasks/1793778003070.md b/issues/feature/62/tasks/1793778003070.md
new file mode 100644
index 0000000..7564bde
--- /dev/null
+++ b/issues/feature/62/tasks/1793778003070.md
@@ -0,0 +1,21 @@
+# 최종 검증과 남은 통합 검증
+
+## 수행 결과
+
+- 선택된 개발 환경과 `PYTHONIOENCODING=utf-8`로 계획의 단위·회귀 5개 모듈을 재실행했다. 48개 중 44개 통과, 실제 브로커용 4개 skip이며 실행 명령은 [앞선 기록](./1793777847993.md)에 있다.
+- 실제 유휴 테스트는 `RABBITMQ_TESTS=1`에서 `python -m unittest tests.rabbitmq.test_rabbitmq_timers.RabbitMQTimerIntegrationTests.test_idle_timer_runs_once_on_owner_thread -v`로 별도 시도했다. 테스트 연결의 유한 제한 안에서 `AMQPConnectionError`가 발생하여 setUp 단계에서 실패했다.
+- 기본 실행의 skip을 통합 검증 통과로 취급하지 않는다. 실제 서버를 설치하거나 다른 서버에 임의로 접속하지 않았다.
+- 마지막 Python 변경 후 Pylance 타입 오류는 없었다. 테스트의 미사용 인자 정보성 진단만 남아 있으며 제품 소스에는 진단이 없다.
+- 두 언어의 타이머 예제 두 개가 같은 코드이며 Python 3.9 문법 파싱·컴파일이 가능함을 확인했다. 반복 예제는 기존 일회성 예약과 동시에 등록하는 것이 아니라 해당 부분을 교체하도록 설명했다.
+- [구현 종합](../IMPLEMENTS.md), [작업 종합](../TASKS.md), [변경 로그](../CHANGELOG.md)와 계획·README를 실제 결과에 맞춰 정리했다.
+
+## 최종 문서·범위 검증
+
+- 이슈 문서와 변경 매뉴얼 18개의 최상위 제목·코드 펜스·상대 링크 대상 파일을 확인했고, 이슈 문서에 로컬 절대 경로가 없음을 검사했다.
+- README의 이슈 원문이 작업 전과 동일하고 Consumer·Worker·Ping·Pool·Executor·Event·Redis 구현에 변경이 없음을 확인했다.
+- 변경 Python 파일 5개를 Python 3.9 문법으로 파싱했다. 최종 한국어·영어 예제도 동일하며 파싱·컴파일이 통과했다. 이는 Python 3.9에서 실제 테스트를 실행했다는 뜻은 아니다.
+- 공통·RabbitMQ 연결 소스의 Pylance 구문 검사와 저장소 루트의 `git diff --check`가 통과했다. 실제 브랜치는 `feature/62`이며 커밋·푸시는 하지 않았다.
+
+## 남은 검증
+
+연결 가능한 테스트용 RabbitMQ에서 `RABBITMQ_TESTS=1`과 `python -m unittest tests.rabbitmq.test_rabbitmq_timers.RabbitMQTimerIntegrationTests -v`로 통합 테스트 4개를 실행한다. 실제 브로커 인수 확인 전이므로 전체 완료로 표시하지 않는다.
diff --git a/issues/feature/62/tasks/1793778440735.md b/issues/feature/62/tasks/1793778440735.md
new file mode 100644
index 0000000..6a516b5
--- /dev/null
+++ b/issues/feature/62/tasks/1793778440735.md
@@ -0,0 +1,9 @@
+# 타이머·Stream 샘플 검증
+
+- 사용자 요청으로 실행 가능한 타이머 샘플과 기존 Stream 샘플의 연결 타이머 전환을 추가했다. 제품 Consumer·Worker·Executor는 변경하지 않았다.
+- Mock 기반 샘플 실행에서 최초 예약 취소, 3회 콜백 실행과 재등록, 최신 핸들 취소, 소비 중단·채널·큐·연결 정리를 검증했다.
+- 계획의 관련 5개 모듈은 `python -m unittest tests.test_protocol_contracts tests.rabbitmq.test_rabbitmq_timers tests.rabbitmq.test_rabbitmq_broker_methods tests.rabbitmq.test_rabbitmq_logging tests.rabbitmq.test_rabbitmq_timeouts -q`로 실행해 49개 중 45개 통과, 실제 브로커용 4개 skip이었다.
+- 샘플 핸들러의 이벤트 타입을 기반 EventHandler 계약과 맞지 않는 구체 RabbitMQ Event에서 공통 Event로 수정했다. 두 샘플의 최종 Pylance 진단은 없고, 신규 샘플 구문 검사와 두 샘플·변경 테스트의 Python 3.9 문법 파싱도 통과했다.
+- 타입 수정 후 샘플 동작 테스트와 기존 샘플 import·로깅 경계 테스트 2개를 묶어 다시 실행하여 통과했다. 실제 실행 환경은 앞선 기록과 같은 Python 3.12.10이며 출력만 UTF-8로 지정했다.
+- 한국어·영어 매뉴얼에 샘플 링크·실행 명령·로컬 서버 전제 조건과 자원 정리 범위를 반영했다. 실제 RabbitMQ 서버 연결 제한은 해소되지 않았으므로 실제 샘플 실행까지 통과했다고 판단하지 않는다.
+- 최종 문서 20개의 구조·샘플을 포함한 상대 링크·로컬 절대 경로 미포함을 확인했다. `git diff --check`가 통과했으며 제외한 제품 코드에 변경이 없고 실제 브랜치는 `feature/62`임을 확인했다.
diff --git a/samples/Broker/RabbitMQ/Sample.Stream.py b/samples/Broker/RabbitMQ/Sample.Stream.py
index e9fa136..d8b880e 100644
--- a/samples/Broker/RabbitMQ/Sample.Stream.py
+++ b/samples/Broker/RabbitMQ/Sample.Stream.py
@@ -5,8 +5,8 @@
 	Configuration,
 	Event,
 	EventHandler,
+	Timer,
 )
-from threading import Timer
 from random import randint
 from logging import (
 	DEBUG,
@@ -77,17 +77,13 @@ def __call__(
 			offset=0,
 		)
 	
-		def stop(timer):
-			consumer.stop()
-			return
-		
-		Timer(
-			1,
-			lambda: stop(None),
-		).start()
-	
 		consumer.subs('stream')
-		consumer.run()
+		timer: Timer = con.set_timer(consumer.stop, 1)
+		try:
+			consumer.run()
+		finally:
+			timer.remove()
+			consumer.close()
 
 	logger.info('Starting Stream read loop')
 	stream = con.stream('stream')
diff --git a/samples/Broker/RabbitMQ/Sample.Timer.py b/samples/Broker/RabbitMQ/Sample.Timer.py
new file mode 100644
index 0000000..20bb0e1
--- /dev/null
+++ b/samples/Broker/RabbitMQ/Sample.Timer.py
@@ -0,0 +1,90 @@
+from logging import (
+	DEBUG,
+	Filter,
+	Formatter,
+	Logger,
+	StreamHandler,
+	getLogger,
+)
+from uuid import uuid4
+
+from DefStream.Broker import (
+	Event,
+	Timer,
+)
+from DefStream.Broker.Implements.RabbitMQ import (
+	Configuration,
+	Connection as RabbitConnection,
+	Consumer,
+)
+
+logger: Logger = getLogger('Sample')
+
+
+def main() -> None:
+	con: RabbitConnection = RabbitConnection(Configuration(
+		host='127.0.0.1',
+		port=5672,
+		username='guest',
+		password='guest',
+	))
+	try:
+		con.connect()
+		queue: str = 'defstream-timer-sample-' + uuid4().hex
+		con.create_queue(queue)
+		try:
+			def receive(event: Event) -> None:
+				logger.info('Message received')
+				event.ack()
+				return
+
+			consumer: Consumer = con.consumer(handler=receive)
+			try:
+				consumer.subs(queue)
+
+				def cancelled_callback() -> None:
+					raise RuntimeError('Cancelled timer was executed')
+
+				cancelled: Timer = con.set_timer(cancelled_callback, 1)
+				cancelled.remove()
+				logger.info('Cancelled a one-shot timer before consumption')
+				remaining: int = 3
+				timer: Timer
+
+				def tick() -> None:
+					nonlocal remaining, timer
+					remaining -= 1
+					logger.info('Timer fired: remaining=%s', remaining)
+					if remaining:
+						timer = con.set_timer(tick, 1)
+					else:
+						consumer.stop()
+					return
+
+				timer = con.set_timer(tick, 1)
+				try:
+					consumer.run()
+				finally:
+					timer.remove()
+			finally:
+				consumer.close()
+		finally:
+			con.delete_queue(queue)
+	finally:
+		con.close()
+	logger.info('RabbitMQ timer sample finished')
+	return
+
+
+if __name__ == '__main__':
+	root: Logger = getLogger()
+	root.setLevel(DEBUG)
+	console: StreamHandler = StreamHandler()
+	console.setLevel(DEBUG)
+	sample_filter: Filter = Filter('Sample')
+	rabbitmq_filter: Filter = Filter('Broker.RabbitMQ')
+	console.addFilter(lambda record: sample_filter.filter(record) or rabbitmq_filter.filter(record))
+	console.setFormatter(Formatter('%(asctime)s %(levelname)s %(name)s %(message)s'))
+	root.addHandler(console)
+	logger.info('Starting RabbitMQ timer sample')
+	main()
diff --git a/src/Broker/Connection.py b/src/Broker/Connection.py
index ba7bbf9..8121fe2 100644
--- a/src/Broker/Connection.py
+++ b/src/Broker/Connection.py
@@ -15,6 +15,9 @@
 __all__ = (
 	'Connection',
 	'BrokerFactory',
+	'Timer',
+	'TimerCallback',
+	'SetTimer',
 	'GetExchange',
 	'GetQueue',
 	'GetStream',
@@ -41,6 +44,38 @@ class BrokerFactory(Protocol):
 	def __call__(self) -> Connection: ...
 
 
+@runtime_checkable
+class Timer(Protocol):
+	"""
+	Cancelable timer handle.
+	"""
+
+	@abstractmethod
+	def remove(self) -> None: ...
+
+
+class TimerCallback(Protocol):
+	"""
+	Callback invoked when a timer expires.
+	"""
+
+	def __call__(self) -> None: ...
+
+
+@runtime_checkable
+class SetTimer(Protocol):
+	"""
+	Connection capability for registering one-shot timers.
+	"""
+
+	@abstractmethod
+	def set_timer(
+		self,
+		cb: TimerCallback,
+		seconds: int,
+	) -> Timer: ...
+
+
 @runtime_checkable
 class GetExchange(Protocol):
 	@abstractmethod
diff --git a/src/Broker/Implements/RabbitMQ/Connection.py b/src/Broker/Implements/RabbitMQ/Connection.py
index dc4797c..7481dee 100644
--- a/src/Broker/Implements/RabbitMQ/Connection.py
+++ b/src/Broker/Implements/RabbitMQ/Connection.py
@@ -2,6 +2,9 @@
 
 from DefStream.Broker import (
 	Connection as BaseConnection,
+	Timer as BaseTimer,
+	TimerCallback,
+	SetTimer,
 	GetExchange,
 	GetQueue,
 	GetStream,
@@ -55,9 +58,29 @@
 	'Parameters',
 	'All',
 	'Any',
+	'Timer',
 )
 
 
+class Timer(BaseTimer):
+	"""
+	Timer handle bound to the Pika connection that registered it.
+	"""
+
+	def __init__(
+		self,
+		connection: BlockingConnection,
+		timeout: int,
+	) -> None:
+		self.connection = connection
+		self.timeout = timeout
+		return
+
+	def remove(self) -> None:
+		self.connection.remove_timeout(self.timeout)
+		return
+
+
 class ExchangeType(str, Enum):
 	Direct = 'direct'
 	FanOut = 'fanout'
@@ -108,6 +131,7 @@ def values(self) -> ValuesView:
 
 class Connection(
 	BaseConnection,
+	SetTimer,
 	GetExchange,
 	GetQueue,
 	GetStream,
@@ -168,6 +192,19 @@ def connected(self):
 			raise RuntimeError('Connect the RabbitMQ connection before use')
 		return self.connection
 
+	def set_timer(
+		self,
+		cb: TimerCallback,
+		seconds: int,
+	) -> Timer:
+		"""
+		Register a one-shot timer using Pika's owner-thread event dispatch.
+		"""
+		connection: BlockingConnection = self.connected()
+		with self.observe('Timer registration'):
+			timeout: int = connection.call_later(seconds, cb)
+		return Timer(connection, timeout)
+
 	def declare(
 		self,
 		parameters: AnyValue,
diff --git a/src/Broker/Implements/RabbitMQ/__init__.py b/src/Broker/Implements/RabbitMQ/__init__.py
index 5fb97e7..9cc4f9c 100644
--- a/src/Broker/Implements/RabbitMQ/__init__.py
+++ b/src/Broker/Implements/RabbitMQ/__init__.py
@@ -7,6 +7,7 @@
 	Parameters,
 	All,
 	Any,
+	Timer,
 )
 from .Exchange import Exchange
 from .Queue import Queue
@@ -31,6 +32,7 @@
 	'Parameters',
 	'All',
 	'Any',
+	'Timer',
 	'Exchange',
 	'Queue',
 	'Stream',
diff --git a/src/Broker/__init__.py b/src/Broker/__init__.py
index 1dc894a..7caa6ab 100644
--- a/src/Broker/__init__.py
+++ b/src/Broker/__init__.py
@@ -4,6 +4,9 @@
 from .Connection import (
 	Connection,
 	BrokerFactory,
+	Timer,
+	TimerCallback,
+	SetTimer,
 	GetExchange,
 	GetQueue,
 	GetStream,
@@ -33,6 +36,9 @@
 	'Configuration',
 	'Connection',
 	'BrokerFactory',
+	'Timer',
+	'TimerCallback',
+	'SetTimer',
 	'GetExchange',
 	'GetQueue',
 	'GetStream',
diff --git a/tests/rabbitmq/test_rabbitmq_timers.py b/tests/rabbitmq/test_rabbitmq_timers.py
new file mode 100644
index 0000000..462c30e
--- /dev/null
+++ b/tests/rabbitmq/test_rabbitmq_timers.py
@@ -0,0 +1,288 @@
+from os import environ
+from pathlib import Path
+from runpy import run_path
+from threading import get_ident
+from time import monotonic
+from typing import (
+	Any,
+	Dict,
+	List,
+	Tuple,
+)
+from unittest import (
+	TestCase,
+	main,
+	skipUnless,
+)
+from unittest.mock import (
+	Mock,
+	call,
+	patch,
+	sentinel,
+)
+from uuid import uuid4
+
+from pika import BlockingConnection
+
+from DefStream.Broker import (
+	Event,
+	Timer,
+	TimerCallback,
+)
+from DefStream.Broker.Implements.RabbitMQ import (
+	Configuration,
+	Connection,
+	Consumer,
+)
+from tests._support import rabbitmq_test_port
+
+
+class TimerTests(TestCase):
+	def setUp(self) -> None:
+		self.broker: Connection = Connection(Configuration(
+			host='127.0.0.1',
+			port=5672,
+			username='guest',
+			password='guest',
+		))
+		self.connection: Mock = Mock(spec=BlockingConnection)
+		self.connection.call_later.return_value = sentinel.timeout
+		self.broker.connection = self.connection
+		self.addCleanup(self.broker.close)
+		return
+
+	def test_positional_and_keyword_registration(self) -> None:
+		cb: Mock = Mock()
+		for keyword in (
+			False,
+			True,
+		):
+			with self.subTest(keyword=keyword):
+				self.connection.reset_mock()
+				timer: Timer
+				if keyword:
+					timer = self.broker.set_timer(cb=cb, seconds=5)
+				else:
+					timer = self.broker.set_timer(cb, 5)
+				self.connection.call_later.assert_called_once_with(5, cb)
+				cb.assert_not_called()
+				self.assertIsInstance(timer, Timer)
+				timer.remove()
+				self.connection.remove_timeout.assert_called_once_with(sentinel.timeout)
+		return
+
+	def test_functions_and_callable_objects_are_passed_unchanged(self) -> None:
+		calls: List[str] = []
+
+		def callback() -> None:
+			calls.append('function')
+			return
+
+		class Callback:
+			def __call__(self) -> None:
+				calls.append('object')
+				return
+
+		callbacks: Tuple[TimerCallback, ...] = (
+			callback,
+			Callback(),
+		)
+		for cb in callbacks:
+			with self.subTest(callback=cb):
+				self.connection.reset_mock()
+				self.broker.set_timer(cb, 5)
+				self.connection.call_later.assert_called_once_with(5, cb)
+				self.assertIs(self.connection.call_later.call_args.args[1], cb)
+		self.assertEqual(calls, [])
+		return
+
+	def test_registration_uses_existing_open_connection_check(self) -> None:
+		self.connection.is_open = False
+		with self.assertLogs('Broker.RabbitMQ.Connection', level='ERROR'):
+			with self.assertRaisesRegex(RuntimeError, 'before use'):
+				self.broker.set_timer(Mock(), 5)
+		self.connection.call_later.assert_not_called()
+		return
+
+	def test_removal_keeps_registration_connection(self) -> None:
+		timer: Timer = self.broker.set_timer(Mock(), 5)
+		replacement: Mock = Mock(spec=BlockingConnection)
+		self.broker.connection = replacement
+		timer.remove()
+		self.connection.remove_timeout.assert_called_once_with(sentinel.timeout)
+		replacement.remove_timeout.assert_not_called()
+		return
+
+	def test_repeated_removal_is_delegated(self) -> None:
+		timer: Timer = self.broker.set_timer(Mock(), 5)
+		timer.remove()
+		timer.remove()
+		self.assertEqual(
+			self.connection.remove_timeout.call_args_list,
+			[
+				call(sentinel.timeout),
+				call(sentinel.timeout),
+			],
+		)
+		return
+
+	def test_delay_values_are_not_reinterpreted(self) -> None:
+		cb: Mock = Mock()
+		for seconds in (
+			-1,
+			0,
+			5,
+		):
+			with self.subTest(seconds=seconds):
+				self.connection.reset_mock()
+				self.broker.set_timer(cb, seconds)
+				self.connection.call_later.assert_called_once_with(seconds, cb)
+		return
+
+	def test_registration_error_is_propagated(self) -> None:
+		failure: ValueError = ValueError('Timer registration failed')
+		self.connection.call_later.side_effect = failure
+		with self.assertLogs('Broker.RabbitMQ.Connection', level='ERROR'):
+			with self.assertRaises(ValueError) as raised:
+				self.broker.set_timer(Mock(), 5)
+		self.assertIs(raised.exception, failure)
+		return
+
+	def test_removal_error_is_propagated(self) -> None:
+		timer: Timer = self.broker.set_timer(Mock(), 5)
+		failure: RuntimeError = RuntimeError('Timer removal failed')
+		self.connection.remove_timeout.side_effect = failure
+		with self.assertRaises(RuntimeError) as raised:
+			timer.remove()
+		self.assertIs(raised.exception, failure)
+		return
+
+	def test_timer_sample_registers_cancels_and_repeats(self) -> None:
+		broker: Mock = Mock(spec=Connection)
+		consumer: Mock = broker.consumer.return_value
+		timers: Tuple[Mock, ...] = tuple(Mock(spec=Timer) for _ in range(4))
+		broker.set_timer.side_effect = timers
+
+		def consume() -> None:
+			for _ in range(3):
+				cb: TimerCallback = broker.set_timer.call_args.args[0]
+				cb()
+			return
+
+		consumer.run.side_effect = consume
+		path: Path = Path(__file__).resolve().parents[2] / 'samples' / 'Broker' / 'RabbitMQ' / 'Sample.Timer.py'
+		with patch('DefStream.Broker.Implements.RabbitMQ.Connection', return_value=broker):
+			namespace: Dict[str, Any] = run_path(str(path), run_name='sample_import_test')
+			namespace['main']()
+		self.assertEqual(broker.set_timer.call_count, 4)
+		for registration in broker.set_timer.call_args_list:
+			self.assertEqual(registration.args[1], 1)
+		timers[0].remove.assert_called_once()
+		timers[-1].remove.assert_called_once()
+		for timer in timers[1:-1]:
+			timer.remove.assert_not_called()
+		consumer.stop.assert_called_once()
+		consumer.close.assert_called_once()
+		queue: str = broker.create_queue.call_args.args[0]
+		broker.delete_queue.assert_called_once_with(queue)
+		broker.close.assert_called_once()
+		return
+
+
+@skipUnless(environ.get('RABBITMQ_TESTS') == '1', 'Opt-in local RabbitMQ integration')
+class RabbitMQTimerIntegrationTests(TestCase):
+	def setUp(self) -> None:
+		self.broker: Connection = Connection(Configuration(
+			host='127.0.0.1',
+			port=rabbitmq_test_port(),
+			username='guest',
+			password='guest',
+		))
+		self.broker.parameters.socket_timeout = 2
+		self.broker.parameters.stack_timeout = 3
+		self.addCleanup(self.broker.close)
+		self.broker.connect()
+		queue: str = 'defstream-timer-test-' + uuid4().hex
+		self.broker.create_queue(queue)
+		self.addCleanup(self.broker.delete_queue, queue)
+		self.consumer: Consumer = self.broker.consumer()
+		self.addCleanup(self.consumer.close)
+		self.consumer.subs(queue, self.unexpected_message)
+		connection: BlockingConnection = self.broker.connected()
+		watchdog: int = connection.call_later(5, self.expired)
+		self.addCleanup(connection.remove_timeout, watchdog)
+		return
+
+	def unexpected_message(
+		self,
+		event: Event,
+	) -> None:
+		self.fail('The timer test queue must remain empty')
+
+	def expired(self) -> None:
+		self.fail('Timer test did not finish within five seconds')
+
+	def test_idle_timer_runs_once_on_owner_thread(self) -> None:
+		threads: List[int] = []
+		owner: int = get_ident()
+		started: float = monotonic()
+
+		def callback() -> None:
+			threads.append(get_ident())
+			self.consumer.stop()
+			return
+
+		timer: Timer = self.broker.set_timer(callback, 1)
+		self.addCleanup(timer.remove)
+		self.consumer.run()
+		self.assertEqual(threads, [owner])
+		self.assertGreaterEqual(monotonic() - started, 1)
+		return
+
+	def test_cancelled_timer_does_not_run(self) -> None:
+		callback: Mock = Mock()
+		timer: Timer = self.broker.set_timer(callback, 1)
+		timer.remove()
+		timer.remove()
+		stop: Timer = self.broker.set_timer(self.consumer.stop, 2)
+		self.addCleanup(stop.remove)
+		self.consumer.run()
+		callback.assert_not_called()
+		return
+
+	def test_callback_can_register_next_timer(self) -> None:
+		threads: List[int] = []
+		owner: int = get_ident()
+
+		def callback() -> None:
+			threads.append(get_ident())
+			if len(threads) == 3:
+				self.consumer.stop()
+			else:
+				next_timer: Timer = self.broker.set_timer(callback, 1)
+				self.addCleanup(next_timer.remove)
+			return
+
+		timer: Timer = self.broker.set_timer(callback, 1)
+		self.addCleanup(timer.remove)
+		self.consumer.run()
+		self.assertEqual(threads, [owner] * 3)
+		return
+
+	def test_callback_exception_reaches_consumption_loop(self) -> None:
+		failure: ValueError = ValueError('Timer callback failed')
+
+		def callback() -> None:
+			raise failure
+
+		timer: Timer = self.broker.set_timer(callback, 1)
+		self.addCleanup(timer.remove)
+		with self.assertLogs('Broker.RabbitMQ.Consumer', level='ERROR'):
+			with self.assertRaises(ValueError) as raised:
+				self.consumer.run()
+		self.assertIs(raised.exception, failure)
+		return
+
+
+if __name__ == '__main__':
+	main()
diff --git a/tests/test_protocol_contracts.py b/tests/test_protocol_contracts.py
index 6a73faa..6415210 100644
--- a/tests/test_protocol_contracts.py
+++ b/tests/test_protocol_contracts.py
@@ -1,5 +1,10 @@
-from inspect import isabstract
+from inspect import (
+	Parameter,
+	isabstract,
+	signature,
+)
 from typing import (
+	Mapping,
 	Optional,
 	get_type_hints,
 )
@@ -12,6 +17,9 @@
 from DefStream.Broker import (
 	Connection,
 	BrokerFactory,
+	Timer,
+	TimerCallback,
+	SetTimer,
 	Consumer,
 	Event,
 	Emittable,
@@ -24,7 +32,12 @@
 	Recoverable,
 )
 from DefStream.Broker import __all__ as broker_exports
-from DefStream.Broker.Connection import __all__ as connection_exports
+from DefStream.Broker.Connection import (
+	Timer as ConnectionTimer,
+	TimerCallback as ConnectionTimerCallback,
+	SetTimer as ConnectionSetTimer,
+	__all__ as connection_exports,
+)
 from DefStream.Inspector.Extensions import Play
 from DefStream.Broker.Implements.RabbitMQ import __all__ as rabbit_exports
 from DefStream.Broker.Implements.Redis import __all__ as redis_exports
@@ -89,6 +102,90 @@ def consumer(
 
 
 class ProtocolContractTests(TestCase):
+	def test_timer_protocols_and_public_contracts(self) -> None:
+		for public, contract in (
+			(
+				Timer,
+				ConnectionTimer,
+			),
+			(
+				TimerCallback,
+				ConnectionTimerCallback,
+			),
+			(
+				SetTimer,
+				ConnectionSetTimer,
+			),
+		):
+			with self.subTest(contract=contract):
+				self.assertIs(public, contract)
+				self.assertIn(contract.__name__, connection_exports)
+				self.assertIn(contract.__name__, broker_exports)
+
+		parameters: Mapping[str, Parameter] = signature(SetTimer.set_timer).parameters
+		self.assertEqual(
+			tuple(parameters),
+			(
+				'self',
+				'cb',
+				'seconds',
+			),
+		)
+		for parameter in parameters.values():
+			self.assertIs(parameter.kind, Parameter.POSITIONAL_OR_KEYWORD)
+			self.assertIs(parameter.default, Parameter.empty)
+		self.assertEqual(
+			get_type_hints(SetTimer.set_timer),
+			{
+				'cb': TimerCallback,
+				'seconds': int,
+				'return': Timer,
+			},
+		)
+		self.assertEqual(get_type_hints(Timer.remove), {'return': type(None)})
+		self.assertEqual(get_type_hints(TimerCallback.__call__), {'return': type(None)})
+
+		class IncompleteTimer(Timer):
+			pass
+
+		class IncompleteProvider(SetTimer):
+			pass
+
+		for incomplete in (
+			IncompleteTimer,
+			IncompleteProvider,
+		):
+			self.assertTrue(isabstract(incomplete))
+			with self.assertRaises(TypeError):
+				call_with_invalid_input(incomplete)
+
+		class StructuralTimer:
+			def remove(self) -> None:
+				return
+
+		class StructuralProvider:
+			def set_timer(
+				self,
+				cb: TimerCallback,
+				seconds: int,
+			) -> Timer:
+				return StructuralTimer()
+
+		self.assertIsInstance(StructuralTimer(), Timer)
+		self.assertIsInstance(StructuralProvider(), SetTimer)
+		self.assertNotIsInstance(object(), Timer)
+		self.assertNotIsInstance(object(), SetTimer)
+		self.assertIn(SetTimer, RabbitConnection.__bases__)
+		self.assertTrue(issubclass(RabbitConnection, SetTimer))
+		for implementation in (
+			Connection,
+			Consumer,
+			RedisConnection,
+		):
+			self.assertFalse(issubclass(implementation, SetTimer))
+			self.assertFalse(hasattr(implementation, 'set_timer'))
+		return
+
 	def test_recovery_capability_is_structural_and_opt_in(self):
 		class StructuralRecovery:
 			def is_recoverable(self, error: BaseException) -> bool: