Process data; returns when the reader signals EOF.
        Callback is notified when any data changes.
337469
def process(self, data_changed_callback):
        """"""
        # pylint: disable=too-many-locals,too-many-branches,too-many-statements
        while True:
            byte = self._reader.read(1)

            while True:
                # Search for FRAME_DLE + FRAME_STX
                if not byte:
                    return
                if byte[0] == self.FRAME_DLE:
                    next_byte = self._reader.read(1)
                    if not next_byte:
                        return
                    if next_byte[0] == self.FRAME_STX:
                        break
                    else:
                        continue
                byte = self._reader.read(1)

            frame = bytearray()
            byte = self._reader.read(1)

            while True:
                if not byte:
                    return
                if byte[0] == self.FRAME_DLE:
                    # Should be FRAME_ETX or 0 according to
                    # the AQ-CO-SERIAL manual
                    next_byte = self._reader.read(1)
                    if not next_byte:
                        return
                    if next_byte[0] == self.FRAME_ETX:
                        break
                    elif next_byte[0] != 0:
                        # Error?
                        pass

                frame.append(byte[0])
                byte = self._reader.read(1)

            # Verify CRC
            frame_crc = int.from_bytes(frame[-2:], byteorder='big')
            frame = frame[:-2]

            calculated_crc = self.FRAME_DLE + self.FRAME_STX
            for byte in frame:
                calculated_crc += byte

            if frame_crc != calculated_crc:
                _LOGGER.warning('Bad CRC')
                continue

            frame_type = frame[0:2]
            frame = frame[2:]

            if frame_type == self.FRAME_TYPE_KEEP_ALIVE:
                # Keep alive
                # If a frame has been queued for transmit, send it.
                if not self._send_queue.empty():
                    data = self._send_queue.get(block=False)
                    self._writer.write(data['frame'])
                    self._writer.flush()
                    _LOGGER.info('Sent: %s', binascii.hexlify(data['frame']))

                    try:
                        if data['desired_states'] is not None:
                            # Set a timer to verify the state changes
                            # Wait 2 seconds as it can take a while for
                            # the state to change.
                            Timer(2.0, self._check_state, [data]).start()
                    except KeyError:
                        pass

                continue
            elif frame_type == self.FRAME_TYPE_KEY_EVENT:
                _LOGGER.info('Key: %s', binascii.hexlify(frame))
            elif frame_type == self.FRAME_TYPE_LEDS:
                _LOGGER.debug('LEDs: %s', binascii.hexlify(frame))
                # First 4 bytes are the LEDs that are on;
                # second 4 bytes_ are the LEDs that are flashing
                states = int.from_bytes(frame[0:4], byteorder='little')
                flashing_states = int.from_bytes(frame[4:8],
                                                 byteorder='little')
                states |= flashing_states
                if (states != self._states
                        or flashing_states != self._flashing_states):
                    self._states = states
                    self._flashing_states = flashing_states
                    data_changed_callback(self)
            elif frame_type == self.FRAME_TYPE_PUMP_SPEED_REQUEST:
                value = int.from_bytes(frame[0:2], byteorder='big')
                _LOGGER.debug('Pump speed request: %d%%', value)
                if self._pump_speed != value:
                    self._pump_speed = value
                    data_changed_callback(self)
            elif frame_type == self.FRAME_TYPE_PUMP_STATUS:
                # Pump status messages sent out by Hayward VSP pumps
                self._multi_speed_pump = True
                speed = frame[2]
                # Power is in BCD
                power = ((((frame[3] & 0xf0) >> 4) * 1000)
                         + (((frame[3] & 0x0f)) * 100)
                         + (((frame[4] & 0xf0) >> 4) * 10)
                         + (((frame[4] & 0x0f))))
                _LOGGER.debug('Pump speed: %d%%, power: %d watts',
                              speed, power)
                if self._pump_power != power:
                    self._pump_power = power
                    data_changed_callback(self)
            elif frame_type == self.FRAME_TYPE_DISPLAY_UPDATE:
                parts = frame.decode('latin-1').split()
                _LOGGER.debug('Display update: %s', parts)

                try:
                    if parts[0] == 'Pool' and parts[1] == 'Temp':
                        # Pool Temp <temp>°[C|F]
                        value = int(parts[2][:-2])
                        if self._pool_temp != value:
                            self._pool_temp = value
                            self._is_metric = parts[2][-1:] == 'C'
                            data_changed_callback(self)
                    elif parts[0] == 'Spa' and parts[1] == 'Temp':
                        # Spa Temp <temp>°[C|F]
                        value = int(parts[2][:-2])
                        if self._spa_temp != value:
                            self._spa_temp = value
                            self._is_metric = parts[2][-1:] == 'C'
                            data_changed_callback(self)
                    elif parts[0] == 'Air' and parts[1] == 'Temp':
                        # Air Temp <temp>°[C|F]
                        value = int(parts[2][:-2])
                        if self._air_temp != value:
                            self._air_temp = value
                            self._is_metric = parts[2][-1:] == 'C'
                            data_changed_callback(self)
                    elif parts[0] == 'Pool' and parts[1] == 'Chlorinator':
                        # Pool Chlorinator <value>%
                        value = int(parts[2][:-1])
                        if self._pool_chlorinator != value:
                            self._pool_chlorinator = value
                            data_changed_callback(self)
                    elif parts[0] == 'Spa' and parts[1] == 'Chlorinator':
                        # Spa Chlorinator <value>%
                        value = int(parts[2][:-1])
                        if self._spa_chlorinator != value:
                            self._spa_chlorinator = value
                            data_changed_callback(self)
                    elif parts[0] == 'Salt' and parts[1] == 'Level':
                        # Salt Level <value> [g/L|PPM|
                        value = float(parts[2])
                        if self._salt_level != value:
                            self._salt_level = value
                            self._is_metric = parts[3] == 'g/L'
                            data_changed_callback(self)
                    elif parts[0] == 'Check' and parts[1] == 'System':
                        # Check System <msg>
                        value = ' '.join(parts[2:])
                        if self._check_system_msg != value:
                            self._check_system_msg = value
                            data_changed_callback(self)
                except ValueError:
                    pass
            else:
                _LOGGER.info('Unknown frame: %s %s',
                             binascii.hexlify(frame_type),
                             binascii.hexlify(frame))
