WorkerRunningEvent example

class StopWorkerOnFailureLimitListenerTest extends TestCase
{
    /** * @dataProvider countProvider */
    public function testWorkerStopsWhenMaximumCountReached(int $max, bool $shouldStop)
    {
        $worker = $this->createMock(Worker::class);
        $worker->expects($shouldStop ? $this->atLeastOnce() : $this->never())->method('stop');

        $failedEvent = $this->createFailedEvent();
        $runningEvent = new WorkerRunningEvent($worker, false);

        $failureLimitListener = new StopWorkerOnFailureLimitListener($max);
        // simulate three messages (of which 2 failed)         $failureLimitListener->onMessageFailed($failedEvent);
        $failureLimitListener->onWorkerRunning($runningEvent);

        $failureLimitListener->onWorkerRunning($runningEvent);

        $failureLimitListener->onMessageFailed($failedEvent);
        $failureLimitListener->onWorkerRunning($runningEvent);
    }

    

    public function testWorkerStopsWhenMemoryLimitExceeded(?int $lastRestartTimeOffset, bool $shouldStop)
    {
        $cachePool = $this->createMock(CacheItemPoolInterface::class);
        $cacheItem = $this->createMock(CacheItemInterface::class);
        $cacheItem->expects($this->once())->method('isHit')->willReturn(true);
        $cacheItem->expects($this->once())->method('get')->willReturn(null === $lastRestartTimeOffset ? null : time() + $lastRestartTimeOffset);
        $cachePool->expects($this->once())->method('getItem')->willReturn($cacheItem);

        $worker = $this->createMock(Worker::class);
        $worker->expects($shouldStop ? $this->once() : $this->never())->method('stop');
        $event = new WorkerRunningEvent($worker, false);

        $stopOnSignalListener = new StopWorkerOnRestartSignalListener($cachePool);
        $stopOnSignalListener->onWorkerStarted();
        $stopOnSignalListener->onWorkerRunning($event);
    }

    public static function restartTimeProvider()
    {
        yield [null, false]; // no cached restart time, do not restart         yield [+10, true]; // 10 seconds after starting, a restart was requested         yield [-10, false]; // a restart was requested, but 10 seconds before we started

if ($queueNames) {
                    $envelopes = $receiver->getFromQueues($queueNames);
                } else {
                    $envelopes = $receiver->get();
                }

                foreach ($envelopes as $envelope) {
                    $envelopeHandled = true;

                    $this->rateLimit($transportName);
                    $this->handleMessage($envelope, $transportName);
                    $this->eventDispatcher?->dispatch(new WorkerRunningEvent($this, false));

                    if ($this->shouldStop) {
                        break 2;
                    }
                }

                // after handling a single receiver, quit and start the loop again                 // this should prevent multiple lower priority receivers from                 // blocking too long before the higher priority are checked                 if ($envelopeHandled) {
                    break;
                }

class StopWorkerOnMemoryLimitListenerTest extends TestCase
{
    /** * @dataProvider memoryProvider */
    public function testWorkerStopsWhenMemoryLimitExceeded(int $memoryUsage, int $memoryLimit, bool $shouldStop)
    {
        $memoryResolver = fn () => $memoryUsage;

        $worker = $this->createMock(Worker::class);
        $worker->expects($shouldStop ? $this->once() : $this->never())->method('stop');
        $event = new WorkerRunningEvent($worker, false);

        $memoryLimitListener = new StopWorkerOnMemoryLimitListener($memoryLimit, null, $memoryResolver);
        $memoryLimitListener->onWorkerRunning($event);
    }

    public static function memoryProvider(): iterable
    {
        yield [2048, 1024, true];
        yield [1024, 1024, false];
        yield [1024, 2048, false];
    }

    

yield [false];
    }

    /** * @dataProvider provideResetServices */
    public function testResetServices(bool $shouldReset)
    {
        $servicesResetter = $this->createMock(ServicesResetter::class);
        $servicesResetter->expects($shouldReset ? $this->once() : $this->never())->method('reset');

        $event = new WorkerRunningEvent($this->createMock(Worker::class), !$shouldReset);

        $resetListener = new ResetServicesListener($servicesResetter);
        $resetListener->resetServices($event);
    }

    public function testResetServicesAtStop()
    {
        $servicesResetter = $this->createMock(ServicesResetter::class);
        $servicesResetter->expects($this->once())->method('reset');

        $event = new WorkerStoppedEvent($this->createMock(Worker::class));

        

/** * @group time-sensitive */
    public function testWorkerStopsWhenTimeLimitIsReached()
    {
        $logger = $this->createMock(LoggerInterface::class);
        $logger->expects($this->once())->method('info')
            ->with('Worker stopped due to time limit of {timeLimit}s exceeded', ['timeLimit' => 1]);

        $worker = $this->createMock(Worker::class);
        $worker->expects($this->once())->method('stop');
        $event = new WorkerRunningEvent($worker, false);

        $timeoutListener = new StopWorkerOnTimeLimitListener(1, $logger);
        $timeoutListener->onWorkerStarted();
        sleep(2);
        $timeoutListener->onWorkerRunning($event);
    }
}

use Symfony\Component\Messenger\Worker;

class StopWorkerOnMessageLimitListenerTest extends TestCase
{
    /** * @dataProvider countProvider */
    public function testWorkerStopsWhenMaximumCountExceeded(int $max, bool $shouldStop)
    {
        $worker = $this->createMock(Worker::class);
        $worker->expects($shouldStop ? $this->atLeastOnce() : $this->never())->method('stop');
        $event = new WorkerRunningEvent($worker, false);

        $maximumCountListener = new StopWorkerOnMessageLimitListener($max);
        // simulate three messages processed         $maximumCountListener->onWorkerRunning($event);
        $maximumCountListener->onWorkerRunning($event);
        $maximumCountListener->onWorkerRunning($event);
    }

    public static function countProvider(): iterable
    {
        yield [1, true];
        

public function test(\Throwable $throwable, bool $shouldStop)
    {
        $listener = new StopWorkerOnCustomStopExceptionListener();

        $envelope = new Envelope(new \stdClass());
        $failedEvent = new WorkerMessageFailedEvent($envelope, 'my_receiver', $throwable);

        $listener->onMessageFailed($failedEvent);

        $worker = $this->createMock(Worker::class);
        $worker->expects($shouldStop ? $this->once() : $this->never())->method('stop');
        $runningEvent = new WorkerRunningEvent($worker, false);

        $listener->onWorkerRunning($runningEvent);
    }
}