[Messenger] Add a way to no ack message automatically #42873

lyrixx · 2021-09-03T10:50:55Z

Q	A
Branch?	5.4
Bug fix?	no
New feature?	yes
Deprecations?	no
Tickets	Fix #36910
License	MIT
Doc PR

This PR add a way to manually controller auto ACK from the handler.
It open the door to buffering.

Here is the code one must write to buffer message, and process the buffer each 10s. No more code is needed!

<?php

namespace App\MessageHandler;

use App\Message\HelloMessage;
use Psr\Container\ContainerInterface;
use Psr\Log\LoggerInterface;
use Symfony\Component\EventDispatcher\EventSubscriberInterface;
use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\Event\WorkerRunningEvent;
use Symfony\Component\Messenger\Event\WorkerStoppedEvent;
use Symfony\Component\Messenger\Handler\ConfigureAutoAckInterface;
use Symfony\Component\Messenger\Handler\MessageHandlerInterface;
use Symfony\Component\Messenger\Stamp\ReceivedStamp;
use Symfony\Component\Messenger\Transport\Receiver\ReceiverInterface;

final class HelloMessageHandler implements MessageHandlerInterface, ConfigureAutoAckInterface, EventSubscriberInterface
{
    private $buffer = [];
    private $lastFlushedAt = null;

    public function __construct(
        private ContainerInterface $receiverLocator,
        private LoggerInterface $logger,
    ) {
        $this->lastFlushedAt = time();
    }

    public function __invoke(HelloMessage $message, Envelope $envelope)
    {
        if ($this->isAutoAckDisabled($envelope)) {
            $this->logger->info('Add message to buffer.');
            $this->buffer[] = $envelope;
            $this->flushIfNeeded();

            return;
        }

        $this->logger->info('process message.');

        // Do regular processing
    }

    public function isAutoAckDisabled(Envelope $envelope): bool
    {
        // This handler could be used by many transports.
        // But only the async one should be manually controlled // bufferised 
        return $envelope->last(ReceivedStamp::class)->getTransportName() === 'async';
    }

    public function flushIfNeeded()
    {
        $this->logger->info('Flush buffer if needed.');
        if (time() < $this->lastFlushedAt + 10) {
            return;
        }
        $this->flush()
    }

    public function flush()
    {
        $this->logger->info('Flush buffer.');

        $this->lastFlushedAt = time();

        // Do your custom processing on the buffer

        try {
            foreach ($this->buffer as $envelope) {
                /** @var ReceiverInterface */
                $receiver = $this->receiverLocator->get($envelope->last(ReceivedStamp::class)->getTransportName());
                $receiver->ack($envelope);
            }
        } finally {
            $this->buffer = [];
        }
    }

    public static function getSubscribedEvents()
    {
        return [
            WorkerRunningEvent::class =>  'flushIfNeeded',
            WorkerStoppedEvent::class =>  'flush',
        ];
    }
}

# services.yaml
services:
    App\MessageHandler\HelloMessageHandler:
        arguments:
            $receiverLocator: '@messenger.receiver_locator'

sponsored by arte.tv

carsonbot · 2021-09-04T06:53:33Z

Hey!

I think @tyx has recently worked with this code. Maybe they can help review this?

Cheers!

Carsonbot

jderusse

I like this feature!.

Beware in your example, flushing in not an option on consumer stop

-WorkerStoppedEvent::class =>  'flushIfNeeded',
+WorkerStoppedEvent::class =>  'forceFlush',

src/Symfony/Component/Messenger/Handler/ConfigureAutoAckInterface.php

src/Symfony/Component/Messenger/Middleware/HandleMessageMiddleware.php

src/Symfony/Component/Messenger/Handler/ConfigureAutoAckInterface.php

lyrixx · 2021-09-22T08:47:04Z

Hello @symfony/mergers

The initial issue got 8 👍🏼 and, this PR got 5 👍🏼 and @jderusse liked this feature.

So without feedback, I'm planning to merge this PR by the end of the month, before the feature freeze.
If you are against it, it's time to raise your voice :)

okwinza · 2021-09-22T09:02:55Z

This is indeed a feature I would like to see in 5.4. 👍

Btw, @lyrixx any chance you could also take a look at my #42335 before the feature freeze is upon us? 😅

lyrixx · 2021-10-01T09:09:29Z

I have rebased the PR (to fix conflict on CHANGELOG.md file)

nicolas-grekas

This makes handlers aware of the transport layer, which is something that we refused to do as of now, see eg #42005.
By doing so, this approach is also giving the responsibility of buffering to handlers. I'm wondering if handlers are the right place to do that. I don't think so for now.

Instead, what about adding a middleware that does the buffering, and communicates with the worker via a stamp as done here? Buffering would be enabled/disabled by either setting a limit on the number of messages per batch, or by injecting a buffering decision strategy if we want something super flexible.