623892
public boolean checkRead()
    {
        if (!inActive) {
            return false;

        }
        if (state != State.ACTIVE && state != State.WAITING_FOR_DELIMITER) {
            return false;
        }

        //  Check if there's an item in the pipe.
        if (!inpipe.checkRead()) {
            inActive = false;
            return false;
        }

        //  If the next item in the pipe is message delimiter,
        //  initiate termination process.
        if (isDelimiter(inpipe.probe())) {
            Msg msg = inpipe.read();
            assert (msg != null);
            processDelimiter();
            return false;
        }

        return true;
    }
120932
public function eof() {
			if (!$this->hasRead) {
				$this->mark();

				$ret = fgetc($this->handle);

				if ($ret === false)
					return true;

				$this->reset();
			}

			return feof($this->handle);
		}
723127
def wait(self) -> None:
        """"""
        if self._out_thread.is_alive():
            self._out_thread.join()
        if self._err_thread.is_alive():
            self._err_thread.join()

        # Handle case where the process ended before the last read could be done.
        # This will return None for the streams that weren't pipes.
        out, err = self._proc.communicate()

        if out:
            self._write_bytes(self._stdout, out)
        if err:
            self._write_bytes(self._stderr, err)
617011
def readout(self, fifo, no_data_timeout=None):
        '''
        '''
        logging.info('Starting readout thread for %s', fifo)
        time_last_data = time()
        time_wait = 0.0
        empty_reads = 0
        while not self.force_stop[fifo].wait(time_wait if time_wait >= 0.0 else 0.0):
            time_read = time()
            try:
                if no_data_timeout and time_last_data + no_data_timeout < get_float_time():
                    raise NoDataTimeout('Received no data for %0.1f second(s) from %s' % (no_data_timeout, fifo))
                raw_data = self.read_raw_data_from_fifo(fifo)
            except NoDataTimeout:
                no_data_timeout = None  # raise exception only once
                if self.errback:
                    self.errback(sys.exc_info())
                else:
                    raise
            except Exception:
                if self.errback:
                    self.errback(sys.exc_info())
                else:
                    raise
                if self.stop_readout.is_set():  # in case of a exception, break immediately
                    break
            else:
                n_data_words = raw_data.shape[0]
                if n_data_words > 0:
                    time_last_data = time()
                    empty_reads = 0
                    time_start_read, time_stop_read = self.update_timestamp(fifo)
                    status = 0
                    self._fifo_data_deque[fifo].append((raw_data, time_start_read, time_stop_read, status))
                    with self._fifo_conditions[fifo]:
                        self._fifo_conditions[fifo].notify_all()
                elif self.stop_readout.is_set():
                    if empty_reads == self._n_empty_reads:
                        break
                    else:
                        empty_reads += 1
            finally:
                # ensure that the readout interval does not depend on the processing time of the data
                # and stays more or less constant over time
                time_wait = self.readout_interval - (time() - time_read)
        self._fifo_data_deque[fifo].append(None)  # last item, None will stop worker
        with self._fifo_conditions[fifo]:
            self._fifo_conditions[fifo].notify_all()
        logging.info('Stopping readout thread for %s', fifo)
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 ]);
        }
    }
