Transport: Do Work Later (Async)

Yo Tomáš S.!

Sorry for the slow reply! There's no built-in mechanism inside of Messenger to do this specifically... in part because that's not a normal function of a queueing system. So, you have a few options for this:

A) Delay the message precisely. We know that we can DELAY a message. And so, you could definitely calculate that "tomorrow at 8pm" is, for example, 20 hours away... and then apply a DelayStamp for that amount of time. I can't see any problem with that, other than you need to do the math to figure out how long to do the delay ;).

B) Your other option is to work outside of Messenger. For example, save some record in the database (or some flag) that this messages needs to be sent. Then run a CRON job (e.g. every 30 minutes or 5 minutes... depending on the need) that is constantly looking for "messages that are ready to be sent". We do this kind of thing, for example, to send "your subscription will renew soon" reminders to annual subscribers. This is a case where we can't use Messenger, because it would involve sending the message when the user first subscribes... but delaying it for around 360 days... and also we would need to *remove* that message if the user cancelled. Obviously, not a clean use-case for Messenger :).

Let me know if this helps!

Cheers!

EDIT: I can execute the command bin/console messenger:setup-transports that will create the table :)

Hey Dirk

I'm glad to hear that you could find the solution to your problem. If I recall correctly, Messenger should create that table by default if you have the auto_setup option enabled

Cheers!

Hey ahmedbhs

I believe you're looking for the --sleep option of the messenger:consume command, so you can tell your worker to wait X seconds before fetching new messages after not finding any.

It does not overload the BDD this kind of practice. How we can override that 1 minute ? And is there a good practice about the hole stroy !!

Usually in an integration test, what you test for is if the message was sent or not. And, in an unit test you test the handler's behavior

Cheers!

Hey Taken,

Wow, a lot of questions at once... have you watched this course up to the end? Because I suppose you will get answers on almost all your questions in it later :) OK, a few thoughts about your questions from me:

1. To handle messages you have to run "bin/console messenger:consume-messages" command, where you can specify memory limit with "--memory-limit" option. And this command tracks memory usage, so when the command handles a new message - it will check if the memory limit you specified is reached. As soon as it found the memory limit was hit - it will exit. That's why you need to specify a memory limit lower than your max memory. If you set 128MB memory limit - command will be stopped only when the memory is over this limit. E.g. used memory after 8th message is 121MB, 121 < 128, so the command will start handling a new message, but e.g. used memory after 9th message is 136MB, 136 > 128, so the command will exit. As you see, if you max allowed memory size would be 128 - the 9th message will fail because of free no memory. Usually --memory-limit is 2x lower than max memory limit to avoid "max memory reached" errors.

2. It depends, by following this course you would see when it's better to use messenger. In some cases, your code might become more clean with messenger even if you use it sync, but sometimes it just may complicate things and lead to bugs if you will do it incorrectly. The advantage is that if you don't need the async handling yet, but switched to messenger architecture using sync, you would be easier to switch your project to async message handling later if needed - it's just like change one line of code in your configuration to use a new transport.

3. It depends on your use cases :) If you have 1000 new users every day, and all them you sends welcome emails after registration - that's totally ok to do it sync, like after each successful registration send the email immediately. But if you need at some point send 1000 emails to your subscribers - that's another question, and that's why messenger really may help.

4. Haha, please, watch the course till the end, I bet you will answer this question yourself ;) But basically, maybe you even don't need it at all. You don't have use ALL Symfony components in your project :) Usually, people use messenger to do some heavy operations async, that's the best benefit of using messenger.

5. Once again, it depends. If those tasks are heavy - you may think of doing them async to return your users response faster.

6. If we're talking about messenger component - it depends on which transport you use. If you put some tasks to your queue but use sync transport - messenger will handle the message in the same process, so your users will wait the response until messenger finished processing the message. For example, after user registration you send a welcome email. If you will do it with messenger but with sync transport. User will see your "You're successfully registered" message only when the email is actually been sent. But if you will do it async - messenger will just add that message to the queue and you will show the user "You're successfully registered" message faster, even before the message is actually been sent to them. And another process, that handles those messages will eventually handles that message later and actually sent the email. So as a result, we may consider the 2nd option with async as better user experience as you return the response a little faster to user and then your system handles the remaining work.

I hope this is a bit clearer for you. But I'd really suggest you to watch the course completely to understand the Messenger better.

