final public function run(int $maxRuntime = 300): void
    {
        if ($this->isRunning) {
            $this->logger->notice(sprintf('Consumer [%s] is already running.', $this->consumerName));
            return;
        }

        /*
         * In many distributed systems multiple consumers are created and these often get
         * started at the exact same time by something like upstart.  Randomizing start
         * times helps with processes depending on live consumers (like gearman) that
         * come and go.  Ensuring they don't come and go all close together makes it less
         * likely that no service will be available while consumers restart.
         */
        $jitter = mt_rand(0, 3000);
        usleep($jitter * 1000);

        $maxRuntime += mt_rand(0, 20);
        $maxRuntime = NumberUtils::bound($maxRuntime, 10, 86400);
        $start = time();
        $this->logger->notice(
            sprintf('Starting [%s] consumer, will run for up to [%d] seconds.', $this->consumerName, $maxRuntime)
        );
        $this->setup();

        $this->isRunning = true;
        declare(ticks=1);
        if (function_exists('pcntl_signal')) {
            pcntl_signal(SIGCHLD, SIG_IGN); // Zombies
            pcntl_signal(SIGTERM, [$this, 'handleSignals']); // Kill
            pcntl_signal(SIGINT, [$this, 'handleSignals']); // Control + C (from shell)
        }
        $dispatcher = $this->locator->getDispatcher();

        do {
            try {
                $this->work();
            } catch (\Throwable $e) {
                $this->stop();
                $this->logger->critical(
                    sprintf(
                        '%s::Consumer [%s] caught an exception, shutting down.',
                        ClassUtils::getShortName($e),
                        $this->consumerName
                    ),
                    [
                        'exception' => $e,
                        'consumer'  => $this->consumerName,
                    ]
                );

                $dispatcher->dispatch(PbjxEvents::CONSUMER_HANDLING_EXCEPTION);
                break;
            }

            if ($this->isRunning && $maxRuntime > 0 && time() - $start > $maxRuntime) {
                $this->stop();
                $this->logger->notice(
                    sprintf(
                        'Consumer [%s] has been running for more than [%d] seconds, shutting down.',
                        $this->consumerName,
                        $maxRuntime
                    )
                );
                break;
            }
        } while ($this->isRunning());

        $this->teardown();
        $dispatcher->dispatch(PbjxEvents::CONSUMER_AFTER_TEARDOWN);
        $this->stop();
    }
final public function run(int $maxRuntime = 300): void
    {
        if ($this->isRunning) {
            $this->logger->notice(sprintf('Consumer [%s] is already running.', $this->consumerName));
            return;
        }

        /*
         * In many distributed systems multiple consumers are created and these often get
         * started at the exact same time by something like upstart.  Randomizing start
         * times helps with processes depending on live consumers (like gearman) that
         * come and go.  Ensuring they don't come and go all close together makes it less
         * likely that no service will be available while consumers restart.
         */
        $jitter = mt_rand(0, 3000);
        usleep($jitter * 1000);

        $maxRuntime += mt_rand(0, 20);
        $maxRuntime = NumberUtils::bound($maxRuntime, 10, 86400);
        $start = time();
        $this->logger->notice(
            sprintf('Starting [%s] consumer, will run for up to [%d] seconds.', $this->consumerName, $maxRuntime)
        );
        $this->setup();

        $this->isRunning = true;
        declare(ticks=1);
        if (function_exists('pcntl_signal')) {
            pcntl_signal(SIGCHLD, SIG_IGN); // Zombies
            pcntl_signal(SIGTERM, [$this, 'handleSignals']); // Kill
            pcntl_signal(SIGINT, [$this, 'handleSignals']); // Control + C (from shell)
        }
        $dispatcher = $this->locator->getDispatcher();

        do {
            try {
                $this->work();
            } catch (\Throwable $e) {
                $this->stop();
                $this->logger->critical(
                    sprintf(
                        '%s::Consumer [%s] caught an exception, shutting down.',
                        ClassUtils::getShortName($e),
                        $this->consumerName
                    ),
                    [
                        'exception' => $e,
                        'consumer'  => $this->consumerName,
                    ]
                );

                $dispatcher->dispatch(PbjxEvents::CONSUMER_HANDLING_EXCEPTION);
                break;
            }

            if ($this->isRunning && $maxRuntime > 0 && time() - $start > $maxRuntime) {
                $this->stop();
                $this->logger->notice(
                    sprintf(
                        'Consumer [%s] has been running for more than [%d] seconds, shutting down.',
                        $this->consumerName,
                        $maxRuntime
                    )
                );
                break;
            }
        } while ($this->isRunning());

        $this->teardown();
        $dispatcher->dispatch(PbjxEvents::CONSUMER_AFTER_TEARDOWN);
        $this->stop();
    }
final public function run ( int $ maxRuntime = 300 ) : void { if ( $ this -> isRunning ) { $ this -> logger -> notice ( sprintf ( 'Consumer [%s] is already running.' , $ this -> consumerName ) ) ; return ; } /*
         * In many distributed systems multiple consumers are created and these often get
         * started at the exact same time by something like upstart.  Randomizing start
         * times helps with processes depending on live consumers (like gearman) that
         * come and go.  Ensuring they don't come and go all close together makes it less
         * likely that no service will be available while consumers restart.
         */ $ jitter = mt_rand ( 0 , 3000 ) ; usleep ( $ jitter * 1000 ) ; $ maxRuntime += mt_rand ( 0 , 20 ) ; $ maxRuntime = NumberUtils :: bound ( $ maxRuntime , 10 , 86400 ) ; $ start = time ( ) ; $ this -> logger -> notice ( sprintf ( 'Starting [%s] consumer, will run for up to [%d] seconds.' , $ this -> consumerName , $ maxRuntime ) ) ; $ this -> setup ( ) ; $ this -> isRunning = true ; declare ( ticks = 1 ) ; if ( function_exists ( 'pcntl_signal' ) ) { pcntl_signal ( SIGCHLD , SIG_IGN ) ; // Zombies pcntl_signal ( SIGTERM , [ $ this , 'handleSignals' ] ) ; // Kill pcntl_signal ( SIGINT , [ $ this , 'handleSignals' ] ) ; // Control + C (from shell) } $ dispatcher = $ this -> locator -> getDispatcher ( ) ; do { try { $ this -> work ( ) ; } catch ( \ Throwable $ e ) { $ this -> stop ( ) ; $ this -> logger -> critical ( sprintf ( '%s::Consumer [%s] caught an exception, shutting down.' , ClassUtils :: getShortName ( $ e ) , $ this -> consumerName ) , [ 'exception' => $ e , 'consumer' => $ this -> consumerName , ] ) ; $ dispatcher -> dispatch ( PbjxEvents :: CONSUMER_HANDLING_EXCEPTION ) ; break ; } if ( $ this -> isRunning && $ maxRuntime > 0 && time ( ) - $ start > $ maxRuntime ) { $ this -> stop ( ) ; $ this -> logger -> notice ( sprintf ( 'Consumer [%s] has been running for more than [%d] seconds, shutting down.' , $ this -> consumerName , $ maxRuntime ) ) ; break ; } } while ( $ this -> isRunning ( ) ) ; $ this -> teardown ( ) ; $ dispatcher -> dispatch ( PbjxEvents :: CONSUMER_AFTER_TEARDOWN ) ; $ this -> stop ( ) ; }
Runs the consumer until a signal is caught or the max runtime is reached.

@param int $maxRuntime
Runs the consumer until a signal is caught or the max runtime is reached .