SymfonyCasts stands united with the people of Ukraine
gstreamer0.10-ffmpeg
gstreamer0.10-plugins-good packages.
So far, we've separated the instructions of what we want to do - we want to add Ponka to this ImagePost - from the logic that actually does that work. And... it's a nice coding pattern: it's easy to test and if we need to add Ponka to an image from anywhere else in our system, it will be super pleasant.
But this pattern unlocks some serious possibilities. Think about it: now that we've isolate the instructions on what we want to do, instead of handling the command object immediately, couldn't we, in theory, "save" that object somewhere... then read and process it later? That's... basically how a queuing system works. The advantage is that, depending on your setup, you could put less load on your web server and give users a faster experience. Like, right now, when a user clicks to upload a file, it takes a few seconds before it finally pops over here. It's not the biggest deal, but it's not ideal. If we can fix that easily, why not?
In Messenger, the key to "saving work for later" is a system called transports. Open up config/packages/messenger.yaml. See that transports key? The details are actually configured in .env.
Here's the idea: we're going to say to Messenger:
Yo! When I create an
AddPonkaToImageobject, instead of handling it immediately, I want you to send it somewhere else.
That "somewhere else" is a transport. And a transport is usually a "queue". If you're new to queueing, the idea is refreshingly simple. A queue is an external system that "holds" onto information in a big list. In our case, it will hold onto serialized message objects. When we send it another message, it adds it to the list. Later, you can read those messages from the queue one-by-one, handle them and, when you're done, the queue will remove it from the list.
Sure... robust queuing systems have a lot of other bells and whistles... but that really is the main concept.
There are a bunch of queueing systems available, like RabbitMQ, Amazon SQS, Kafka, and queueing at the supermarket. Out-of-the box, Messenger supports three: amqp - which basically means RabbitMQ, but technically means any system that implements the "AMQP" spec - doctrine and redis. AMQP is the most powerful... but unless you're already a queueing pro and want to do something crazy, these all work exactly the same.
Oh, and if you need to talk to some unsupported transport, Messenger integrates with another library called Enqueue, which supports a bunch more.
Because I'm already using Doctrine in this project, let's use the doctrine transport. Uncomment the environment variable for that.
| ... lines 1 - 29 | |
| ###> symfony/messenger ### | |
| ... lines 31 - 32 | |
| MESSENGER_TRANSPORT_DSN=doctrine://default | |
| ... line 34 | |
| ### |
See this ://default part? That tells the Doctrine transport that we want to use the default Doctrine connection. Yep, it'll re-use the connection you've already set up in your app to store the message inside a new table. More on that soon.
Tip
Starting in symfony 5.1, the code behind the Doctrine transport was moved to its own package.
The only difference is that you should now also run this command:
composer require symfony/doctrine-messenger
Now, back in messenger.yaml, uncomment this async transport, which uses that MESSENGER_TRANSPORT_DSN environment variable we just created. The name - async - isn't important - that could be anything. But, in a second, we'll start referencing that name.
| framework: | |
| messenger: | |
| ... lines 3 - 5 | |
| transports: | |
| # https://symfony.com/doc/current/messenger.html#transports | |
| async: '%env(MESSENGER_TRANSPORT_DSN)%' | |
| ... lines 9 - 16 |
At this point... yay! We've told Messenger that we have an async transport. And if we want back and uploaded a file now, it would... make absolutely no difference: it would still be processed immediately. Why?
Because we need to tell Messenger that this message should be sent to that transport, instead of being handled right now.
Back in messenger.yaml, see this routing key? When we dispatch a message, Messenger looks at all of the classes in this list... which is zero right now if you don't count the comment... and looks for our class - AddPonkaToImage. If it doesn't find the class, it handles the message immediately.
Let's tell Messenger to instead send that to the async transport. Set App\Message\AddPonkaToImage to async.
| framework: | |
| messenger: | |
| ... lines 3 - 12 | |
| routing: | |
| # Route your messages to the transports | |
| 'App\Message\AddPonkaToImage': async |
As soon as we do that... it makes a huge difference. Watch how fast the image loads on the right after uploading. Boom! That was faster than before and... Ponka isn't there! Gasp!
Actually, let's try one more - that first image was a little bit slow because Symfony was rebuilding its cache. This one should be nearly instant. It is! Instead of calling our handler immediately, Messenger is sending our message to the Doctrine transport.
And... um... what does that actually mean? Find your terminal... or whatever tool you like to use to play with databases. I'll use the mysql client to connect to the messenger_tutorial database. Inside, let's:
SHOW TABLES;
Woh! We expected migration_versions and image_post... but suddenly we have a third table called messenger_messages. Let's see what's in there:
SELECT * FROM messenger_messages;
Nice! It has two rows for our two messages! Let's use the magic \G to format this nicer:
SELECT * FROM messenger_messages \G
Cool! The body holds our object: it's been serialized using PHP's serialize() function... though that can be configured. The object is wrapped inside something called an Envelope... but inside... we can see our AddPonkaToImage object and the ImagePost inside of that... complete with the filename, createdAt date, etc.
Wait... but where did this table come from? By default, if it's not there, Messenger creates it for you. If you don't want that, there's a config option called auto_setup to disable this - I'll show you how later. If you did disable auto setup, you could then use the handy setup-transports command on deploy to create that table for you.
php bin/console messenger:setup-transports
This doesn't do anything now... because the table is already there.
Hey! This was a huge step! Whenever we upload images... they are not being handled immediately: when we upload two more... they're being sent to Doctrine and it is keeping track of them. Thanks Doctrine!
Next, it's time to read those messages one-by-one and start handling them. We do that with a console command called a "worker".
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!
Hi,
for some reason the database table messenger_messages does not get created automatically. I tried generated the statements with make:migration but to no avail. I'm using Symfony 4.4. Any idea what I'm doing wrong?
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!
Hi, thanks for this tutorial!
Doctrine transport is good no need to install more dependencies.
But I'm wondering about one thing message under my logs, I think each 1 minute (not sure):
SELECT m.* FROM messenger_messages m WHERE (m.delivered_at is null OR
m.delivered_at < ?) AND (m.available_at <= ?) AND (m.queue_name = ?) ORDER BY available_at ASC LIMIT 1 FOR UPDATE ["2020-09-15 14:04:21","2020-09-15 15:04:21","default"] []
So each 1 minute I'm calling the database to fetch messages.
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 !!
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!
Helo guys i'm new to symfony cast, thank you for the amazing tuto:
1/ By what criteria Symfony messenger knows that this is the right time to process the following message. How the broker knows that there is some memory available and that the handler has finished processing all these messages to go to the next message! How ke broker knows that the handler has finished his job. I'm afraid that in some case it can lead to memory problems.
2/ Is it good way to decouple code via messenger or event ubscriber is the traitment is synchrone?
3/ What is the best way to send 1000 emails per a day.
4/ Is there some usescase where it's not necessary to use messenger ! I still dont understand why why use symfony messenger !
5/ For example, do we use the messenger when we are treating 3 dependent tasks to each other or not ?
6/ Is queing système is synchronous traitment or asynchronous ? I mean queing is also synchronous he trait thing in FIFO mode. It's clear that the user gonna not wait for the response and that help for user expirience.
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!
Hi Rayen thank you so mush for the tuto,
* There's some point of view that saying "a database is not a message queue." is this is true. I'm sure that, a database makes it very easy to centralize information, and we know very well how to save and redundate it.
A lot of online recommendation say that we should not use a database as a message queue because problems can be appeared during concurrency, for exemple: if we made a SELECT request to retrieve the identifier of the next task to be processed, then an UPDATE request to indicate that this task is being treatment ; obviously, if two programs do this at the same time, the task may be processed twice.
* The fact remains that it requires polling data Fromm database for each task, that is to say, scanning the database very regularly to see if there are new tasks waiting. It is true that this is not ideal, because it unnecessarily loads the base, but above all implies latency in the treatments
I know that doctrine transport is amazing for development experience. But I'm wondring is symfony core team already solved the's problems.
PS: This is the slide that talk about we should not use database as a queue système https://fr.slideshare.net/R...
Thank you!
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!
Hi! Is there a way to save all the messages even if they were consumed successfully? It may be quite useful. For example, I want to store history about all messages.
This moment I can guess that I can create my own Transport, but maybe there is a more elegant and simple way?
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!
Hello ! Thanks for this tuto (amazing work!),
I faced "Serialization of 'Symfony\Component\HttpFoundation\File\UploadedFile' is not allowed", trying to process upload files but without saving them on hosting? The goal is to merge all of this stuff asynchronously, and avoid waiting time for users, but I really want to avoid saving...
Actually my handler works perfectly while messenger is synchronous but find the error on async. Could you help me? <3
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)
Hi, thanks for this tutorial.
I have created some messages and I used doctrine transport like in this tutorial , but I want to ask you as you are an expert in symfony , better using doctrine transport or rabbitmq ? which one is the most efficient and faster ? Thanks
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!
Hi! Right now when I type bin/console messenger, I only see debug:messenger and messenger:consume-messages, so I'm missing the rest of commands, any idea why? Thanks!
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!
I used this tutorial to build a queueing system about two years ago and it worked OK in the beginning. The system receives inputs from several different sources and a lot of the messages are duplicates. As we grow, the duplicate messages are causing problems because our server communicates with another system via an API and they limit our service if the the number of duplicate requests exceeds a certain threshold. The more duplicates we receive, the longer the queue gets and the longer it takes to process them all.
I'm trying to figure out how to detect duplicate messages and avoid dispatching them. Right now I've developed a hack that runs from cron once every minute and deletes the duplicates from the database. However, this is just a workaround only works if I'm using the database to store messages. I'd like create a generic solution that would work with other transports like RabbitMQ or Redis.
I've been toying with idea of using the `DelayStamp` to force the messages to be stored in queue (say for 30 seconds) before anything happens. Then I could create a "deduplication" process that would continually run in the background and check the queue for duplicate messages (like every 10 seconds). This basically what my SQL script hack is doing right now, but I'd like to create a generic solution that would work with other transports.
I've started looking through the messenger library code, but am not sure where to start. This might be a feature that could be useful to others and I don't mind contributing to the project.
Any ideas on how to do this?
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.
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.
// composer.json
{
"require": {
"php": "^7.1.3",
"ext-ctype": "*",
"ext-iconv": "*",
"composer/package-versions-deprecated": "^1.11", // 1.11.99
"doctrine/annotations": "^1.0", // v1.8.0
"doctrine/doctrine-bundle": "^1.6.10", // 1.11.2
"doctrine/doctrine-migrations-bundle": "^1.3|^2.0", // v2.0.0
"doctrine/orm": "^2.5.11", // v2.6.3
"intervention/image": "^2.4", // 2.4.2
"league/flysystem-bundle": "^1.0", // 1.1.0
"phpdocumentor/reflection-docblock": "^3.0|^4.0", // 4.3.1
"sensio/framework-extra-bundle": "^5.3", // v5.3.1
"symfony/console": "4.3.*", // v4.3.2
"symfony/dotenv": "4.3.*", // v4.3.2
"symfony/flex": "^1.9", // v1.18.7
"symfony/framework-bundle": "4.3.*", // v4.3.2
"symfony/messenger": "4.3.*", // v4.3.4
"symfony/property-access": "4.3.*", // v4.3.2
"symfony/property-info": "4.3.*", // v4.3.2
"symfony/serializer": "4.3.*", // v4.3.2
"symfony/validator": "4.3.*", // v4.3.2
"symfony/webpack-encore-bundle": "^1.5", // v1.6.2
"symfony/yaml": "4.3.*" // v4.3.2
},
"require-dev": {
"easycorp/easy-log-handler": "^1.0.7", // v1.0.7
"symfony/debug-bundle": "4.3.*", // v4.3.2
"symfony/maker-bundle": "^1.0", // v1.12.0
"symfony/monolog-bundle": "^3.0", // v3.4.0
"symfony/stopwatch": "4.3.*", // v4.3.2
"symfony/twig-bundle": "4.3.*", // v4.3.2
"symfony/var-dumper": "4.3.*", // v4.3.2
"symfony/web-profiler-bundle": "4.3.*" // v4.3.2
}
}
Hello,
I want to ask how to delay consumption for concrete message at specific date and time. Lets assume the message sends an email and I want to do at tommorrow at 8pm.
Cheers Tomas