Main processing loop. Also called from send(), you should call this in
your own application in order to continue reading events and responses
from ami.
332182
public function process()
    {
        $msgs = $this->getMessages();
        foreach ($msgs as $aMsg) {
            $this->logger->debug(
                '------ Received: ------ ' . "\n" . $aMsg . "\n\n"
            );
            $resPos = strpos($aMsg, 'Response:');
            $evePos = strpos($aMsg, 'Event:');
            if (($resPos !== false) &&
              (($resPos < $evePos) || $evePos === false)
            ) {
                $response = $this->messageToResponse($aMsg);
                $this->incomingQueue[$response->getActionId()] = $response;
            } elseif ($evePos !== false) {
                $event = $this->messageToEvent($aMsg);
                $response = $this->findResponse($event);
                if ($response === false || $response->isComplete()) {
                    $this->dispatch($event);
                } else {
                    $response->addEvent($event);
                }
            } else {
                // broken ami.. sending a response with events without
                // Event and ActionId
                $bMsg = 'Event: ResponseEvent' . "\r\n";
                $bMsg .= 'ActionId: ' . $this->lastActionId . "\r\n" . $aMsg;
                $event = $this->messageToEvent($bMsg);
                $response = $this->findResponse($event);
                $response->addEvent($event);
            }
            $this->logger->debug('----------------');
        }
    }
304878
public void run() {
        synchronized (this) {
            started = true;
        }
        finished = false;
        finish = false;

        try {
            while (true) {
                waitForInput(is);

                if (finish || Thread.interrupted()) {
                    break;
                }

                int availableBytes = is.available();
                byte[] buf = new byte[availableBytes];
                int actualBytesRead = is.read(buf, 0, availableBytes);

                if (actualBytesRead <= 0 || finish || Thread.interrupted()) {
                    break;
                }
                os.write(buf, 0, actualBytesRead);
                if (autoflush) {
                    os.flush();
                }
            }
            os.flush();
        } catch (InterruptedException ie) {
            // likely PumpStreamHandler trying to stop us
        } catch (Exception e) {
            synchronized (this) {
                exception = e;
            }
        } finally {
            if (closeWhenExhausted) {
                FileUtils.close(os);
            }
            finished = true;
            synchronized (this) {
                notifyAll();
            }
        }
    }
184665
public void run()
    {
        try
        {
            // Open connection to URL calling this interceptor.
            HttpURLConnection connection = (HttpURLConnection) url.openConnection();
            // Pass in the session id cookie to maintain the same session.
            log.trace("passing in cookies: ", cookies);
            connection.setRequestProperty("Cookie", cookies);
            // Invoke event in background.
            byte[] buff = new byte[1024 * 4];
            InputStream input = connection.getInputStream();
            try {
                while (input.read(buff) > -1);
            } finally {
                input.close();
            }
        }
        catch (Exception e)
        {
            // Log any exception that could have occurred.
            log.error(e);
        }
    }
1064066
def run(self):
        
        self._stop = False # Allow re-starting the event loop
        while not self._stop:
            try:
                self._buffer += self.socket.recv(4096)
            except socket.error:
                raise

            lines = self._buffer.split("\n")
            self._buffer = lines.pop() # Last line may not have been fully read
            for line in lines:
                line = line.rstrip("\r")
                _log.debug("%s --> %s", self.server.host, line)
                self.dispatch_event("LINE", line)
                self.dispatch_event("ACTIVITY")
896805
public static void main(String[] args){

        // Listeners are subscribed by passing them to the #subscribe() method
        bus.subscribe(new ListenerDefinition.SyncAsyncListener());

        // #subscribe() is idem-potent => Multiple calls to subscribe do NOT add the listener more than once (set semantics)
        Object listener = new ListenerDefinition.SyncAsyncListener();
        bus.subscribe(listener);
        bus.subscribe(listener);

        // Classes without handlers will be silently ignored
        bus.subscribe(new Object());
        bus.subscribe(new String());

        bus.publishAsync(new File("/tmp/random.csv")); //returns immediately, publication will continue asynchronously
        bus.post(new File("/tmp/random.csv")).asynchronously(); // same as above

        bus.publish("some message");   // will return after each handler has been invoked
        bus.post("some message").now(); // same as above


    }
303425
def begin_application(self, req, res):
        
        # Add the middleware processing to the event loop - this *should*
        # change the call stack so any server errors do not link back to this
        # function
        self.loop.create_task(self.http_application.handle_client_request(req, res))
