Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
83 changes: 69 additions & 14 deletions bubus/service.py
Original file line number Diff line number Diff line change
Expand Up @@ -453,17 +453,7 @@ def on(
f'Invalid handler: {handler}, must be a sync or async function or method'
)

# Determine event key
event_key: str
if event_pattern == '*':
event_key = '*'
elif isinstance(event_pattern, type) and issubclass(event_pattern, BaseEvent): # pyright: ignore[reportUnnecessaryIsInstance]
event_key = event_pattern.__name__ # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType]
else:
event_key = str(event_pattern)

# Ensure event_key is definitely a string at this point
assert isinstance(event_key, str)
event_key = self._get_event_key(event_pattern)
Comment thread
cubic-dev-ai[bot] marked this conversation as resolved.

# Check for duplicate handler names
new_handler_name = get_handler_name(handler)
Expand All @@ -482,6 +472,73 @@ def on(
self.handlers[event_key].append(handler) # type: ignore
logger.debug(f'👂 {self}.on({event_key}, {get_handler_name(handler)}) Registered event handler')

def _get_event_key(self, event_pattern: EventPatternType) -> str:
"""Resolve an event pattern (type name string, event class, or '*') to the internal handler key."""
if event_pattern == '*':
return '*'
elif isinstance(event_pattern, type) and issubclass(event_pattern, BaseEvent): # pyright: ignore[reportUnnecessaryIsInstance]
return event_pattern.__name__ # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType]
return str(event_pattern)

# Overloads mirror on() so handlers typed for a concrete event remain valid.
@overload
def remove(self, event_pattern: EventPatternType, handler: EventHandlerFunc[T_Event]) -> bool: ...

@overload
def remove(self, event_pattern: EventPatternType, handler: AsyncEventHandlerFunc[T_Event]) -> bool: ...

@overload
def remove(self, event_pattern: EventPatternType, handler: EventHandlerMethod[T_Event]) -> bool: ...

@overload
def remove(self, event_pattern: EventPatternType, handler: AsyncEventHandlerMethod[T_Event]) -> bool: ...

@overload
def remove(self, event_pattern: EventPatternType, handler: EventHandlerClassMethod['BaseEvent[Any]']) -> bool: ...

@overload
def remove(self, event_pattern: EventPatternType, handler: AsyncEventHandlerClassMethod['BaseEvent[Any]']) -> bool: ...

def remove(
self,
event_pattern: EventPatternType,
handler: (
EventHandlerFunc[T_Event]
| AsyncEventHandlerFunc['BaseEvent[Any]']
| EventHandlerMethod[T_Event]
| AsyncEventHandlerMethod['BaseEvent[Any]']
| EventHandlerClassMethod['BaseEvent[Any]']
| AsyncEventHandlerClassMethod['BaseEvent[Any]']
),
) -> bool:
"""
Unsubscribe a previously registered handler from events matching a pattern.

Args:
event_pattern: The event type string, event model class, or '*' that was passed to `on()`.
handler: The exact handler function or method previously registered with `on()`.

Returns:
True if the handler was found and removed, False otherwise.

Example:
eventbus.on(TaskStartedEvent, handler)
...
eventbus.remove(TaskStartedEvent, handler)
"""
event_key = self._get_event_key(event_pattern)
registered_handlers = self.handlers.get(event_key, [])
registered_handler = cast(ContravariantEventHandler['BaseEvent[Any]'], handler)
if registered_handler in registered_handlers:
registered_handlers.remove(registered_handler)
if not registered_handlers:
del self.handlers[event_key]
logger.debug(
f'👂 {self}.remove({event_key}, {get_handler_name(registered_handler)}) Unregistered event handler'
)
return True
return False

def dispatch(self, event: T_ExpectedEvent) -> T_ExpectedEvent:
"""
Enqueue an event for processing and immediately return an Event(status='pending') version (synchronous).
Expand Down Expand Up @@ -677,9 +734,7 @@ def notify_expect_handler(event: 'BaseEvent[Any]') -> None:
return await future
finally:
# Clean up handler
event_key: str = event_type.__name__ if isinstance(event_type, type) else str(event_type) # pyright: ignore[reportUnknownMemberType, reportPartialTypeErrors]
if event_key in self.handlers and notify_expect_handler in self.handlers[event_key]:
self.handlers[event_key].remove(notify_expect_handler)
self.remove(event_type, notify_expect_handler)

def _start(self) -> None:
"""Start the event bus if not already running"""
Expand Down
91 changes: 91 additions & 0 deletions tests/test_remove_handler.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,91 @@
"""Tests for EventBus.remove() handler unsubscription."""

from bubus import BaseEvent, EventBus


class PingEvent(BaseEvent[str]):
"""Minimal event used to verify handler removal."""


async def test_remove_handler_by_event_class_stops_delivery():
"""Removing a handler registered by event class stops it from receiving future events."""
bus = EventBus(name='remove_class')
calls: list[str] = []

async def handler(event: PingEvent) -> str:
calls.append(event.event_type)
return 'ok'

bus.on(PingEvent, handler)
assert bus.remove(PingEvent, handler) is True
assert PingEvent.__name__ not in bus.handlers
assert bus.remove(PingEvent, handler) is False # second removal is a no-op

await bus.dispatch(PingEvent())
await bus.wait_until_idle()

assert calls == []
await bus.stop()


async def test_remove_handler_by_string_name():
"""Removing a handler registered by event type string works too."""
bus = EventBus(name='remove_string')
calls: list[str] = []

async def handler(event: BaseEvent) -> str:
calls.append(event.event_type)
return 'ok'

bus.on('PingEvent', handler)
assert bus.remove('PingEvent', handler) is True
assert bus.remove('PingEvent', handler) is False

await bus.dispatch(PingEvent())
await bus.wait_until_idle()

assert calls == []
await bus.stop()


async def test_remove_only_targets_specific_handler():
"""Removing one handler leaves other handlers for the same event intact."""
bus = EventBus(name='remove_specific')
calls: list[str] = []

async def keep(event: PingEvent) -> str:
calls.append('keep')
return 'keep'

async def drop(event: PingEvent) -> str:
calls.append('drop')
return 'drop'

bus.on(PingEvent, keep)
bus.on(PingEvent, drop)
assert bus.remove(PingEvent, drop) is True

await bus.dispatch(PingEvent())
await bus.wait_until_idle()

assert calls == ['keep']
await bus.stop()


async def test_remove_wildcard_handler():
"""Removing a '*' wildcard handler stops it from receiving all events."""
bus = EventBus(name='remove_wildcard')
calls: list[str] = []

async def handler(event: BaseEvent) -> str:
calls.append(event.event_type)
return 'ok'

bus.on('*', handler)
assert bus.remove('*', handler) is True

await bus.dispatch(PingEvent())
await bus.wait_until_idle()

assert calls == []
await bus.stop()