[Messenger] Custom method argument $onHandledCallback for worker with command messenger:consume

[B-Galati]

Description
At the moment it does not look possible to have a custom $onHandledCallback on the Worker used by messenger:consume command.

See ->

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

Line 93 in 52e9fb9

$onHandled($envelope);

My use case is that I would like to reset monolog buffered handler at this very moment to send my error logs in batch to Sentry.

I thought about using WorkerMessageFailedEvent but then I would miss important logging messages such as :

• symfony/src/Symfony/Component/Messenger/Worker.php

  Line 145 in 52e9fb9

  $this->logger->error('Error thrown while handling message {class}. Dispatching for retry #{retryCount} using {delay} ms delay. Error: "{error}"', $context + ['retryCount' => $retryCount, 'delay' => $delay, 'error' => $throwable->getMessage(), 'exception' => $throwable]);

• symfony/src/Symfony/Component/Messenger/Worker.php

  Line 159 in 52e9fb9

  $this->logger->critical('Error thrown while handling message {class}. Removing from transport after {retryCount} retries. Error: "{error}"', $context + ['retryCount' => $retryCount, 'error' => $throwable->getMessage(), 'exception' => $throwable]);

Possible solution
Perhaps by replacing the callable with an interface that could be autowired in the command, not sure.

[lyrixx]

IMHO, the Symfony should be reset after each message. I do that in my custom worker and I Opened an issue in swarrot to do the same.

The status of the message should not be taken in consideration.

[B-Galati]

@weaverryan one idea would be to dispatch an event to replace the callback, WDYT?

[Tobion]

I thought about using WorkerMessageFailedEvent but then I would miss important logging messages

Why would you miss them? They would get sent with the next batch of messages, wouldn't they?

[B-Galati]

@Tobion You're right, but they would be sent with unrelated logs records like:

---- Nothing before because it's been flushed ----
ERROR related to my message
Handling a new message unrelated to my error

[B-Galati]

trying to address that issue #32614

[Tobion]

they would be sent with unrelated logs records

If you sent messages in batches there are always messages in one batch that might not belong together. But what is the problem you have? Logs are time based not group based.

[B-Galati]

they would be sent with unrelated logs records

If you sent messages in batches there are always messages in one batch that might not belong together. But what is the problem you have? Logs are time based not group based.

I would like to send all logs that are related to the same HTTP request or handled message in one batch. For instance I am using this monolog config. The problem I have is that I cannot achieve this behavior right now with messenger worker while it's perfectly fine with HTTP request.

Another solution that works would be to reset the kernel after each message the worker handled.

monolog:
  handlers:
    sentry_buffer:
      type: fingers_crossed
      action_level: warning
      handler: buffer
      channels: ["!event", "!security"]
    buffer:
      type: buffer
      handler: sentry
    sentry:
      type: service
      id: 'sentry_handler'