DelayStamp example

$bus = new MessageBus([
            $firstMiddleware,
            $secondMiddleware,
        ]);

        $bus->dispatch($envelope);
    }

    public function testItAddsTheStamps()
    {
        $finalEnvelope = (new MessageBus())->dispatch(new \stdClass(), [new DelayStamp(5), new BusNameStamp('bar')]);
        $this->assertCount(2, $finalEnvelope->all());
    }

    public function testItAddsTheStampsToEnvelope()
    {
        $finalEnvelope = (new MessageBus())->dispatch(new Envelope(new \stdClass()), [new DelayStamp(5), new BusNameStamp('bar')]);
        $this->assertCount(2, $finalEnvelope->all());
    }

    public static function provideConstructorDataStucture(): iterable
    {
        

$connection->expects($this->once())->method('send')->with($encoded['body'], $encoded['headers'], 0);

        $serializer = $this->createMock(SerializerInterface::class);
        $serializer->method('encode')->with($envelope)->willReturnOnConsecutiveCalls($encoded);

        $sender = new BeanstalkdSender($connection, $serializer);
        $sender->send($envelope);
    }

    public function testSendWithDelay()
    {
        $envelope = (new Envelope(new DummyMessage('Oy')))->with(new DelayStamp(500));
        $encoded = ['body' => '...', 'headers' => ['type' => DummyMessage::class]];

        $connection = $this->createMock(Connection::class);
        $connection->expects($this->once())->method('send')->with($encoded['body'], $encoded['headers'], 500);

        $serializer = $this->createMock(SerializerInterface::class);
        $serializer->method('encode')->with($envelope)->willReturnOnConsecutiveCalls($encoded);

        $sender = new BeanstalkdSender($connection, $serializer);
        $sender->send($envelope);
    }
}

$retryCount = RedeliveryStamp::getRetryCountFromEnvelope($envelope);
        if ($shouldRetry) {
            $event->setForRetry();

            ++$retryCount;

            $delay = $retryStrategy->getWaitingTime($envelope, $throwable);

            $this->logger?->warning('Error thrown while handling message {class}. Sending for retry #{retryCount} using {delay} ms delay. Error: "{error}"', $context + ['retryCount' => $retryCount, 'delay' => $delay, 'error' => $throwable->getMessage(), 'exception' => $throwable]);

            // add the delay and retry stamp info             $retryEnvelope = $this->withLimitedHistory($envelope, new DelayStamp($delay), new RedeliveryStamp($retryCount));

            // re-send the message for retry             $this->getSenderForTransport($event->getReceiverName())->send($retryEnvelope);

            $this->eventDispatcher?->dispatch(new WorkerMessageRetriedEvent($retryEnvelope, $event->getReceiverName()));
        } else {
            $this->logger?->critical('Error thrown while handling message {class}. Removing from transport after {retryCount} retries. Error: "{error}"', $context + ['retryCount' => $retryCount, 'error' => $throwable->getMessage(), 'exception' => $throwable]);
        }
    }

    /** * Adds stamps to the envelope by keeping only the First + Last N stamps. */

