Method called everytime stream is considered "read ready".
Please note that method can be also called for stream errors (e.g. remote disconnection) - it's how streams are
handled by PHP itself.

@return void
770869
public function doRead()
    {
        $data = fread($this->stream, static::READ_CHUNK_SIZE);

        if ($data === '') {
            fclose($this->stream);
            $this->stream = null; //fclose() will only leave stream resource in unknown state

        } elseif ($this->isDegenerated === false) {
            $this->readBuffer .= $data;

            //For explanation read "important note" for processInputBuffer()
            //@formatter:off
            while ($this->processInputBuffer() === false);
            //@formatter:on
        }
    }
879358
public function onRead()
    {
        $packet = $this->read(1024);
        $orig = $packet;
        $type = Binary::getByte($packet);
        $code = Binary::getByte($packet);
        $checksum = Binary::getStrWord($packet);
        $id = Binary::getWord($packet);
        $seq = Binary::getWord($packet);
        if ($checksum !== self::checksum(substr_replace($orig, "\x00\x00", 2, 2))) {
            $status = 'badChecksum';
        } elseif ($type === 0x03) {
            $status = isset(static::$unreachableCodes[$code]) ? static::$unreachableCodes[$code] : 'unk' . $code . 'unreachable';
        } else {
            $status = 'unknownType0x' . dechex($type);
        }
        while (!$this->onResponse->isEmpty()) {
            $el = $this->onResponse->shift();
            if ($el instanceof CallbackWrapper) {
                $el = $el->unwrap();
            }
            list($cb, $st) = $el;
            $cb(microtime(true) - $st, $status);
        }
        $this->finish();
    }
162928
protected function ready()
    {
        if ($this->emittedReady) {
            return false;
        }

        $this->logger->info('client is ready');
        $this->emit('ready', [$this]);

        foreach ($this->unparsedPackets as $parser) {
            $parser();
        }
    }
498143
public function onReady()
    {
        \PHPDaemon\Clients\IMAP\Pool::getInstance()->open(
            $this->config->host->value,
            $this->config->login->value,
            $this->config->password->value,
            function ($conn) {
                if (!$conn) {
                    $this->log('Fail to open IMAP connection');
                    return;
                }
                $this->log('open IMAP connection success');
                $conn->getRawMessage(
                    function ($conn, $isSuccess, $raw) {
                        $this->log(print_r($raw, true));
                        $conn->logout();
                    },
                    1,
                    false
                );
            },
            true
        );
    }
426238
public function onReady()
    {
        $this->createXMLStream();
        $this->startXMLStream();
        $this->keepaliveTimer = setTimeout(function ($timer) {
            $this->ping();
        }, 1e6 * 30);
    }
719725
protected function onRead()
    {
        start:
        if ($this->state === static::STATE_STANDBY) { // outside of packet
            while (($l = $this->readline()) !== null) {
                if ($l === '') {
                    continue;
                }
                $char = $l{0};
                $val = mb_orig_substr($l, 1);
                if ($char === ':') { // inline integer
                    $this->pushValue((int)$val);
                    goto start;
                } elseif (($char === '+') || ($char === '-')) { // inline string
                    $this->error = $char === '-';
                    $this->pushValue($val);
                    goto start;
                } elseif ($char === '*') { // defines number of elements of incoming array
                    $length = (int)$val;
                    if ($length <= 0) {
                        $this->pushValue([]);
                        goto start;
                    }

                    $ptr = [];

                    if (is_array($this->ptr)) {
                        $this->ptr[] =& $ptr;
                    }

                    $this->ptr =& $ptr;
                    $this->stack[] = [&$ptr, $length];
                    $this->levelLength = $length;
                    unset($ptr);

                    goto start;
                } elseif ($char === '$') { // defines size of the data block
                    if ($l{1} === '-') {
                        $this->pushValue(null);
                        goto start;
                    }
                    $this->valueLength = (int)$val;
                    if ($this->valueLength + 2 > $this->pool->maxAllowedPacket) {
                        $this->log('max-allowed-packet (' . $this->pool->config->maxallowedpacket->getHumanValue() . ') exceed, aborting connection');
                        $this->finish();
                        return;
                    }
                    $this->setWatermark($this->valueLength + 2);
                    $this->state = static::STATE_BINARY; // binary data block
                    break; // stop reading line-by-line
                }
            }
        }

        if ($this->state === static::STATE_BINARY) { // inside of binary data block
            if ($this->getInputLength() < $this->valueLength + 2) {
                return; //we do not have a whole packet
            }
            $value = $this->read($this->valueLength);
            if ($this->read(2) !== $this->EOL) {
                $this->finish();
                return;
            }
            $this->state = static::STATE_STANDBY;
            $this->setWatermark(3);
            $this->pushValue($value);
            goto start;
        }
    }
913666
public function onRead()
    {
        if (false === ($pct = $this->readExact($this->lowMark))) {
            return; // not readed yet
        }
        if ($pct === "<policy-file-request/>\x00") {
            if ($this->pool->policyData) {
                $this->write($p = $this->pool->policyData . "\x00");
            } else {
                $this->write("<error/>\x00");
            }
        }
        $this->finish();
    }
