Read data from the specified stream until it is closed by the remote
        peer. If the stream is never ended, this never returns.
9828
async def read_data(self, stream_id: int) -> bytes:
        
        frames = [f async for f in self.stream_frames(stream_id)]
        return b''.join(frames)
471291
public int read() throws IOException {
    if (this.bytesRead == this.length) {
      // All chunks and final additional chunk are read.
      // This means we have reached EOF.
      return -1;
    }

    try {
      // Read a chunk from given input stream when
      // it is first chunk or all bytes in chunk body is read
      if (this.streamBytesRead == 0 || this.chunkPos == this.chunkBody.length) {
        // Check if there are data available to read from given input stream.
        if (this.streamBytesRead != this.streamSize) {
          // Send all data chunks.
          int chunkSize = CHUNK_SIZE;
          if (this.streamBytesRead + chunkSize > this.streamSize) {
            chunkSize = this.streamSize - this.streamBytesRead;
          }

          if (readChunk(chunkSize) < 0) {
            return -1;
          }

          this.streamBytesRead += chunkSize;
        } else {
          // Send final additional chunk to complete chunk upload.
          byte[] chunk = new byte[0];
          createChunkBody(chunk);
        }
      }

      this.bytesRead++;
      // Value must be between 0 to 255.
      int value = this.chunkBody[this.chunkPos] & 0xFF;
      this.chunkPos++;
      return value;
    } catch (NoSuchAlgorithmException | InvalidKeyException | InsufficientDataException | InternalException e) {
      throw new IOException(e.getCause());
    }
  }
557852
private void readMoreBytesFromStream() throws IOException {
        if (!innerStreamHasMoreData) {
            return;
        }

        int bufferSpaceAvailable = buffer.length - bytesInBuffer;
        if (bufferSpaceAvailable <= 0) {
            return;
        }

        int bytesRead =
                stream.read(buffer, bytesInBuffer, bufferSpaceAvailable);

        if (bytesRead == -1) {
            innerStreamHasMoreData = false;
        } else {
            bytesInBuffer += bytesRead;
        }
    }
640573
public ReadStreamOld openRead()
    throws IOException
  {
    closeWrite();
    
    TempReadStream read = new TempReadStream(_head);
    read.setFreeWhenDone(true);
    _head = null;
    _tail = null;
    
    return new ReadStreamOld(read);
  }
474562
def _read(self, length):
        
        bytes = self.read(length)

        if len(bytes) != length:
            self.seek(0 - len(bytes), 1)

            raise IOError("Tried to read %d byte(s) from the stream" % length)

        return bytes
1351984
private void closeStream(InputStream stream) {

        if (stream != null) {
            try {
                stream.close();
            } catch (Exception ex) {
                LOG.error(ex.getLocalizedMessage(), ex);
            }
        }
    }
1104343
private String readAndClose(final InputStream stream) throws IOException {
		String contents = null;

		if (stream != null) {
			contents = CharStreams.toString(new InputStreamReader(new BufferedInputStream(stream), "UTF-8"));
			stream.close();
		}

		return contents;
	}
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);
                                }
                            });
                        }
                    });
        }
    }
390536
def _try_inline_read(self) -> None:
        
        # See if we've already got the data from a previous read
        pos = self._find_read_pos()
        if pos is not None:
            self._read_from_buffer(pos)
            return
        self._check_closed()
        pos = self._read_to_buffer_loop()
        if pos is not None:
            self._read_from_buffer(pos)
            return
        # We couldn't satisfy the read inline, so make sure we're
        # listening for new data unless the stream is closed.
        if not self.closed():
            self._add_io_state(ioloop.IOLoop.READ)
