Does Watermill support publishing delayed (set to a a future time) messages? about watermill

[m110]

Hey @preslavrachev!

Some Pub/Subs may support this. I know RabbitMQ has a plugin for delayed messages, I’ve been using it fot message retries.

But frankly, I think the best way would be to create a separate component for this. It could implement the Publisher and Subscriber interfaces, so you can use it plug-and-play with Watermill.

The publisher would just save the message in the storage, along with the desired publish time (could be a metadata or context field). The subscriber would filter the messages based on the current time. It should be quite easy to implement on top of the SQL Pub/Sub, but that’s not ideal if you don’t use a database at the moment.

What do you think?

[preslavrachev]

@m110 funny you mention SQL, because this is exactly what I set out to use :) May app is fairly small at the moment, and the heavy weightlifting is done by another service. My idea was to introduce persistent messaging as a form of retrial, in case something in any of the use cases goes wrong. A real message queue is more or less something I did not want to manage at the moment.

Anyway, I had a look at the DefaultPostgreSQLSchema and I saw that everything about creating the tables, inserting and querying is there in one place. I suppose, I could shamelessly copy it, and add support for a date in the future (which may be tucked somewhere in the Message metadata).

I'll report on my success. Expect a nice blog post about my experience with Watermill if it works out fine ;)

[m110]

@preslavrachev That’s sounds like a great fit then! 👍 It should be straightforward - you just need to provide a schema adapter based on the default one. If you feel like contributing, this could be added to watermill-sql.

[preslavrachev]

@m110 I was able to add delayed message sending - it was not difficult at all (I guess, that speaks to Watermill's extensibility)!

We can think of contributing the adapter changes to watermill-sql, of course. My only problem is the following - adding the delay right now happens as part of adding a message metadata field. This field is then read before inserting and if present, the future execution gets stored in a separate column. I want to somehow avoid confusing people who would think that delayed messages work with every adapter, and then wonder why those get executed right away.

Let us think about some more edge cases like that. Otherwise, I'd be happy to contribute.

[m110]

Great to hear that @preslavrachev! 🙌

One idea I can think of is exposing a public function like msg = sql.WithDelay(msg, time.Hour) with a doc saying it works only with a specific adapter. It could also add a context field if you don’t want to modify the metadata, as the insert is done before any network transfer, so all context fields will be there.

[stong1994]

In my situation, I utilize Redis zset for handling 'delayed messages.' Since zset doesn't support 'consumer groups,' I move the messages from the zset to RabbitMQ using a router. You can check out an example at: https://github.com/stong1994/watermill-rediszset/tree/main/example/timer