Add support for unsubscribing after using `EventBus.on`

[chowder]

Firstly, thanks for developing this library, the API is great.

What feels somewhat missing though is an easy way to unsubscribe after using EventBus.on

I'd like to propose two patterns for this, both of which I think would be very trivial to implement:

bus.on(MyEvent, handler)
# unsubscribe with the same handler and event
bus.remove(MyEvent, handler)

cancel = bus.on(MyEvent, handler)
# unsubscribe using the cancellation callback returned by .on
cancel()

Another small nicety is that this would let you simplify the cleanup in EventBus.expect here: https://github.com/browser-use/bubus/blob/main/bubus/service.py#L611-L614

Happy to be the one raising a PR for this as well.

[pirate]

Thanks for the kind words. It was actually an intentional design choice to not implement handler unsubscribing. I've always thought it was a pattern that encourages bad codebase design.

I wanted to encourage people to use expect() so that ephemeral handlers are always inline awaits somewhere that only last for the duration of a function. I really encourage people to use only expect() or to create a whole new EventBus instance with its own handler set if you need the handler list to change.

I find allowing dyanamic handler registration/unregistration leads to code thats really hard to reason about, people tend to abuse it to write hooks for consuming one-off events, or for cleanup before re-use of a bus. It's much better to just destroy the entire bus and create a new one (ideally with a different name) with a new set of handlers.

Re-using the same bus with a mutable handler set leads to it being difficult to determine what code will run when a given event is fired after-the-fact, and breaks time-travel debugging. Unless you serialize the entire list of active handlers on every event and save it to the WAL as well, it becomes hard to know deterministically what ran. I'd prefer handlers are defined as close to statically as possible / only at startup, dynamic handler is better as expect or if: branches within existing handler functions.

Can you describe the use-case you need it for? If it's reasonable I could reluctantly be convinced to add unsubscribing...

[chowder]

Ah okay - so everything you described does make sense.

The use case where I'm finding it awkward is where clients/actors are ephemeral - e.g. imagine a web server serving events over SSE, or a websocket - it would've been nice to do something like:

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import asyncio
import bubus

app = FastAPI()
event_bus = bubus.EventBus()

@app.get("/events")
async def sse(request: Request):
    queue = asyncio.Queue()

    async def publish(event):
        await queue.put(event)

    event_bus.on("SomeEvent", publish)

    async def event_generator():
        try:
            while True:
                event = await queue.get()
                yield f"data: {event}\n\n"
        # runs if/when the client disconnects 
        finally:
            event_bus.off("SomeEvent", publish)

    return StreamingResponse(event_generator(), media_type="text/event-stream")

What do you think?

[pirate]

I don't really understand the code you shared, nothing is run except event_generator()? It seems like nothing is put in the queue and no events are ever fired.

In general you should avoid dynamically registering and unregistering handlers, instead just use brand new EventBus objects that have a fixed set of handlers. Event busses are cheap! and no special cleanup is required, just let unused EventBus objects get garbage collected when they go out of scope.

By changing the handlers registered at runtime you cant reason about the event stream after-the-fact / replicate it by replaying it. You wouldn't be able to know what runs as a result of SomeEvent unless you have the whole python process in memory snapshotted at every point in time too.

By only setting up the handlers once, you know on the publisher_bus, all SomeEvents always result in publish() being called, and you cant have contention between multiple threads or async tasks fighting eachchother by registering/unregistering stuff.

queue = asyncio.Queue()
async def publish(event):
	await queue.put(event)

pushblisher_bus = EventBus()
pushblisher_bus.on("SomeEvent", publish)
# if you need to sync events with any other bus:
#  publisher_bus.on('*', any_other_eventbus)
#  any_other_eventbus.on('*', publisher_bus)

[chowder]

Sorry, that example perhaps wasn't very clear, I assumed it would've been apparent that you'd have something dispatching the event.

Let me give a more concrete example, let's say I have just a simple long-running process that does some work:

#!/usr/bin/env -S uv run --script
#
# /// script
# dependencies = [
#   "bubus",
# ]
# ///
import asyncio
import bubus


class StateUpdate(bubus.BaseEvent):
    pass


class Application:
    """
    Business logic contained here
    """

    def __init__(self):
        # Root event bus
        self.event_bus = bubus.EventBus()

    async def run_to_completion(self):
        # Simulate work being done; the internals here can be as complicated
        # as it needs to be 
        while True:
            await asyncio.sleep(5)
            await self.event_bus.dispatch(StateUpdate())


async def main():
    app = Application()
    await app.run_to_completion()


if __name__ == "__main__":
    asyncio.run(main())

All OK.

Then perhaps at some point I decide to expose a websocket/SSE endpoint where I can stream these TaskCompleted events, so I'll run a webserver along my application that shares the event bus:

#!/usr/bin/env -S uv run --script
#
# /// script
# dependencies = [
#   "bubus",
#   "fastapi",
#   "uvicorn",
# ]
# ///

import asyncio
import bubus
import uvicorn

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse


class StateUpdate(bubus.BaseEvent):
    pass