BTW, how are we sure that messages are flushed eventually, when we don't know when messages can arrive? Should we inject a delayed message that would trigger the flush if nothing happened before it?

About naming, "DelayedAckStamp" made me think: "delayed? when?". Maybe this should be named "DisableAutoAckStamp"?

lyrixx · 2021-10-06T10:16:54Z

Instead, what about adding a middleware that does the buffering, and communicates with the worker via a stamp as done here? Buffering would be enabled/disabled by either setting a limit on the number of messages per batch, or by injecting a buffering decision strategy if we want something super flexible.

Hmm, indeed it might be possible. But ATM, I fail to see how the middleware could actually call the "code that process the buffer". Should we introduce a new method (via an interface) on the handler (Like it's done on monolog handler)?

So basically, you want me to move the the ConfigurableAutoAckInterface::isAutoAckDisabled() to a middleware instead of the handler itself ? (I'm asking to be sure I fully understand)

BTW, how are we sure that messages are flushed eventually, when we don't know when messages can arrive? Should we inject a delayed message that would trigger the flush if nothing happened before it?

As you can see in the PR body, user need to register also a listener but a timer to flush after a configured time. But this could leave in the middleware

About naming, "DelayedAckStamp" made me think: "delayed? when?". Maybe this should be named "DisableAutoAckStamp"?

I agree 👍🏼

But I see a drawback with your idea: a new middleware implies a new bus because one doesn't want to buffer all message: https://symfony.com/doc/current/messenger.html#middleware

So publishing a message will be more complexe because one will have to choose the right buses when dispatching the message. IMHO, it's error prone.

So you want to move the "handlers [is] aware of the transport layer" to the publisher is aware of the transport layer. Is it really better?

nicolas-grekas · 2021-10-06T11:56:20Z

how the middleware could actually call the "code that process the buffer".

can't the middleware call the handler many times in a loop, once per buffered message?
the middleware would do that when the Nth message arrives (aka when the buffer is full).

So basically, you want me to move the the ConfigurableAutoAckInterface::isAutoAckDisabled() to a middleware instead of the handler itself ?

I think we won't need any new interface if we do this. The stamp might be enough.

a new middleware implies a new bus because one doesn't want to buffer all message

we could have a config option that gives the size of the buffer, and if the size is > 1, then we add this middleware? (or we remove it if it's <= 1? or we make it a no-op if we can't skip registering it).

lyrixx · 2021-10-06T12:00:38Z

can't the middleware call the handler many times in a loop, once per buffered message?
the middleware would do that when the Nth message arrives (aka when the buffer is full).

nope, because you don't know how it's the "last message". You need all messages at once.

I think we won't need any new interface if we do this. The stamp might be enough.

You don't have access to the stamp in the handler. Adding it is what you don't want.

we could have a config option that gives the size of the buffer, and if the size is > 1, then we add this middleware? (or we remove it if it's <= 1? or we make it a no-op if we can't skip registering it).

I don't understand. Middlerware a setup for a busses. not for a transport or handler. See the doc to better understand.

nicolas-grekas

From the PR description:

__invoke(HelloMessage $message, Envelope $envelope)

~~the envelope is not passed as 2nd arg so this doesn't work~~ it would be in this PR :)
to be well thought as we rejected doing so in the past

$envelope->last(ReceivedStamp::class)->getTransportName() === 'async';

hardcoding the name of the transport in the class should be really avoided

I'm putting this on hold while looking for alternative approaches.

nicolas-grekas · 2021-10-06T21:23:20Z

Here is another proposal on the topic: #43354

nicolas-grekas · 2021-10-27T19:32:20Z

Closing in favor of #43354, thanks for pushing this @lyrixx!

…las-grekas) This PR was merged into the 5.4 branch. Discussion ---------- [Messenger] allow processing messages in batches | Q | A | ------------- | --- | Branch? | 5.4 | Bug fix? | no | New feature? | yes | Deprecations? | no | Tickets | #36910 | License | MIT | Doc PR | - This replaces #42873 as it proposes an alternative approach to handling messages in batch. `BatchHandlerInterface` says it all: if a handler implements this interface, then it should expect a new `$ack` optional argument to be provided when `__invoke()` is called. When `$ack` is not provided, `__invoke()` is expected to handle the message synchronously as usual. But when `$ack` is provided, `__invoke()` is expected to buffer the message and its `$ack` function, and to return the number of pending messages in the batch. Batch handlers are responsible for deciding when they flush their buffers, calling the `$ack` functions while doing so. Best reviewed [ignoring whitespaces](https://github.com/symfony/symfony/pull/43354/files?w=1). Here is what a batch handler might look like: ```php class MyBatchHandler implements BatchHandlerInterface { use BatchHandlerTrait; public function __invoke(MyMessage $message, Acknowledger $ack = null) { return $this->handle($message, $ack); } private function process(array $jobs): void { foreach ($jobs as [$job, $ack]) { try { // [...] compute $result from $job $ack->ack($result); } catch (\Throwable $e) { $ack->nack($e); } } } } ``` By default, `$jobs` contains the messages to handle, but it can be anything as returned by `BatchHandlerTrait::schedule()` (eg a Symfony HttpClient response derived from the message, a promise, etc.). The size of the batch is controlled by `BatchHandlerTrait::shouldProcess()` (defaults to 10). The transport is acknowledged in batch, *after* the bus returned from dispatching (unlike what is done in #42873). This is especially important when considering transactions since we don't want to ack unless the transaction committed successfully. By default, pending batches are flushed when the worker is idle and when it stops. Commits ------- 81e52b2 [Messenger] allow processing messages in batches

lyrixx requested a review from sroze as a code owner September 3, 2021 10:50

carsonbot added Status: Needs Review Feature Messenger labels Sep 3, 2021

lyrixx mentioned this pull request Sep 3, 2021

[Messenger] Consume batch of messages (manually, instead of worker), consuming multiple messages at once, BULK message processing at once #36910

Closed

lyrixx force-pushed the messenger-batch branch from 89bb008 to 682e7ca Compare September 3, 2021 12:30

lyrixx added this to the 5.4 milestone Sep 3, 2021

jderusse reviewed Sep 4, 2021

View reviewed changes

src/Symfony/Component/Messenger/Handler/ConfigureAutoAckInterface.php Outdated Show resolved Hide resolved

src/Symfony/Component/Messenger/Middleware/HandleMessageMiddleware.php Outdated Show resolved Hide resolved

lyrixx force-pushed the messenger-batch branch from 682e7ca to a77c2c5 Compare September 6, 2021 15:30

lyrixx requested review from jderusse, chalasr, Tobion and xabbuh September 7, 2021 12:47

okwinza reviewed Sep 11, 2021

View reviewed changes

src/Symfony/Component/Messenger/Handler/ConfigureAutoAckInterface.php Outdated Show resolved Hide resolved

lyrixx force-pushed the messenger-batch branch from a77c2c5 to 934dde6 Compare September 13, 2021 09:12

okwinza mentioned this pull request Sep 13, 2021

[Messenger] Pass envelope to message handler as second __invoke parameter #42005

Closed

lyrixx requested a review from fabpot September 15, 2021 13:27

[Messenger] Add a way to no ack message automatically

4b255b0

lyrixx force-pushed the messenger-batch branch from 934dde6 to 4b255b0 Compare October 1, 2021 09:09

nicolas-grekas reviewed Oct 6, 2021

View reviewed changes

nicolas-grekas requested changes Oct 6, 2021

View reviewed changes

carsonbot removed the Status: Needs Review label Oct 6, 2021

carsonbot added the Status: Needs Work label Oct 6, 2021

nicolas-grekas mentioned this pull request Oct 6, 2021

[Messenger] allow processing messages in batches #43354

Merged

nicolas-grekas closed this Oct 27, 2021

lyrixx deleted the messenger-batch branch October 28, 2021 07:15

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Messenger] Add a way to no ack message automatically #42873

