How to delete a redis stream after success processing

Question

I use redis as transport in messenger, I thought that after processing a flow the deletion was automatic but alas not. I do not know how to delete a repeat stream when the processing has been carried out with success.

I use symfony 4.4.latest and redis server 6.0

Thanks

Do you mean delete the all stream or only the processed messages? — Guy Korland, May 14 '20 at 11:29
@GuyKorland I wish when I process a message that it can be deleted from the queue — FOKO THierry, May 14 '20 at 13:35

score 4 · Answer 1 · answered May 14 '20 at 14:21

4

The way to do it is by using XTRIM command.

You can call you process couple of messages you trim the stream to retain only the messages that were not processed. By, calling XLEN you can get the stream size and if you subtract the amount of messages you processed you should be left with the right argument for the XTRIM.

answered May 14 '20 at 14:21

Guy Korland

9,139
14
59
106

1

Thank you I find it hard to understand how to do it. Is there a way when the symfony handler retrieves the message to process, how to retrieve the message id so that at the end of processing by the handler symfony, I can remove this message from the redis stream. Thank you – FOKO THierry May 14 '20 at 16:43

score 0 · Answer 2 · answered May 24 '21 at 05:27

0

Simple steps:

Read a stream message and keep its ID.
When you are done processing the message use the XDEL command to delete specific entries from the redis stream.

BTW redis streams are not supposed to be used like this, better use pub/sub functionality as @Dudo mentioned. This is a good introduction to redis streams: https://redis.io/topics/streams-intro

answered May 24 '21 at 05:27

rsjethani

2,179
6
24
30

problem with pub/sub is that it does not guarantee at-least-once delivery – sodik Jun 14 '23 at 11:29

score -1 · Answer 3 · answered Jun 10 '22 at 11:58

Want to add here that in Symfony 5 ?delete_after_ack=true was added.

MESSENGER_TRANSPORT_DSN=redis://localhost:6379/messages?delete_after_ack=true

Why the default in Symfony 5 is false. The default in Symfony 6 is true so Symfony 6 will automatically remove acked message.

See also: https://symfony.com/doc/current/messenger.html#redis-transport

There are also other parameters like delete_after_reject or stream_max_entries which trims the stream. Keep in mind stream_max_entries is trim messages aways which also are not processed. So it should be a high enough value.

How to delete a redis stream after success processing

3 Answers3