Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly
Appearance settings

Commit 8b8e52c

Browse filesBrowse files
alli83alli83
committed
[Messenger] implementation of messenger:consume, which processes messages concurrently
1 parent b167190 commit 8b8e52c
Copy full SHA for 8b8e52c

File tree

11 files changed

+236
-14
lines changed
Filter options

11 files changed

+236
-14
lines changed

‎src/Symfony/Bundle/FrameworkBundle/DependencyInjection/FrameworkExtension.php

Copy file name to clipboardExpand all lines: src/Symfony/Bundle/FrameworkBundle/DependencyInjection/FrameworkExtension.php
+1-1Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2158,7 +2158,7 @@ private function registerMessengerConfiguration(array $config, ContainerBuilder
21582158

21592159
if ($busId === $config['default_bus']) {
21602160
$container->setAlias('messenger.default_bus', $busId)->setPublic(true);
2161-
$container->setAlias(MessageBusInterface::class, $busId);
2161+
$container->setAlias(MessageBusInterface::class, $busId)->setPublic(true);
21622162
} else {
21632163
$container->registerAliasForArgument($busId, MessageBusInterface::class);
21642164
}

‎src/Symfony/Bundle/FrameworkBundle/Resources/config/messenger.php

Copy file name to clipboardExpand all lines: src/Symfony/Bundle/FrameworkBundle/Resources/config/messenger.php
+9Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@
3333
use Symfony\Component\Messenger\Middleware\SendMessageMiddleware;
3434
use Symfony\Component\Messenger\Middleware\TraceableMiddleware;
3535
use Symfony\Component\Messenger\Middleware\ValidationMiddleware;
36+
use Symfony\Component\Messenger\ParallelMessageBus;
3637
use Symfony\Component\Messenger\Retry\MultiplierRetryStrategy;
3738
use Symfony\Component\Messenger\RoutableMessageBus;
3839
use Symfony\Component\Messenger\Transport\InMemory\InMemoryTransportFactory;
@@ -54,6 +55,7 @@
5455
abstract_arg('per message senders map'),
5556
abstract_arg('senders service locator'),
5657
])
58+
5759
->set('messenger.middleware.send_message', SendMessageMiddleware::class)
5860
->abstract()
5961
->args([
@@ -134,6 +136,13 @@
134136
])
135137
->tag('messenger.transport_factory')
136138

139+
->set('parallel_bus', ParallelMessageBus::class)
140+
->args([
141+
[],
142+
service('kernel'),
143+
])
144+
->tag('messenger.bus')
145+
137146
->set('messenger.transport.in_memory.factory', InMemoryTransportFactory::class)
138147
->tag('messenger.transport_factory')
139148
->tag('kernel.reset', ['method' => 'reset'])

‎src/Symfony/Component/Messenger/Command/ConsumeMessagesCommand.php

Copy file name to clipboardExpand all lines: src/Symfony/Component/Messenger/Command/ConsumeMessagesCommand.php
+3Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -85,6 +85,7 @@ protected function configure(): void
8585
new InputOption('queues', null, InputOption::VALUE_REQUIRED | InputOption::VALUE_IS_ARRAY, 'Limit receivers to only consume from the specified queues'),
8686
new InputOption('no-reset', null, InputOption::VALUE_NONE, 'Do not reset container services after each message'),
8787
new InputOption('all', null, InputOption::VALUE_NONE, 'Consume messages from all receivers'),
88+
new InputOption('parallel-limit', 'p', InputOption::VALUE_OPTIONAL, 'The number of concurrent processes', 10),
8889
])
8990
->setHelp(<<<'EOF'
9091
The <info>%command.name%</info> command consumes messages and dispatches them to the message bus.
@@ -250,6 +251,8 @@ protected function execute(InputInterface $input, OutputInterface $output): int
250251
$options['queues'] = $queues;
251252
}
252253

