async def METHOD_NAME(self, on_event: EventCallback, join: bool = True) -> AsyncGenerator[None, None]:
    """Start a task in the background that sends events from the subscriber's stream
    to ``socket`` as they become available. On exit, it will prevent the stream from
    accepting new events and wait until the currently enqueued ones are processed.
    Should the context be left with an exception, the task will be cancelled
    immediately.
    Args:
        on_event: Callback to invoke with the event data for every event
        join: If ``True``, wait for all items in the stream to be processed before
            stopping the worker. Note that an error occurring within the context
            will always lead to the immediate cancellation of the worker
    """
    self._start_in_background(on_event=on_event)
    async with AsyncExitStack() as exit_stack:
        exit_stack.push_async_callback(self.stop, join=False)
        yield
        exit_stack.pop_all()
        await self.stop(join=join)