class Application:
    """
    Business logic contained here
    """

    def __init__(self, event_bus: bubus.EventBus):
        self.event_bus = event_bus

    async def run_to_completion(self):
        # Simulate work being done
        while True:
            await asyncio.sleep(5)
            await self.event_bus.dispatch(StateUpdate())


class WebApp:
    def __init__(self, event_bus: bubus.EventBus, fastapi_app: FastAPI):
        self.event_bus = event_bus
        self.app = fastapi_app
        self.app.get("/stream")(self.stream)

    def stream(self):
        queue = asyncio.Queue()

        async def publish(event):
            await queue.put(event)

        self.event_bus.on(StateUpdate, publish)

        async def event_generator():
            # Maybe publish some initial state here, I dunno 
            yield f"data: <some initial state>\n\n"
            try:
                # And then any incremental updates after
                while True:
                    _ = await queue.get()
                    yield f"data: <update>\n\n"

            # Runs if/when the client disconnects
            finally:
                self.event_bus.off(StateUpdate, publish)

        return StreamingResponse(event_generator(), media_type="text/event-stream")


async def main():
    # Lift the root event bus up to this scope
    event_bus = bubus.EventBus()
    app = Application(event_bus=event_bus)

    # So that I can pass it into the web application
    fastapi_app = FastAPI()
    web = WebApp(event_bus=event_bus, fastapi_app=fastapi_app)

    # Run both in the same event loop
    await asyncio.gather(
        app.run_to_completion(),
        uvicorn.Server(
            uvicorn.Config(
                app=fastapi_app,
                host="127.0.0.1",
                port=8080,
            ),
        ).serve(),
    )


if __name__ == "__main__":
    asyncio.run(main())

I can't see a way to accomodate the above workflow (even with event bus chaining), the subscription in WebApp.stream has to be ephemeral as far as I can tell.

But perhaps that just means bubus isn't a good fit for what I'm trying to do here? (i.e. where your event actors are transient)

[pirate]

Ah ok, then this is a perfect example to demonstrate the concept.

So my understanding is you want a cross-request IPC / broadcast channel, and adding/removing listeners as they connect and disconnect keeps your main bus clean so you queue up events that are actually being published somewhere. In this case, this is the shared state you have to manage:

• the list of state updates (already handled via StateUpdates over main bus)
• the list of websocket clients actively subscribed to state updates

In your example subscribing/unsubscribing to StateUpdate is actually a obscuring the fact that the list of websockets that are subscribed is legitimate shared state that should be treated as such.

Mutable shared state hidden in callback relationships / implicitly tied to function scope is a recipie for pain unless you have at least a data structure like _active_listeners: dict[str, Queue] to track it, and ideally enforce changes only through events.

The golden rule for event-based system is: never write to any shared mutable state without an event. (direct reads from shared state without events are fine, it's writes that matter most)

# Set up a save little event-driven container for your mutable state.
class StateRelay:
    # could even just be a global dict with no class, but definitely use a data structure
    _active_listeners = {
        'some-user-id': asyncio.Queue(),
        'other-user-id': asyncio.Queue(),
        ...
    }
    
    async def on_state_update(self, event: StateUpdate) -> None:
        if event.user_id in active_listeners:
            await self._active_listeners[event.user_id].put(event.content)
    
    async def on_subscribe(self, event: SubscribeEvent) -> asyncio.Queue:
        _active_listeners[event.user_id] = asyncio.Queue()
        return self._active_listeners[event.user_id]
    
    async def on_unsubscribe(self, event: UnsubscribeEvent) -> None:
        del self._active_listeners[event.user_id]

Hook up your shared main event bus to your state relay service.

event_bus.on(StateUpdate, StateRelay.on_state_update)
event_bus.on(Subscribe, StateRelay.on_subscribe)
event_bus.on(Unsubscribe, StateRelay.on_unsubscribe)

Then have your invidiual requests subscribe/unsubscribe using events to the state relay service.

class WebApp:
    ...

    def stream(self):
        async def event_generator():
            queue = await event_bus.dispatch(Subscribe(user_id='13245')).event_result()

            yield f"data: <some initial state for user 13245>\n\n"
            try:
                # And then any incremental updates after
                while True:
                    next_state_update = await queue.get()
                    yield f"latest data for user 13245: {next_state_update}\n\n"

            # Runs if/when the client disconnects
            finally:
                await self.event_bus.dispatch(Unsubscribe(user_id='13245'))

        return StreamingResponse(event_generator(), media_type="text/event-stream")

Here is some more of our writing and others on the philosophy behind this approach:

• https://monadical.com/posts/functional-declarative-animations.html
• https://monadical.com/posts/Linearizing-database-operations-django.html
• https://monadical.com/posts/Designing-A-Banking-System.html#Designing-a-linearizable-queue-for-transactions
• https://monadical.com/posts/event-driven-architecture-1.html
• https://www.cosmicpython.com/book/chapter_08_events_and_message_bus.html#message_bus_diagram
• https://www.youtube.com/watch?v=LKtk3HCgTa8 Rich Hickey: Simple Made Easy + https://www.youtube.com/watch?v=rI8tNMsozo0

[pirate]

I implemented .off() support over in my fork: https://github.com/ArchiveBox/abxbus

also added a typescript version of the library with bridges between py and ts