371084
async def _send_loop(self):
        
        try:
            while self._connected:
                self._send(await self._send_queue.get())
                await self._writer.drain()
        except asyncio.CancelledError:
            pass
        except Exception as e:
            if isinstance(e, IOError):
                self._log.info('The server closed the connection while sending')
            else:
                self._log.exception('Unexpected exception in the send loop')

            await self.disconnect()
677339
def main_loop():
    ''''''

    global screensaver_cookie

    if not mpstate.status.setup_mode and not opts.nowait:
        for master in mpstate.mav_master:
            if master.linknum != 0:
                break
            print("Waiting for heartbeat from %s" % master.address)
            send_heartbeat(master)
            master.wait_heartbeat(timeout=0.1)
        set_stream_rates()

    while True:
        if mpstate is None or mpstate.status.exit:
            return

        # enable or disable screensaver:
        if (mpstate.settings.inhibit_screensaver_when_armed and
            screensaver_interface is not None):
            if mpstate.status.armed and screensaver_cookie is None:
                # now we can inhibit the screensaver
                screensaver_cookie = screensaver_interface.Inhibit("MAVProxy",
                                                             "Vehicle is armed")
            elif not mpstate.status.armed and screensaver_cookie is not None:
                # we can also restore it
                screensaver_interface.UnInhibit(screensaver_cookie)
                screensaver_cookie = None

        while not mpstate.input_queue.empty():
            line = mpstate.input_queue.get()
            mpstate.input_count += 1
            cmds = line.split(';')
            if len(cmds) == 1 and cmds[0] == "":
                  mpstate.empty_input_count += 1
            for c in cmds:
                process_stdin(c)

        for master in mpstate.mav_master:
            if master.fd is None:
                if master.port.inWaiting() > 0:
                    process_master(master)

        periodic_tasks()

        rin = []
        for master in mpstate.mav_master:
            if master.fd is not None and not master.portdead:
                rin.append(master.fd)
        for m in mpstate.mav_outputs:
            rin.append(m.fd)
        for sysid in mpstate.sysid_outputs:
            m = mpstate.sysid_outputs[sysid]
            rin.append(m.fd)
        if rin == []:
            time.sleep(0.0001)
            continue

        for fd in mpstate.select_extra:
            rin.append(fd)
        try:
            (rin, win, xin) = select.select(rin, [], [], mpstate.settings.select_timeout)
        except select.error:
            continue

        if mpstate is None:
            return

        for fd in rin:
            if mpstate is None:
                  return
            for master in mpstate.mav_master:
                  if fd == master.fd:
                        process_master(master)
                        if mpstate is None:
                              return
                        continue
            for m in mpstate.mav_outputs:
                if fd == m.fd:
                    process_mavlink(m)
                    if mpstate is None:
                          return
                    continue

            for sysid in mpstate.sysid_outputs:
                m = mpstate.sysid_outputs[sysid]
                if fd == m.fd:
                    process_mavlink(m)
                    if mpstate is None:
                          return
                    continue

            # this allow modules to register their own file descriptors
            # for the main select loop
            if fd in mpstate.select_extra:
                try:
                    # call the registered read function
                    (fn, args) = mpstate.select_extra[fd]
                    fn(args)
                except Exception as msg:
                    if mpstate.settings.moddebug == 1:
                        print(msg)
                    # on an exception, remove it from the select list
                    mpstate.select_extra.pop(fd)
724335
protected function onRead()
    {
        if (!empty($this->boundary)) {
            $this->req->attrs->input->parseMultipart();
        }
        if (($this->req->attrs->contentLength <= $this->readed) && !$this->EOF) {
            $this->sendEOF();
        }
    }
787928
@Override
	public void run() {
		while (running) {
			try {
				// Read the request
				service.execute(new HttpSession(this, socket.accept()));
			} catch (IOException e) {
				// Ignore mostly.
			}
		}
	}
