[Messenger] Adding support for record messages #27844

Nyholm · 2018-07-04T13:46:54Z

Q	A
Branch?	master
Bug fix?	no
New feature?	yes
BC breaks?	no
Deprecations?	no
Tests pass?	yes
Fixed tickets
License	MIT
Doc PR	symfony/symfony-docs#10015

With inspiration from this blog post, I created a message recorder. This is needed when you handle messages in the same doctrine transaction. Ie, you have a Command and the CommandHandler dispatches Events. Those events should be handled in a new doctrine transaction.

Scenario

If the command handler creates a User with uuid X and dispatches UserCreated event. If they are handled in the same transaction then this code will find nothing:

$this->em->getRepository(User::class)->findOneBy(['uuid'=>X]);

Example configuration:

framework:
    messenger:
        default_bus: messenger.bus.command
        buses:
            messenger.bus.command:
                middleware:
                    # Make sure recorded events are handled by the event bus. You may also configure the message recorder with a second parameter. 
                    - app.messenger.middleware.record_messages
                    - doctrine_transaction_middleware: ['default']
        # -- more buses

services: 
    app.messenger.middleware.record_messages:
        parent: messenger.middleware.record_messages
        arguments: ['@messenger.bus.event']

Nyholm · 2018-07-04T13:48:13Z

src/Symfony/Component/Messenger/DependencyInjection/MessengerPass.php

@@ -296,6 +296,11 @@ private function registerBusMiddleware(ContainerBuilder $container, string $busI
                $childDefinition = new ChildDefinition($messengerMiddlewareId);
                $count = \count($definition->getArguments());
                foreach (array_values($arguments ?? array()) as $key => $argument) {
+                    // Convert argument to references if needed.
+                    if (is_string($argument) && $argument[0] === '@') {
+                        $argument = new Reference(substr($argument, 1));


Ping @ogizanagi, This is the best way I found to support references. Im not sure if this is needed if you configure framework.messenger with XML.

Nyholm · 2018-07-04T14:13:47Z

src/Symfony/Component/Messenger/DependencyInjection/MessengerPass.php

@@ -296,6 +296,11 @@ private function registerBusMiddleware(ContainerBuilder $container, string $busI
                $childDefinition = new ChildDefinition($messengerMiddlewareId);
                $count = \count($definition->getArguments());
                foreach (array_values($arguments ?? array()) as $key => $argument) {
+                    // Convert argument to references if needed.
+                    if (is_string($argument) && '@' === $argument[0]) {
+                        $argument = new Reference(substr($argument, 1));


Ping @ogizanagi, This is the best way I found to support references. Im not sure if this is needed if you configure framework.messenger with XML.

Or should I go the long way of creating a Factory that uses @container as a dependency?

The idea of factories weren't to replace DI config when more suitable, only to solve most simple needs for configurable middlewares. Therefore, I don't think we should add services references support.
If really desired, we could indeed use dedicated service locators or perhaps let the user wiring it itself by creating the middleware service using DI config for now.

Hm. I tried using a Factory that depends on @service_container. That does not work since the buses are private.

I also tried injecting all services tagged with message_bus into the factory. That did not work because of circular dependencies.

Circular reference detected for service "debug.traced.messenger.bus.command", path: "debug.traced.messenger.bus.command -
debug.traced.messenger.bus.command.inner -> messenger.bus.command.middleware.messenger.middleware.handles_recorded_mess
ages -> messenger.middleware.handles_recorded_messages.factory -> debug.traced.messenger.bus.command".

Im saying that I do not know a better (working) solution that this. Im sorry.

Injecting the Symfony service container now belongs to another era 😃
We should register dedicated service locators instead. This would happen in the MessengerPass, using ServiceLocatorTagPass::register().

But as I said, we could let the user create the concrete instance of this middleware in userland for now.

Done. I've updated the configuration example

ogizanagi · 2018-07-04T21:55:07Z

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

+        $this->messageRecorder->eraseMessages();
+
+        foreach ($recordedMessages as $recordedMessage) {
+            $this->messageBus->dispatch($recordedMessage);


What if a recorded message recorded another one?
Should we clear any recorded message before middleware execution, so it's obvious it's not supported & undesired by design, and no left over message is dispatched when reusing the bus to handle another message? (am I clear? 😅)

Also should an exception while dispatching a recorded message really interrupt everything?

Interesting.

I think I agree with all your questions. I'll try to make some changes to address them.

BenjaminRbt · 2018-07-05T07:27:52Z

src/Symfony/Component/Messenger/MessageRecorderInterface.php

+     *
+     * @return object[]
+     */
+    public function recordedMessages(): array;


Why not naming this function fetch() instead of recordedMessages(), as you already name below the method record() (without messages at the end of the name) ?

Same thing for eraseMessages().

I took the names from SimpleBus. But you are correct, your naming seams easier

Nyholm · 2018-07-05T08:09:21Z

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

+                try {
+                    $this->messageBus->dispatch($recordedMessage);
+                } catch (\Throwable $exception) {
+                    $exceptions[] = $exception;


Note that if one handler to message A throws an exception, then other handlers may not run (depending on the order).

This will just make sure that handlers to message B will run even though a handler to A throws an exception.

This slightly ties into what I proposed in #27215. I think all handlers should be run even if some fail. Also, I think catching all the exceptions should be done in a middleware on the event bus.

ogizanagi · 2018-07-05T08:32:08Z

src/Symfony/Component/Messenger/MessageRecorder.php

+    /**
+     * {@inheritdoc}
+     */
+    public function erase(): void


reset? Would be a legit use-case for the ResettableInterface too I guess (for now using the kernel.reset tag manually until #27093 is approved and merged)

ogizanagi · 2018-07-05T08:39:40Z

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

+        }
+
+        $exceptions = array();
+        while(!empty($recordedMessages = $this->messageRecorder->fetch())) {


if rather than while?

ogizanagi · 2018-07-05T08:40:34Z

src/Symfony/Component/Messenger/Exception/MessageRecorderHandlerException.php

+    public static function create(array $exceptions): self
+    {
+        $message = sprintf(
+            "One or more handlers for reordered messages threw an exception. Their messages were: \n\n%s",


s/reordered/recorded

ogizanagi · 2018-07-05T08:40:55Z

src/Symfony/Component/Messenger/Exception/MessageRecorderHandlerException.php

+    {
+        return $this->exceptions;
+    }
+}


Missing new line

ogizanagi · 2018-07-05T08:44:49Z

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

+        }
+
+        if (!empty($exceptions)) {
+            throw MessageRecorderHandlerException::create($exceptions);


We don't use named construct for exceptions in core ^^

Wild thoughts:
Should it really throw? Make it configurable? Logged as error? Should the MessageRecorder store the exceptions itself?

Hm.. I think we should throw.
But making it configurable sounds like a reasonable future update.

javiereguiluz · 2018-07-05T09:19:14Z

src/Symfony/Bundle/FrameworkBundle/Resources/config/messenger.xml

@@ -34,6 +34,11 @@
            <argument type="service" id="validator" />
        </service>

+        <service id="messenger.middleware.handles_recorded_messages" class="Symfony\Component\Messenger\Middleware\HandlesRecordedMessagesMiddleware" abstract="true">


Looking at the names of other middlewares, this one maybe should be called messenger.middleware.record_messages

BenjaminRbt · 2018-07-19T07:55:27Z

src/Symfony/Component/Messenger/MessageRecorderInterface.php

+    public function fetch(): array;
+
+    /**
+     * Erase messages that were recorded since the last call to eraseMessages().


As your function is called reset(), I would have say:

Reset messages that were recorded since the last call to eraseMessages().

Reset messages that were recorded since the last call to reset().

😄

You can actually extend the ResettableInterface from symfony/contracts now.

BenjaminRbt · 2018-07-19T07:56:22Z

src/Symfony/Component/Messenger/MessageRecorderInterface.php

+     *
+     * @param object $message
+     */
+    public function record($message);


Add : void as function signature (because of interface).

~~I don't see any value in adding : void tbh~~. It might be an issue if some recorders want to return something? (which might be needed in some case?)

sroze

In this pull-request, there is nothing that actually records the messages, right? 🤔

As far as I understand, the problem you are trying to solve is to wrap messages into some "transaction", so that dispatched message will be handled after the "current" one, right? If I understand it correctly then I believe we should work on the naming to clarify this and we can use the term "transactions", it isn't necessarily database-only.

sroze · 2018-07-20T09:31:08Z

src/Symfony/Component/Messenger/MessageRecorderInterface.php

+    public function fetch(): array;
+
+    /**
+     * Erase messages that were recorded since the last call to eraseMessages().


You can actually extend the ResettableInterface from symfony/contracts now.

sroze · 2018-07-20T09:31:28Z

src/Symfony/Component/Messenger/MessageRecorderInterface.php

+     *
+     * @param object $message
+     */
+    public function record($message);


~~I don't see any value in adding : void tbh~~. It might be an issue if some recorders want to return something? (which might be needed in some case?)

sroze · 2018-07-20T09:32:10Z

src/Symfony/Component/Messenger/Exception/MessageRecorderHandlerException.php

+namespace Symfony\Component\Messenger\Exception;
+
+/**
+ * When handling recorded messaged one or more handlers caused an exception.


sroze · 2018-07-20T09:37:23Z

src/Symfony/Component/Messenger/Exception/MessageRecorderHandlerException.php

+    public function __construct(array $exceptions)
+    {
+        $message = sprintf(
+            "One or more handlers for recorded messages threw an exception. Their messages were: \n\n%s",


If there is only one exception, let's not confuse the user and display it without wrapping it to "one or more".

sroze · 2018-07-20T09:38:10Z

src/Symfony/Component/Messenger/Exception/MessageRecorderHandlerException.php

+ *
+ * @author Tobias Nyholm <tobias.nyholm@gmail.com>
+ */
+class MessageRecorderHandlerException extends \RuntimeException implements ExceptionInterface


MessageHandlingException? This has nothing to do with the "recorder" AFAIK

sroze · 2018-07-20T09:40:19Z

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

+
+        $exceptions = array();
+        while (!empty($recordedMessages = $this->messageRecorder->fetch())) {
+            $this->messageRecorder->erase();


If the use-case is always fetch and erase... I'm pretty sure we can find better than fetch + erase/reset.

Im not sure what you mean here.

sroze · 2018-07-20T09:40:38Z

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

+    public function handle($message, callable $next)
+    {
+        // Make sure the recorder is empty before we begin
+        $this->messageRecorder->erase();


Why would you? 🤔

To avoid any side effects. See #27844 (comment)

Nyholm · 2018-07-23T18:51:17Z

In this pull-request, there is nothing that actually records the messages, right? 🤔

True. "record" is a word from SimpleBus.

As far as I understand, the problem you are trying to solve is to wrap messages into some "transaction", so that dispatched message will be handled after the "current" one, right? If I understand it correctly then I believe we should work on the naming to clarify this and we can use the term "transactions", it isn't necessarily database-only.

Correct. I want to wrap my commands in one transaction and handle my events later. So when used with doctrine transaction middleware, if my event fails, it should not roll back the command.

gubler · 2018-08-21T19:03:43Z

I implemented this pull request in an application I'm working on after it was pointed out to me by @ogizanagi in Slack and @sroze asked me to write up a use case.

I am working on App with a Command Bus with doctrine transactions and had an Event Bus. When injecting the Event Bus directly to the Command Handler, any error in the doctrine transaction caused a rollback as it was supposed to, but the Events were still dispatched to the Event Bus.

I copied the contents of this pull request into my app's namespace - just changing the namespaces for everything and it worked flawlessly. I did have to add the reset method to the MessageRecorderInterface because I am on Symfony 4.1 and not using symfony/contracts.

Now I inject the message recorder and if my transaction fails, the event is not dispatched. Here is an example Command Handler:

namespace App\Domain\CommandHandler\Facility;

use App\Domain\Command\Facility\NewFacilityCommand;
use App\Domain\Event\Facility\FacilityCreatedEvent;
use App\Lib\Messenger\MessageRecorder\MessageRecorderInterface;
use App\Repository\FacilityRepository;

/**
 * Class NewFacilityHandler.
 */
class NewFacilityHandler
{
    /** @var FacilityRepository */
    protected $repository;
    /** @var MessageRecorderInterface */
    protected $recorder;

    /**
     * @param FacilityRepository       $repository
     * @param MessageRecorderInterface $recorder
     */
    public function __construct(FacilityRepository $repository, MessageRecorderInterface $recorder)
    {
        $this->repository = $repository;
        $this->recorder = $recorder;
    }

    /**
     * @param NewFacilityCommand $command
     */
    public function __invoke(NewFacilityCommand $command)
    {
        $facility = $this->repository->persist($command);

        $this->recorder->record(
            new FacilityCreatedEvent(
                $facility,
                $command->generator
            )
        );
    }
}

I kept all the names the same so when this is (hopefully) merged and added to a tagged release, I can update all of the use statements and my services.yaml file and everything should continue to work the same.

I did have one problem until I read the documentation PR that mentioned that the message recorder middleware needed to be declared before the transaction middleware, but once I fixed that, it worked fine.

Here is my messages.yaml if it helps:

framework:
    messenger:
        default_bus: messenger.bus.commands

        buses:
            messenger.bus.commands:
                middleware:
                    - validation
                    - record_messages
                    - doctrine_transaction_middleware
                    - 'App\Domain\BusMiddleware\SecurityMiddleware'
            messenger.bus.events:
                middleware:
                    - validation
                    - allow_no_handler
                    - doctrine_transaction_middleware
                    - 'App\Domain\BusMiddleware\SecurityMiddleware'
                    - 'App\Domain\EventBusMiddleware\FlashMessageMiddleware'
            messenger.bus.queries:
                middleware:
                    - validation
                    - doctrine_transaction_middleware
                    - 'App\Domain\BusMiddleware\SecurityMiddleware'

and the relevant parts of my services.yaml. I had to add the method argument to the tags for message.recorder - I assume this is bacause I am using Symfony 4.1 and not using symfony/contracts.

 # doctrine middleware for command bus
    doctrine.orm.messenger.middleware_factory.transaction:
        class: Symfony\Bridge\Doctrine\Messenger\DoctrineTransactionMiddlewareFactory
        arguments: ['@doctrine']

    messenger.middleware.doctrine_transaction_middleware:
        class: Symfony\Bridge\Doctrine\Messenger\DoctrineTransactionMiddleware
        factory: ['@doctrine.orm.messenger.middleware_factory.transaction', 'createMiddleware']
        abstract: true
        arguments: ['default']

    messenger.middleware.record_messages:
        class: App\Domain\BusMiddleware\HandlesRecordedMessagesMiddleware

    messenger.recorder:
        class: App\Lib\Messenger\MessageRecorder\MessageRecorder
        tags:
            - name: 'kernel.reset'
              method: 'reset'

    App\Lib\Messenger\MessageRecorder\MessageRecorderInterface:
        alias: "messenger.recorder"

sroze · 2018-08-01T07:40:52Z

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

+        $this->messageRecorder->reset();
+
+        try {
+            $next($message);


You need to capture the results as well to return them.

You are correct. Thank you. I've updated the PR

arnolanglade · 2018-09-07T07:32:50Z

src/Symfony/Component/Messenger/MessageRecorderInterface.php

+ * @author Tobias Nyholm <tobias.nyholm@gmail.com>
+ * @author Matthias Noback <matthiasnoback@gmail.com>
+ */
+interface MessageRecorderInterface extends ResetInterface


SimpleBus split this interface into 2 small ones. The first one ContainsRecordedMessages is implement by our models because you need to tell them to give us the events raise during the transaction. The second one RecordsMessages is used to record event at this end of the transaction. For instance, I use it in my repository when my aggregate is flushed. I guess this two distinct responsibilities: I don't want that my model have a public method record.

BTW, I prefer the naming of simple bus (declarative vs imperative programming). It is better to describer what happens instead how it happens :)

I thought about slitting them. But I was not sure. Thank you for commenting!

arnolanglade · 2018-09-07T07:41:15Z

src/Symfony/Component/Messenger/MessageRecorderInterface.php

+     *
+     * @return object[]
+     */
+    public function fetch(): array;


This name seems weird to me. Like I said previously my model will implement this interface. For example, a product model. My product will have Product::fetch() and Product::reset(). In term of business that does not have any sense. I think we should use eraseMessage() and recordedMessage() it more explicit. I think it is a good compromise between technical and business stuff.

arnolanglade · 2018-09-07T07:43:56Z

src/Symfony/Component/Messenger/Exception/MessageHandlingException.php

+        parent::__construct($message);
+    }
+
+    public function getExceptions(): array


toArray would be better.

Hm, Im not sure

src/Symfony/Component/Messenger/MessageRecorder.php

arnolanglade · 2018-09-07T07:53:08Z

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

+        }
+
+        if (!empty($exceptions)) {
+            if (1 === count($exceptions)) {


Why do you do that? I think it will be better to throw a single type of exception. It will be easier to handle it.

The MessageHandlingException is only for grouped exceptions. If the group only have one element, there is no need for a group.

I think an earlier review suggested it to be this way.

arnolanglade · 2018-09-07T07:58:41Z

src/Symfony/Bundle/FrameworkBundle/Resources/config/messenger.xml

@@ -34,6 +34,11 @@
            <argument type="service" id="validator" />
        </service>

+        <service id="messenger.middleware.record_messages" class="Symfony\Component\Messenger\Middleware\HandlesRecordedMessagesMiddleware" abstract="true">
+            <argument type="service" id="message_bus" />


message_bus? Will you dispatch the events into the same bus than the command?

This will be overwritten by your configuration. It is just abstract.

Nyholm · 2018-09-09T14:16:30Z

Thank you for the reviews. I've made some updates.

chalasr · 2018-09-09T15:19:29Z

Could you add some tests? Something like https://github.com/symfony/symfony/blob/master/src/Symfony/Bundle/FrameworkBundle/Tests/DependencyInjection/FrameworkExtensionTest.php#L615 would help understanding the wiring of this feature for me.

Koc · 2018-09-15T11:07:59Z

@Nyholm consider another approach, when aggregate root emits events and they are dispatched after transaction commited. See https://beberlei.de/2013/07/24/doctrine_and_domainevents.html for more details.

With current approach we need emit events manually from command handlers.

Nyholm · 2018-09-15T11:46:55Z

That is a follow up PR :) It is just a new listener and a messageContainer :)

…

On 15 Sep 2018, at 04:08, Konstantin Myakshin ***@***.***> wrote: @Nyholm consider another approach, when aggregate root emits events and they are dispatched after transaction commited. See https://beberlei.de/2013/07/24/doctrine_and_domainevents.html for more details. With current approach we need emit events manually from command handlers. — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub, or mute the thread.

Nyholm · 2018-09-15T12:06:45Z

I've squashed and rebased. I'm ready for a final review.

Changes made

sroze · 2018-09-15T13:52:58Z

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

+    public function handle($message, callable $next)
+    {
+        // Make sure the recorder is empty before we begin
+        $this->messageRecorder->resetRecordedMessages();


Let's imagine the following example:

Controller dispatches a FirstMessage message.

FirstMessageHandler gets called and it does the following:
$recorder->record(new SecondMessage); $bus->dispatch(new SaveToSecondaryDatabase); $recorder->record(new ThirdMessage);

The dispatch will call the middleware and the just recorded message will be erased, right?

This is correct IF the current bus and $bus is using the same MessageRecorder.

I think I should remove this line. Are you okey with that @ogizanagi? It was added by your suggestion.

It might be fine removing it, but the case described above is a more complex topic then.
For instance, if there is any issue with FirstMessageHandler after SaveToSecondaryDatabase, SecondMessage
(and possibly those recorded by SaveToSecondaryDatabase) would still have been executed while I would expect it does only if FirstMessageHandler is fully successful.
Actually, to me recorded messages should be treated if and only if the highest dispatch call is successful. Possibly by using an internal flag inside the middleware? (Thus, we could even keep the reset for safety, only when the flag is off).

Perhaps the LockingMiddleware of Tactician could be a source of inspiration:

http://tactician.thephpleague.com/plugins/locking-middleware/

https://github.com/thephpleague/tactician/blob/master/src/Plugins/LockingMiddleware.php

Which is equivalent to the FinishesHandlingMessageBeforeHandlingNext middleware of SimpleBus. (I prefer SimpleBus' naming in here, it's more explicit)

sroze · 2018-09-15T13:54:13Z

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

+        }
+
+        $exceptions = array();
+        while (!empty($recordedMessages = $this->messageRecorder->getRecordedMessages())) {


What's the reasoning of get + reset instead of pop?

I want to make sure nobody runs $this->messageRecorder->resetRecordedMessages(); while I'm handling the recorded messages.

nicolas-grekas · 2018-09-16T19:24:41Z

(deps=low tests are red, but they should never be to me.)

Nyholm · 2018-09-17T00:15:24Z

I rebased and everything is green.

weaverryan · 2018-09-18T15:12:04Z

@Nyholm is the documentation PR more-or-less up-to-date with the most recent changes? I'm want to "wrap my head" around this PR - I was starting by reading your docs :)

Nyholm · 2018-09-18T15:58:00Z

Yes. It’s up to date.

javiereguiluz · 2018-09-21T07:19:12Z

@Nyholm is "record" a common/required word in this context? Personally I would always say: "save messages" or "store messages" to refer to something like this but never "record messages".

Nyholm · 2018-09-22T14:34:44Z

That is a fair point. "Record" is used from SimpleBus. As far as I know it is made up by them. Using "record" will make it easier for SimpleBus users to migrate (if they would like to).

Im all for using another word but it has to be more clear than using "record". I think that using "Store" or "Save" would cause for confusion since the component support adding things to a queue or "queue like".

Since I'm not "saving" any messages anywhere (just having them in a temporary memory queue) I don't think it is a good word.

sroze · 2018-09-22T14:42:33Z

Then, what about this?

$this->bus->dispatch(Envelope::wrap($message)->with(new ToBeDispatchedAfterCurrentMessageHandling))

All we need is a middleware that is doing the "locking" (i.e. tactitian term) or "FinishesHandlingMessageBeforeHandlingNext" (SimpleBus) only if the following envelope item is present.

Nyholm · 2018-09-22T15:02:33Z

With an EventRecorder you can collect events from Doctrine (just as @arnolanglade is doing).

class User {
  use EventRecorderTrait; // To be created in a following PR

  public function setUsername($username) 
  { 
    $this->username = $username;
    $this->record(new UsernameWasUpdated());
  }
  // ...
}

Using the minor modification in my entity as shown above and a DoctrineEventRecorder (Doctrine subscriber) I can make sure that domain events like these eventually ends up on the bus.

You will also have the flexibility to decide which events that should be recorded and which should be handled now.

public function handle(FooCommand $command)
{
    $this->recorder->record(new Bar()); // Handle later
    $this->eventBus->dispatch(new Biz()); // Handle now
}

This is not possible with a middleware like FinishesHandlingMessageBeforeHandlingNext.

I know "EventRecorder" and "record" is a new concept to this component. But it is not a new concept for message busses. SimpleBus has successfully been using this for years.

sroze · 2018-09-22T15:07:06Z

You will also have the flexibility to decide which events that should be recorded and which should be handled now.

That would exactly be the point of this Envelope item in this comment, right?

Using the minor modification in my entity as shown above and a DoctrineEventRecorder (Doctrine subscriber) I can make sure that domain events like these eventually ends up on the bus.

But you could do the same with the other option which does not introduce this extra notion of "recorder", isn't it?

Nyholm · 2018-09-22T15:33:39Z

No, you can’t. (As far as I know). How and when would you dispatch new messages to the bus?
Sure, you can create a middleware called “DispatchMessagesCollectedByDoctrineSubscriber”. That would possibly achieve the same thing but will less abstraction and flexibility.

I don’t know why we are hesitant of introducing this concept to the MessengerComponent. The argument cannot be DX and/or to keep the component simple, because the alternative solutions proposed are neighter simpler or more intuitive.

Btw, I like that we are challenging this, and I don’t think we should have redundant ways of doing the same thing.

Koc · 2018-09-22T17:55:54Z

How it possible to use this MessageRecorder for event sourcing and store all triggered events in db in same transaction? But queue them to the bus after transaction is complete.

Nyholm · 2018-09-22T18:02:34Z

This is how you keep them in different transaction:

# config/packages/messenger.yaml
framework:
    messenger:
        default_bus: messenger.bus.command
        buses:
            messenger.bus.command:
                middleware:
                    - messenger.middleware.validation
                    - messenger.middleware.handles_recorded_messages
                      # Doctrine transaction must be after handles_recorded_messages middleware
                    - doctrine_transaction_middleware: ['default']
            messenger.bus.event:
                middleware:
                    - messenger.middleware.allow_no_handler
                    - messenger.middleware.validation

If you change the order fo the middlewares, you can get them in the same middleware:

            messenger.bus.command:
                middleware:
                    - messenger.middleware.validation
                    - doctrine_transaction_middleware: ['default']
                    - messenger.middleware.handles_recorded_messages

If you are talking about events from a CommandHandler you would obviously not use this event recorder but the event bus directly. But if you are referring to the events from your entities, then the solution above it for you.

If you want to use both, just define more services for the same classes.

Koc · 2018-09-22T18:53:30Z

If you are talking about events from a CommandHandler you would obviously not use this event recorder but the event bus directly.

No, I cann't. Command handing wrapped in transaction and we shouldn't send any events until it commited.

emit events in aggregate root
save aggregate root and emited events in db in same transaction
after transaction commited - send emited events also to bus.

If you want to use both, just define more services for the same classes.

Not sure that I understand what services should be defined twice. Can you elaborate?

fabpot · 2019-03-03T21:30:50Z

Closing in favor of #28849

…t bus is finished (Nyholm) This PR was merged into the 4.3-dev branch. Discussion ---------- [Messenger] Support for handling messages after current bus is finished | Q | A | ------------- | --- | Branch? | master | Bug fix? | no | New feature? | yes | BC breaks? | no | Deprecations? | no | Tests pass? | yes | Fixed tickets | | License | MIT | Doc PR | symfony/symfony-docs#10015 This is a replacement for #27844. We achieve the same goals without introducing the new concept of "recorder". ```php class CreateUserHandler { private $em; private $eventBus; public function __construct(MessageBus $eventBus, EntityManagerInterface $em) { $this->eventBus = $eventBus; $this->em = $em; } public function __invoke(CreateUser $command) { $user = new User($command->getUuid(), $command->getName(), $command->getEmail()); $this->em->persist($user); $message = new UserCreatedEvent($command->getUuid(); $this->eventBus->dispatch((new Envelope($message))->with(new DispatchAfterCurrentBus())); } } ``` Note that this `DispatchAfterCurrentBusMiddleware` is added automatically as the first middleware. 2019-03-13: I updated the PR description. Commits ------- 903355f Support for handling messages after current bus is finished

…rentBusMiddleware (Nyholm, gubler, OskarStark) This PR was submitted for the master branch but it was merged into the 4.3 branch instead (closes #10015). Discussion ---------- [Messenger] Added documentation about DispatchAfterCurrentBusMiddleware Documentation to PR: symfony/symfony#27844 ![Screenshot 2019-03-17 at 11 43 23](https://user-images.githubusercontent.com/1275206/54489171-edee4880-48a9-11e9-8028-ac501e62dfcb.png) ![Screenshot 2019-03-17 at 11 43 34](https://user-images.githubusercontent.com/1275206/54489173-edee4880-48a9-11e9-877f-f749f242b502.png) ![Screenshot 2019-03-17 at 11 43 39](https://user-images.githubusercontent.com/1275206/54489172-edee4880-48a9-11e9-84f1-d78e68474c7a.png) Commits ------- 0d12880 minor updates according to feedback ec3e1d7 updates according to feedback 3e99663 Update messenger/message-recorder.rst fe81abc Update messenger/message-recorder.rst c1cef9e Update messenger/message-recorder.rst cfa56f7 Update messenger/message-recorder.rst 06ec8a4 Added PHP and XML config 797f530 Minor updates 30e958c Explain HandleMessageInNewTransaction middleware 84b81e3 Added fixes ca22c3c a user fff8659 According to feedback afdddbf Added documentation about message recorder

Nyholm requested a review from sroze as a code owner July 4, 2018 13:46

carsonbot added Status: Needs Review Messenger Feature labels Jul 4, 2018

Nyholm commented Jul 4, 2018

View reviewed changes

This was referenced Jul 4, 2018

[Messenger] Show how to configure a doctrine transaction middleware symfony/symfony-docs#10013

Closed

[Messenger] Added documentation about DispatchAfterCurrentBusMiddleware symfony/symfony-docs#10015

Closed

nicolas-grekas added this to the next milestone Jul 4, 2018

ogizanagi reviewed Jul 4, 2018

View reviewed changes

BenjaminRbt reviewed Jul 5, 2018

View reviewed changes

Nyholm commented Jul 5, 2018

View reviewed changes

ogizanagi reviewed Jul 5, 2018

View reviewed changes

javiereguiluz reviewed Jul 5, 2018

View reviewed changes

BenjaminRbt reviewed Jul 19, 2018

View reviewed changes

sroze previously requested changes Jul 20, 2018

View reviewed changes

carsonbot added Status: Needs Work and removed Status: Needs Review labels Jul 20, 2018

Nyholm force-pushed the record-messages branch from d345a98 to 55b9fff Compare July 23, 2018 18:49

ogizanagi added Status: Needs Review Status: Needs Work and removed Status: Needs Work Status: Needs Review labels Aug 21, 2018

sroze reviewed Aug 26, 2018

View reviewed changes

arnolanglade reviewed Sep 7, 2018

View reviewed changes

Nyholm force-pushed the record-messages branch 2 times, most recently from 57a9089 to f81f82b Compare September 15, 2018 12:06

sroze reviewed Sep 15, 2018

View reviewed changes

Adding support for record messages

3cc6857

Nyholm force-pushed the record-messages branch from f81f82b to 3cc6857 Compare September 17, 2018 00:02

Nyholm mentioned this pull request Oct 13, 2018

[Messenger] Support for handling messages after current bus is finished #28849

Merged

fabpot closed this Mar 3, 2019

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

fabpot mentioned this pull request Aug 30, 2020

Rename messenger/message-recorder.rst symfony/symfony-docs#14148

Closed

[Messenger] Adding support for record messages #27844

[Messenger] Adding support for record messages #27844

Conversation

Nyholm commented Jul 4, 2018 • edited

Scenario

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ogizanagi Jul 5, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ogizanagi Jul 4, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kbond Jul 5, 2018 • edited

Choose a reason for hiding this comment

ogizanagi Jul 5, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ogizanagi Jul 5, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sroze Jul 20, 2018 • edited

Choose a reason for hiding this comment

sroze left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sroze Jul 20, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Nyholm commented Jul 23, 2018

gubler commented Aug 21, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Nyholm commented Sep 9, 2018

chalasr commented Sep 9, 2018

Koc commented Sep 15, 2018

Nyholm commented Sep 15, 2018 via email

Nyholm commented Sep 15, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nicolas-grekas commented Sep 16, 2018

Nyholm commented Sep 17, 2018

weaverryan commented Sep 18, 2018

Nyholm commented Sep 18, 2018

javiereguiluz commented Sep 21, 2018

Nyholm commented Sep 22, 2018

Nyholm commented Jul 4, 2018 •

edited

ogizanagi Jul 5, 2018 •

edited

ogizanagi Jul 4, 2018 •

edited

kbond Jul 5, 2018 •

edited

ogizanagi Jul 5, 2018 •

edited

ogizanagi Jul 5, 2018 •

edited

sroze Jul 20, 2018 •

edited

sroze Jul 20, 2018 •

edited

sroze commented Sep 22, 2018 •

edited

sroze commented Sep 22, 2018 •

edited

Koc commented Sep 22, 2018 •

edited