1084418
public function onReady()
    {
        if ($this->pool->identd) {
            /** @noinspection PhpParamsInspection */
            $this->getSocketName();
            $this->pool->identd->registerPair($this->locAddr, $this->locPort, ['UNIX', $this->user]);
        }
        list($this->nick, $this->realname) = explode('/', $this->path . '/John Doe');
        $this->command('USER', $this->user, 0, '*', $this->realname);
        $this->command('NICK', $this->nick);
        if (mb_orig_strlen($this->password)) {
            $this->message('NickServ', 'IDENTIFY ' . $this->password);
        }
    }
123887
public function onRead()
    {
        while (!is_null($line = $this->readline())) {
            $finish =
                (mb_orig_strpos($line, $s = "\xff\xf4\xff\xfd\x06") !== false)
                || (mb_orig_strpos($line, $s = "\xff\xec") !== false)
                || (mb_orig_strpos($line, $s = "\x03") !== false)
                || (mb_orig_strpos($line, $s = "\x04") !== false);

            $e = explode(' ', rtrim($line, "\r\n"), 2);

            $cmd = trim($e[0]);

            if ($cmd === 'ping') {
                $this->writeln('pong');
            } elseif (
                ($cmd === 'exit')
                || ($cmd === 'quit')
            ) {
                $this->finish();
            } else {
                $this->writeln('Unknown command "' . $cmd . '"');
            }

            if (
                (mb_orig_strlen($line) > 1024)
                || $finish
            ) {
                $this->finish();
            }
        }
    }
558167
public function onReady()
    {
        if (static::$requestCommandListFlipped === null) {
            static::$requestCommandListFlipped = array_flip(static::$requestCommandList);
        }
        if (static::$responseCommandListFlipped === null) {
            static::$responseCommandListFlipped = array_flip(static::$responseCommandList);
        }
        parent::onReady();
    }
222305
public function handleRead()
    {
        try
        {
            if (!$this->reading)
            {
                return;
            }

            $length = $this->bufferSize;
            $ret = fread($this->resource, $length);

            if ($ret === false)
            {
                $this->emit('error', [ $this, new ReadException('Error occurred while reading from the stream resource.') ]);
                return;
            }

            if ($ret !== '')
            {
                $this->emit('data', [ $this, $ret ]);

                if (strlen($ret) < $length)
                {
                    $this->reading = false;
                    $this->emit('end', [ $this ]);
                }
            }
        }
        catch (Error $ex)
        {
            $this->emit('error', [ $this, $ex ]);
        }
        catch (Exception $ex)
        {
            $this->emit('error', [ $this, $ex ]);
        }
    }
764330
public function onRead()
    {
        start:
        if ($this->type === 'udp') {
            $this->onUdpPacket($this->read($this->getInputLength()));
            return;
        }
        if ($this->state === self::STATE_ROOT) {
            if (false === ($hdr = $this->readExact(2))) {
                return; // not enough data
            }
            $this->pctSize = Binary::bytes2int($hdr);
            $this->setWatermark($this->pctSize);
            $this->state = self::STATE_PACKET;
        }
        if ($this->state === self::STATE_PACKET) {
            if (false === ($pct = $this->readExact($this->pctSize))) {
                return; // not enough data
            }
            $this->state = self::STATE_ROOT;
            $this->setWatermark(2);
            $this->onUdpPacket($pct);
        }
        goto start;
    }
577052
protected function onRead()
    {
        while (($line = $this->readline()) !== null) {
            $e = explode(' , ', $line);
            if ((sizeof($e) !== 2) || !ctype_digit($e[0]) || !ctype_digit($e[1])) {
                $this->writeln($line . ' : ERROR : INVALID-PORT');
                $this->finish();
                return;
            }
            $local = (int)$e[0];
            $foreign = (int)$e[1];
            if ($user = $this->pool->findPair($local, $foreign)) {
                $this->writeln($line . ' : USERID : ' . $user);
            } else {
                $this->writeln($line . ' : ERROR : NO-USER');
            }
        }
    }
