Skip to content

Commit d1d0943

Browse files
fabpotfabpot
committed
bug #32497 [Messenger] DispatchAfterCurrentBusMiddleware does not cancel messages from delayed handlers (Nyholm, BastienClement)
This PR was merged into the 4.3 branch. Discussion ---------- [Messenger] DispatchAfterCurrentBusMiddleware does not cancel messages from delayed handlers | Q | A | ------------- | --- | Branch? | 4.3 for bug fixes | Bug fix? | yes | New feature? | no | BC breaks? | no | Deprecations? |no | Tests pass? | yes (thanks @Nyholm) | Fixed tickets | #32370 | License | MIT | Doc PR | - This is a fix for #32370. There is no need for anything sophisticated. There is no recursion or fancy stuff going on, just a queue of message handled sequentially. A simple variable is enough to keep track of the queue state. Thanks @Nyholm for the test, it would clearly have been the hardest part of the job. Commits ------- 1f5c8a6 Cancel delayed message if handler fails 35c76a3 Added tests for #32370
2 parents 431ead2 + 1f5c8a6 commit d1d0943

File tree

2 files changed

+84
-0
lines changed

2 files changed

+84
-0
lines changed

‎src/Symfony/Component/Messenger/Middleware/DispatchAfterCurrentBusMiddleware.php

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -81,12 +81,16 @@ public function handle(Envelope $envelope, StackInterface $stack): Envelope
8181
// "Root dispatch" call is finished, dispatch stored messages.
8282
$exceptions = [];
8383
while (null !== $queueItem = array_shift($this->queue)) {
84+
// Save how many messages are left in queue before handling the message
85+
$queueLengthBefore = \count($this->queue);
8486
try {
8587
// Execute the stored messages
8688
$queueItem->getStack()->next()->handle($queueItem->getEnvelope(), $queueItem->getStack());
8789
} catch (\Exception $exception) {
8890
// Gather all exceptions
8991
$exceptions[] = $exception;
92+
// Restore queue to previous state
93+
$this->queue = \array_slice($this->queue, 0, $queueLengthBefore);
9094
}
9195
}
9296

‎src/Symfony/Component/Messenger/Tests/Middleware/DispatchAfterCurrentBusMiddlewareTest.php

Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -99,6 +99,86 @@ public function testThrowingEventsHandlingWontStopExecution()
9999
$messageBus->dispatch($message);
100100
}
101101

102+
public function testLongChainWithExceptions()
103+
{
104+
$command = new DummyMessage('Level 0');
105+
106+
$eventL1a = new DummyEvent('Event level 1A');
107+
$eventL1b = new DummyEvent('Event level 1B'); // will dispatch 2 more events
108+
$eventL1c = new DummyEvent('Event level 1C');
109+
110+
$eventL2a = new DummyEvent('Event level 2A'); // Will dispatch 1 event and throw exception
111+
$eventL2b = new DummyEvent('Event level 2B'); // Will dispatch 1 event
112+
113+
$eventL3a = new DummyEvent('Event level 3A'); // This should never get handled.
114+
$eventL3b = new DummyEvent('Event level 3B');
115+
116+
$middleware = new DispatchAfterCurrentBusMiddleware();
117+
$handlingMiddleware = $this->createMock(MiddlewareInterface::class);
118+
119+
$eventBus = new MessageBus([
120+
$middleware,
121+
$handlingMiddleware,
122+
]);
123+
124+
// The command bus will dispatch 3 events.
125+
$commandBus = new MessageBus([
126+
$middleware,
127+
new DispatchingMiddleware($eventBus, [
128+
new Envelope($eventL1a, [new DispatchAfterCurrentBusStamp()]),
129+
new Envelope($eventL1b, [new DispatchAfterCurrentBusStamp()]),
130+
new Envelope($eventL1c, [new DispatchAfterCurrentBusStamp()]),
131+
]),
132+
$handlingMiddleware,
133+
]);
134+
135+
// Expect main dispatched message to be handled first:
136+
$this->expectHandledMessage($handlingMiddleware, 0, $command);
137+
138+
$this->expectHandledMessage($handlingMiddleware, 1, $eventL1a);
139+
140+
// Handling $eventL1b will dispatch 2 more events
141+
$handlingMiddleware->expects($this->at(2))->method('handle')->with($this->callback(function (Envelope $envelope) use ($eventL1b) {
142+
return $envelope->getMessage() === $eventL1b;
143+
}))->willReturnCallback(function ($envelope, StackInterface $stack) use ($eventBus, $eventL2a, $eventL2b) {
144+
$envelope1 = new Envelope($eventL2a, [new DispatchAfterCurrentBusStamp()]);
145+
$eventBus->dispatch($envelope1);
146+
$eventBus->dispatch(new Envelope($eventL2b, [new DispatchAfterCurrentBusStamp()]));
147+
148+
return $stack->next()->handle($envelope, $stack);
149+
});
150+
151+
$this->expectHandledMessage($handlingMiddleware, 3, $eventL1c);
152+
153+
// Handle $eventL2a will dispatch event and throw exception
154+
$handlingMiddleware->expects($this->at(4))->method('handle')->with($this->callback(function (Envelope $envelope) use ($eventL2a) {
155+
return $envelope->getMessage() === $eventL2a;
156+
}))->willReturnCallback(function ($envelope, StackInterface $stack) use ($eventBus, $eventL3a) {
157+
$eventBus->dispatch(new Envelope($eventL3a, [new DispatchAfterCurrentBusStamp()]));
158+
159+
throw new \RuntimeException('Some exception while handling Event level 2a');
160+
});
161+
162+
// Make sure $eventL2b is handled, since it was dispatched from $eventL1b
163+
$handlingMiddleware->expects($this->at(5))->method('handle')->with($this->callback(function (Envelope $envelope) use ($eventL2b) {
164+
return $envelope->getMessage() === $eventL2b;
165+
}))->willReturnCallback(function ($envelope, StackInterface $stack) use ($eventBus, $eventL3b) {
166+
$eventBus->dispatch(new Envelope($eventL3b, [new DispatchAfterCurrentBusStamp()]));
167+
168+
return $stack->next()->handle($envelope, $stack);
169+
});
170+
171+
// We dont handle exception L3a since L2a threw an exception.
172+
$this->expectHandledMessage($handlingMiddleware, 6, $eventL3b);
173+
174+
// Note: $eventL3a should not be handled.
175+
176+
$this->expectException(DelayedMessageHandlingException::class);
177+
$this->expectExceptionMessage('RuntimeException: Some exception while handling Event level 2a');
178+
179+
$commandBus->dispatch($command);
180+
}
181+
102182
public function testHandleDelayedEventFromQueue()
103183
{
104184
$message = new DummyMessage('Hello');

0 commit comments

Comments
 (0)