菜单
Implement the Observer pattern for event-driven decoupled communication. Outputs event bus implementations, typed event systems, and async observer patterns.
默认使用会先检查来源的 Prompt；你也可以切换为直接命令，或下载本地副本。
决定是否安装前，请先阅读 SKILL.md，以及 SkillsMP 当前展示的配套文件。
用 Codex 或 Claude 帮你安装 复制这段 Prompt，粘贴到 Codex、Claude 或其他助手里，让它检查 Skill 页面并帮你完成安装。
直接命令不会经过审查 Prompt；运行前请先检查来源。
npx skills add https://github.com/kalilurrahman/kr-claudiator-skills --skill observer-pattern命令会保持在同一行。复制前请横向滚动并检查完整内容。
想先保存到本地？可下载 SkillsMP 当前能够提供的文件。
正在显示 SKILL.md
基于 SOC 职业分类
| name | observer-pattern |
| description | Implement the Observer pattern for event-driven decoupled communication. Outputs event bus implementations, typed event systems, and async observer patterns. |
| argument-hint | ["language","event types","synchronous vs async","existing pub/sub infrastructure"] |
| allowed-tools | Read, Write |
The Observer pattern defines a one-to-many dependency so that when one object changes state, all dependents are notified automatically. It decouples event producers from consumers — neither knows about the other.
from abc import ABC, abstractmethod
from typing import Callable, Any
from collections import defaultdict
import asyncio
import logging
# Typed event system
from dataclasses import dataclass
from datetime import datetime
@dataclass
class :
order_id:
customer_id:
total_cents:
occurred_at: datetime
:
order_id:
amount_cents:
payment_id:
occurred_at: datetime
:
():
._handlers: [, []] = defaultdict()
._logger = logging.getLogger(__name__)
() -> :
._handlers[event_type].append(handler)
._logger.debug()
() -> :
._handlers[event_type].remove(handler)
() -> :
event_type = (event)
handlers = ._handlers.get(event_type, [])
handlers:
._logger.warning()
handler handlers:
:
handler(event)
Exception e:
._logger.error()
bus = EventBus()
():
email_service.send(event.customer_id, , event.order_id)
():
inventory_service.reserve(event.order_id)
():
fulfilment_service.create(event.order_id)
bus.subscribe(OrderPlacedEvent, send_confirmation_email)
bus.subscribe(OrderPlacedEvent, reserve_inventory)
bus.subscribe(PaymentCapturedEvent, create_fulfilment_request)
bus.publish(OrderPlacedEvent(
order_id=,
customer_id=,
total_cents=,
occurred_at=datetime.utcnow(),
))
class AsyncEventBus:
def __init__(self):
self._handlers: dict[type, list[Callable]] = defaultdict(list)
def subscribe(self, event_type: type, handler: Callable) -> None:
self._handlers[event_type].append(handler)
async def publish(self, event: Any) -> None:
handlers = self._handlers.get(type(event), [])
# Run all handlers concurrently
results = await asyncio.gather(
*[self._call(h, event) for h in handlers],
return_exceptions=True,
)
for h, r in zip(handlers, results):
if isinstance(r, Exception):
logging.error(f"Async handler {h.__name__} failed: {r}")
async def _call(self, handler: Callable, event: Any):
if asyncio.iscoroutinefunction(handler):
return await handler(event)
return handler(event)
async_bus = AsyncEventBus()
async def async_send_email(event: OrderPlacedEvent):
await email_client.send_async(event.customer_id, "Order confirmed")
async_bus.subscribe(OrderPlacedEvent, async_send_email)
class EventBusWithDecorators(EventBus):
def on(self, event_type: type):
def decorator(fn: Callable) -> Callable:
self.subscribe(event_type, fn)
return fn
return decorator
bus = EventBusWithDecorators()
@bus.on(OrderPlacedEvent)
def handle_order_placed(event: OrderPlacedEvent):
print(f"Order {event.order_id} placed for {event.total_cents} cents")
@bus.on(OrderPlacedEvent)
def notify_analytics(event: OrderPlacedEvent):
analytics.track("order_placed", {"order_id": event.order_id})
| Anti-Pattern | Problem | Fix |
|---|---|---|
| Catching exceptions silently | Failed handlers hide bugs | Log all errors; consider retry or dead letter |
| Order-dependent handlers | Brittle; implicit coupling | Handlers must be independent and order-agnostic |
| Leaking domain objects in events | Consumers coupled to domain internals | Events carry only primitive data or IDs |
| Synchronous long-running handlers | Slow handlers block publisher | Async handlers or queue for slow work |
| Global mutable event bus | Testing nightmare | Inject bus; reset between tests |
The sections below extend this skill into a complete operating playbook so it can run end-to-end inside Claude Code, CoWork, or any agentic tool without further prompting. Pull only the sections you need for a given engagement.
Before producing any output, the skill confirms:
If any of these are missing, the skill asks targeted clarifying questions before generating output. It never invents constraints the user did not state.
The canonical workflow for Observer Pattern runs in five stages. Each stage has an explicit exit criterion so the skill knows when to advance.
Stage 1 — Frame. Restate the problem in one paragraph. Name the decision, the deadline, the stakeholders, and the success metric. Surface assumptions explicitly so they can be challenged.
Stage 2 — Diagnose. Inventory the current state with concrete evidence: metrics, quotes, screenshots, configs, tickets. Separate facts from interpretations. Identify the two or three root causes that explain most of the gap, not the long tail of symptoms.
Stage 3 — Design. Generate at least two viable options. For each option, capture: what changes, who owns it, what it costs, what it unblocks, what it risks, and how it could fail. Recommend one with a written rationale.
Stage 4 — Execute. Convert the chosen option into a sequenced plan: milestones, owners, dependencies, gating checks, communication cadence, and rollback triggers. Anything that cannot be assigned an owner and a date is not yet a plan.
Stage 5 — Validate. Define how success will be measured, when the measurement happens, and what action follows each possible result. Schedule the retrospective before the work starts, not after.
Depending on the request, the skill returns one or more of:
Every artifact uses clear headings, short paragraphs, and tables where comparison helps. No filler. No restating the prompt. No hedging language when a recommendation is warranted.
The skill applies the following heuristics when choices are not obvious:
| Anti-pattern | Why it fails | What the skill does instead |
|---|---|---|
| Generic best-practice list with no context | Reader cannot act on it | Tailors recommendations to the stated constraints |
| Recommendation without trade-offs | Hides the cost of being wrong | Names the price paid for the recommendation |
| Plan with no owners or dates | Cannot be executed or tracked | Assigns a named owner and a date to every action |
| Metrics theatre | Measures activity, not outcome | Ties every metric back to the user or business outcome |
| Boil-the-ocean scope | Nothing ships | Cuts scope to the smallest valuable slice |
| Buried recommendation | Reader misses the point | Leads with the recommendation in the first paragraph |
The skill self-checks each output against these gates before returning it:
If any gate fails, the skill rewrites the section before returning it.
Context: Implement the Observer pattern for event-driven decoupled communication. Outputs event bus implementations, typed event systems, and async observer patterns.
Frame: the team needs a defensible recommendation within five working days; the audience is a cross-functional steering group; the cost of delay is higher than the cost of being slightly wrong.
Diagnose: the dominant constraint is decision latency, not analytical depth. Existing data is sufficient for a directional call.
Design: two viable options surfaced. Option A optimises for speed and reversibility. Option B optimises for completeness but slips the deadline by two weeks.
Execute: Option A recommended. Plan sequenced into a two-week sprint with named owners, a mid-point checkpoint, and a clear rollback trigger.
Validate: success measured against a single leading indicator at day 30 and a single lagging indicator at day 90. Retrospective scheduled for day 35.
A one-shot artifact rarely changes outcomes. The skill recommends a lightweight cadence to keep the work alive: