minor #34155 Revert SyncTransport simplification and fix properly (weaverryan)

This PR was squashed before being merged into the 4.4 branch (closes #34155).

Discussion
----------

Revert SyncTransport simplification and fix properly

| Q             | A
| ------------- | ---
| Branch?       | 4.4
| Bug fix?      | yes
| New feature?  | no
| Deprecations? | no
| Tickets       | Fix #34115 (and also related to #34066)
| License       | MIT
| Doc PR        | Not needed

In #34069, I made `SyncTransport` simpler by removing that transport class and making the whole things a config trick. I felt GREAT about that solution... until i realized two big problems:

1) It kills using env vars for `sync://` because we read the config values at build time - #34115 - that could probably be fixed by adding a factory, but then there is also the next problem

2) If someone routed a message to `[async, sync]` (weird, but allowed), my #34069 config solution basically maps this internally to `[async]`, which actually causes the message to *not* be handled immediately. Basically, my solution only worked if you route a message ONLY to one sync transport, but fails if you route to multiple transports.

So... this fixes things in a less-cool, but sensible way:

A) The first commit reverts #34069 exactly
B) The second commit solves the issue that we need to know if a message is being handled in a "worker" context or not, so middleware can decide if they should reset things before/after handling things. Previously we were using `ReceivedStamp` to know this. But because `SyncTransport` also "receives" the message and adds this stamp, it's not enough. To fix this, I added a new `ConsumedByWorkerStamp` that clearly means: "This message is being handled by a worker" (and so, you might want to "reset" some things before/after handling).

Thanks!

Commits
-------

01a9fefe77 Adding ConsumedByWorkerStamp as way to mark a message in a "worker context"
38f19a960c Revert "[Messenger] Removing "sync" transport and replacing it with much nicer config trick"

Author: Tobion

DependencyInjection/FrameworkExtension.php

@@ -1755,27 +1755,18 @@ private function registerMessengerConfiguration(array $config, ContainerBuilder
             $container->setAlias('messenger.default_serializer', $config['serializer']['default_serializer']);
         }
 
-        $senderReferences = [];
-        $syncTransports = [];
+        $senderAliases = [];
         $transportRetryReferences = [];
         foreach ($config['transports'] as $name => $transport) {
             $serializerId = $transport['serializer'] ?? 'messenger.default_serializer';
 
-            if (0 === strpos($transport['dsn'], 'sync://')) {
-                $syncTransports[] = $name;
-            } else {
-                $transportDefinition = (new Definition(TransportInterface::class))
-                    ->setFactory([new Reference('messenger.transport_factory'), 'createTransport'])
-                    ->setArguments([$transport['dsn'], $transport['options'] + ['transport_name' => $name], new Reference($serializerId)])
-                    ->addTag('messenger.receiver', ['alias' => $name])
-                ;
-                $container->setDefinition($transportId = 'messenger.transport.'.$name, $transportDefinition);
-
-                // alias => service_id
-                $senderReferences[$name] = new Reference($transportId);
-                // service_id => service_id
-                $senderReferences[$transportId] = new Reference($transportId);
-            }
+            $transportDefinition = (new Definition(TransportInterface::class))
+                ->setFactory([new Reference('messenger.transport_factory'), 'createTransport'])
+                ->setArguments([$transport['dsn'], $transport['options'] + ['transport_name' => $name], new Reference($serializerId)])
+                ->addTag('messenger.receiver', ['alias' => $name])
+            ;
+            $container->setDefinition($transportId = 'messenger.transport.'.$name, $transportDefinition);
+            $senderAliases[$name] = $transportId;
 
             if (null !== $transport['retry_strategy']['service']) {
                 $transportRetryReferences[$name] = new Reference($transport['retry_strategy']['service']);
@@ -1793,25 +1784,30 @@ private function registerMessengerConfiguration(array $config, ContainerBuilder
             }
         }
 
+        $senderReferences = [];
+        // alias => service_id
+        foreach ($senderAliases as $alias => $serviceId) {
+            $senderReferences[$alias] = new Reference($serviceId);
+        }
+        // service_id => service_id
+        foreach ($senderAliases as $serviceId) {
+            $senderReferences[$serviceId] = new Reference($serviceId);
+        }
+
         $messageToSendersMapping = [];
         foreach ($config['routing'] as $message => $messageConfiguration) {
             if ('*' !== $message && !class_exists($message) && !interface_exists($message, false)) {
                 throw new LogicException(sprintf('Invalid Messenger routing configuration: class or interface "%s" not found.', $message));
             }
 
-            // filter out "sync" senders
-            $realSenders = [];
+            // make sure senderAliases contains all senders
             foreach ($messageConfiguration['senders'] as $sender) {
-                if (isset($senderReferences[$sender])) {
-                    $realSenders[] = $sender;
-                } elseif (!\in_array($sender, $syncTransports, true)) {
+                if (!isset($senderReferences[$sender])) {
                     throw new LogicException(sprintf('Invalid Messenger routing configuration: the "%s" class is being routed to a sender called "%s". This is not a valid transport or service id.', $message, $sender));
                 }
             }
 
-            if ($realSenders) {
-                $messageToSendersMapping[$message] = $realSenders;
-            }
+            $messageToSendersMapping[$message] = $messageConfiguration['senders'];
         }
 
         $container->getDefinition('messenger.senders_locator')

Resources/config/messenger.xml

@@ -75,6 +75,11 @@
             <tag name="messenger.transport_factory" />
         </service>
 
+        <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">
             <tag name="messenger.transport_factory" />
             <tag name="kernel.reset" method="reset" />

Tests/DependencyInjection/Fixtures/php/messenger_sync_transport.php

@@ -16,8 +16,6 @@
 use Symfony\Bundle\FrameworkBundle\DependencyInjection\Compiler\AddAnnotationsCachedReaderPass;
 use Symfony\Bundle\FrameworkBundle\DependencyInjection\FrameworkExtension;
 use Symfony\Bundle\FrameworkBundle\Tests\Fixtures\Messenger\DummyMessage;
-use Symfony\Bundle\FrameworkBundle\Tests\Fixtures\Messenger\FooMessage;
-use Symfony\Bundle\FrameworkBundle\Tests\Fixtures\Messenger\SecondMessage;
 use Symfony\Bundle\FrameworkBundle\Tests\TestCase;
 use Symfony\Bundle\FullStack;
 use Symfony\Component\Cache\Adapter\AdapterInterface;
@@ -784,17 +782,6 @@ public function testMessengerInvalidTransportRouting()
         $this->createContainerFromFile('messenger_routing_invalid_transport');
     }
 
-    public function testMessengerSyncTransport()
-    {
-        $container = $this->createContainerFromFile('messenger_sync_transport');
-        $senderLocatorDefinition = $container->getDefinition('messenger.senders_locator');
-
-        $sendersMapping = $senderLocatorDefinition->getArgument(0);
-        $this->assertEquals(['amqp'], $sendersMapping[DummyMessage::class]);
-        $this->assertArrayNotHasKey(SecondMessage::class, $sendersMapping);
-        $this->assertEquals(['amqp'], $sendersMapping[FooMessage::class]);
-    }
-
     public function testTranslator()
     {
         $container = $this->createContainerFromFile('full');