Need help with unusual use case: publish-only app

[fyber]

Hi!
I'm planning to use faststream in my application which only publishes messages to Kafka and doesn't listen to any topic. It polls a database table and publishes new rows to Kafka in an infinite loop. However, I see that in documentation there's only consumer or consumer+publisher examples. I managed to create a direct publisher (https://faststream.airt.ai/latest/getting-started/publishing/direct/) and it works, but I struggle with running it.

1. How to correctly launch a broker? Who should manage event loop?
2. Do I even need an app instance? Or KafkaBroker instance is enough?
3. How to launch it? With the code below, looks like a standard command faststream run app will do the import, but the event loop will be held by my infinite for loop. Is that okay?

import asyncio

from faststream import FastStream

from app.core.config import get_settings

settings = get_settings()

# create_app is our factory that creates FastAPI application
app: FastStream = create_app(settings=settings)

broker = app.broker

publisher = broker.publisher(settings.APP.TARGET_TOPIC)


async def relay_to_kafka() -> None:
    async with broker:
        for n in range(100):  # for loop is just for test, it should be `while True`
            # grab a message from DB table
            msg = f"Ping {n}"
            await publisher.publish(key=str(n), message=msg)
            await asyncio.sleep(0.1)


def main() -> None:
    asyncio.run(relay_to_kafka())

main()

Thanks!

[Lancetnik]

If you want to just a publish messages, you can use smth like

async with KafkaBroker() as broker:
    await broker.publish(...)

In any code you want.

Direct broker.publisher().publish(...) is just a trick to generate AsyncAPI schema based on your app.

I think, the preffered way in your case is do not use FastStream application object (if you need no AsyncAPI documentation of your service) and just write your own service with broker.publish(...) usage

But, if you want to have a doc, you should create regular FastStream app (withour any subscribers) and start endless task in app.on_startup hook