165498
function readStream (stream, encoding, length, limit, callback) {
  var complete = false
  var sync = true

  // check the length and limit options.
  // note: we intentionally leave the stream paused,
  // so users should handle the stream themselves.
  if (limit !== null && length !== null && length > limit) {
    return done(createError(413, 'request entity too large', {
      expected: length,
      length: length,
      limit: limit,
      type: 'entity.too.large'
    }))
  }

  // streams1: assert request encoding is buffer.
  // streams2+: assert the stream encoding is buffer.
  //   stream._decoder: streams1
  //   state.encoding: streams2
  //   state.decoder: streams2, specifically < 0.10.6
  var state = stream._readableState
  if (stream._decoder || (state && (state.encoding || state.decoder))) {
    // developer error
    return done(createError(500, 'stream encoding should not be set', {
      type: 'stream.encoding.set'
    }))
  }

  var received = 0
  var decoder

  try {
    decoder = getDecoder(encoding)
  } catch (err) {
    return done(err)
  }

  var buffer = decoder
    ? ''
    : []

  // attach listeners
  stream.on('aborted', onAborted)
  stream.on('close', cleanup)
  stream.on('data', onData)
  stream.on('end', onEnd)
  stream.on('error', onEnd)

  // mark sync section complete
  sync = false

  function done () {
    var args = new Array(arguments.length)

    // copy arguments
    for (var i = 0; i < args.length; i++) {
      args[i] = arguments[i]
    }

    // mark complete
    complete = true

    if (sync) {
      process.nextTick(invokeCallback)
    } else {
      invokeCallback()
    }

    function invokeCallback () {
      cleanup()

      if (args[0]) {
        // halt the stream on error
        halt(stream)
      }

      callback.apply(null, args)
    }
  }

  function onAborted () {
    if (complete) return

    done(createError(400, 'request aborted', {
      code: 'ECONNABORTED',
      expected: length,
      length: length,
      received: received,
      type: 'request.aborted'
    }))
  }

  function onData (chunk) {
    if (complete) return

    received += chunk.length

    if (limit !== null && received > limit) {
      done(createError(413, 'request entity too large', {
        limit: limit,
        received: received,
        type: 'entity.too.large'
      }))
    } else if (decoder) {
      buffer += decoder.write(chunk)
    } else {
      buffer.push(chunk)
    }
  }

  function onEnd (err) {
    if (complete) return
    if (err) return done(err)

    if (length !== null && received !== length) {
      done(createError(400, 'request size did not match content length', {
        expected: length,
        length: length,
        received: received,
        type: 'request.size.invalid'
      }))
    } else {
      var string = decoder
        ? buffer + (decoder.end() || '')
        : Buffer.concat(buffer)
      done(null, string)
    }
  }

  function cleanup () {
    buffer = null

    stream.removeListener('aborted', onAborted)
    stream.removeListener('data', onData)
    stream.removeListener('end', onEnd)
    stream.removeListener('error', onEnd)
    stream.removeListener('close', cleanup)
  }
}
7776
@Override
    public int read(byte[] buffer, int offset, int length) throws IOException {
        abortIfNeeded();
        int numBytesRead;
        // If no more bytes are available, do not read any bytes into the buffer
        if(this.virtualAvailable <= 0) {
            numBytesRead = -1;
        } else {
            // If the desired read length is greater than the number of available bytes,
            // shorten the read length to the number of available bytes.
            if(length > this.virtualAvailable) {
                // If the number of available bytes is greater than the maximum value of a 32 bit int, then
                // read as many bytes as an int can.
                length = (this.virtualAvailable < Integer.MAX_VALUE) ? (int)this.virtualAvailable : Integer.MAX_VALUE;
            }
            // Read bytes into the buffer.
            numBytesRead = this.decryptedContents.read(buffer, offset, length);
        }
        // If we were able to read bytes, decrement the number of bytes available to be read.
        if(numBytesRead != -1) {
            this.virtualAvailable -= numBytesRead;
        } else {
            // If we've reached the end of the stream, close it
            this.virtualAvailable = 0;
            close();
        }
        return numBytesRead;
    }
185967
private void closeStream(String msg, long length) {
    if (wrappedStream != null) {
      long remaining = remainingInCurrentRequest();
      boolean shouldAbort = remaining > readahead;
      if (!shouldAbort) {
        try {
          wrappedStream.close();
        } catch (IOException e) {
          LOG.debug("When closing {} stream for {}", uri, msg, e);
          shouldAbort = true;
        }
      }
      if (shouldAbort) {
        wrappedStream.abort();
      }
      LOG.trace("Close stream {} {}: {}; streamPos={}, nextReadPos={},"
          + " request range {}-{} length={}", uri, (shouldAbort ? "aborted" : "closed"), msg,
          pos, nextReadPos, contentRangeStart, contentRangeFinish, length);
      wrappedStream = null;
    }
  }
537169
@Override
    public int read() throws IOException {
        abortIfNeeded();
        int result;
        // If there are no more available bytes, mark that we are at the end of the stream.
        if (this.virtualAvailable <= 0) {
            result = -1;
        } else {
            // Otherwise, read a byte.
            result = this.decryptedContents.read();
        }

        // If we have not reached the end of the stream, decrement the number of available bytes.
        if (result != -1) {
            this.virtualAvailable--;
        } else {
            // If we are at the end of the stream, close it.
            this.virtualAvailable = 0;
            close();
        }

        return result;
    }
