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
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:
@@ -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.
|
||||
|
||||
|
||||
Reference in New Issue
Block a user