254+
$options['parallel-limit'] = $input->getOption('parallel-limit');
255+
253256
try {
254257
$this->worker->run($options);
255258
} finally {

‎src/Symfony/Component/Messenger/DispatchTask.php

Copy file name to clipboard
+82Lines changed: 82 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,82 @@
1+
<?php
2+
3+
/*
4+
* This file is part of the Symfony package.
5+
*
6+
* (c) Fabien Potencier <fabien@symfony.com>
7+
*
8+
* For the full copyright and license information, please view the LICENSE
9+
* file that was distributed with this source code.
10+
*/
11+
12+
/*
13+
* This file is part of the Symfony package.
14+
*
15+
* (c) Fabien Potencier <fabien@symfony.com>
16+
*
17+
* For the full copyright and license information, please view the LICENSE
18+
* file that was distributed with this source code.
19+
*/
20+
21+
namespace Symfony\Component\Messenger;
22+
23+
use Amp\Cache\LocalCache;
24+
use Amp\Cancellation;
25+
use Amp\Parallel\Worker\Task;
26+
use Amp\Sync\Channel;
27+
use App\Kernel;
28+
use Psr\Container\ContainerInterface;
29+
use Symfony\Component\Dotenv\Dotenv;
30+
use Symfony\Component\Messenger\Exception\LogicException;
31+
use Symfony\Component\Messenger\Stamp\AckStamp;
32+
33+
class DispatchTask implements Task
34+
{
35+
private static ?LocalCache $cache = null;
36+
37+
public function __construct(private Envelope $envelope, private array $stamps, private readonly string $env, private readonly bool $isDebug)
38+
{
39+
}
40+
41+
public function run(Channel $channel, Cancellation $cancellation): mixed
42+
{
43+
$container = $this->getContainer();
44+
$envelope = $this->dispatch($container, $channel);
45+
46+
return $envelope->withoutStampsOfType(AckStamp::class);
47+
}
48+
49+
private function dispatch(ContainerInterface $container, $channel)
50+
{
51+
if (!$container->has(MessageBusInterface::class)) {
52+
throw new LogicException(sprintf("%s can't be found.", MessageBusInterface::class));
53+
}
54+
55+
$traceable = $container->get(MessageBusInterface::class);
56+
$messageBus = $traceable->getMessageBus();
57+
58+
return $messageBus->dispatch($this->envelope, $this->stamps);
59+
}
60+
61+
private function getContainer()
62+
{
63+
$cache = self::$cache ??= new LocalCache();
64+
$container = $cache->get('cache-container');
65+
66+
// if not in cache, create container
67+
if (!$container) {
68+
$kernel = new Kernel($this->env, $this->isDebug);
69+
$kernel->boot();
70+
71+
$container = $kernel->getContainer();
72+
$cache->set('cache-container', $container);
73+
74+
(new Dotenv())
75+
->setProdEnvs(['prod'])
76+
->usePutenv(false)
77+
->bootEnv('.env');
78+
}
79+
80+
return $container;
81+
}
82+
}

‎src/Symfony/Component/Messenger/MessageBusInterface.php

Copy file name to clipboardExpand all lines: src/Symfony/Component/Messenger/MessageBusInterface.php
+1-1Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,5 +27,5 @@ interface MessageBusInterface
2727
*
2828
* @throws ExceptionInterface
2929
*/
30-
public function dispatch(object $message, array $stamps = []): Envelope;
30+
public function dispatch(object $message, array $stamps = []): Envelope|array;
3131
}

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