Cheers!

Hey ahmedbhs

As far as I know Symfony Messenger covers the concurrency problems when using Doctrine (the database) as a message transport. Actually, we use it on Symfonycasts, we have a couple of workers to consume all the messages and we haven't got any issues related to consume the same message twice

Cheers!

Hey Roman,

Interesting question :) Hm, creating a new transport and duplicate messages to it does not mean the messages were successfully handled in other transports, so most probably it's not something you need. Actually, the everything should be logged, so maybe it's enough for you to have those logs? I think you should be able to hook the Messenger, it has some events, you can take a look at them here: https://github.com/symfony/... . Or, as an alternative solution you can create your own middleware and then do whatever you want in it, like storing successfully handled messages in a separate table, etc.

I hope this helps!

Cheers!

Yo @Rémi!

Ah, really cool question. I totally get what you're doing - you want to process the file upload later so the user doesn't have to wait. Hmm, here's how I would do this, at least approximately - and you can let me know what you think ;).

1) Move the temporary-uploaded file somewhere that the worker will have access to. If you have just one server, this is easy: just move it somewhere on your filesystem that is not the temporary directory (because I believe PHP deletes the tmp uploaded file at the end of the request - so that's no good!).

2) Create a new message object that is, sort of, similar to the UploadedFile. It would contain anything you find relevant: like the original filename, the path to where you moved the file on the filesystem... and anything else. This will absolutely be serializable, because it will contain very simple data.

Another option it to put the binary contents of the file itself onto the message class. However, I think you may run into problems if you do this with the Doctrine transport (which, out-of-the-box doesn't like binary content, though you could base64_encode it to get around this). Also, this might slow things down even worse (not sure, just guessing), because now you would need to send the (for example) 2mb of data to your queue and your user would need to wait for this to happen.

Let me know what you think!

Cheers!

Good options ! And I had tested two of them yet... ;)
1) I don't really want to save temporary-uploaded files (whooo,so scary, am I working on secrets stuffs? shhhh!) but i guess I will resignate because as you said PHP delete them at the end of the request. Life is sad but I know this is the right thing to do.
2) Yes, doing this already, it's fine.
3) Mmm agree, on Doctrine transport it'll definitively take too long and i'm working on heavy stuff.

You gave me to think, thank you so much (and I have an other question for you, asking it in the right tuto)

Hey hous04!

Cool question :). Short answer: both are fine. Slightly-less short answer: RabbitMQ is better, faster and more powerful.

Here's an analogy. Suppose you want to cache something. Should you use "filesystem" cache or Redis? For many situations, both are just fine. If you have a huge, heavy-traffic, distributed app, Redis will start to make a difference. Or, if you use some advanced functionality or Redis itself (beyond just using it as a key-value store for data), then you'll need Redis.

The exact same is true for RabbitMQ. It's definitely superior to Doctrine. But, unless you are processing many, many messages, I don't think you would notice much performance difference. Using the Doctrine transport will result in frequent (but fast) queries made to fetch messages... but that's about it. RabbitMQ also has many more advanced features & workflows... which if you don't know about them, then you don't need them... at least not yet ;).

So, stick with Doctrine unless you have the need for more complex use-case, already know Rabbit or are excited to learn and try it.

Let me know if this clears things up!

Cheers!

It's very clear thank you. I'll test both ;)

Hi Again! I was following the videos using my own project, which it was on 4.2.7 symfony version...So, upgrading to 4.3 (as the downloaded code) did the trick. Cheers!

Hey Alex,

Glad you found the problem yourself and was able to fix it! Yeah, each new version might bring new commands, it's a good idea to check your version first if you don't see some of them in the list.

And thank you for your feedback! It might be useful for others ;)

Cheers!

Hey Jeff,

I probably does not understand the your case completely, but if you get a lot of duplicates - it sounds like you're missing validation. I think you should validate the form that sends the data that might be duplicated. The most widely used Symfony constraint for this is the UniqueEntity, see docs: https://symfony.com/doc/cur... . In theory, if you will reject duplicated data on form submit because of invalid form - you won't add duplicated messages to the queue and so it will handle much limited set of "good" messages.

I hope this helps!

Cheers!

Sorry, but the inputs are not generated by a form. They are triggered by various activities and they need to be buffered and deduplicated before a message is dispatched.

Hey Jeff,

