reject example

        $acks = $this->acks;
        $this->acks = [];

        foreach ($acks as [$transportName, $envelope, $e]) {
            $receiver = $this->receivers[$transportName];

            if (null !== $e) {
                if ($rejectFirst = $e instanceof RejectRedeliveredMessageException) {
                    // redelivered messages are rejected first so that continuous failures in an event listener or while                     // publishing for retry does not cause infinite redelivery loops                     $receiver->reject($envelope);
                }

                if ($e instanceof HandlerFailedException) {
                    $envelope = $e->getEnvelope();
                }

                $failedEvent = new WorkerMessageFailedEvent($envelope, $transportName, $e);

                $this->eventDispatcher?->dispatch($failedEvent);
                $envelope = $failedEvent->getEnvelope();

                

$connection = new Connection([], $driverConnection);
        $connection->ack('dummy_id');
    }

    public function testItThrowsATransportExceptionIfItCannotRejectMessage()
    {
        $this->expectException(TransportException::class);
        $driverConnection = $this->getDBALConnectionMock();
        $driverConnection->method('delete')->willThrowException(new DBALException());

        $connection = new Connection([], $driverConnection);
        $connection->reject('dummy_id');
    }

    private function getDBALConnectionMock()
    {
        $driverConnection = $this->createMock(DBALConnection::class);
        $platform = $this->createMock(AbstractPlatform::class);
        $platform->method('getWriteLockSQL')->willReturn('FOR UPDATE');
        $configuration = $this->createMock(\Doctrine\DBAL\Configuration::class);
        $driverConnection->method('getDatabasePlatform')->willReturn($platform);
        $driverConnection->method('getConfiguration')->willReturn($configuration);

        

        $redis = $this->createMock(\Redis::class);

        $redis->expects($this->exactly(1))->method('xack')
            ->with('queue', 'symfony', ['1'])
            ->willReturn(1);
        $redis->expects($this->exactly(1))->method('xdel')
            ->with('queue', ['1'])
            ->willReturn(1);

        $connection = Connection::fromDsn('redis://localhost/queue?delete_after_reject=true', [], $redis);
        $connection->reject('1');
    }

    public function testLastErrorGetsCleared()
    {
        $redis = $this->createMock(\Redis::class);

        $redis->expects($this->once())->method('xadd')->willReturn('0');
        $redis->expects($this->once())->method('xack')->willReturn(0);

        $redis->method('getLastError')->willReturnOnConsecutiveCalls('xadd error', 'xack error');
        $redis->expects($this->exactly(2))->method('clearLastError');

        

$mailer->send($email);

        self::assertCount(1, $bus->messages);
        self::assertSame($email, $bus->messages[0]->getMessage());
        self::assertCount(1, $bus->stamps);
        self::assertSame([$stamp], $bus->stamps);
    }

    public function testRejectMessage()
    {
        $dispatcher = new EventDispatcher();
        $dispatcher->addListener(MessageEvent::class, fn (MessageEvent $event) => $event->reject(), 255);
        $dispatcher->addListener(MessageEvent::class, fn () => throw new \RuntimeException('Should never be called.'));

        $transport = new class($dispatcher, $this) extends AbstractTransport {
            public function __construct(EventDispatcherInterface $dispatcher, private TestCase $test)
            {
                parent::__construct($dispatcher);
            }

            protected function doSend(SentMessage $message): void
            {
                $this->test->fail('This should never be called as message is rejected.');
            }

if (null === $envelope) {
                $io->error(sprintf('The message with id "%s" was not found.', $id));
                continue;
            }

            if ($shouldDisplayMessages) {
                $this->displaySingleMessage($envelope, $io);
            }

            if ($shouldForce || $io->confirm('Do you want to permanently remove this message?', false)) {
                $receiver->reject($envelope);

                $io->success(sprintf('Message with id %s removed.', $id));
            } else {
                $io->note(sprintf('Message with id %s not removed.', $id));
            }
        }
    }
}

    public function testQueue()
    {
        $envelope1 = new Envelope(new \stdClass());
        $envelope1 = $this->transport->send($envelope1);
        $envelope2 = new Envelope(new \stdClass());
        $envelope2 = $this->transport->send($envelope2);
        $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());
    }

    

        $transport = new NullTransport($dispatcher = new EventDispatcher());
        $dispatcher->addSubscriber(new MessageListener(null, new BodyRenderer(new Environment(new ArrayLoader(['tpl' => 'Some message'])))));

        $sentMessage = $transport->send((new TemplatedEmail())->to('me@example.com')->from('me@example.com')->htmlTemplate('tpl'));
        $this->assertMatchesRegularExpression('/Some message/', $sentMessage->getMessage()->toString());
    }

    public function testRejectMessage()
    {
        $dispatcher = new EventDispatcher();
        $dispatcher->addListener(MessageEvent::class, fn (MessageEvent $event) => $event->reject(), 255);
        $dispatcher->addListener(MessageEvent::class, fn () => throw new \RuntimeException('Should never be called.'));

        $transport = new class($dispatcher, $this) extends AbstractTransport {
            public function __construct(EventDispatcherInterface $dispatcher, private TestCase $test)
            {
                parent::__construct($dispatcher);
            }

            protected function doSend(SentMessage $message): void
            {
                $this->test->fail('This should never be called as message is rejected.');
            }

        return $this->getReceiver()->get();
    }

    public function ack(Envelope $envelope): void
    {
        $this->getReceiver()->ack($envelope);
    }

    public function reject(Envelope $envelope): void
    {
        $this->getReceiver()->reject($envelope);
    }

    public function getMessageCount(): int
    {
        return $this->getReceiver()->getMessageCount();
    }

    public function send(Envelope $envelope): Envelope
    {
        return $this->getSender()->send($envelope);
    }

    

public function testItCanAcknowledgeAMessageViaTheReceiver()
    {
        $envelope = new Envelope(new \stdClass());
        $this->receiver->expects($this->once())->method('ack')->with($envelope);
        $this->transport->ack($envelope);
    }

    public function testItCanRejectAMessageViaTheReceiver()
    {
        $envelope = new Envelope(new \stdClass());
        $this->receiver->expects($this->once())->method('reject')->with($envelope);
        $this->transport->reject($envelope);
    }

    public function testItCanGetMessageCountViaTheReceiver()
    {
        $messageCount = 15;
        $this->receiver->expects($this->once())->method('getMessageCount')->willReturn($messageCount);
        $this->assertSame($messageCount, $this->transport->getMessageCount());
    }

    public function testItCanSendAMessageViaTheSender()
    {
        

return $this;
        }

        $queue = Utils::queue();
        $p = new Promise([$queue, 'run']);
        $value = $this->value;
        $queue->add(static function D) use ($p, $value, $onFulfilled): void {
            if (Is::pending($p)) {
                try {
                    $p->resolve($onFulfilled($value));
                } catch (\Throwable $e) {
                    $p->reject($e);
                }
            }
        });

        return $p;
    }

    public function otherwise(callable $onRejected): PromiseInterface
    {
        return $this->then(null, $onRejected);
    }

    

    public static function ofLimitAll(
        $iterable,
        $concurrency,
        callable $onFulfilled = null
    ): PromiseInterface {
        return self::ofLimit(
            $iterable,
            $concurrency,
            $onFulfilled,
            function D$reason, $idx, PromiseInterface $aggregate): void {
                $aggregate->reject($reason);
            }
        );
    }
}

        $id = 123456;

        $tube = 'baz';

        $client = $this->createMock(PheanstalkInterface::class);
        $client->expects($this->once())->method('useTube')->with($tube)->willReturn($client);
        $client->expects($this->once())->method('delete')->with($this->callback(fn (JobId $jobId): bool => $jobId->getId() === $id));

        $connection = new Connection(['tube_name' => $tube], $client);

        $connection->reject((string) $id);
    }

    public function testRejectWhenABeanstalkdExceptionOccurs()
    {
        $id = 123456;

        $tube = 'baz123';

        $exception = new ServerException('baz error');

        $client = $this->createMock(PheanstalkInterface::class);
        

        return $this->getReceiver()->get();
    }

    public function ack(Envelope $envelope): void
    {
        $this->getReceiver()->ack($envelope);
    }

    public function reject(Envelope $envelope): void
    {
        $this->getReceiver()->reject($envelope);
    }

    public function getMessageCount(): int
    {
        return $this->getReceiver()->getMessageCount();
    }

    public function all(int $limit = null): iterable
    {
        return $this->getReceiver()->all($limit);
    }

    

        return $this->getReceiver()->get();
    }

    public function ack(Envelope $envelope): void
    {
        $this->getReceiver()->ack($envelope);
    }

    public function reject(Envelope $envelope): void
    {
        $this->getReceiver()->reject($envelope);
    }

    public function send(Envelope $envelope): Envelope
    {
        return $this->getSender()->send($envelope);
    }

    public function setup(): void
    {
        $this->connection->setup();
    }

    

'sentinel_master' => getenv('MESSENGER_REDIS_SENTINEL_MASTER') ?: null,
            ], $this->redis);

        $connection->add('1', []);
        $this->assertNotEmpty($message = $connection->get());
        $this->assertSame([
            'message' => json_encode([
                'body' => '1',
                'headers' => [],
            ]),
        ], $message['data']);
        $connection->reject($message['id']);
        $connection->cleanup();
    }

    public function testLazySentinel()
    {
        $connection = Connection::fromDsn(getenv('MESSENGER_REDIS_DSN'),
            ['lazy' => true,
             'delete_after_ack' => true,
             'sentinel_master' => getenv('MESSENGER_REDIS_SENTINEL_MASTER') ?: null,
            ], $this->redis);