fromDsn example

    /** * @group integration */
    public function testCreateTransport()
    {
        $this->skipIfRedisUnavailable();

        $factory = new RedisTransportFactory();
        $serializer = $this->createMock(SerializerInterface::class);
        $expectedTransport = new RedisTransport(Connection::fromDsn('redis://'.getenv('REDIS_HOST'), ['stream' => 'bar', 'delete_after_ack' => true]), $serializer);

        $this->assertEquals($expectedTransport, $factory->createTransport('redis://'.getenv('REDIS_HOST'), ['stream' => 'bar', 'delete_after_ack' => true], $serializer));
    }

    private function skipIfRedisUnavailable()
    {
        try {
            (new \Redis())->connect(...explode(':', getenv('REDIS_HOST')));
        } catch (\Exception $e) {
            self::markTestSkipped($e->getMessage());
        }
    }

private \Redis|Relay|null $redis = null;
    private ?Connection $connection = null;

    protected function setUp(): void
    {
        if (!getenv('MESSENGER_REDIS_DSN')) {
            $this->markTestSkipped('The "MESSENGER_REDIS_DSN" environment variable is required.');
        }

        try {
            $this->redis = $this->createRedisClient();
            $this->connection = Connection::fromDsn(getenv('MESSENGER_REDIS_DSN'), ['sentinel_master' => getenv('MESSENGER_REDIS_SENTINEL_MASTER') ?: null], $this->redis);
            $this->connection->cleanup();
            $this->connection->setup();
        } catch (\Exception $e) {
            self::markTestSkipped($e->getMessage());
        }
    }

    public function testConnectionSendAndGet()
    {
        $this->connection->add('{"message": "Hi"}', ['type' => DummyMessage::class]);
        $message = $this->connection->get();
        

class ConnectionTest extends TestCase
{
    private const DEFAULT_EXCHANGE_NAME = 'messages';

    public function testItCannotBeConstructedWithAWrongDsn()
    {
        $this->expectException(\InvalidArgumentException::class);
        $this->expectExceptionMessage('The given AMQP DSN is invalid.');
        Connection::fromDsn('amqp://:');
    }

    public function testItCanBeConstructedWithDefaults()
    {
        $this->assertEquals(
            new Connection([
                'host' => 'localhost',
                'port' => 5672,
                'vhost' => '/',
            ], [
                'name' => self::DEFAULT_EXCHANGE_NAME,
            ],

/** * @requires extension redis */
class ConnectionTest extends TestCase
{
    public function testFromInvalidDsn()
    {
        $this->expectException(\InvalidArgumentException::class);
        $this->expectExceptionMessage('The given Redis DSN is invalid.');

        Connection::fromDsn('redis://');
    }

    public function testFromDsn()
    {
        $this->assertEquals(
            new Connection([
                'stream' => 'queue',
                'host' => 'localhost',
                'port' => 6379,
            ], $this->createMock(\Redis::class)),
            Connection::fromDsn('redis://localhost/queue?', [], $this->createMock(\Redis::class))
        );

use Symfony\Component\Messenger\Transport\Serialization\Serializer;
use Symfony\Component\Messenger\Worker;
use Symfony\Component\Serializer as SerializerComponent;
use Symfony\Component\Serializer\Encoder\JsonEncoder;
use Symfony\Component\Serializer\Normalizer\ArrayDenormalizer;
use Symfony\Component\Serializer\Normalizer\ObjectNormalizer;

$serializer = new Serializer(
    new SerializerComponent\Serializer([new ObjectNormalizer(), new ArrayDenormalizer()], ['json' => new JsonEncoder()])
);

$connection = Connection::fromDsn(getenv('DSN'));
$receiver = new AmqpReceiver($connection, $serializer);
$eventDispatcher = new EventDispatcher();
$eventDispatcher->addSubscriber(new StopWorkerOnSignalsListener());
$eventDispatcher->addSubscriber(new DispatchPcntlSignalListener());

$worker = new Worker(['the_receiver' => $receiver], new class() implements MessageBusInterface {
    public function dispatch($envelope, array $stamps = []): Envelope
    {
        echo 'Get envelope with message: '.$envelope->getMessage()::class."\n";
        echo sprintf("with stamps: %s\n", json_encode(array_keys($envelope->all()), \JSON_PRETTY_PRINT));

        

$this->assertFalse($factory->supports('invalid-dsn', []));
    }

    /** * @requires extension amqp */
    public function testItCreatesTheTransport()
    {
        $factory = new AmqpTransportFactory();
        $serializer = $this->createMock(SerializerInterface::class);

        $expectedTransport = new AmqpTransport(Connection::fromDsn('amqp://localhost', ['host' => 'localhost']), $serializer);

        $this->assertEquals($expectedTransport, $factory->createTransport('amqp://localhost', ['host' => 'localhost'], $serializer));
    }
}

$this->assertTrue($factory->supports('beanstalkd://127.0.0.1', []));
        $this->assertFalse($factory->supports('doctrine://127.0.0.1', []));
    }

    public function testCreateTransport()
    {
        $factory = new BeanstalkdTransportFactory();
        $serializer = $this->createMock(SerializerInterface::class);

        $this->assertEquals(
            new BeanstalkdTransport(Connection::fromDsn('beanstalkd://127.0.0.1'), $serializer),
            $factory->createTransport('beanstalkd://127.0.0.1', [], $serializer)
        );
    }
}

parent::setUp();

        if (!getenv('MESSENGER_AMQP_DSN')) {
            $this->markTestSkipped('The "MESSENGER_AMQP_DSN" environment variable is required.');
        }
    }

    public function testItSendsAndReceivesMessages()
    {
        $serializer = $this->createSerializer();

        $connection = Connection::fromDsn(getenv('MESSENGER_AMQP_DSN'));
        $connection->setup();
        $connection->purgeQueues();

        $sender = new AmqpSender($connection, $serializer);
        $receiver = new AmqpReceiver($connection, $serializer);

        $sender->send($first = new Envelope(new DummyMessage('First')));
        $sender->send($second = new Envelope(new DummyMessage('Second')));

        $envelopes = iterator_to_array($receiver->get());
        $this->assertCount(1, $envelopes);
        

yield 'mixed transport' => [
            'roundrobin(dummy://a failover(dummy://b dummy://a) dummy://b)',
            new RoundRobinTransport([$transportA, new FailoverTransport([$transportB, $transportA]), $transportB]),
        ];
    }

    /** * @dataProvider fromDsnProvider */
    public function testFromDsn(string $dsn, TransportInterface $transport)
    {
        $this->assertEquals($transport, Transport::fromDsn($dsn));
    }

    public static function fromDsnProvider(): iterable
    {
        yield 'multiple transports' => [
            'failover(smtp://a smtp://b)',
            new FailoverTransport([new Transport\Smtp\EsmtpTransport('a'), new Transport\Smtp\EsmtpTransport('b')]),
        ];
    }

    /** * @dataProvider fromWrongStringProvider */

use Psr\Log\NullLogger;
use Symfony\Component\HttpClient\MockHttpClient;
use Symfony\Component\HttpClient\Response\MockResponse;
use Symfony\Component\Messenger\Bridge\AmazonSqs\Transport\Connection;
use Symfony\Contracts\HttpClient\HttpClientInterface;

class ConnectionTest extends TestCase
{
    public function testExtraOptions()
    {
        $this->expectException(\InvalidArgumentException::class);
        Connection::fromDsn('sqs://default/queue', [
            'extra_key',
        ]);
    }

    public function testExtraParamsInQuery()
    {
        $this->expectException(\InvalidArgumentException::class);
        Connection::fromDsn('sqs://default/queue?extra_param=some_value');
    }

    public function testConfigureWithCredentials()
    {

private ?LoggerInterface $logger;

    public function __construct(LoggerInterface $logger = null)
    {
        $this->logger = $logger;
    }

    public function createTransport(#[\SensitiveParameter] string $dsn, array $options, SerializerInterface $serializer): TransportInterface     {
        unset($options['transport_name']);

        return new AmazonSqsTransport(Connection::fromDsn($dsn, $options, null, $this->logger), $serializer);
    }

    public function supports(#[\SensitiveParameter] string $dsn, array $options): bool     {
        return str_starts_with($dsn, 'sqs://') || preg_match('#^https://sqs\.[\w\-]+\.amazonaws\.com/.+#', $dsn);
    }
}

/** * @author Samuel Roze <samuel.roze@gmail.com> * * @implements TransportFactoryInterface<AmqpTransport> */
class AmqpTransportFactory implements TransportFactoryInterface
{
    public function createTransport(#[\SensitiveParameter] string $dsn, array $options, SerializerInterface $serializer): TransportInterface     {
        unset($options['transport_name']);

        return new AmqpTransport(Connection::fromDsn($dsn, $options), $serializer);
    }

    public function supports(#[\SensitiveParameter] string $dsn, array $options): bool     {
        return str_starts_with($dsn, 'amqp://') || str_starts_with($dsn, 'amqps://');
    }
}

use Symfony\Component\Messenger\Bridge\Beanstalkd\Transport\Connection;
use Symfony\Component\Messenger\Exception\InvalidArgumentException as MessengerInvalidArgumentException;
use Symfony\Component\Messenger\Exception\TransportException;

final class ConnectionTest extends TestCase
{
    public function testFromInvalidDsn()
    {
        $this->expectException(\InvalidArgumentException::class);
        $this->expectExceptionMessage('The given Beanstalkd DSN is invalid.');

        Connection::fromDsn('beanstalkd://');
    }

    public function testFromDsn()
    {
        $this->assertEquals(
            $connection = new Connection([], Pheanstalk::create('127.0.0.1', 11300)),
            Connection::fromDsn('beanstalkd://127.0.0.1')
        );

        $configuration = $connection->getConfiguration();

        

public function testConnectionSendAndGet()
    {
        if (!getenv('MESSENGER_SQS_DSN')) {
            $this->markTestSkipped('The "MESSENGER_SQS_DSN" environment variable is required.');
        }

        $this->execute(getenv('MESSENGER_SQS_DSN'));
    }

    private function execute(string $dsn): void
    {
        $connection = Connection::fromDsn($dsn, []);
        $connection->setup();
        $this->clearSqs($dsn);

        $connection->send('{"message": "Hi"}', ['type' => DummyMessage::class, DummyMessage::class => 'special']);
        $this->assertSame(1, $connection->getMessageCount());

        $wait = 0;
        while ((null === $encoded = $connection->get()) && $wait++ < 200) {
            usleep(5000);
        }

        

class RedisTransportFactory implements TransportFactoryInterface
{
    public function createTransport(#[\SensitiveParameter] string $dsn, array $options, SerializerInterface $serializer): TransportInterface     {
        unset($options['transport_name']);

        return new RedisTransport(Connection::fromDsn($dsn, $options), $serializer);
    }

    public function supports(#[\SensitiveParameter] string $dsn, array $options): bool     {
        return str_starts_with($dsn, 'redis://') || str_starts_with($dsn, 'rediss://');
    }
}