Implements an event loop based on a buffer activity queue.

@internal
762977
public static function stream(iterable $responses, float $timeout = null): \Generator
    {
        $runningResponses = [];

        foreach ($responses as $response) {
            self::schedule($response, $runningResponses);
        }

        $lastActivity = microtime(true);
        $isTimeout = false;

        while (true) {
            $hasActivity = false;
            $timeoutMax = 0;
            $timeoutMin = $timeout ?? INF;

            /** @var ClientState $multi */
            foreach ($runningResponses as $i => [$multi]) {
                $responses = &$runningResponses[$i][1];
                self::perform($multi, $responses);

                foreach ($responses as $j => $response) {
                    $timeoutMax = $timeout ?? max($timeoutMax, $response->timeout);
                    $timeoutMin = min($timeoutMin, $response->timeout, 1);
                    $chunk = false;

                    if (isset($multi->handlesActivity[$j])) {
                        // no-op
                    } elseif (!isset($multi->openHandles[$j])) {
                        unset($responses[$j]);
                        continue;
                    } elseif ($isTimeout) {
                        $multi->handlesActivity[$j] = [new ErrorChunk($response->offset)];
                    } else {
                        continue;
                    }

                    while ($multi->handlesActivity[$j] ?? false) {
                        $hasActivity = true;
                        $isTimeout = false;

                        if (\is_string($chunk = array_shift($multi->handlesActivity[$j]))) {
                            $response->offset += \strlen($chunk);
                            $chunk = new DataChunk($response->offset, $chunk);
                        } elseif (null === $chunk) {
                            $e = $multi->handlesActivity[$j][0];
                            unset($responses[$j], $multi->handlesActivity[$j]);
                            $response->close();

                            if (null !== $e) {
                                $response->info['error'] = $e->getMessage();

                                if ($e instanceof \Error) {
                                    throw $e;
                                }

                                $chunk = new ErrorChunk($response->offset, $e);
                            } else {
                                $chunk = new LastChunk($response->offset);
                            }
                        } elseif ($chunk instanceof ErrorChunk) {
                            unset($responses[$j]);
                            $isTimeout = true;
                        } elseif ($chunk instanceof FirstChunk && $response->logger) {
                            $info = $response->getInfo();
                            $response->logger->info(sprintf('Response: "%s %s"', $info['http_code'], $info['url']));
                        }

                        yield $response => $chunk;
                    }

                    unset($multi->handlesActivity[$j]);

                    if ($chunk instanceof FirstChunk && null === $response->initializer) {
                        // Ensure the HTTP status code is always checked
                        $response->getHeaders(true);
                    } elseif ($chunk instanceof ErrorChunk && !$chunk->didThrow()) {
                        // Ensure transport exceptions are always thrown
                        $chunk->getContent();
                    }
                }

                if (!$responses) {
                    unset($runningResponses[$i]);
                }

                // Prevent memory leaks
                $multi->handlesActivity = $multi->handlesActivity ?: [];
                $multi->openHandles = $multi->openHandles ?: [];
            }

            if (!$runningResponses) {
                break;
            }

            if ($hasActivity) {
                $lastActivity = microtime(true);
                continue;
            }

            switch (self::select($multi, $timeoutMin)) {
                case -1: usleep(min(500, 1E6 * $timeoutMin)); break;
                case 0: $isTimeout = microtime(true) - $lastActivity > $timeoutMax; break;
            }
        }
    }
138400
private void executeListeners() {
    if (!monitor.isOccupiedByCurrentThread()) {
      // iterate by index to avoid concurrent modification exceptions
      for (int i = 0; i < listeners.size(); i++) {
        listeners.get(i).execute();
      }
    }
  }
616669
def set_event_loop(self, event_loop):
        """"""
        assert event_loop is None or isinstance(event_loop, AbstractEventLoop)
        self._event_loop = event_loop
286184
protected function loop(): void
    {
        $job = $this->getJob();

        if (!$job) {
            $this->sleep();

            return;
        }

        $this->log->info(sprintf('got Job %s', $job->getId()));

        $this->getEventManager()->trigger('beforeFork', $job);
        $this->setWorkingOn($job);

        $this->child = $this->fork();

        // We are the Child Process
        if ($this->child === 0 || $this->child === false) {
            $this->runAsChild($job);
        }

        // We are the parent Process
        if ($this->child > 0) {
            $this->runAsParent($job);
        }

        $this->child = null;
        $this->finishedWorking();
    }
145815
public function activate(): void
    {
        if ($this->stopBuffering) {
            $this->buffering = false;
        }
        if (!$this->handler instanceof HandlerInterface) {
            $record = end($this->buffer) ?: null;

            $this->handler = call_user_func($this->handler, $record, $this);
            if (!$this->handler instanceof HandlerInterface) {
                throw new \RuntimeException("The factory callable should return a HandlerInterface");
            }
        }
        $this->handler->handleBatch($this->buffer);
        $this->buffer = [];
    }
