How to Stop and Clean Up an EventBus Instance in bubus: Complete Guide

Call await bus.stop(timeout=None, clear=False) to gracefully shut down an EventBus, cancel pending queue operations, and optionally purge all internal state.

The bubus library from browser-use/bubus provides a robust asynchronous event bus for Python applications. When you need to stop and clean up an EventBus instance in bubus, the framework offers a comprehensive shutdown API that prevents memory leaks and eliminates dangling tasks.

The EventBus Shutdown API

The primary method for stopping an EventBus is stop(), defined in bubus/service.py at lines 742–714. It accepts two optional parameters:

await bus.stop(timeout: float | None = None, clear: bool = False)

• timeout: When set to a positive number, the bus waits for in-flight events to complete via wait_until_idle before proceeding. A value of 0 skips the wait entirely, while None forces immediate shutdown.
• clear: When True, the method performs a deep clean, clearing event history, handler registries, and removing the instance from the global EventBus.all_instances weak set.

Internal Shutdown Sequence

The stop() method executes a ten-step teardown process to ensure resources are released properly:

1. Guard-Rail Check

If the bus is not running (self._is_running is False), the method returns immediately without action. This prevents double-stop scenarios.

2. Graceful Idle Wait

When timeout is provided, the bus awaits self.wait_until_idle(timeout=timeout) (lines 754–558). This allows current handlers to finish processing before shutdown commences.

3. Pending Work Diagnostics

If events remain queued or in progress, a debug log entry is emitted (lines 560–566) to help diagnose stuck handlers or event backlogs.

4. Running Flag Termination

The method sets self._is_running = False (line 668), signaling the background run-loop to stop pulling new events from the queue.

5. Queue Shutdown

The custom CleanShutdownQueue.shutdown() method is invoked (lines 672–673). Unlike standard asyncio queues, this implementation safely cancels pending get() and put() futures with a QueueShutDown exception, preventing "Event loop is closed" warnings.

6. Background Task Cancellation

The internal _runloop_task is cancelled with a safety timeout. The code awaits asyncio.wait({self._runloop_task}, timeout=0.1) then calls self._runloop_task.cancel() (lines 676–683).

7. Reference Release

Internal handles are cleared: self._runloop_task = None and self._on_idle.set() (lines 686–688), allowing garbage collection.

8. Deep Clean (Optional)

When clear=True, the method purges self.event_history, self.handlers, and removes the instance from EventBus.all_instances and the loop-level _eventbus_instances set (lines 690–702).

9. Final Logging

A debug message confirms shutdown completion, noting whether it was graceful or forced (lines 706–707).

10. Memory Safety Net

A final call to self._check_total_memory_usage() runs in a try/except block (lines 710–714), ensuring diagnostics do not abort the shutdown sequence.

Practical Code Examples

Immediate Shutdown

For fire-and-forget scenarios where you need to stop the bus instantly:

import asyncio
from bubus import EventBus

async def main():
    bus = EventBus(name="mybus")
    # ... dispatch events, register handlers ...

    await bus.stop()  # immediate shutdown

asyncio.run(main())

Graceful Shutdown with Timeout

Wait for in-flight events to complete before stopping:

async def main():
    bus = EventBus()
    # start long-running handlers ...

    await bus.stop(timeout=5.0)  # wait up to 5 seconds

If the timeout expires, the method proceeds to cancel the background task and shut down the queue, preventing indefinite hangs.

Complete Memory Cleanup

Remove all internal references and global tracking:

async def main():
    bus = EventBus()
    # ... use the bus ...

    await bus.stop(clear=True)  # clears history, handlers, and global registry

    # Instance is now eligible for garbage collection

Manual Idle Check

Inspect bus state before deciding to stop:

async def main():
    bus = EventBus()
    # ... dispatch many events ...

    await bus.wait_until_idle(timeout=10)  # block until queue empty or 10s passes

    await bus.stop()  # stop immediately after idle confirmed

Stopping from Within a Handler

Since stop() is async, it can be awaited from any coroutine, including event handlers:

@bus.on("ShutdownRequest")
async def handle_shutdown(event):
    await bus.stop(timeout=2.0)  # graceful stop triggered by event

Key Implementation Files

File	Relevant Section	Link
bubus/service.py	EventBus.stop implementation (lines 742–714)	service.py
bubus/service.py	CleanShutdownQueue.shutdown (lines 63–78)	service.py
bubus/service.py	EventBus.wait_until_idle (used by stop for graceful shutdown)	service.py
bubus/service.py	Global registration hook in _start (ensures stop on loop close)	service.py

Summary

• Use await bus.stop() to halt an EventBus immediately, cancel pending queue operations, and shut down the background run-loop.
• Provide a timeout (e.g., await bus.stop(timeout=5.0)) to allow in-flight handlers to complete before termination.
• Set clear=True to purge event history, handler registries, and remove the instance from global tracking sets, enabling full garbage collection.
• The custom CleanShutdownQueue ensures no "Event loop is closed" warnings by resolving pending futures with QueueShutDown exceptions.
• Automatic cleanup occurs via a loop-level hook that calls stop() when the asyncio loop closes, preventing resource leaks in long-running applications.

Frequently Asked Questions

What happens if I call stop() on an already stopped EventBus?

The method returns immediately without action. An internal guard-rail check at lines 742–749 in bubus/service.py verifies self._is_running; if False, the function exits, preventing double-stop errors or exceptions.

Does stop() wait for event handlers to finish by default?

No. By default (timeout=None), stop() initiates immediate shutdown. To wait for handlers, pass a positive timeout value (e.g., timeout=5.0), which invokes wait_until_idle to allow in-flight events to complete before the queue shuts down.

What is the difference between stop() and stop(clear=True)?

stop() halts the event loop and cancels pending queue operations but retains the event history, handler registry, and global instance tracking. stop(clear=True) performs a deep clean, clearing self.event_history, self.handlers, and removing the instance from EventBus.all_instances, making the object eligible for garbage collection.

Can I stop the EventBus from within an event handler?

Yes. Since stop() is an async method, you can await it from any coroutine, including event handlers. The bus will finish processing the current event before executing the shutdown sequence, allowing for graceful termination triggered by application logic.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}