897360
def background_reader(stream, loop: asyncio.AbstractEventLoop, callback):
    

    for line in iter(stream.readline, b''):
        loop.call_soon_threadsafe(loop.create_task, callback(line))
746114
async def read(self, *, decode: bool=False) -> Any:
        
        if self._at_eof:
            return b''
        data = bytearray()
        while not self._at_eof:
            data.extend((await self.read_chunk(self.chunk_size)))
        if decode:
            return self.decode(data)
        return data
469424
public function onRead()
    {
        if (!$this->client) {
            return;
        }
        do {
            $this->client->writeFromBuffer($this->bev->input, $this->bev->input->length);
        } while ($this->bev->input->length > 0);
    }
900941
def is_eof(self):
        
        pos = self._stream.tell()
        byte = self._stream.read(1)
        self._stream.seek(pos, 0)

        return utils.binary(byte) == utils.binary("")
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
557119
def data_received(self, data):
        
        try:
            self.data_received_count += 1
            while data:
                consumer = self.current_consumer()
                if not consumer.request:
                    consumer.start()
                toprocess = consumer.feed_data(data)
                consumer.fire_event('data_processed', data=data, exc=None)
                data = toprocess
            self.changed()
        except Exception:
            if self.transport:
                self.transport.abort()
            raise
573421
def process_data(self):

        
        self.trigger("process-data", self.received_data)
        self.received_data = ""
        if self.dispatcher:
            self.time = self.dispatcher.time
1071616
def read_blocking(self):
        
        while True:
            data = self._read()
            if data != None:
                break

        return self._parse_message(data)
191382
function processData(err, data) {
              if (enc) { file.text = data.toString(enc); } else { file.buffer = data; }
              if (entry.sha) { file.sha = entry.sha; }
              append(err, file);
              readDone();
            }
905529
function onRead( error, data ) {
		if ( error ) {
			return done( error );
		}
		data = parse( data, opts.reviver );
		if ( data instanceof Error ) {
			return done( data );
		}
		done( null, data );
	}
844529
function() {
				this.result = null;
				
				if (Basic.inArray(this.readyState, [FileReader.EMPTY, FileReader.DONE]) !== -1) {
					return;
				} else if (this.readyState === FileReader.LOADING) {
					this.readyState = FileReader.DONE;
				}

				if (_fr) {
					_fr.getRuntime().exec.call(this, 'FileReader', 'abort');
				}
				
				this.trigger('abort');
				this.trigger('loadend');
			}