136795
public function run()
    {
        $this->inputAttemptsUsed = 0;
        if ($this->executeBeforeRun !== null) {
            $callback = $this->executeBeforeRun;
            $callback($this);
        }
        for ($attempts = 0; $attempts < $this->totalAttemptsForInput; $attempts++) {
            $this->doInput();
            if ($this->wasCancelled()) {
                if ($this->cancelWithInputRetriesInput && $this->hasInput()) {
                    $this->logDebug("Cancelled input, retrying");
                    continue;
                }
                $this->logDebug("Cancelled node, quitting");
                break;
            }
            // dont play on last attempt by default
            if ($this->onNoInputMessage !== null
                && !$this->hasInput()
                && ($this->playOnNoInputInLastAttempt || $attempts != ($this->totalAttemptsForInput - 1))
            ) {
                $this->addPrePromptMessage($this->onNoInputMessage);
                continue;
            }
            if ($this->hasInput()) {
                if ($this->validate()) {
                    $this->logDebug("Input validated");
                    if ($this->executeOnValidInput !== null) {
                        $callback = $this->executeOnValidInput;
                        $this->beforeOnValidInput();
                        $callback($this);
                    }
                    break;
                } elseif ($this->executeAfterFailedValidation !== null) {
                    $callback = $this->executeAfterFailedValidation;
                    $callback($this);
                }
            }
        }
        $result = $this->playPrePromptMessages();
        if ($this->minInput > 0 && $attempts == $this->totalAttemptsForInput) {
            $this->logDebug("Max attempts reached");
            $this->state = self::STATE_MAX_INPUTS_REACHED;
            if ($this->onMaxValidInputAttempts !== null) {
                $this->callClientMethods(array(
                    array('streamFile' => array($this->onMaxValidInputAttempts, self::DTMF_ANY))
                ));
            }
        }
        if (!$this->isComplete() && $this->executeOnInputFailed !== null) {
            $callback = $this->executeOnInputFailed;
            $this->beforeOnInputFailed();
            $callback($this);
        }
        if ($this->executeAfterRun !== null) {
            $callback = $this->executeAfterRun;
            $callback($this);
        }

        $this->logDebug($this);
        return $this;
    }
254554
private static function event()
    {
        $files = ['Event', 'EventListener'];
        $folder = static::$root.'Event'.'/';

        self::call($files, $folder);

        $files = ['ManyListenersArgumentsException'];
        $folder = static::$root.'Event/Exceptions'.'/';

        self::call($files, $folder);
    }
944176
public function execute()
    {
        $events = $this->eventQueue->getEvents($this->chunkSize);
        if (!empty($events)) {
            $this->eventIndex->indexEvents($events);
            $this->sessionIndex->indexEvents($events);
            $this->eventQueue->deleteEvents(array_column($events, 'event_id'));
        }
    }
79459
def process(self):
        
        assert self.queue is not None
        while True:
            event = self.queue.get()
            if self.pre_process_event(event):
                self.invoke_handlers(event)
            self.queue.task_done()
371583
private void runQueuedActions() {
        V view = viewRef == null ? null : viewRef.get();
        if (view != null) {
            while (!viewActionQueue.isEmpty()) {
                ViewAction<V> action = viewActionQueue.poll();
                action.run(view);
            }
        }
    }
1263782
protected function perform()
    {
        if (!$this->requests) {
            return;
        }

        // Initialize the handles with a very quick select timeout
        $active = $mrc = null;
        $this->executeHandles($active, $mrc, 0.001);
        $event = new Event(array('curl_multi' => $this));
        $this->processMessages();

        while ($this->requests) {

            // Notify each request as polling
            $blocking = $total = 0;
            foreach ($this->requests as $request) {
                ++$total;
                $event['request'] = $request;
                $request->getEventDispatcher()->dispatch(self::POLLING_REQUEST, $event);
                // The blocking variable just has to be non-falsey to block the loop
                if ($request->getParams()->hasKey(self::BLOCKING)) {
                    ++$blocking;
                }
            }

            if ($blocking == $total) {
                // Sleep to prevent eating CPU because no requests are actually pending a select call
                usleep(500);
            } else {
                do {
                    $this->executeHandles($active, $mrc, 1);
                } while ($active);
            }
            $this->processMessages();
        }
    }
165709
def event_loop(self):
        
        if hasattr(self.loop, '._run_once'):
            self.loop._thread_id = threading.get_ident()
            try:
                self.loop._run_once()
            finally:
                self.loop._thread_id = None
        else:
            self.loop.call_soon(self.loop.stop)
            self.loop.run_forever()
718053
private synchronized void pushEvent(Event event, boolean flushBuffer)
  {
    eventBuffer.add(event);

    if (flushBuffer || eventBuffer.size() >= maxEntries)
    {
      this.flushEventBuffer();
    }
  }
330020
public function process()
    {
        for ($i = $this->io->inReadQueue(); $i > 0; $i--) {
            $this->eventsProducer->parseLine($this->io->pop());
        }
    }
1282632
public void doEvents() {
		while (!isInterrupted()) {
			Iterator<Map.Entry<Timer, Runnable>> it = timerMap.entrySet().iterator();
			if (it.hasNext()) {
				long timeout = it.next().getKey().uptime - System.currentTimeMillis();
				doEvents(timeout > 0 ? timeout : 0);
			} else {
				doEvents(-1);
			}
		}
	}
828369
def edit_loop
      while @working
        @edit_window.sync_window(edit_buffer, edit_posn)
        @edit_window.sync_cursor(edit_posn)
        process_keystroke(MiniTerm.get_mapped_char)
      end

      edit_buffer
    end
331211
public function executeAction()
    {
        // get event queue id
        $eventQueueId = $this->getRequest()->getParam('event_queue_id');
        $this->_executeEventQueue($eventQueueId);

        // return back to the view
        $this->_redirect('*/*/');
    }