127986
def process_client(app)
      loop do
        client = nil
        begin
          make_interruptable do
            client = @listener.synchronize do
              begin
                @listener && @listener.accept()
              rescue => e
                return # this means we've been turned off, so exit the loop.
              end
            end
            if (!client)
              return # nil return means we're quitting, exit loop.
            end
          end
          
          req = Http::Parser.new()
          buf = ""
          chunked_read(client, @timeout) do |data|
            begin
              buf << data
              req.parse!(buf)
              if (req.done?)
                process_request(client, req, app)
                req = Http::Parser.new()
                if (@listener.closed?)
                  return # ignore any more requests from this client if we're shutting down.
                end
              end
            rescue Http::ParserError => e
              err_str = "<h2>#{e.code} #{e.message}</h2>"
              client.write("HTTP/1.1 #{e.code} #{e.message}\r\n")
              client.write("Connection: close\r\n")
              client.write("Content-Length: #{err_str.length}\r\n")
              client.write("Content-Type: text/html\r\n")
              client.write("\r\n")
              client.write(err_str)
              client.close_write            
            end
          end
        rescue Errno::EPIPE
          # do nothing, just let the connection close.
        rescue SystemExit, GracefulExit
          raise # pass it on.  
        rescue Object => e
          $stderr.puts("Unhandled error #{e}:")
          e.backtrace.each do |line|
            $stderr.puts(line)
          end
        ensure
          client.close if (client && !client.closed?)
        end
      end
    end
359450
def __receiver_loop(self):
        
        log.info("Starting receiver loop")
        notify_disconnected = True
        try:
            while self.running:
                try:
                    while self.running:
                        frames = self.__read()

                        for frame in frames:
                            f = utils.parse_frame(frame)
                            if f is None:
                                continue
                            if self.__auto_decode:
                                f.body = decode(f.body)
                            self.process_frame(f, frame)
                except exception.ConnectionClosedException:
                    if self.running:
                        #
                        # Clear out any half-received messages after losing connection
                        #
                        self.__recvbuf = b''
                        self.running = False
                        notify_disconnected = True
                    break
                finally:
                    self.cleanup()
        finally:
            with self.__receiver_thread_exit_condition:
                self.__receiver_thread_exited = True
                self.__receiver_thread_exit_condition.notifyAll()
            log.info("Receiver loop ended")
            self.notify('receiver_loop_completed')
            if notify_disconnected:
                self.notify('disconnected')
            with self.__connect_wait_condition:
                self.__connect_wait_condition.notifyAll()
750882
protected void processRequest() {
        final int timeout = getHTTPContext().getReadTimeout();
        final TCPReadCompletedCallback callback = HttpICLReadCallback.getRef();

        // keep looping on processing information until we fully parse the message
        // and hand it off, or until the reads for more data go async
        VirtualConnection rc = null;
        do {
            if (handleNewInformation()) {
                // new information triggered an error message, so we're done
                return;
            }
            // Note: handleNewInformation will allocate the read buffers
            if (!isPartiallyParsed()) {
                // we're done parsing at this point. Note: cannot take any action
                // with information after this call because it may go all the way
                // from the channel above us back to the persist read, must exit
                // this callstack immediately
                handleNewRequest();
                return;
            }
            rc = this.myTSC.getReadInterface().read(1, callback, false, timeout);
        } while (null != rc);
    }
58550
function send () {
      sent = true;
      // next tick it to avoid express catching runtime error in the callback
      //   getting caught by express and triggering app's error handler
      process.nextTick(function () {
        res.emit('finish');
        cb(null, res, body);
      });
    }
