fix(events): add unsubscribe() to EventBus protocol and implementations (#10356)
CI / push-validation (pull_request) Successful in 45s
CI / helm (pull_request) Successful in 57s
CI / build (pull_request) Successful in 1m14s
CI / lint (pull_request) Successful in 1m47s
CI / quality (pull_request) Successful in 1m54s
CI / typecheck (pull_request) Successful in 1m55s
CI / security (pull_request) Successful in 1m56s
CI / integration_tests (pull_request) Successful in 7m37s
CI / unit_tests (pull_request) Failing after 9m18s
CI / coverage (pull_request) Has been skipped
CI / docker (pull_request) Has been skipped
CI / status-check (pull_request) Failing after 3s

Add EventBus.unsubscribe(event_type, handler) -> bool method to the
EventBus structural Protocol and both concrete implementations
(ReactiveEventBus and LoggingEventBus). Returns True if a matching
handler was found and removed, False otherwise.

Resolves #10356
This commit is contained in:
2026-05-16 09:04:50 +00:00
committed by CleverThis
parent 23d73e7fb2
commit b4cc6cd66b
4 changed files with 90 additions and 0 deletions
+1
View File
@@ -267,6 +267,7 @@ def step_protocol_stubs_callable(ctx: Context) -> None:
event = _make_event("plan.created")
obj.emit(event)
obj.subscribe(EventType.PLAN_CREATED, lambda e: None)
obj.unsubscribe(EventType.PLAN_CREATED, lambda e: None)
# ---------------------------------------------------------------------------
@@ -105,5 +105,38 @@ class LoggingEventBus:
raise TypeError("handler must be callable")
self._subscriptions.setdefault(event_type, []).append(handler)
def unsubscribe(
self,
event_type: EventType,
handler: Callable[[DomainEvent], None],
) -> bool:
"""Remove *handler* from events of *event_type*.
Removes only the **first** occurrence of *handler* (in case it was
registered multiple times). Returns ``True`` if a matching handler
was found and removed, ``False`` otherwise.
Args:
event_type: The :class:`EventType` whose handler list to modify.
handler: The previously-registered callable to remove.
Returns:
``True`` if a handler was removed; ``False`` if no match found.
"""
if not isinstance(event_type, EventType):
raise TypeError(
f"event_type must be an EventType, got {type(event_type).__name__!r}"
)
if not callable(handler):
raise TypeError("handler must be callable")
handlers = self._subscriptions.get(event_type)
if handlers is None:
return False
try:
handlers.remove(handler)
return True
except ValueError:
return False
__all__ = ["LoggingEventBus"]
@@ -49,5 +49,26 @@ class EventBus(Protocol):
"""
...
def unsubscribe(
self,
event_type: EventType,
handler: Callable[[DomainEvent], None],
) -> bool:
"""Remove *handler* from the subscriber list for *event_type*.
Removes only the **first** occurrence of *handler* in the list for
*event_type* (in case it was registered multiple times). Returns
``True`` if a matching handler was found and removed, ``False``
otherwise.
Args:
event_type: The :class:`EventType` whose handler list to modify.
handler: The previously-registered callable to remove.
Returns:
``True`` if a handler was removed; ``False`` if no match found.
"""
...
__all__ = ["EventBus"]
@@ -184,6 +184,41 @@ class ReactiveEventBus:
"""
return self._stream
def unsubscribe(
self,
event_type: EventType,
handler: Callable[[DomainEvent], None],
) -> bool:
"""Remove *handler* from events of *event_type*.
Removes only the **first** occurrence of *handler* (in case it was
registered multiple times). Returns ``True`` if a matching handler
was found and removed, ``False`` otherwise.
Args:
event_type: The :class:`EventType` whose handler list to modify.
handler: The previously-registered callable to remove.
Returns:
``True`` if a handler was removed; ``False`` if no match found.
"""
if not isinstance(event_type, EventType):
raise TypeError(
f"event_type must be an EventType, got {type(event_type).__name__!r}"
)
if not callable(handler):
raise TypeError("handler must be callable")
handlers = self._subscriptions.get(event_type)
if handlers is None:
return False
try:
handlers.remove(handler)
return True
except ValueError:
# Handler was not found in list (duplicates registered but
# this specific reference is not present).
return False
def close(self) -> None:
"""Complete the RxPY stream and prevent further event emission.