original_string: 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());
      }
code: 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());
      }
code_tokens: 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 ( ) ) ; }
docstring: 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
docstring_tokens: Receive the messages and dispatch them to the bus .