Copy file name to clipboardExpand all lines: src/Symfony/Component/Messenger/Middleware/HandleMessageMiddleware.php
+7-1Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
use Symfony\Component\Messenger\Handler\HandlerDescriptor;
2121
use Symfony\Component\Messenger\Handler\HandlersLocatorInterface;
2222
use Symfony\Component\Messenger\Stamp\AckStamp;
23+
use Symfony\Component\Messenger\Stamp\BusNameStamp;
2324
use Symfony\Component\Messenger\Stamp\FlushBatchHandlersStamp;
2425
use Symfony\Component\Messenger\Stamp\HandledStamp;
2526
use Symfony\Component\Messenger\Stamp\HandlerArgumentsStamp;
@@ -32,6 +33,8 @@ class HandleMessageMiddleware implements MiddlewareInterface
3233
{
3334
use LoggerAwareTrait;
3435

36+
private const PARALLEL_BUS = 'parallel_bus';
37+
3538
public function __construct(
3639
private HandlersLocatorInterface $handlersLocator,
3740
private bool $allowNoHandlers = false,
@@ -64,6 +67,10 @@ public function handle(Envelope $envelope, StackInterface $stack): Envelope
6467

6568
/** @var AckStamp $ackStamp */
6669
if ($batchHandler && $ackStamp = $envelope->last(AckStamp::class)) {
70+
if (self::PARALLEL_BUS === $envelope->last(BusNameStamp::class)->getBusName()) {
71+
throw new HandlerFailedException($envelope, [new \Exception("Parallel bus can't be used for batch messages")]);
72+
}
73+
6774
$ack = new Acknowledger(get_debug_type($batchHandler), static function (?\Throwable $e = null, $result = null) use ($envelope, $ackStamp, $handlerDescriptor) {
6875
if (null !== $e) {
6976
$e = new HandlerFailedException($envelope, [$handlerDescriptor->getName() => $e]);
@@ -75,7 +82,6 @@ public function handle(Envelope $envelope, StackInterface $stack): Envelope
7582
});
7683

7784
$result = $this->callHandler($handler, $message, $ack, $envelope->last(HandlerArgumentsStamp::class));
78-
7985
if (!\is_int($result) || 0 > $result) {
8086
throw new LogicException(sprintf('A handler implementing BatchHandlerInterface must return the size of the current batch as a positive integer, "%s" returned from "%s".', \is_int($result) ? $result : get_debug_type($result), get_debug_type($batchHandler)));
8187
}

‎src/Symfony/Component/Messenger/ParallelMessageBus.php

Copy file name to clipboard
+42Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,42 @@
1+
<?php
2+
3+
/*
4+
* This file is part of the Symfony package.
5+
*
6+
* (c) Fabien Potencier <fabien@symfony.com>
7+
*
8+
* For the full copyright and license information, please view the LICENSE
9+
* file that was distributed with this source code.
10+
*/
11+
12+
namespace Symfony\Component\Messenger;
13+
14+
use Amp\Parallel\Worker\ContextWorkerPool;
15+
use Symfony\Component\HttpKernel\KernelInterface;
16+
17+
use function Amp\async;
18+
use function Amp\Parallel\Worker\workerPool;
19+
20+
class ParallelMessageBus implements MessageBusInterface
21+
{
22+
public function __construct(private array $something, private readonly KernelInterface $kernel)
23+
{
24+
}
25+
26+
public static ?ContextWorkerPool $worker = null;
27+
28+
public function dispatch(object $message, array $stamps = []): Envelope|array
29+
{
30+
$worker = self::$worker ?: workerPool();
31+
self::$worker = $worker;
32+
33+
$envelope = Envelope::wrap($message, $stamps);
34+
$task = new DispatchTask($envelope, $stamps, $this->kernel->getEnvironment(), $this->kernel->isDebug());
35+
36+
$future = async(function () use ($worker, $task) {
37+
return $worker->submit($task);
38+
});
39+
40+
return [$future, $envelope];
41+
}
42+
}

‎src/Symfony/Component/Messenger/RoutableMessageBus.php

Copy file name to clipboardExpand all lines: src/Symfony/Component/Messenger/RoutableMessageBus.php
+1-1Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,7 @@ public function __construct(ContainerInterface $busLocator, ?MessageBusInterface
3434
$this->fallbackBus = $fallbackBus;
3535
}
3636

37-
public function dispatch(object $envelope, array $stamps = []): Envelope
37+
public function dispatch(object $envelope, array $stamps = []): Envelope|array
3838
{
3939
if (!$envelope instanceof Envelope) {
4040
throw new InvalidArgumentException('Messages passed to RoutableMessageBus::dispatch() must be inside an Envelope.');

‎src/Symfony/Component/Messenger/TraceableMessageBus.php

Copy file name to clipboardExpand all lines: src/Symfony/Component/Messenger/TraceableMessageBus.php
+9-2Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,7 @@ public function __construct(MessageBusInterface $decoratedBus)
2424
$this->decoratedBus = $decoratedBus;
2525
}
2626

27-
public function dispatch(object $message, array $stamps = []): Envelope
27+
public function dispatch(object $message, array $stamps = []): Envelope|array
2828
{
2929
$envelope = Envelope::wrap($message, $stamps);
3030
$context = [
@@ -41,7 +41,9 @@ public function dispatch(object $message, array $stamps = []): Envelope
4141

4242
throw $e;
4343
} finally {
44-
$this->dispatchedMessages[] = $context + ['stamps_after_dispatch' => array_merge([], ...array_values($envelope->all()))];
44+
if ($envelope instanceof Envelope) {
45+
$this->dispatchedMessages[] = $context + ['stamps_after_dispatch' => array_merge([], ...array_values($envelope->all()))];
46+
}
4547
}
4648
}
4749

@@ -102,4 +104,9 @@ private function getCaller(): array
102104
'line' => $line,
103105
];
104106
}
107+
108+
public function getMessageBus(): MessageBusInterface
109+
{
110+
return $this->decoratedBus;
111+
}
105112
}

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

Copy file name to clipboardExpand all lines: src/Symfony/Component/Messenger/Worker.php
+79-7Lines changed: 79 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
use Symfony\Component\Messenger\Exception\RejectRedeliveredMessageException;
2828
use Symfony\Component\Messenger\Exception\RuntimeException;
2929
use Symfony\Component\Messenger\Stamp\AckStamp;
30+
use Symfony\Component\Messenger\Stamp\BusNameStamp;
3031
use Symfony\Component\Messenger\Stamp\ConsumedByWorkerStamp;
3132
use Symfony\Component\Messenger\Stamp\FlushBatchHandlersStamp;
3233
use Symfony\Component\Messenger\Stamp\NoAutoAckStamp;
@@ -49,6 +50,8 @@ class Worker
4950
private array $acks = [];
5051
private \SplObjectStorage $unacks;
5152

53+
private static $futures = [];
54+
5255
/**
5356
* @param ReceiverInterface[] $receivers Where the key is the transport name
5457
*/
@@ -103,11 +106,16 @@ public function run(array $options = []): void
103106
$envelopes = $receiver->get();
104107
}
105108

109+
if (!$envelopes) {
110+
// flush
111+
$this->handleFutures($transportName);
112+
}
113+
106114
foreach ($envelopes as $envelope) {
107115
$envelopeHandled = true;
108116

109117
$this->rateLimit($transportName);
110-
$this->handleMessage($envelope, $transportName);
118+
$this->handleMessage($envelope, $transportName, $options['parallel-limit']);
111119
$this->eventDispatcher?->dispatch(new WorkerRunningEvent($this, false));
112120

113121
if ($this->shouldStop) {
@@ -142,7 +150,7 @@ public function run(array $options = []): void
142150
$this->eventDispatcher?->dispatch(new WorkerStoppedEvent($this));
143151
}
144152

145-
private function handleMessage(Envelope $envelope, string $transportName): void
153+
private function handleMessage(Envelope $envelope, string $transportName, $parallelProcessesLimit): void
146154
{
147155
$event = new WorkerMessageReceivedEvent($envelope, $transportName);
148156
$this->eventDispatcher?->dispatch($event);
@@ -153,17 +161,49 @@ private function handleMessage(Envelope $envelope, string $transportName): void
153161
}
154162

155163
$acked = false;
156-
$ack = function (Envelope $envelope, ?\Throwable $e = null) use ($transportName, &$acked) {
157-
$acked = true;
158-
$this->acks[] = [$transportName, $envelope, $e];
159-
};
164+
$busNameStamp = $envelope->last(BusNameStamp::class);
165+
166+
// "non concurrent" behaviour
167+
if ('parallel_bus' !== $busNameStamp->getBusName()) {
168+
$ack = function (Envelope $envelope, ?\Throwable $e = null) use ($transportName, &$acked) {
169+
$acked = true;
170+
$this->acks[] = [$transportName, $envelope, $e];
171+
};
172+
}
160173

161174
try {
162175
$e = null;
163-
$envelope = $this->bus->dispatch($envelope->with(new ReceivedStamp($transportName), new ConsumedByWorkerStamp(), new AckStamp($ack)));
176+
177+
$envelope = $envelope->with(new ReceivedStamp($transportName), new ConsumedByWorkerStamp());
178+
if ('parallel_bus' !== $busNameStamp->getBusName()) {
179+
$envelope->with(new AckStamp($ack));
180+
}
181+
182+
$future = $this->bus->dispatch($envelope);
183+
184+
// "non concurrent" behaviour
185+
if ($future instanceof Envelope) {
186+
$this->preAck($envelope, $transportName, $acked, $e);
187+
188+
return;
189+
}
190+
191+
self::$futures[] = $future;
164192
} catch (\Throwable $e) {
193+
$this->preAck($envelope, $transportName, $acked, $e);
194+
195+
return;
165196
}
166197

198+
if (\count(self::$futures) < $parallelProcessesLimit) {
199+
return;
200+
}
201+
202+
$this->handleFutures($transportName);
203+
}
204+
205+
private function preAck(Envelope $envelope, string $transportName, bool $acked, $e): void
206+
{
167207
$noAutoAckStamp = $envelope->last(NoAutoAckStamp::class);
168208

169209
if (!$acked && !$noAutoAckStamp) {
@@ -282,4 +322,36 @@ public function getMetadata(): WorkerMetadata
282322
{
283323
return $this->metadata;
284324
}
325+
326+
public function handleFutures(string $transportName): void
327+
{
328+
$toHandle = self::$futures;
329+
self::$futures = [];
330+
$errors = [];
331+
332+
foreach ($toHandle as $future) {
333+
$e = null;
334+
try {
335+
$execution = $future[0]->await();
336+
$envelope = $execution->await();
337+
} catch (\Throwable $e) {
338+
$errors[] = [$future[1], $e];
339+
340+
continue;
341+
}
342+
343+
$this->preAck($envelope, $transportName, false, null);
344+
}
345+
346+
if (!$errors) {
347+
return;
348+
}
349+
350+
// otherwise, there might be an issue with the Messenger event as failedEvent (TO DO: test better)
351+
foreach ($errors as $combo) {
352+
[$envelope, $e] = $combo;
353+
354+
$this->preAck($envelope, $transportName, false, $e);
355+
}
356+
}
285357
}

0 commit comments

Comments
0 (0)
Morty Proxy This is a proxified and sanitized view of the page, visit original site.