Skip to content

[Messenger] Simplifying SyncTransport and fixing bug with handlers transport #31401

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
May 9, 2019
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,7 @@

<service id="messenger.transport.sync.factory" class="Symfony\Component\Messenger\Transport\Sync\SyncTransportFactory">
<tag name="messenger.transport_factory" />
<argument type="service" id="messenger.routable_message_bus" />
</service>

<service id="messenger.transport.in_memory.factory" class="Symfony\Component\Messenger\Transport\InMemoryTransportFactory">
Expand Down
3 changes: 1 addition & 2 deletions src/Symfony/Component/Messenger/CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -24,8 +24,7 @@ CHANGELOG
to stop all `messenger:consume` workers.
* [BC BREAK] The `TransportFactoryInterface::createTransport()` signature
changed: a required 3rd `SerializerInterface` argument was added.
* Added a new `SyncTransport` along with `ForceCallHandlersStamp` to
explicitly handle messages synchronously.
* Added a new `SyncTransport` to explicitly handle messages synchronously.
* Added `AmqpStamp` allowing to provide a routing key, flags and attributes on message publishing.
* [BC BREAK] Removed publishing with a `routing_key` option from queue configuration, for
AMQP. Use exchange `default_publish_routing_key` or `AmqpStamp` instead.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,6 @@
use Psr\Log\NullLogger;
use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\Event\SendMessageToTransportsEvent;
use Symfony\Component\Messenger\Stamp\ForceCallHandlersStamp;
use Symfony\Component\Messenger\Stamp\ReceivedStamp;
use Symfony\Component\Messenger\Stamp\RedeliveryStamp;
use Symfony\Component\Messenger\Stamp\SentStamp;
Expand Down Expand Up @@ -78,12 +77,6 @@ public function handle(Envelope $envelope, StackInterface $stack): Envelope
$envelope = $sender->send($envelope->with(new SentStamp(\get_class($sender), \is_string($alias) ? $alias : null)));
}

// if the message was marked (usually by SyncTransport) that it handlers
// MUST be called, mark them to be handled.
if (null !== $envelope->last(ForceCallHandlersStamp::class)) {
$handle = true;
}