501923
def run(self, stdin, callbacks):
        
        assert isinstance(stdin, Input)
        assert isinstance(callbacks, EventLoopCallbacks)
        assert not self._running

        if self.closed:
            raise Exception('Event loop already closed.')

        self._running = True
        self._callbacks = callbacks

        inputstream = InputStream(callbacks.feed_key)
        current_timeout = [INPUT_TIMEOUT]  # Nonlocal

        # Create reader class.
        stdin_reader = PosixStdinReader(stdin.fileno())

        # Only attach SIGWINCH signal handler in main thread.
        # (It's not possible to attach signal handlers in other threads. In
        # that case we should rely on a the main thread to call this manually
        # instead.)
        if in_main_thread():
            ctx = call_on_sigwinch(self.received_winch)
        else:
            ctx = DummyContext()

        def read_from_stdin():
            " Read user input. "
            # Feed input text.
            data = stdin_reader.read()
            inputstream.feed(data)

            # Set timeout again.
            current_timeout[0] = INPUT_TIMEOUT

            # Quit when the input stream was closed.
            if stdin_reader.closed:
                self.stop()

        self.add_reader(stdin, read_from_stdin)
        self.add_reader(self._schedule_pipe[0], None)

        with ctx:
            while self._running:
                # Call inputhook.
                if self._inputhook_context:
                    with TimeIt() as inputhook_timer:
                        def ready(wait):
                            " True when there is input ready. The inputhook should return control. "
                            return self._ready_for_reading(current_timeout[0] if wait else 0) != []
                        self._inputhook_context.call_inputhook(ready)
                    inputhook_duration = inputhook_timer.duration
                else:
                    inputhook_duration = 0

                # Calculate remaining timeout. (The inputhook consumed some of the time.)
                if current_timeout[0] is None:
                    remaining_timeout = None
                else:
                    remaining_timeout = max(0, current_timeout[0] - inputhook_duration)

                # Wait until input is ready.
                fds = self._ready_for_reading(remaining_timeout)

                # When any of the FDs are ready. Call the appropriate callback.
                if fds:
                    # Create lists of high/low priority tasks. The main reason
                    # for this is to allow painting the UI to happen as soon as
                    # possible, but when there are many events happening, we
                    # don't want to call the UI renderer 1000x per second. If
                    # the eventloop is completely saturated with many CPU
                    # intensive tasks (like processing input/output), we say
                    # that drawing the UI can be postponed a little, to make
                    # CPU available. This will be a low priority task in that
                    # case.
                    tasks = []
                    low_priority_tasks = []
                    now = None  # Lazy load time. (Fewer system calls.)

                    for fd in fds:
                        # For the 'call_from_executor' fd, put each pending
                        # item on either the high or low priority queue.
                        if fd == self._schedule_pipe[0]:
                            for c, max_postpone_until in self._calls_from_executor:
                                if max_postpone_until is None:
                                    # Execute now.
                                    tasks.append(c)
                                else:
                                    # Execute soon, if `max_postpone_until` is in the future.
                                    now = now or _now()
                                    if max_postpone_until < now:
                                        tasks.append(c)
                                    else:
                                        low_priority_tasks.append((c, max_postpone_until))
                            self._calls_from_executor = []

                            # Flush all the pipe content.
                            os.read(self._schedule_pipe[0], 1024)
                        else:
                            handler = self._read_fds.get(fd)
                            if handler:
                                tasks.append(handler)

                    # When there are high priority tasks, run all these.
                    # Schedule low priority tasks for the next iteration.
                    if tasks:
                        for t in tasks:
                            t()

                        # Postpone low priority tasks.
                        for t, max_postpone_until in low_priority_tasks:
                            self.call_from_executor(t, _max_postpone_until=max_postpone_until)
                    else:
                        # Currently there are only low priority tasks -> run them right now.
                        for t, _ in low_priority_tasks:
                            t()

                else:
                    # Flush all pending keys on a timeout. (This is most
                    # important to flush the vt100 'Escape' key early when
                    # nothing else follows.)
                    inputstream.flush()

                    # Fire input timeout event.
                    callbacks.input_timeout()
                    current_timeout[0] = None

        self.remove_reader(stdin)
        self.remove_reader(self._schedule_pipe[0])

        self._callbacks = None
361146
def run(self, limit=None, timeout=None):
        
        counter = 0
        self.stream = sseclient.SSEClient(self.endpoint)
        while True:
            with utils.Timeout(timeout):
                try:
                    event = next(self.stream)
                except StopIteration:
                    continue
                else:
                    if not event.data:
                        continue
                    self.main_callback(event.data)
                    counter += 1
                    if limit and counter >= limit:
                        return
332554
def process(self):
        

        while self.running:
            if self._processor_lock.acquire(blocking=False):

                if self.ping_timer:
                    try:
                        self._check_ping()
                    except TimeoutError:
                        log.exception("BitfinexWSS.ping(): TimedOut! (%ss)" %
                                      self.ping_timer)
                    except (WebSocketConnectionClosedException,
                            ConnectionResetError):
                        log.exception("BitfinexWSS.ping(): Connection Error!")
                        self.conn = None
                if not self.conn:
                    # The connection was killed - initiate restart
                    self._controller_q.put('restart')

                skip_processing = False

                try:
                    ts, data = self.receiver_q.get(timeout=0.1)
                except queue.Empty:
                    skip_processing = True
                    ts = time.time()
                    data = None

                if not skip_processing:
                    log.debug("Processing Data: %s", data)
                    if isinstance(data, list):
                        self.handle_data(ts, data)
                    else:  # Not a list, hence it could be a response
                        try:
                            self.handle_response(ts, data)
                        except UnknownEventError:

                            # We don't know what event this is- Raise an
                            # error & log data!
                            log.exception("main() - UnknownEventError: %s",
                                          data)
                            log.info("main() - Shutting Down due to "
                                     "Unknown Error!")
                            self._controller_q.put('stop')
                        except ConnectionResetError:
                            log.info("processor Thread: Connection Was reset, "
                                     "initiating restart")
                            self._controller_q.put('restart')

                self._check_heartbeats(ts)
                self._processor_lock.release()
            else:
                time.sleep(0.5)