Ah, ok then. But in this case I think you can still apply some validation on the server side before dispatching that message. You don't have to work with forms - you can work with symfony validator directly. Here're some examples from the docs: https://symfony.com/doc/cur...

Otherwise, you may probably want different messages... first message handler will deduplicated the messages, and then the 2nd will process already deduplicated messages further - it's like 2 different events (processes). I hope this makes sense to you for your specific case.

I hope this helps!

Cheers!

I'm familiar with Symfony Validator, but I don't think it's the correct tool for this job.

The SQL to delete the duplicate messages is fairly simple:
DELETE t1 FROM messenger_messages t1<br />INNER JOIN messenger_messages t2<br />WHERE t1.id < t2.id<br />AND t1.body = t2.body<br />AND t1.headers = t2.headers<br />AND t1.queue_name = t2.queue_name<br />AND t1.queue_name != 'failed'

What I'm really looking for is a deeper understanding of the Symfony Messenger architecture and how to create a generic dedupe that would work on various transports (not just the database). The deduplication could possibly be done in several different places. I could filter out any duplicates before doing the dispatch, or I could do the normal dispatch with a delay and and then do the deduplication before the event is actually processed. I'm trying to figure out the best place in the code to handle it. It would be nice if I had something like a PRE_EVENT_PROCESS event that could the trigger the dedupe (or message ignore) before the event is actually processed.

It might also be possible to create an IgnoreDuplicateStamp() stamp that could handle things. This stamp would cause messenger to first look for an existing copy of the message and then discard the current message if one is found. In order to make a generic solution I need to figure out how to search for messages in Redis, RabbitMQ and other transports. I don't know if there is existing code in Messenger to do this or whether I would need to write it.

Hey Jeff G.

That's an interesting question. My first recommendation is to avoid duplicating messages from your application, just add appropiate checks before dispatching the message.
You can achieve that by leveraging Messenger middlewares (we talk about them later in the tutorial)
If that's not an option because there is more than one system dispatching messages, then, the logic for detecting duplicates has to happen on the message handling side (once the message is received and it's about to be processed). There's a message event that you can hook into Symfony\Component\Messenger\Event\WorkerMessageReceivedEvent, so you just have to write an event listener and decide if such message should be handled or not

Hope it helps, cheers!

I've been looking into that already. I think that I can trigger things with the SendMessageToTransportEvent event and use that to call a routine that checks for duplicates in the queue. It's fairly simple to write a query to do this when using the messenger_messages table, but I haven't figured out how to do it when using Redis.

I'm afraid you'll need a custom implementation for each service (DB, RabbitMq, Redis, etc). You can add a custom stamp to the message so you can know which service dispatched the message

Is there a way to determine the transport from the message? For example, could I call getEnvelope() and figure out the transport? It appears that most of my messages have the BusNameStamp, so it looks like I can get the bus name stamp, which I can probably link back to a transport.

Or here's an idea that is even simpler...

Since it's unlikely I'll be using multiple transports at same time, I can just use the MESSENGER_TRANSPORT_DSN environment variable. That will give me the transport and I can write some custom code to do the duplicate message lookup for each one. If I really want to do things the correct way, I can create a FindDuplicateMessageInterface class and use that to build classes for each transport I need to support.

An IgnoreDuplicateMessageStamp could determine the transport from MESSAGE_TRANSPORT_DSN and then provide a findDuplicate() function that could be called.

> Is there a way to determine the transport from the message?

I don't think you can tell what transport was used by looking at the message object just out of the box. I think you can add a TransportNameStamp to your messages if you really need that flexibility, but if you're going to use only one transport, then, you can simplify things by assuming the transport.

Cheers!

I'll add an option to my IgnoreDuplicateMessageStamp class that will let you specify the transport. If not set, it will default to the one specified by MESSAGE_TRANSPORT_DSN. I think that should cover things in case I decide to use multiple transports.

Sounds good! However, as a personal recommendation I suggest to avoid adding unnecessary flexibility, it's always better to wait for a real use-case than invest early in something that you don't know if you'll ever use.

Yes, that's always a trade-off (have been a programmer for 40+ years). At least my architecture should support it.

Wow, 40 years, congrats that's not easy to say! Keep up the good work

Cheers!

Actually longer, if I count the time before graduating from university. I think I wrote my first program in 1970. Still enjoy it. Working at a startup now with other programmers half my age.