[Messenger] ReceiverInterface::handle() to get() & Worker with prioritized transports #30708

weaverryan · 2019-03-26T13:16:28Z

Q	A
Branch?	master
Bug fix?	no
New feature?	yes
BC breaks?	no
Deprecations?	no
Tests pass?	yes
Fixed tickets	Helps with #30699
License	MIT
Doc PR	TODO

Highlights:

messenger:consume can now consume messages from multiple transports with priority ❗️

bin/console messenger:consume amqp_high amqp_medium amqp_low

How long you want to sleep before checking more messages is now an option to messenger:consume
ReceiverInterface::receive() is replaced with ReceiverInterface::get()
Logic for looping & sleeping is moved into Worker

weaverryan · 2019-03-27T03:18:16Z

src/Symfony/Component/Messenger/Worker.php

-                }
+                $handled(null);
+
+                usleep(1000000);


Two changes here:

I changed from .2 seconds to 1 second. Waiting only 200ms between when there is not a message is quite rapid. For reference, in Laravel, this defaults to 3 and is configurable on the command/worker.

weaverryan · 2019-03-27T03:26:53Z

This is ready!

tl;dr In recent changes, the ReceiverInterface is not really the layer that's handling the "looping and waiting for messages" layer - the Worker is doing that. This completes that transformation. It will also allow us to create a Worker that listens to multiple queues at once, in a priority order (#30699).

src/Symfony/Component/Messenger/CHANGELOG.md

sroze · 2019-03-27T12:35:04Z

The reasoning of having handle($handler) was for the transports to be able to catch the exceptions from the handlers and deal with the errors (i.e. ack, nack, etc...). Since we moved this to the worker, I'm happy to move back to get one message. Though, why wouldn't we return an iterable instead to match the behaviour of prefetching (i.e. some transports would have loaded X messages, not just one)?

src/Symfony/Component/Messenger/WorkerInterface.php

src/Symfony/Component/Messenger/Worker/StopWhenTimeLimitIsReachedWorker.php

src/Symfony/Component/Messenger/Worker/StopWhenMessageCountIsExceededWorker.php

weaverryan · 2019-03-28T03:18:30Z

src/Symfony/Component/Messenger/Transport/Receiver/ReceiverInterface.php

-     * Stop receiving some messages.
-     */
-    public function stop(): void;
+    public function get(): iterable;


The iterable, like always, was kind of a pain, compared to just returning one. But, it is more flexible and end-users won't touch this method much anyways.

src/Symfony/Component/Messenger/Worker.php

weaverryan · 2019-03-28T03:57:14Z

This is ready for review! Description updated.

