How to programmatically stop the faststream app?

[fyber]

This question relates to my previous one: #1133 (comment)
I went with the option where I create a FastStream app, and in @app.after_startup I create and register an infinite coroutine that does its job:

@app.after_startup
async def after_startup(context: ContextRepo) -> None:
    my_task = asyncio.create_task(...)
    context.set_global("my_task", my_task)

@app.on_shutdown
async def on_shutdown(my_task: asyncio.Task = Context()) -> None:
    my_task.cancel()

And it works.

Inside my task, I use not only FastStream but also a database. In case of any errors inside my task (e.g. DB connection broken, some other critical error) I want to terminate the entire program and let Kubernetes re-schedule my app on another pod.
How to properly achieve that? I tried this:

1. Call app._shutdown(). Looks like this closes broker connections, but the program does not terminate.
2. Send SIGINT to myself with os.kill(os.getpid(), signal.SIGINT). If my program is running in single process mode, this kinda works. But if I run it under the master faststream process or with reloader (--reload option), it doesn't -- it just gets respawned (as it should under supervisor).

Both described methods don't look natural. Do I miss something? Is there a better way to terminate my process?

[Lancetnik]

@fyber, sorry, for now we have no public API to stop application, but you can patch FastStream object to implement it by yourself

import anyio

from faststream import FastStream
from faststream.cli.supervisors.utils import set_exit
from faststream.nats import NatsBroker


class StoppableApp(FastStream):
    async def _stop(self, *args) -> None:
        event = self.event = anyio.Event()
        set_exit(lambda *_: event.set())
        await event.wait()

        await self._shutdown()

broker = NatsBroker()
app = StoppableApp(broker)

And call app.event.set() anywhere

It is a dirty trick, but working and you can use it right now