// on a redelivery, only send back to queue: never call local handlers
if (null !== $redeliveryStamp) {
$handle = false;
Expand Down
27 changes: 0 additions & 27 deletions src/Symfony/Component/Messenger/Stamp/ForceCallHandlersStamp.php

This file was deleted.

Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,6 @@
use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\Event\SendMessageToTransportsEvent;
use Symfony\Component\Messenger\Middleware\SendMessageMiddleware;
use Symfony\Component\Messenger\Stamp\ForceCallHandlersStamp;
use Symfony\Component\Messenger\Stamp\ReceivedStamp;
use Symfony\Component\Messenger\Stamp\RedeliveryStamp;
use Symfony\Component\Messenger\Stamp\SentStamp;
Expand Down Expand Up @@ -88,8 +87,6 @@ public function testItSendsTheMessageToMultipleSenders()
public function testItSendsToOnlyOneSenderOnRedelivery()
{
$envelope = new Envelope(new DummyMessage('Hey'), [new RedeliveryStamp(5, 'bar')]);
// even with a ForceCallHandlersStamp, the next middleware won't be called
$envelope = $envelope->with(new ForceCallHandlersStamp());
$sender = $this->getMockBuilder(SenderInterface::class)->getMock();
$sender2 = $this->getMockBuilder(SenderInterface::class)->getMock();

Expand Down Expand Up @@ -270,21 +267,6 @@ public function testItDoesNotDispatchOnRedeliver()
$middleware->handle($envelope, $this->getStackMock(false));
}

public function testItHandlesWithForceCallHandlersStamp()
{
$envelope = new Envelope(new DummyMessage('original envelope'));
$envelope = $envelope->with(new ForceCallHandlersStamp());

$sender = $this->getMockBuilder(SenderInterface::class)->getMock();
$sender->expects($this->once())->method('send')->willReturn($envelope);

$sendersLocator = $this->createSendersLocator([DummyMessage::class => ['foo']], ['foo' => $sender]);
$middleware = new SendMessageMiddleware($sendersLocator);

// next handler *should* be called
$middleware->handle($envelope, $this->getStackMock(true));
}

private function createSendersLocator(array $sendersMap, array $senders, array $sendAndHandle = [])
{
$container = $this->createMock(ContainerInterface::class);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
namespace Symfony\Component\Messenger\Tests\Transport\AmqpExt;

use PHPUnit\Framework\TestCase;
use Symfony\Component\Messenger\MessageBusInterface;
use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
use Symfony\Component\Messenger\Transport\Sync\SyncTransport;
use Symfony\Component\Messenger\Transport\Sync\SyncTransportFactory;
Expand All @@ -21,7 +22,8 @@ class SyncTransportFactoryTest extends TestCase
public function testCreateTransport()
{
$serializer = $this->createMock(SerializerInterface::class);
$factory = new SyncTransportFactory();
$bus = $this->createMock(MessageBusInterface::class);
$factory = new SyncTransportFactory($bus);
$transport = $factory->createTransport('sync://', [], $serializer);
$this->assertInstanceOf(SyncTransport::class, $transport);
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -13,18 +13,29 @@

use PHPUnit\Framework\TestCase;
use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\Stamp\ForceCallHandlersStamp;
use Symfony\Component\Messenger\MessageBusInterface;
use Symfony\Component\Messenger\Stamp\ReceivedStamp;
use Symfony\Component\Messenger\Transport\Sync\SyncTransport;

class SyncTransportTest extends TestCase
{
public function testSend()
{
$bus = $this->createMock(MessageBusInterface::class);
$bus->expects($this->once())
->method('dispatch')
->with($this->callback(function ($arg) {
$this->assertInstanceOf(Envelope::class, $arg);

return true;
}))
->willReturnArgument(0);
$message = new \stdClass();
$envelope = new Envelope($message);
$transport = new SyncTransport();
$transport = new SyncTransport($bus);
$envelope = $transport->send($envelope);

$this->assertSame($message, $envelope->getMessage());
$this->assertNotNull($envelope->last(ForceCallHandlersStamp::class));
$this->assertNotNull($envelope->last(ReceivedStamp::class));
}
}
21 changes: 18 additions & 3 deletions src/Symfony/Component/Messenger/Transport/Sync/SyncTransport.php
Original file line number Diff line number Diff line change
Expand Up @@ -13,18 +13,27 @@

use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\Exception\InvalidArgumentException;
use Symfony\Component\Messenger\Stamp\ForceCallHandlersStamp;
use Symfony\Component\Messenger\MessageBusInterface;
use Symfony\Component\Messenger\Stamp\ReceivedStamp;
use Symfony\Component\Messenger\Stamp\SentStamp;
use Symfony\Component\Messenger\Transport\TransportInterface;

/**
* A "fake" transport that marks messages to be handled immediately.
* Transport that immediately marks messages as received and dispatches for handling.
*
* @experimental in 4.3
*
* @author Ryan Weaver <ryan@symfonycasts.com>
*/
class SyncTransport implements TransportInterface
{
private $messageBus;

public function __construct(MessageBusInterface $messageBus)
{
$this->messageBus = $messageBus;
}

public function get(): iterable
{
throw new InvalidArgumentException('You cannot receive messages from the Messenger SyncTransport.');
Expand All @@ -47,6 +56,12 @@ public function reject(Envelope $envelope): void

public function send(Envelope $envelope): Envelope
{
return $envelope->with(new ForceCallHandlersStamp());
/** @var SentStamp|null $sentStamp */
$sentStamp = $envelope->last(SentStamp::class);
$alias = null === $sentStamp ? 'sync' : $sentStamp->getSenderAlias() ?: $sentStamp->getSenderClass();

$envelope = $envelope->with(new ReceivedStamp($alias));

return $this->messageBus->dispatch($envelope);
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@

namespace Symfony\Component\Messenger\Transport\Sync;

use Symfony\Component\Messenger\MessageBusInterface;
use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
use Symfony\Component\Messenger\Transport\TransportFactoryInterface;
use Symfony\Component\Messenger\Transport\TransportInterface;
Expand All @@ -22,9 +23,16 @@
*/
class SyncTransportFactory implements TransportFactoryInterface
{
private $messageBus;

public function __construct(MessageBusInterface $messageBus)
{
$this->messageBus = $messageBus;
}

public function createTransport(string $dsn, array $options, SerializerInterface $serializer): TransportInterface
{
return new SyncTransport();
return new SyncTransport($this->messageBus);
}

public function supports(string $dsn, array $options): bool
Expand Down