847314
public int read(byte []buf, int offset, int length) throws IOException
  {
    if (_stream != null)
      return _stream.read(buf, offset, length);
    else
      return -1;
  }
731247
private synchronized int read(long timeout, boolean isPeek) throws IOException {
        /*
         * If the thread hit an IOException, we report it.
         */
        if (exception != null) {
            assert ch == -2;
            IOException toBeThrown = exception;
            if (!isPeek)
                exception = null;
            throw toBeThrown;
        }

        /*
         * If there was a pending character from the thread, then
         * we send it. If the timeout is 0L or the thread was shut down
         * then do a local read.
         */
        if (ch >= -1) {
            assert exception == null;
        }
        else if ((timeout == 0L || isShutdown) && !threadIsReading) {
            ch = in.read();
        }
        else {
            /*
             * If the thread isn't reading already, then ask it to do so.
             */
            if (!threadIsReading) {
                threadIsReading = true;
                notify();
            }

            boolean isInfinite = (timeout <= 0L);

            /*
             * So the thread is currently doing the reading for us. So
             * now we play the waiting game.
             */
            while (isInfinite || timeout > 0L)  {
                long start = System.currentTimeMillis ();

                try {
                    wait(timeout);
                }
                catch (InterruptedException e) {
                    /* IGNORED */
                }

                if (exception != null) {
                    assert ch == -2;

                    IOException toBeThrown = exception;
                    if (!isPeek)
                        exception = null;
                    throw toBeThrown;
                }

                if (ch >= -1) {
                    assert exception == null;
                    break;
                }

                if (!isInfinite) {
                    timeout -= System.currentTimeMillis() - start;
                }
            }
        }

        /*
         * ch is the character that was just read. Either we set it because
         * a local read was performed or the read thread set it (or failed to
         * change it).  We will return it's value, but if this was a peek
         * operation, then we leave it in place.
         */
        int ret = ch;
        if (!isPeek) {
            ch = -2;
        }
        return ret;
    }
260412
public static <OUT> Iterator<OUT> collect(DataStream<OUT> stream) throws IOException {

		TypeSerializer<OUT> serializer = stream.getType().createSerializer(
				stream.getExecutionEnvironment().getConfig());

		SocketStreamIterator<OUT> iter = new SocketStreamIterator<OUT>(serializer);

		//Find out what IP of us should be given to CollectSink, that it will be able to connect to
		StreamExecutionEnvironment env = stream.getExecutionEnvironment();
		InetAddress clientAddress;

		if (env instanceof RemoteStreamEnvironment) {
			String host = ((RemoteStreamEnvironment) env).getHost();
			int port = ((RemoteStreamEnvironment) env).getPort();
			try {
				clientAddress = ConnectionUtils.findConnectingAddress(new InetSocketAddress(host, port), 2000, 400);
			}
			catch (Exception e) {
				throw new IOException("Could not determine an suitable network address to " +
						"receive back data from the streaming program.", e);
			}
		} else if (env instanceof LocalStreamEnvironment) {
			clientAddress = InetAddress.getLoopbackAddress();
		} else {
			try {
				clientAddress = InetAddress.getLocalHost();
			} catch (UnknownHostException e) {
				throw new IOException("Could not determine this machines own local address to " +
						"receive back data from the streaming program.", e);
			}
		}

		DataStreamSink<OUT> sink = stream.addSink(new CollectSink<OUT>(clientAddress, iter.getPort(), serializer));
		sink.setParallelism(1); // It would not work if multiple instances would connect to the same port

		(new CallExecute(env, iter)).start();

		return iter;
	}
161344
InputStream readChannel(int channel)
        throws IOException
    {
        while (!isClosed) {
            if (inputReady[channel]) {
                inputReady[channel] = false;
                return is;
            }

            boolean canRead = false;
            synchronized (READ_LOCK) {
                if (!isReadLocked) {
                    isReadLocked = true;
                    canRead = true;
                }
                else {
                    try {
                        READ_LOCK.wait(5000);
                    } catch (Exception e) {
                    }
                }
            }

            if (canRead) {
                try {
                    readData();
                } catch (IOException e) {
                    close();
                }
            }
        }

        return null;
    }