Turn a stream into an iterator returning even-sized results.

@param StreamInterface $stream
@param int $bufferSize
@return \Iterator

@throws \InvalidArgumentException When the stream is not readable.
504030
public static function bufferedReader(StreamInterface $stream, $bufferSize = 8192)
	{
		if(!$stream->isReadable())
		{
			throw new \InvalidArgumentException(sprintf('Stream is not readable: %s', get_class($stream)));
		}
		
		$buffer = '';
		
		while(!$stream->eof())
		{
			$buffer .= $stream->read($bufferSize);
			
			while(strlen($buffer) >= $bufferSize)
			{
				yield (string)substr($buffer, 0, $bufferSize);
				
				$buffer = (string)substr($buffer, $bufferSize);
			}
		}
		
		if($buffer !== '')
		{
			yield $buffer;
		}
	}
256843
public function pregSplit(string $regex, int $limit = -1): StreamInterface
    {
        $strings = \preg_split($regex, $this->value, $limit);
        $stream = new Stream(self::class);

        foreach ($strings as $string) {
            $stream = $stream->add(new self($string, $this->encoding));
        }

        return $stream;
    }
1311784
public function chunk(int $size = 1): StreamInterface
    {
        $stream = new Stream(self::class);
        $string = $this;

        while ($string->length() > 0) {
            $stream = $stream->add($string->substring(0, $size));
            $string = $string->substring($size);
        }

        return $stream;
    }
117254
public function compressStream($stream)
    {
        if (!is_resource($stream)) {
            throw new InvalidArgumentException('Argument is not a resource');
        }

        rewind($stream);
        $string = stream_get_contents($stream);
        fclose($stream);
        $compressedString = $this->compressString($string);
        $stream = fopen('php://memory', 'b+');
        fwrite($stream, $compressedString);
        rewind($stream);

        return $stream;
    }
956411
public function chunk($chunksize)
    {
        if ($chunksize < 1) {
            throw new \InvalidArgumentException("'{$chunksize}' is not a valid chunk size.");
        }

        return Linq::from(new ChunkIterator($this->iterator, $chunksize));
    }
806625
public static function fromStream(StreamInterface $stream)
    {
        if (! $stream->isReadable() || ! $stream->isSeekable()) {
            throw new InvalidArgumentException('Message stream must be both readable and seekable');
        }

        $stream->rewind();

        list($version, $status, $reasonPhrase) = self::getStatusLine($stream);
        list($headers, $body)                  = self::splitStream($stream);

        return (new Response($body, $status, $headers))
            ->withProtocolVersion($version)
            ->withStatus($status, $reasonPhrase);
    }
789323
public function read(AbstractStream $stream)
    {
        return $stream->getBuffer()->read($this->getSize(), $this->getEndian());
    }
1350277
public function run(): StreamInterface
    {
        // Extract some important objects
        $router = $this->container('router');
        $request = $this->container('request');
        $response = $this->container('response');

        // Get current matched route with and extract variables with callback
        $route = $router->getRoute();

        return $this->exec($route, $request, $response);
    }
1305430
private function newLimitStream(StreamInterface $stream, $length, $start)
    {
        return new SeekingLimitStream(
            $this->newNonClosingStream($stream),
            $length,
            $start
        );
    }
802262
public static void copy(InputStream input,
      OutputStream output,
      int bufferSize)
      throws IOException {
    byte[] buf = new byte[bufferSize];
    int bytesRead = input.read(buf);
    while (bytesRead != -1) {
      output.write(buf, 0, bytesRead);
      bytesRead = input.read(buf);
    }
    output.flush();
  }
145745
public static byte[] toByteArray(InputStream in, int bufferSize) throws IOException {
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        copyData(in, out, bufferSize);
        return out.toByteArray();
    }
1172367
public static function html(StreamInterface $input, StreamInterface $output)
    {
        $output->write(Minify_HTML::minify((string) $input, ['jsCleanComments' => true]));

        return $output;
    }
130785
protected function prepareChunk(StreamInterface $stream)
    {
        if ($this->hasSearchedUntilPos > -1 && $this->hasSearchedUntilPos < (strlen($this->workingBlob) - 1)) {
            // More work to do
            return true;
        }

        $chunk = $stream->getChunk();

        if ($chunk === false) {
            // EOF
            if ($this->hasSearchedUntilPos === -1 && strlen($this->workingBlob) > 0) {
                // EOF, but we haven't even started searching, special case that probably means we're dealing with a file of less size than the stream buffer
                // Therefore, keep looping
                return true;
            }
            return false;
        } else {
            // New chunk fetched
            if (!$this->firstNodeFound && !$this->options["extractContainer"]) {
                // Prevent a memory leak if we never find our first node, throw away our old stuff
                // but keep some letters to not cut off a first node
                $this->workingBlob = substr($this->workingBlob, -1 * strlen("<" . $this->options["uniqueNode"] . ">")) . $chunk;
            } else {
                $this->workingBlob .= $chunk;
            }
            
            return true;
        }
    }
217521
public function seek(int $offset): StreamInterface
    {
        if ( $offset < 0 || $offset + 1 > $this->getSize() ) {
            throw new StreamException(
                'offset is out of bounds'
            );
        }

        $this->position = $offset;

        return $this;
    }
1063748
protected function createStream($socket, RequestInterface $request, ResponseInterface $response): Stream
    {
        $size = null;

        if ($response->hasHeader('Content-Length')) {
            $size = (int) $response->getHeaderLine('Content-Length');
        }

        return new Stream($request, $socket, $size);
    }
74270
public static function copyContent($inputFile, $outputFile, $bufferSize = 1024)
    {
        $input = fopen($inputFile, 'r');
        $output = fopen($outputFile, "w");
        while ($data = fread($input, $bufferSize)) {
            fwrite($output, $data);
        }
        fclose($input);
        fclose($output);
    }
701757
protected function callCallback($stream, $callback, $waitTime = 250)
    {
        if (!is_callable($callback)) {
            throw new \InvalidArgumentException('$callback must be a callable');
        }

        $stdio  = ssh2_fetch_stream($stream, SSH2_STREAM_STDIO);
        $stderr = ssh2_fetch_stream($stream, SSH2_STREAM_STDERR);

        if ($this->isInBlock) {
            stream_set_blocking($stdio, 0);
            stream_set_blocking($stderr, 0);

            do {
                // Hacky way to retrieve ssh stream, which is not select()able
                usleep($waitTime * 1000);
                $io  = stream_get_contents($stdio);
                $err = stream_get_contents($stderr);
                if ($io || $err) {
                    call_user_func($callback, $io, $err);
                }
            } while ($io || $err);

            return $this;
        }

        stream_set_blocking($stdio, 1);
        stream_set_blocking($stderr, 1);
        call_user_func($callback, stream_get_contents($stdio), stream_get_contents($stderr));

        return $this;
    }