$listener = new SendFailedMessageForRetryListener($senderLocator, $retryStrategyLocator);

        $event = new WorkerMessageFailedEvent($envelope, 'my_receiver', $exception);

        $listener->onMessageFailed($event);
    }

    public function testEnvelopeKeepOnlyTheLast10Stamps()
    {
        $exception = new \Exception('no!');
        $stamps = array_merge(
            array_fill(0, 15, new DelayStamp(1)),
            array_fill(0, 3, new RedeliveryStamp(1))
        );
        $envelope = new Envelope(new \stdClass(), $stamps);

        $sender = $this->createMock(SenderInterface::class);
        $sender->expects($this->once())->method('send')->willReturnCallback(function DEnvelope $envelope) {
            $delayStamps = $envelope->all(DelayStamp::class);
            $redeliveryStamps = $envelope->all(RedeliveryStamp::class);

            $this->assertCount(10, $delayStamps);
            $this->assertCount(4, $redeliveryStamps);

            

public function testItRoutesToTheCorrectBus()
    {
        $envelope = new Envelope(new \stdClass(), [new BusNameStamp('foo_bus')]);

        $bus1 = $this->createMock(MessageBusInterface::class);
        $bus2 = $this->createMock(MessageBusInterface::class);

        $container = $this->createMock(ContainerInterface::class);
        $container->expects($this->once())->method('has')->with('foo_bus')->willReturn(true);
        $container->expects($this->once())->method('get')->willReturn($bus2);

        $stamp = new DelayStamp(5);
        $bus1->expects($this->never())->method('dispatch');
        $bus2->expects($this->once())->method('dispatch')->with($envelope, [$stamp])->willReturn($envelope);

        $routableBus = new RoutableMessageBus($container);
        $this->assertSame($envelope, $routableBus->dispatch($envelope, [$stamp]));
    }

    public function testItRoutesToDefaultBus()
    {
        $envelope = new Envelope(new \stdClass());
        $stamp = new DelayStamp(5);
        

use Symfony\Component\Messenger\Tests\Fixtures\AnEnvelopeStamp;
use Symfony\Component\Messenger\Tests\Fixtures\DummyMessage;
use Symfony\Component\Messenger\Tests\Fixtures\TestTracesWithHandleTraitAction;
use Symfony\Component\Messenger\TraceableMessageBus;

class TraceableMessageBusTest extends TestCase
{
    public function testItTracesDispatch()
    {
        $message = new DummyMessage('Hello');

        $stamp = new DelayStamp(5);
        $bus = $this->createMock(MessageBusInterface::class);
        $bus->expects($this->once())->method('dispatch')->with($message, [$stamp])->willReturn(new Envelope($message, [$stamp]));

        $traceableBus = new TraceableMessageBus($bus);
        $line = __LINE__ + 1;
        $traceableBus->dispatch($message, [$stamp]);
        $this->assertCount(1, $tracedMessages = $traceableBus->getDispatchedMessages());
        $actualTracedMessage = $tracedMessages[0];
        unset($actualTracedMessage['callTime']); // don't check, too variable         $this->assertEquals([
            'message' => $message,
            

    public function testWithReturnsNewInstance()
    {
        $envelope = new Envelope(new DummyMessage('dummy'));

        $this->assertNotSame($envelope, $envelope->with(new ReceivedStamp('transport')));
    }

    public function testWithoutAll()
    {
        $envelope = new Envelope(new DummyMessage('dummy'), [new ReceivedStamp('transport1'), new ReceivedStamp('transport2'), new DelayStamp(5000)]);

        $envelope = $envelope->withoutAll(ReceivedStamp::class);

        $this->assertEmpty($envelope->all(ReceivedStamp::class));
        $this->assertCount(1, $envelope->all(DelayStamp::class));
    }

    public function testWithoutAllWithNonExistentStampClass()
    {
        $envelope = new Envelope(new DummyMessage('dummy'));

        

$this->assertSame([$envelope1, $envelope2], $this->transport->get());
        $this->transport->ack($envelope1);
        $this->assertSame([$envelope2], $this->transport->get());
        $this->transport->reject($envelope2);
        $this->assertSame([], $this->transport->get());
    }

    public function testQueueWithDelay()
    {
        $envelope1 = new Envelope(new \stdClass());
        $envelope1 = $this->transport->send($envelope1);
        $envelope2 = (new Envelope(new \stdClass()))->with(new DelayStamp(10_000));
        $envelope2 = $this->transport->send($envelope2);
        $this->assertSame([$envelope1], $this->transport->get());
    }

    public function testQueueWithSerialization()
    {
        $envelope = new Envelope(new \stdClass());
        $envelopeDecoded = Envelope::wrap(new DummyMessage('Hello.'));
        $this->serializer
            ->method('encode')
            ->with($this->equalTo($envelope->with(new TransportMessageIdStamp(1))))
            

$sender = new DoctrineSender($connection, $serializer);
        $actualEnvelope = $sender->send($envelope);

        /** @var TransportMessageIdStamp $transportMessageIdStamp */
        $transportMessageIdStamp = $actualEnvelope->last(TransportMessageIdStamp::class);
        $this->assertNotNull($transportMessageIdStamp);
        $this->assertSame('15', $transportMessageIdStamp->getId());
    }

    public function testSendWithDelay()
    {
        $envelope = (new Envelope(new DummyMessage('Oy')))->with(new DelayStamp(500));
        $encoded = ['body' => '...', 'headers' => ['type' => DummyMessage::class]];

        $connection = $this->createMock(Connection::class);
        $connection->expects($this->once())->method('send')->with($encoded['body'], $encoded['headers'], 500);

        $serializer = $this->createMock(SerializerInterface::class);
        $serializer->method('encode')->with($envelope)->willReturnOnConsecutiveCalls($encoded);

        $sender = new DoctrineSender($connection, $serializer);
        $sender->send($envelope);
    }
}

$failureSender = $this->failureSenders->get($event->getReceiverName());

        $envelope = $event->getEnvelope();

        // avoid re-sending to the failed sender         if (null !== $envelope->last(SentToFailureTransportStamp::class)) {
            return;
        }

        $envelope = $envelope->with(
            new SentToFailureTransportStamp($event->getReceiverName()),
            new DelayStamp(0),
            new RedeliveryStamp(0)
        );

        $this->logger?->info('Rejected message {class} will be sent to the failure transport {transport}.', [
            'class' => $envelope->getMessage()::class,
            'transport' => $failureSender::class,
        ]);

        $failureSender->send($envelope);
    }

    

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

        // send a first message         $sender->send($first = new Envelope(new DummyMessage('First')));

        // receive it immediately and imitate a redeliver with 2 second delay         $envelopes = iterator_to_array($receiver->get());
        /** @var Envelope $envelope */
        $envelope = $envelopes[0];
        $newEnvelope = $envelope
            ->with(new DelayStamp(2000))
            ->with(new RedeliveryStamp(1));
        $sender->send($newEnvelope);
        $receiver->ack($envelope);

        // send a 2nd message with a shorter delay and custom routing key         $customRoutingKeyMessage = new DummyMessage('custom routing key');
        $envelopeCustomRoutingKey = new Envelope($customRoutingKeyMessage, [
            new DelayStamp(1000),
            new AmqpStamp('my_custom_routing_key'),
        ]);
        $sender->send($envelopeCustomRoutingKey);