Better viewed (especially the Worker with ?w=1). The 3 StopWhen*** classes are not new - but were moved & changed enough (they decorate WorkerInterface instead of ReceiverInerface so they look new.

Nyholm

I really like this.

I am not a fan of priority in a message system, but this is priority of transports. That is a different thing. Im all 👍
(after my comments are addressed/answered)

src/Symfony/Component/Messenger/Worker.php

Nyholm

Good. I’m happy with the updated doc block on ReceiverInterface::get().

Thank you

fabpot

minor typos

src/Symfony/Component/Messenger/Transport/Receiver/ReceiverInterface.php

src/Symfony/Component/Messenger/WorkerInterface.php

…tized transports

fabpot · 2019-03-30T07:01:39Z

Thank you @weaverryan.

…ker with prioritized transports (weaverryan) This PR was squashed before being merged into the 4.3-dev branch (closes #30708). Discussion ---------- [Messenger] ReceiverInterface::handle() to get() & Worker with prioritized transports | Q | A | ------------- | --- | Branch? | master | Bug fix? | no | New feature? | yes | BC breaks? | no | Deprecations? | no | Tests pass? | yes | Fixed tickets | Helps with #30699 | License | MIT | Doc PR | TODO Highlights: * `messenger:consume` can now consume messages from multiple transports with priority ❗️ ``` bin/console messenger:consume amqp_high amqp_medium amqp_low ``` * How long you want to sleep before checking more messages is now an option to `messenger:consume` * `ReceiverInterface::receive()` is replaced with `ReceiverInterface::get()` * Logic for looping & sleeping is moved into `Worker` Commits ------- e800bd5 [Messenger] ReceiverInterface::handle() to get() & Worker with prioritized transports

fabpot · 2019-03-30T07:02:43Z

typos fixed in f4176b0

Tobion · 2019-03-30T17:54:03Z

src/Symfony/Component/Messenger/Transport/Receiver/ReceiverInterface.php

@@ -25,21 +25,25 @@ interface ReceiverInterface
    /**
     * Receive some messages to the given handler.


This doc is outdated: There is no "given handler"

@Tobion

…erInterface::get` phpdoc. (sroze) This PR was merged into the 4.3-dev branch. Discussion ---------- [Messenger] Remove the mention of handler in the `ReceiverInterface::get` phpdoc. | Q | A | ------------- | --- | Branch? | master | Bug fix? | no | New feature? | no | BC breaks? | no | Deprecations? | no | Tests pass? | yes | Fixed tickets | #30708 (review) | License | MIT | Doc PR | ø As spotted by @Tobion, we don't have an handler as an argument anymore. Commits ------- 9c63112 Remove the mention of handler in the phpdoc.

…ryan) This PR was squashed before being merged into the 4.3-dev branch (closes #30754). Discussion ---------- [Messenger] New messenger:stop-workers Command | Q | A | ------------- | --- | Branch? | master | Bug fix? | no | New feature? | yes | BC breaks? | no | Deprecations? | no | Tests pass? | yes | Fixed tickets | Kinda of #29451 | License | MIT | Doc PR | symfony/symfony-docs#11236 o/ me again. This requires and is built on top of #30708 When you deploy, all workers need to be stopped and restarted. That's not currently possible, unless you manually track the pids and send a SIGTERM signal. We can make that much easier :). Now run: ``` bin/console messenger:stop-workers ``` And it will signal to all workers (even if they're distributed on other servers) that they should stop, once they finish processing their current message. This is done via a key in `cache.app`. Cheers! Commits ------- 5897162 [Messenger] New messenger:stop-workers Command

Tobion · 2019-05-06T02:00:55Z

Just thinking out loud for reference: The prioritized transports can be used to prioritize messages by publishing messages to different transports and consuming them in order. But it's a different way than rabbitmq uses message priority. If you want to use rabbitmq priorities instead, you can do that by setting the priority attribute on the AmqpStamp as introduced in #30913

weaverryan requested a review from sroze as a code owner March 26, 2019 13:16

carsonbot added Status: Needs Review Messenger Feature labels Mar 26, 2019

weaverryan force-pushed the receiver-no-handler branch from e7c770a to d5fd552 Compare March 26, 2019 13:17

nicolas-grekas added this to the next milestone Mar 26, 2019

weaverryan force-pushed the receiver-no-handler branch 5 times, most recently from 2ee34de to 357cdb9 Compare March 27, 2019 03:16

weaverryan commented Mar 27, 2019

View reviewed changes

weaverryan force-pushed the receiver-no-handler branch from 357cdb9 to 2d1f17b Compare March 27, 2019 03:25

fabpot reviewed Mar 27, 2019

View reviewed changes

src/Symfony/Component/Messenger/CHANGELOG.md Outdated Show resolved Hide resolved

weaverryan force-pushed the receiver-no-handler branch from 2d1f17b to fa4776f Compare March 27, 2019 11:45

weaverryan mentioned this pull request Mar 27, 2019

[WIP][Messenger] Multiple queue support + prioritization #30699

Closed

chalasr reviewed Mar 27, 2019

View reviewed changes

This was referenced Mar 28, 2019

[Messenger][4.3] Tracker for changes symfony/symfony-docs#11236

Closed

[Messenger] Making it Shine #30540

Closed

weaverryan force-pushed the receiver-no-handler branch 2 times, most recently from 3864a11 to 8866804 Compare March 28, 2019 03:14

weaverryan changed the title ~~[Messenger] Changes ReceiverInterface::handle() to get()~~ [Messenger] ReceiverInterface::handle() to get() & Worker with prioritized transports Mar 28, 2019

weaverryan commented Mar 28, 2019

View reviewed changes

weaverryan force-pushed the receiver-no-handler branch from 8866804 to 9c9c4e0 Compare March 28, 2019 03:24

weaverryan mentioned this pull request Mar 28, 2019

[Messenger] New messenger:stop-workers Command #30754

Merged

Nyholm requested changes Mar 29, 2019

View reviewed changes

src/Symfony/Component/Messenger/Worker.php Show resolved Hide resolved

src/Symfony/Component/Messenger/Worker.php Outdated Show resolved Hide resolved

src/Symfony/Component/Messenger/Worker.php Show resolved Hide resolved

src/Symfony/Component/Messenger/Worker.php Show resolved Hide resolved

carsonbot removed the Status: Needs Review label Mar 29, 2019

carsonbot added the Status: Needs Work label Mar 29, 2019

weaverryan force-pushed the receiver-no-handler branch from f5ebe36 to 8356476 Compare March 29, 2019 19:12

Nyholm approved these changes Mar 29, 2019

View reviewed changes

carsonbot added Status: Reviewed and removed Status: Needs Work labels Mar 29, 2019

fabpot approved these changes Mar 30, 2019

View reviewed changes

src/Symfony/Component/Messenger/Transport/Receiver/ReceiverInterface.php Show resolved Hide resolved

src/Symfony/Component/Messenger/WorkerInterface.php Show resolved Hide resolved

src/Symfony/Component/Messenger/WorkerInterface.php Show resolved Hide resolved

[Messenger] ReceiverInterface::handle() to get() & Worker with priori…

e800bd5

…tized transports

fabpot force-pushed the receiver-no-handler branch from 8356476 to e800bd5 Compare March 30, 2019 07:01

fabpot merged commit e800bd5 into symfony:master Mar 30, 2019

Tobion reviewed Mar 30, 2019

View reviewed changes

sroze mentioned this pull request Mar 31, 2019

[Messenger] Remove the mention of handler in the ReceiverInterface::get phpdoc. #30793

Merged

weaverryan deleted the receiver-no-handler branch March 31, 2019 14:42

nicolas-grekas modified the milestones: next, 4.3 Apr 30, 2019

fabpot mentioned this pull request May 9, 2019

Release v4.3.0-BETA1 #31435

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Messenger] ReceiverInterface::handle() to get() & Worker with prioritized transports #30708

[Messenger] ReceiverInterface::handle() to get() & Worker with prioritized transports #30708

weaverryan commented Mar 26, 2019 •

edited

Loading

weaverryan Mar 27, 2019

weaverryan commented Mar 27, 2019 •

edited by ogizanagi

Loading

sroze commented Mar 27, 2019

weaverryan Mar 28, 2019

weaverryan commented Mar 28, 2019

Nyholm left a comment

Nyholm left a comment

fabpot left a comment

fabpot commented Mar 30, 2019

fabpot commented Mar 30, 2019

Tobion Mar 30, 2019

Tobion commented May 6, 2019

		@@ -25,21 +25,25 @@ interface ReceiverInterface
		/**
		* Receive some messages to the given handler.

[Messenger] ReceiverInterface::handle() to get() & Worker with prioritized transports #30708

[Messenger] ReceiverInterface::handle() to get() & Worker with prioritized transports #30708

Conversation

weaverryan commented Mar 26, 2019 • edited Loading

weaverryan Mar 27, 2019

Choose a reason for hiding this comment

weaverryan commented Mar 27, 2019 • edited by ogizanagi Loading

sroze commented Mar 27, 2019

weaverryan Mar 28, 2019

Choose a reason for hiding this comment

weaverryan commented Mar 28, 2019

Nyholm left a comment

Choose a reason for hiding this comment

Nyholm left a comment

Choose a reason for hiding this comment

fabpot left a comment

Choose a reason for hiding this comment

fabpot commented Mar 30, 2019

fabpot commented Mar 30, 2019

Tobion Mar 30, 2019

Choose a reason for hiding this comment

Tobion commented May 6, 2019

weaverryan commented Mar 26, 2019 •

edited

Loading

weaverryan commented Mar 27, 2019 •

edited by ogizanagi

Loading