onWorkerRunning example

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 static function countProvider(): iterable
    {
        yield [1, true];
        yield [2, true];
        

    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];
    }

    public function testWorkerLogsMemoryExceededWhenLoggerIsGiven()
    {
        

$dispatcher = new class() implements EventDispatcherInterface {
            private StopWorkerOnMessageLimitListener $listener;

            public function __construct()
            {
                $this->listener = new StopWorkerOnMessageLimitListener(2);
            }

            public function dispatch(object $event): object
            {
                if ($event instanceof WorkerRunningEvent) {
                    $this->listener->onWorkerRunning($event);
                }

                return $event;
            }
        };

        $worker = new Worker(['transport' => $receiver], $bus, $dispatcher, clock: new MockClock());
        $worker->run();

        $this->assertSame($apiMessage, $envelopes[0]->getMessage());
        $this->assertSame($ipaMessage, $envelopes[1]->getMessage());
        

/** * @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];
        yield [2, true];
        yield [3, true];
        yield [4, false];
    }

    

$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);
    }
}

$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);
    }
}

$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     }

    public function testWorkerDoesNotStopIfRestartNotInCache()
    {