PostgreSQL support

[Vad1mo]

Is your feature request related to a problem? Please describe.

I am aware that PostgreSQL is now a Message Queue per se; however, it can be utilized as such. Many developers realize that it suffices for the majority of tasks, and it comes with zero added costs if a PostgreSQL db is present.

Describe the solution you'd like
Clearly and concisely describe the desired outcome or solution.

Feature code example
Nothing in particular

Describe alternatives you've considered
You can do exactly this with MySQL and SQL server too because they both support SKIP LOCKED.

Additional context

• https://adriano.fyi/posts/2023-09-24-choose-postgres-queue-technology/
• https://www.prequel.co/blog/sql-maxis-why-we-ditched-rabbitmq-and-replaced-it-with-a-postgres-queue
• https://dagster.io/blog/skip-kafka-use-postgres-message-queue

[acaloiaro]

@Vad1mo 👋 I came across this by chance and see you've referenced my write up on the topic of PG as a queue.

If you find Go easy enough to read, you might benefit from reading the neoq source (https://github.com/acaloiaro/neoq) as a reference implementation for the ideas lain out in my post. The whole PG implementation is contained within: https://github.com/acaloiaro/neoq/blob/main/backends/postgres/postgres_backend.go

Cheers!

[Vad1mo]

@Vad1mo 👋 I came across this by chance and see you've referenced my write up on the topic of PG as a queue.

If you find Go easy enough to read, you might benefit from reading the neoq source (https://github.com/acaloiaro/neoq) as a reference implementation for the ideas lain out in my post. The whole PG implementation is contained within: https://github.com/acaloiaro/neoq/blob/main/backends/postgres/postgres_backend.go

Cheers!

Found it already 👀 acaloiaro/neoq#70 🤣

[Lancetnik]

I don't like an idea to poll DB for updates
I am thinking about Listen/Notify PG feature or even CDC (replication journal) consuming

[Vad1mo]

Yes, I was referring to listen/notify. No mention about polling.
Regarding CDC, here I am not sure how or if it is low level, hence if it can be easily setup within the offerings of cloud providers.

[Lancetnik]

CDC available from any pg driver: asyncpg/psycopg/etc
It is just a part of protocol
So, it can be a useful feature as I can image

[mixartemev]

Hey guys! I really need it asap. R u planning to implement it in nearest future or should I search other tools for manage pg listen/notify queues?

[Lancetnik]

Hey guys! I really need it asap. R u planning to implement it in nearest future or should I search other tools for manage pg listen/notify queues?

Sorry, I don't think, that we can add Postgres support this year. Probably, we can do it in the first half of 2025, but can't strongly promise 😢

[mcproger]

Hey @Lancetnik

I've built a simple PoC with the listen/notify approach (took a couple of hours, so not the best code quality) that shows the general idea – link

If you are OK with this way, I would like to try to implement this feature. I'm still tinkering with the source code so if you can point places to start with it would be very helpful (if no - not a problem, just will spend a little bit more time reading the code).

Thanks

[Lancetnik]

@mcproger Hi! Thank you a lot for your interest! I am appriciate any help with Postgres support and will be glad to work on it with you. But, can you wait for 0.6 release? I plan to unlock new brokers support after it and release SQS as a 0.7 one.

Anyway, you can start to work on Postgres already basing on 0.6 branch - it is almost ready to release.

To implement a new broker you should follow these steps

1. Implement broker connect, _connect and start methods
   At this step, FastStream(PostgresBroker()) should be runnable already
2. Then, you should create a subscriber class and register it in the broker
   So, @broker.subscriber(...) should be able to consume a messages already
3. Next step you should impl a Producer class, and broker.publish - it should work already and subscriber can consume your messages already
4. Then you can impl a publish class to use it as a @broker.publisher(...) decorator
5. Finally, you should add AsyncAPI specification support, TestClient with MemoryProducer and add FastAPI support