public function run(array $options = [], callable $onHandledCallback = null): void
    {
        $options = array_merge([
            'sleep' => 1000000,
        ], $options);

        if (\function_exists('pcntl_signal')) {
            pcntl_signal(SIGTERM, function () {
                $this->stop();
            });
        }

        $onHandled = function (?Envelope $envelope) use ($onHandledCallback) {
            if (\function_exists('pcntl_signal_dispatch')) {
                pcntl_signal_dispatch();
            }

            if (null !== $onHandledCallback) {
                $onHandledCallback($envelope);
            }
        };

        while (false === $this->shouldStop) {
            $envelopeHandled = false;
            foreach ($this->receivers as $transportName => $receiver) {
                $envelopes = $receiver->get();

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

                    $this->handleMessage($envelope, $receiver, $transportName, $this->retryStrategies[$transportName] ?? null);
                    $onHandled($envelope);
                }

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

            if (false === $envelopeHandled) {
                $onHandled(null);

                usleep($options['sleep']);
            }
        }

        $this->dispatchEvent(new WorkerStoppedEvent());
    }
public function run(array $options = [], callable $onHandledCallback = null): void
    {
        $options = array_merge([
            'sleep' => 1000000,
        ], $options);

        if (\function_exists('pcntl_signal')) {
            pcntl_signal(SIGTERM, function () {
                $this->stop();
            });
        }

        $onHandled = function (?Envelope $envelope) use ($onHandledCallback) {
            if (\function_exists('pcntl_signal_dispatch')) {
                pcntl_signal_dispatch();
            }

            if (null !== $onHandledCallback) {
                $onHandledCallback($envelope);
            }
        };

        while (false === $this->shouldStop) {
            $envelopeHandled = false;
            foreach ($this->receivers as $transportName => $receiver) {
                $envelopes = $receiver->get();

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

                    $this->handleMessage($envelope, $receiver, $transportName, $this->retryStrategies[$transportName] ?? null);
                    $onHandled($envelope);
                }

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

            if (false === $envelopeHandled) {
                $onHandled(null);

                usleep($options['sleep']);
            }
        }

        $this->dispatchEvent(new WorkerStoppedEvent());
    }
public function run ( array $ options = [ ] , callable $ onHandledCallback = null ) : void { $ options = array_merge ( [ 'sleep' => 1000000 , ] , $ options ) ; if ( \ function_exists ( 'pcntl_signal' ) ) { pcntl_signal ( SIGTERM , function ( ) { $ this -> stop ( ) ; } ) ; } $ onHandled = function ( ? Envelope $ envelope ) use ( $ onHandledCallback ) { if ( \ function_exists ( 'pcntl_signal_dispatch' ) ) { pcntl_signal_dispatch ( ) ; } if ( null !== $ onHandledCallback ) { $ onHandledCallback ( $ envelope ) ; } } ; while ( false === $ this -> shouldStop ) { $ envelopeHandled = false ; foreach ( $ this -> receivers as $ transportName => $ receiver ) { $ envelopes = $ receiver -> get ( ) ; foreach ( $ envelopes as $ envelope ) { $ envelopeHandled = true ; $ this -> handleMessage ( $ envelope , $ receiver , $ transportName , $ this -> retryStrategies [ $ transportName ] ?? null ) ; $ onHandled ( $ envelope ) ; } // 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 ; } } if ( false === $ envelopeHandled ) { $ onHandled ( null ) ; usleep ( $ options [ 'sleep' ] ) ; } } $ this -> dispatchEvent ( new WorkerStoppedEvent ( ) ) ; }
Receive the messages and dispatch them to the bus.

Valid options are:
* sleep (default: 1000000): Time in microseconds to sleep after no messages are found
Receive the messages and dispatch them to the bus .