[Messenger] Add a way to no ack message automatically #42873

lyrixx commented Sep 3, 2021 •

edited

Loading

carsonbot commented Sep 4, 2021

jderusse left a comment

lyrixx commented Sep 22, 2021

okwinza commented Sep 22, 2021

lyrixx commented Oct 1, 2021

nicolas-grekas left a comment

lyrixx commented Oct 6, 2021

nicolas-grekas commented Oct 6, 2021

lyrixx commented Oct 6, 2021

nicolas-grekas left a comment •

edited

Loading

nicolas-grekas commented Oct 6, 2021

nicolas-grekas commented Oct 27, 2021

[Messenger] Add a way to no ack message automatically #42873

[Messenger] Add a way to no ack message automatically #42873

Conversation

lyrixx commented Sep 3, 2021 • edited Loading

carsonbot commented Sep 4, 2021

jderusse left a comment

Choose a reason for hiding this comment

lyrixx commented Sep 22, 2021

okwinza commented Sep 22, 2021

lyrixx commented Oct 1, 2021

nicolas-grekas left a comment

Choose a reason for hiding this comment

lyrixx commented Oct 6, 2021

nicolas-grekas commented Oct 6, 2021

lyrixx commented Oct 6, 2021

nicolas-grekas left a comment • edited Loading

Choose a reason for hiding this comment

nicolas-grekas commented Oct 6, 2021

nicolas-grekas commented Oct 27, 2021

lyrixx commented Sep 3, 2021 •

edited

Loading

nicolas-grekas left a comment •

edited

Loading