1181538
public function onRead()
    {
        if ($this->state === self::CONN_STATE_START) {
            if (($ver = $this->readline()) === null) {
                return;
            }
            $this->pool->setAmiVersion($this->addr, $ver);
            $this->state = self::CONN_STATE_GOT_INITIAL_PACKET;
            $this->auth();
        }

        while (($line = $this->readline()) !== null) {
            //Daemon::log('>>> '.$line);
            if ($line === '') {
                $this->instate = self::INPUT_STATE_END_OF_PACKET;
                $packet =& $this->packets[$this->cnt];
                ++$this->cnt;
            } else {
                $this->instate = self::INPUT_STATE_PROCESSING;
                list($header, $value) = Pool::extract($line);
                $this->packets[$this->cnt][$header] = $value;
            }

            if ((int)$this->state === self::CONN_STATE_AUTH) {
                if ($this->instate === self::INPUT_STATE_END_OF_PACKET) {
                    if ($packet['response'] === 'success') {
                        if ($this->state === self::CONN_STATE_CHALLENGE_PACKET_SENT) {
                            if (is_callable($this->onChallenge)) {
                                $func = $this->onChallenge;
                                $func($this, $packet['challenge']);
                            }
                        } else {
                            if ($packet['message'] === 'authentication accepted') {
                                $this->state = self::CONN_STATE_HANDSHAKED_OK;

                                Daemon::$process->log(
                                    __METHOD__ . ': Authentication ok. Connected to ' .
                                    parse_url($this->addr, PHP_URL_HOST)
                                );

                                if ($this->onConnected) {
                                    $this->connected = true;
                                    $this->onConnected->executeAll($this);
                                    $this->onConnected = null;
                                }

                                $this->event('connected');
                            }
                        }
                    } else {
                        $this->state = self::CONN_STATE_HANDSHAKED_ERROR;

                        Daemon::$process->log(
                            __METHOD__ . ': Authentication failed. Connection to ' .
                            parse_url($this->addr, PHP_URL_HOST) . ' failed.'
                        );

                        if ($this->onConnected) {
                            $this->connected = false;
                            $this->onConnected->executeAll($this);
                            $this->onConnected = null;
                        }

                        $this->finish();
                    }

                    $this->packets = [];
                }
            } elseif ($this->state === self::CONN_STATE_HANDSHAKED_OK) {
                if ($this->instate === self::INPUT_STATE_END_OF_PACKET) {
                    // Event
                    if (isset($packet['event']) && !isset($packet['actionid'])) {
                        $this->event('event_' . $packet['event'], $packet);
                        $this->event('event', $packet);
                    } // Response
                    elseif (isset($packet['actionid'])) {
                        $action_id =& $packet['actionid'];

                        if (isset($this->callbacks[$action_id])) {
                            if (isset($this->assertions[$action_id])) {
                                $this->packets[$action_id][] = $packet;

                                $assertations = count(
                                    array_uintersect_uassoc(
                                        $this->assertions[$action_id],
                                        $packet,
                                        'strcasecmp',
                                        'strcasecmp'
                                    )
                                );
                                if ($assertations === count($this->assertions[$action_id])) {
                                    if (is_callable($this->callbacks[$action_id])) {
                                        $this->callbacks[$action_id]($this, $this->packets[$action_id]);
                                        unset($this->callbacks[$action_id]);
                                    }

                                    unset($this->assertions[$action_id]);
                                    unset($this->packets[$action_id]);
                                }
                            } else {
                                if (is_callable($this->callbacks[$action_id])) {
                                    $this->callbacks[$action_id]($this, $packet);
                                    unset($this->callbacks[$action_id]);
                                }
                            }
                        }
                    }

                    unset($packet);
                    unset($this->packets[$this->cnt - 1]);
                }
            }
        }
    }
252450
private void doRead() {
        if (context == null) {
            context = vertx.getOrCreateContext();
        }
        if (state == STATUS_ACTIVE) {
            final Handler<Buffer> dataHandler = this.dataHandler;
            final Handler<Void> closeHandler = this.closeHandler;
            executor.submit(
                    (Runnable) () -> {
                        try {
                            final byte[] bytes = readChunk();

                            if (bytes == null || bytes.length == 0) {
                                // null or 0 means we reach the end of the stream, invoke the close handler.
                                state = STATUS_CLOSED;
                                IOUtils.closeQuietly(in);
                                context.runOnContext(event -> {
                                    if (closeHandler != null) {
                                        closeHandler.handle(null);
                                    }
                                });
                            } else {
                                // We still have data, dispatch it.
                                context.runOnContext(event -> {
                                    dataHandler.handle(Buffer.buffer(bytes));
                                    // The next chunk will be read in another call, and maybe another thread.
                                    // As the data was already given to the data handler, this is fine.
                                    doRead();
                                });
                            }
                        } catch (final Exception e) {
                            // Error detected, invokes the failure handler.
                            state = STATUS_CLOSED;
                            IOUtils.closeQuietly(in);
                            /**
                             * Invokes the failure handler.
                             * @param event irrelevant
                             */
                            context.runOnContext(event -> {
                                if (failureHandler != null) {
                                    failureHandler.handle(e);
                                }
                            });
                        }
                    });
        }
    }
1149323
public function markRead()
    {
        if ($this->receiver_status == Message::STATUS_NEW) {
            $this->receiver_status = Message::STATUS_READ;
            if ($this->save()) {
                Podium::getInstance()->podiumCache->deleteElement('user.newmessages', $this->receiver_id);
            }
        }
    }
1260118
@Override
  public boolean ready() throws IOException {
    synchronized (lock) {
      ensureOpen();

      /*
       * If newline needs to be skipped and the next char to be read is a
       * newline character, then just skip it right away.
       */
      if (skipLF) {
        /*
         * Note that in.ready() will return true if and only if the next read on
         * the stream will not block.
         */
        if (nextChar >= nChars && in.ready()) {
          fill();
        }
        if (nextChar < nChars) {
          if (cb[nextChar] == '\n')
            nextChar++;
          skipLF = false;
        }
      }
      return (nextChar < nChars) || in.ready();
    }
  }