暂存

2026-05-04 13:15:16 +08:00
parent b8f599a617
commit 093ce2e8bc
776 changed files with 70357 additions and 16 deletions
@@ -0,0 +1,101 @@
+=========
+Changelog
+=========
+
+4.0.0 (2024-06-19)
+------------------
+
+* Changed the minimum PHP version to 7.4+
+* Created .gitattributes to avoid installing test folder
+  https://github.com/ezimuel/guzzlestreams/pull/2
+
+3.0.0 (2014-10-12)
+------------------
+
+* Now supports creating streams from functions and iterators.
+* Supports creating buffered streams and asynchronous streams.
+* Removed ``functions.php``. Use the corresponding functions provided by
+  ``GuzzleHttp\Streams\Utils`` instead.
+* Moved ``GuzzleHttp\Stream\MetadataStreamInterface::getMetadata`` to
+  ``GuzzleHttp\Stream\StreamInterface``. MetadataStreamInterface is no longer
+  used and is marked as deprecated.
+* Added ``attach()`` to ``GuzzleHttp\Stream\StreamInterface`` for PSR-7
+  compatibility.
+* Removed ``flush()`` from StreamInterface.
+* Removed the ``$maxLength`` parameter from
+  ``GuzzleHttp\Stream\StreamInterface::getContents()``. This function now
+  returns the entire remainder of the stream. If you want to limit the maximum
+  amount of data read from the stream, use the
+  ``GuzzleHttp\Stream\Utils::copyToString()`` function.
+* Streams that return an empty string, ``''``, are no longer considered a
+  failure. You MUST return ``false`` to mark the read as a failure, and ensure
+  that any decorators you create properly return ``true`` in response to the
+  ``eof()`` method when the stream is consumed.
+* ``GuzzleHttp\Stream\Stream::__construct``,
+  ``GuzzleHttp\Stream\Stream::factory``, and
+  ``GuzzleHttp\Stream\Utils::create`` no longer accept a size in the second
+  argument. They now accept an associative array of options, including the
+  "size" key and "metadata" key which can be used to provide custom metadata.
+* Added ``GuzzleHttp\Stream\BufferStream`` to add support for buffering data,
+  and when read, shifting data off of the buffer.
+* Added ``GuzzleHttp\Stream\NullBuffer`` which can be used as a buffer that
+  does not actually store any data.
+* Added ``GuzzleHttp\Stream\AsyncStream`` to provide support for non-blocking
+  streams that can be filled by a remote source (e.g., an event-loop). If a
+  ``drain`` option is provided, the stream can also act as if it is a blocking
+  stream.
+
+2.1.0 (2014-08-17)
+------------------
+
+* Added an InflateStream to inflate gzipped or deflated content.
+* Added ``flush`` to stream wrapper.
+* Added the ability to easily register the GuzzleStreamWrapper if needed.
+
+2.0.0 (2014-08-16)
+------------------
+
+* Deprecated functions.php and moved all of those methods to
+  ``GuzzleHttp\Streams\Utils``. Use ``GuzzleHttp\Stream\Stream::factory()``
+  instead of ``GuzzleHttp\Stream\create()`` to create new streams.
+* Added ``flush()`` to ``StreamInterface``. This method is used to flush any
+  underlying stream write buffers.
+* Added ``FnStream`` to easily decorate stream behavior with callables.
+* ``Utils::hash`` now throws an exception when the stream cannot seek to 0.
+
+1.5.1 (2014-09-10)
+------------------
+
+* Stream metadata is grabbed from the underlying stream each time
+  ``getMetadata`` is called rather than returning a value from a cache.
+* Properly closing all underlying streams when AppendStream is closed.
+* Seek functions no longer throw exceptions.
+* LazyOpenStream now correctly returns the underlying stream resource when
+  detached.
+
+1.5.0 (2014-08-07)
+------------------
+
+* Added ``Stream\safe_open`` to open stream resources and throw exceptions
+  instead of raising errors.
+
+1.4.0 (2014-07-19)
+------------------
+
+* Added a LazyOpenStream
+
+1.3.0 (2014-07-15)
+------------------
+
+* Added an AppendStream to stream over multiple stream one after the other.
+
+1.2.0 (2014-07-15)
+------------------
+
+* Updated the ``detach()`` method to return the underlying stream resource or
+  ``null`` if it does not wrap a resource.
+* Multiple fixes for how streams behave when the underlying resource is
+  detached
+* Do not clear statcache when a stream does not have a 'uri'
+* Added a fix to LimitStream
+* Added a condition to ensure that functions.php can be required multiple times
@@ -0,0 +1,19 @@
+Copyright (c) 2014 Michael Dowling, https://github.com/mtdowling <mtdowling@gmail.com>
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.
@@ -0,0 +1,19 @@
+all: clean coverage
+
+release: tag
+	git push origin --tags
+
+tag:
+	chag tag --sign --debug CHANGELOG.rst
+
+test:
+	vendor/bin/phpunit
+
+coverage:
+	vendor/bin/phpunit --coverage-html=artifacts/coverage
+
+view-coverage:
+	open artifacts/coverage/index.html
+
+clean:
+	rm -rf artifacts/*
@@ -0,0 +1,43 @@
+==============
+Guzzle Streams
+==============
+
+**Note:** this is a fork of the original project since it was abandoned.
+
+The main goal of this fork is to offer support for `elastic/elasticsearch-php <https://github.com/elastic/elasticsearch-php>`_
+version 7.x.
+
+## Here the original README
+
+Provides a simple abstraction over streams of data.
+
+This library is used in `Guzzle 5 <https://github.com/guzzle/guzzle>`_, and is
+(currently) compatible with the WIP PSR-7.
+
+Installation
+============
+
+This package can be installed easily using `Composer <http://getcomposer.org>`_.
+Simply add the following to the composer.json file at the root of your project:
+
+.. code-block:: javascript
+
+    {
+      "require": {
+        "guzzlehttp/streams": "~3.0"
+      }
+    }
+
+Then install your dependencies using ``composer.phar install``.
+
+Documentation
+=============
+
+The documentation for this package can be found on the main Guzzle website at
+http://docs.guzzlephp.org/en/guzzle4/streams.html.
+
+Testing
+=======
+
+This library is tested using PHPUnit. You'll need to install the dependencies
+using `Composer <http://getcomposer.org>`_ then run ``make test``.
@@ -0,0 +1,33 @@
+{
+    "name": "ezimuel/guzzlestreams",
+    "description": "Fork of guzzle/streams (abandoned) to be used with elasticsearch-php",
+    "homepage": "http://guzzlephp.org/",
+    "keywords": ["stream", "guzzle"],
+    "license": "MIT",
+    "authors": [
+        {
+            "name": "Michael Dowling",
+            "email": "mtdowling@gmail.com",
+            "homepage": "https://github.com/mtdowling"
+        }
+    ],
+    "require": {
+        "php": ">=7.4.0"
+    },
+    "require-dev": {
+        "phpunit/phpunit": "~9.0",
+        "phpstan/phpstan": "^2.1"
+    },
+    "autoload": {
+        "psr-4": { "GuzzleHttp\\Stream\\": "src/" }
+    },
+    "scripts": {
+        "test": "make test",
+        "phpstan": "vendor/bin/phpstan analyse src"
+    },
+    "extra": {
+        "branch-alias": {
+            "dev-master": "3.0-dev"
+        }
+    }
+}
@@ -0,0 +1,220 @@
+<?php
+namespace GuzzleHttp\Stream;
+
+use GuzzleHttp\Stream\Exception\CannotAttachException;
+
+/**
+ * Reads from multiple streams, one after the other.
+ *
+ * This is a read-only stream decorator.
+ */
+class AppendStream implements StreamInterface
+{
+    /** @var StreamInterface[] Streams being decorated */
+    private $streams = [];
+
+    private $seekable = true;
+    private $current = 0;
+    private $pos = 0;
+    private $detached = false;
+
+    /**
+     * @param StreamInterface[] $streams Streams to decorate. Each stream must
+     *                                   be readable.
+     */
+    public function __construct(array $streams = [])
+    {
+        foreach ($streams as $stream) {
+            $this->addStream($stream);
+        }
+    }
+
+    public function __toString()
+    {
+        try {
+            $this->seek(0);
+            return $this->getContents();
+        } catch (\Exception $e) {
+            return '';
+        }
+    }
+
+    /**
+     * Add a stream to the AppendStream
+     *
+     * @param StreamInterface $stream Stream to append. Must be readable.
+     *
+     * @throws \InvalidArgumentException if the stream is not readable
+     */
+    public function addStream(StreamInterface $stream)
+    {
+        if (!$stream->isReadable()) {
+            throw new \InvalidArgumentException('Each stream must be readable');
+        }
+
+        // The stream is only seekable if all streams are seekable
+        if (!$stream->isSeekable()) {
+            $this->seekable = false;
+        }
+
+        $this->streams[] = $stream;
+    }
+
+    public function getContents()
+    {
+        return Utils::copyToString($this);
+    }
+
+    /**
+     * Closes each attached stream.
+     *
+     * {@inheritdoc}
+     */
+    public function close()
+    {
+        $this->pos = $this->current = 0;
+
+        foreach ($this->streams as $stream) {
+            $stream->close();
+        }
+
+        $this->streams = [];
+    }
+
+    /**
+     * Detaches each attached stream
+     *
+     * {@inheritdoc}
+     */
+    public function detach()
+    {
+        $this->close();
+        $this->detached = true;
+    }
+
+    public function attach($stream)
+    {
+        throw new CannotAttachException();
+    }
+
+    public function tell()
+    {
+        return $this->pos;
+    }
+
+    /**
+     * Tries to calculate the size by adding the size of each stream.
+     *
+     * If any of the streams do not return a valid number, then the size of the
+     * append stream cannot be determined and null is returned.
+     *
+     * {@inheritdoc}
+     */
+    public function getSize()
+    {
+        $size = 0;
+
+        foreach ($this->streams as $stream) {
+            $s = $stream->getSize();
+            if ($s === null) {
+                return null;
+            }
+            $size += $s;
+        }
+
+        return $size;
+    }
+
+    public function eof()
+    {
+        return !$this->streams ||
+            ($this->current >= count($this->streams) - 1 &&
+             $this->streams[$this->current]->eof());
+    }
+
+    /**
+     * Attempts to seek to the given position. Only supports SEEK_SET.
+     *
+     * {@inheritdoc}
+     */
+    public function seek($offset, $whence = SEEK_SET)
+    {
+        if (!$this->seekable || $whence !== SEEK_SET) {
+            return false;
+        }
+
+        $success = true;
+        $this->pos = $this->current = 0;
+
+        // Rewind each stream
+        foreach ($this->streams as $stream) {
+            if (!$stream->seek(0)) {
+                $success = false;
+            }
+        }
+
+        if (!$success) {
+            return false;
+        }
+
+        // Seek to the actual position by reading from each stream
+        while ($this->pos < $offset && !$this->eof()) {
+            $this->read(min(8096, $offset - $this->pos));
+        }
+
+        return $this->pos == $offset;
+    }
+
+    /**
+     * Reads from all of the appended streams until the length is met or EOF.
+     *
+     * {@inheritdoc}
+     */
+    public function read($length)
+    {
+        $buffer = '';
+        $total = count($this->streams) - 1;
+        $remaining = $length;
+
+        while ($remaining > 0) {
+            // Progress to the next stream if needed.
+            if ($this->streams[$this->current]->eof()) {
+                if ($this->current == $total) {
+                    break;
+                }
+                $this->current++;
+            }
+            $buffer .= $this->streams[$this->current]->read($remaining);
+            $remaining = $length - strlen($buffer);
+        }
+
+        $this->pos += strlen($buffer);
+
+        return $buffer;
+    }
+
+    public function isReadable()
+    {
+        return true;
+    }
+
+    public function isWritable()
+    {
+        return false;
+    }
+
+    public function isSeekable()
+    {
+        return $this->seekable;
+    }
+
+    public function write($string)
+    {
+        return false;
+    }
+
+    public function getMetadata($key = null)
+    {
+        return $key ? null : [];
+    }
+}
@@ -0,0 +1,207 @@
+<?php
+namespace GuzzleHttp\Stream;
+
+/**
+ * Represents an asynchronous read-only stream that supports a drain event and
+ * pumping data from a source stream.
+ *
+ * The AsyncReadStream can be used as a completely asynchronous stream, meaning
+ * the data you can read from the stream will immediately return only
+ * the data that is currently buffered.
+ *
+ * AsyncReadStream can also be used in a "blocking" manner if a "pump" function
+ * is provided. When a caller requests more bytes than are available in the
+ * buffer, then the pump function is used to block until the requested number
+ * of bytes are available or the remote source stream has errored, closed, or
+ * timed-out. This behavior isn't strictly "blocking" because the pump function
+ * can send other transfers while waiting on the desired buffer size to be
+ * ready for reading (e.g., continue to tick an event loop).
+ *
+ * @unstable This class is subject to change.
+ */
+class AsyncReadStream implements StreamInterface
+{
+    use StreamDecoratorTrait;
+
+    /** @var callable|null Fn used to notify writers the buffer has drained */
+    private $drain;
+
+    /** @var callable|null Fn used to block for more data */
+    private $pump;
+
+    /** @var int|null Highwater mark of the underlying buffer */
+    private $hwm;
+
+    /** @var bool Whether or not drain needs to be called at some point */
+    private $needsDrain;
+
+    /** @var int The expected size of the remote source */
+    private $size;
+
+    /**
+     * In order to utilize high water marks to tell writers to slow down, the
+     * provided stream must answer to the "hwm" stream metadata variable,
+     * providing the high water mark. If no "hwm" metadata value is available,
+     * then the "drain" functionality is not utilized.
+     *
+     * This class accepts an associative array of configuration options.
+     *
+     * - drain: (callable) Function to invoke when the stream has drained,
+     *   meaning the buffer is now writable again because the size of the
+     *   buffer is at an acceptable level (e.g., below the high water mark).
+     *   The function accepts a single argument, the buffer stream object that
+     *   has drained.
+     * - pump: (callable) A function that accepts the number of bytes to read
+     *   from the source stream. This function will block until all of the data
+     *   that was requested has been read, EOF of the source stream, or the
+     *   source stream is closed.
+     * - size: (int) The expected size in bytes of the data that will be read
+     *   (if known up-front).
+     *
+     * @param StreamInterface $buffer   Buffer that contains the data that has
+     *                                  been read by the event loop.
+     * @param array           $config   Associative array of options.
+     *
+     * @throws \InvalidArgumentException if the buffer is not readable and
+     *                                   writable.
+     */
+    public function __construct(
+        StreamInterface $buffer,
+        array $config = []
+    ) {
+        if (!$buffer->isReadable() || !$buffer->isWritable()) {
+            throw new \InvalidArgumentException(
+                'Buffer must be readable and writable'
+            );
+        }
+
+        if (isset($config['size'])) {
+            $this->size = $config['size'];
+        }
+
+        static $callables = ['pump', 'drain'];
+        foreach ($callables as $check) {
+            if (isset($config[$check])) {
+                if (!is_callable($config[$check])) {
+                    throw new \InvalidArgumentException(
+                        $check . ' must be callable'
+                    );
+                }
+                $this->{$check} = $config[$check];
+            }
+        }
+
+        $this->hwm = $buffer->getMetadata('hwm');
+
+        // Cannot drain when there's no high water mark.
+        if ($this->hwm === null) {
+            $this->drain = null;
+        }
+
+        $this->stream = $buffer;
+    }
+
+    /**
+     * Factory method used to create new async stream and an underlying buffer
+     * if no buffer is provided.
+     *
+     * This function accepts the same options as AsyncReadStream::__construct,
+     * but added the following key value pairs:
+     *
+     * - buffer: (StreamInterface) Buffer used to buffer data. If none is
+     *   provided, a default buffer is created.
+     * - hwm: (int) High water mark to use if a buffer is created on your
+     *   behalf.
+     * - max_buffer: (int) If provided, wraps the utilized buffer in a
+     *   DroppingStream decorator to ensure that buffer does not exceed a given
+     *   length. When exceeded, the stream will begin dropping data. Set the
+     *   max_buffer to 0, to use a NullStream which does not store data.
+     * - write: (callable) A function that is invoked when data is written
+     *   to the underlying buffer. The function accepts the buffer as the first
+     *   argument, and the data being written as the second. The function MUST
+     *   return the number of bytes that were written or false to let writers
+     *   know to slow down.
+     * - drain: (callable) See constructor documentation.
+     * - pump: (callable) See constructor documentation.
+     *
+     * @param array $options Associative array of options.
+     *
+     * @return array Returns an array containing the buffer used to buffer
+     *               data, followed by the ready to use AsyncReadStream object.
+     */
+    public static function create(array $options = [])
+    {
+        $maxBuffer = isset($options['max_buffer'])
+            ? $options['max_buffer']
+            : null;
+
+        if ($maxBuffer === 0) {
+            $buffer = new NullStream();
+        } elseif (isset($options['buffer'])) {
+            $buffer = $options['buffer'];
+        } else {
+            $hwm = isset($options['hwm']) ? $options['hwm'] : 16384;
+            $buffer = new BufferStream($hwm);
+        }
+
+        if ($maxBuffer > 0) {
+            $buffer = new DroppingStream($buffer, $options['max_buffer']);
+        }
+
+        // Call the on_write callback if an on_write function was provided.
+        if (isset($options['write'])) {
+            $onWrite = $options['write'];
+            $buffer = FnStream::decorate($buffer, [
+                'write' => function ($string) use ($buffer, $onWrite) {
+                    $result = $buffer->write($string);
+                    $onWrite($buffer, $string);
+                    return $result;
+                }
+            ]);
+        }
+
+        return [$buffer, new self($buffer, $options)];
+    }
+
+    public function getSize()
+    {
+        return $this->size;
+    }
+
+    public function isWritable()
+    {
+        return false;
+    }
+
+    public function write($string)
+    {
+        return false;
+    }
+
+    public function read($length)
+    {
+        if (!$this->needsDrain && $this->drain) {
+            $this->needsDrain = $this->stream->getSize() >= $this->hwm;
+        }
+
+        $result = $this->stream->read($length);
+
+        // If we need to drain, then drain when the buffer is empty.
+        if ($this->needsDrain && $this->stream->getSize() === 0) {
+            $this->needsDrain = false;
+            $drainFn = $this->drain;
+            $drainFn($this->stream);
+        }
+
+        $resultLen = strlen($result);
+
+        // If a pump was provided, the buffer is still open, and not enough
+        // data was given, then block until the data is provided.
+        if ($this->pump && $resultLen < $length) {
+            $pumpFn = $this->pump;
+            $result .= $pumpFn($length - $resultLen);
+        }
+
+        return $result;
+    }
+}
@@ -0,0 +1,138 @@
+<?php
+namespace GuzzleHttp\Stream;
+
+use GuzzleHttp\Stream\Exception\CannotAttachException;
+
+/**
+ * Provides a buffer stream that can be written to to fill a buffer, and read
+ * from to remove bytes from the buffer.
+ *
+ * This stream returns a "hwm" metadata value that tells upstream consumers
+ * what the configured high water mark of the stream is, or the maximum
+ * preferred size of the buffer.
+ *
+ * @package GuzzleHttp\Stream
+ */
+class BufferStream implements StreamInterface
+{
+    private $hwm;
+    private $buffer = '';
+
+    /**
+     * @param int $hwm High water mark, representing the preferred maximum
+     *                 buffer size. If the size of the buffer exceeds the high
+     *                 water mark, then calls to write will continue to succeed
+     *                 but will return false to inform writers to slow down
+     *                 until the buffer has been drained by reading from it.
+     */
+    public function __construct($hwm = 16384)
+    {
+        $this->hwm = $hwm;
+    }
+
+    public function __toString()
+    {
+        return $this->getContents();
+    }
+
+    public function getContents()
+    {
+        $buffer = $this->buffer;
+        $this->buffer = '';
+
+        return $buffer;
+    }
+
+    public function close()
+    {
+        $this->buffer = '';
+    }
+
+    public function detach()
+    {
+        $this->close();
+    }
+
+    public function attach($stream)
+    {
+        throw new CannotAttachException();
+    }
+
+    public function getSize()
+    {
+        return strlen($this->buffer);
+    }
+
+    public function isReadable()
+    {
+        return true;
+    }
+
+    public function isWritable()
+    {
+        return true;
+    }
+
+    public function isSeekable()
+    {
+        return false;
+    }
+
+    public function seek($offset, $whence = SEEK_SET)
+    {
+        return false;
+    }
+
+    public function eof()
+    {
+        return strlen($this->buffer) === 0;
+    }
+
+    public function tell()
+    {
+        return false;
+    }
+
+    /**
+     * Reads data from the buffer.
+     */
+    public function read($length)
+    {
+        $currentLength = strlen($this->buffer);
+
+        if ($length >= $currentLength) {
+            // No need to slice the buffer because we don't have enough data.
+            $result = $this->buffer;
+            $this->buffer = '';
+        } else {
+            // Slice up the result to provide a subset of the buffer.
+            $result = substr($this->buffer, 0, $length);
+            $this->buffer = substr($this->buffer, $length);
+        }
+
+        return $result;
+    }
+
+    /**
+     * Writes data to the buffer.
+     */
+    public function write($string)
+    {
+        $this->buffer .= $string;
+
+        if (strlen($this->buffer) >= $this->hwm) {
+            return false;
+        }
+
+        return strlen($string);
+    }
+
+    public function getMetadata($key = null)
+    {
+        if ($key == 'hwm') {
+            return $this->hwm;
+        }
+
+        return $key ? null : [];
+    }
+}
@@ -0,0 +1,122 @@
+<?php
+namespace GuzzleHttp\Stream;
+
+use GuzzleHttp\Stream\Exception\SeekException;
+
+/**
+ * Stream decorator that can cache previously read bytes from a sequentially
+ * read stream.
+ */
+class CachingStream implements StreamInterface
+{
+    use StreamDecoratorTrait;
+
+    /** @var StreamInterface Stream being wrapped */
+    private $remoteStream;
+
+    /** @var int Number of bytes to skip reading due to a write on the buffer */
+    private $skipReadBytes = 0;
+
+    /**
+     * We will treat the buffer object as the body of the stream
+     *
+     * @param StreamInterface $stream Stream to cache
+     * @param StreamInterface $target Optionally specify where data is cached
+     */
+    public function __construct(
+        StreamInterface $stream,
+        ?StreamInterface $target = null
+    ) {
+        $this->remoteStream = $stream;
+        $this->stream = $target ?: new Stream(fopen('php://temp', 'r+'));
+    }
+
+    public function getSize()
+    {
+        return max($this->stream->getSize(), $this->remoteStream->getSize());
+    }
+
+    /**
+     * {@inheritdoc}
+     * @throws SeekException When seeking with SEEK_END or when seeking
+     *     past the total size of the buffer stream
+     */
+    public function seek($offset, $whence = SEEK_SET)
+    {
+        if ($whence == SEEK_SET) {
+            $byte = $offset;
+        } elseif ($whence == SEEK_CUR) {
+            $byte = $offset + $this->tell();
+        } else {
+            return false;
+        }
+
+        // You cannot skip ahead past where you've read from the remote stream
+        if ($byte > $this->stream->getSize()) {
+            throw new SeekException(
+                $this,
+                $byte,
+                sprintf('Cannot seek to byte %d when the buffered stream only'
+                    . ' contains %d bytes', $byte, $this->stream->getSize())
+            );
+        }
+
+        return $this->stream->seek($byte);
+    }
+
+    public function read($length)
+    {
+        // Perform a regular read on any previously read data from the buffer
+        $data = $this->stream->read($length);
+        $remaining = $length - strlen($data);
+
+        // More data was requested so read from the remote stream
+        if ($remaining) {
+            // If data was written to the buffer in a position that would have
+            // been filled from the remote stream, then we must skip bytes on
+            // the remote stream to emulate overwriting bytes from that
+            // position. This mimics the behavior of other PHP stream wrappers.
+            $remoteData = $this->remoteStream->read(
+                $remaining + $this->skipReadBytes
+            );
+
+            if ($this->skipReadBytes) {
+                $len = strlen($remoteData);
+                $remoteData = substr($remoteData, $this->skipReadBytes);
+                $this->skipReadBytes = max(0, $this->skipReadBytes - $len);
+            }
+
+            $data .= $remoteData;
+            $this->stream->write($remoteData);
+        }
+
+        return $data;
+    }
+
+    public function write($string)
+    {
+        // When appending to the end of the currently read stream, you'll want
+        // to skip bytes from being read from the remote stream to emulate
+        // other stream wrappers. Basically replacing bytes of data of a fixed
+        // length.
+        $overflow = (strlen($string) + $this->tell()) - $this->remoteStream->tell();
+        if ($overflow > 0) {
+            $this->skipReadBytes += $overflow;
+        }
+
+        return $this->stream->write($string);
+    }
+
+    public function eof()
+    {
+        return $this->stream->eof() && $this->remoteStream->eof();
+    }
+
+    /**
+     * Close both the remote stream and buffer stream
+     */
+    public function close()
+    {
+        $this->remoteStream->close() && $this->stream->close();
+    }
+}
@@ -0,0 +1,42 @@
+<?php
+namespace GuzzleHttp\Stream;
+
+/**
+ * Stream decorator that begins dropping data once the size of the underlying
+ * stream becomes too full.
+ */
+class DroppingStream implements StreamInterface
+{
+    use StreamDecoratorTrait;
+
+    private $maxLength;
+
+    /**
+     * @param StreamInterface $stream    Underlying stream to decorate.
+     * @param int             $maxLength Maximum size before dropping data.
+     */
+    public function __construct(StreamInterface $stream, $maxLength)
+    {
+        $this->stream = $stream;
+        $this->maxLength = $maxLength;
+    }
+
+    public function write($string)
+    {
+        $diff = $this->maxLength - $this->stream->getSize();
+
+        // Begin returning false when the underlying stream is too large.
+        if ($diff <= 0) {
+            return false;
+        }
+
+        // Write the stream or a subset of the stream if needed.
+        if (strlen($string) < $diff) {
+            return $this->stream->write($string);
+        }
+
+        $this->stream->write(substr($string, 0, $diff));
+
+        return false;
+    }
+}
@@ -0,0 +1,4 @@
+<?php
+namespace GuzzleHttp\Stream\Exception;
+
+class CannotAttachException extends \RuntimeException {}
@@ -0,0 +1,27 @@
+<?php
+namespace GuzzleHttp\Stream\Exception;
+
+use GuzzleHttp\Stream\StreamInterface;
+
+/**
+ * Exception thrown when a seek fails on a stream.
+ */
+class SeekException extends \RuntimeException
+{
+    private $stream;
+
+    public function __construct(StreamInterface $stream, $pos = 0, $msg = '')
+    {
+        $this->stream = $stream;
+        $msg = $msg ?: 'Could not seek the stream to position ' . $pos;
+        parent::__construct($msg);
+    }
+
+    /**
+     * @return StreamInterface
+     */
+    public function getStream()
+    {
+        return $this->stream;
+    }
+}
@@ -0,0 +1,147 @@
+<?php
+namespace GuzzleHttp\Stream;
+
+/**
+ * Compose stream implementations based on a hash of functions.
+ *
+ * Allows for easy testing and extension of a provided stream without needing
+ * to create a concrete class for a simple extension point.
+ */
+class FnStream implements StreamInterface
+{
+    /** @var array */
+    private $methods;
+
+    /** @var array Methods that must be implemented in the given array */
+    private static $slots = ['__toString', 'close', 'detach', 'attach',
+        'getSize', 'tell', 'eof', 'isSeekable', 'seek', 'isWritable', 'write',
+        'isReadable', 'read', 'getContents', 'getMetadata'];
+
+    /**
+     * @param array $methods Hash of method name to a callable.
+     */
+    public function __construct(array $methods)
+    {
+        $this->methods = $methods;
+
+        // Create the functions on the class
+        foreach ($methods as $name => $fn) {
+            $this->{'_fn_' . $name} = $fn;
+        }
+    }
+
+    /**
+     * Lazily determine which methods are not implemented.
+     * @throws \BadMethodCallException
+     */
+    public function __get($name)
+    {
+        throw new \BadMethodCallException(str_replace('_fn_', '', $name)
+            . '() is not implemented in the FnStream');
+    }
+
+    /**
+     * The close method is called on the underlying stream only if possible.
+     */
+    public function __destruct()
+    {
+        if (isset($this->_fn_close)) {
+            call_user_func($this->_fn_close);
+        }
+    }
+
+    /**
+     * Adds custom functionality to an underlying stream by intercepting
+     * specific method calls.
+     *
+     * @param StreamInterface $stream  Stream to decorate
+     * @param array           $methods Hash of method name to a closure
+     *
+     * @return FnStream
+     */
+    public static function decorate(StreamInterface $stream, array $methods)
+    {
+        // If any of the required methods were not provided, then simply
+        // proxy to the decorated stream.
+        foreach (array_diff(self::$slots, array_keys($methods)) as $diff) {
+            $methods[$diff] = [$stream, $diff];
+        }
+
+        return new self($methods);
+    }
+
+    public function __toString()
+    {
+        return call_user_func($this->_fn___toString);
+    }
+
+    public function close()
+    {
+        return call_user_func($this->_fn_close);
+    }
+
+    public function detach()
+    {
+        return call_user_func($this->_fn_detach);
+    }
+
+    public function attach($stream)
+    {
+        return call_user_func($this->_fn_attach, $stream);
+    }
+
+    public function getSize()
+    {
+        return call_user_func($this->_fn_getSize);
+    }
+
+    public function tell()
+    {
+        return call_user_func($this->_fn_tell);
+    }
+
+    public function eof()
+    {
+        return call_user_func($this->_fn_eof);
+    }
+
+    public function isSeekable()
+    {
+        return call_user_func($this->_fn_isSeekable);
+    }
+
+    public function seek($offset, $whence = SEEK_SET)
+    {
+        return call_user_func($this->_fn_seek, $offset, $whence);
+    }
+
+    public function isWritable()
+    {
+        return call_user_func($this->_fn_isWritable);
+    }
+
+    public function write($string)
+    {
+        return call_user_func($this->_fn_write, $string);
+    }
+
+    public function isReadable()
+    {
+        return call_user_func($this->_fn_isReadable);
+    }
+
+    public function read($length)
+    {
+        return call_user_func($this->_fn_read, $length);
+    }
+
+    public function getContents()
+    {
+        return call_user_func($this->_fn_getContents);
+    }
+
+    public function getMetadata($key = null)
+    {
+        return call_user_func($this->_fn_getMetadata, $key);
+    }
+}
@@ -0,0 +1,117 @@
+<?php
+namespace GuzzleHttp\Stream;
+
+/**
+ * Converts Guzzle streams into PHP stream resources.
+ */
+class GuzzleStreamWrapper
+{
+    /** @var resource */
+    public $context;
+
+    /** @var StreamInterface */
+    private $stream;
+
+    /** @var string r, r+, or w */
+    private $mode;
+
+    /**
+     * Returns a resource representing the stream.
+     *
+     * @param StreamInterface $stream The stream to get a resource for
+     *
+     * @return resource
+     * @throws \InvalidArgumentException if stream is not readable or writable
+     */
+    public static function getResource(StreamInterface $stream)
+    {
+        self::register();
+
+        if ($stream->isReadable()) {
+            $mode = $stream->isWritable() ? 'r+' : 'r';
+        } elseif ($stream->isWritable()) {
+            $mode = 'w';
+        } else {
+            throw new \InvalidArgumentException('The stream must be readable, '
+                . 'writable, or both.');
+        }
+
+        return fopen('guzzle://stream', $mode, false, stream_context_create([
+            'guzzle' => ['stream' => $stream],
+        ]));
+    }
+
+    /**
+     * Registers the stream wrapper if needed
+     */
+    public static function register()
+    {
+        if (!in_array('guzzle', stream_get_wrappers())) {
+            stream_wrapper_register('guzzle', __CLASS__);
+        }
+    }
+
+    public function stream_open($path, $mode, $options, &$opened_path)
+    {
+        $options = stream_context_get_options($this->context);
+
+        if (!isset($options['guzzle']['stream'])) {
+            return false;
+        }
+
+        $this->mode = $mode;
+        $this->stream = $options['guzzle']['stream'];
+
+        return true;
+    }
+
+    public function stream_read($count)
+    {
+        return $this->stream->read($count);
+    }
+
+    public function stream_write($data)
+    {
+        return (int) $this->stream->write($data);
+    }
+
+    public function stream_tell()
+    {
+        return $this->stream->tell();
+    }
+
+    public function stream_eof()
+    {
+        return $this->stream->eof();
+    }
+
+    public function stream_seek($offset, $whence)
+    {
+        return $this->stream->seek($offset, $whence);
+    }
+
+    public function stream_stat()
+    {
+        static $modeMap = [
+            'r'  => 33060,
+            'r+' => 33206,
+            'w'  => 33188,
+        ];
+
+        return [
+            'dev'     => 0,
+            'ino'     => 0,
+            'mode'    => $modeMap[$this->mode],
+            'nlink'   => 0,
+            'uid'     => 0,
+            'gid'     => 0,
+            'rdev'    => 0,
+            'size'    => $this->stream->getSize() ?: 0,
+            'atime'   => 0,
+            'mtime'   => 0,
+            'ctime'   => 0,
+            'blksize' => 0,
+            'blocks'  => 0
+        ];
+    }
+}
@@ -0,0 +1,27 @@
+<?php
+namespace GuzzleHttp\Stream;
+
+/**
+ * Uses PHP's zlib.inflate filter to inflate deflate or gzipped content.
+ *
+ * This stream decorator skips the first 10 bytes of the given stream to remove
+ * the gzip header, converts the provided stream to a PHP stream resource,
+ * then appends the zlib.inflate filter. The stream is then converted back
+ * to a Guzzle stream resource to be used as a Guzzle stream.
+ *
+ * @link http://tools.ietf.org/html/rfc1952
+ * @link http://php.net/manual/en/filters.compression.php
+ */
+class InflateStream implements StreamInterface
+{
+    use StreamDecoratorTrait;
+
+    public function __construct(StreamInterface $stream)
+    {
+        // Skip the first 10 bytes
+        $stream = new LimitStream($stream, -1, 10);
+        $resource = GuzzleStreamWrapper::getResource($stream);
+        stream_filter_append($resource, 'zlib.inflate', STREAM_FILTER_READ);
+        $this->stream = new Stream($resource);
+    }
+}
@@ -0,0 +1,37 @@
+<?php
+namespace GuzzleHttp\Stream;
+
+/**
+ * Lazily reads or writes to a file that is opened only after an IO operation
+ * take place on the stream.
+ */
+class LazyOpenStream implements StreamInterface
+{
+    use StreamDecoratorTrait;
+
+    /** @var string File to open */
+    private $filename;
+
+    /** @var string $mode */
+    private $mode;
+
+    /**
+     * @param string $filename File to lazily open
+     * @param string $mode     fopen mode to use when opening the stream
+     */
+    public function __construct($filename, $mode)
+    {
+        $this->filename = $filename;
+        $this->mode = $mode;
+    }
+
+    /**
+     * Creates the underlying stream lazily when required.
+     *
+     * @return StreamInterface
+     */
+    protected function createStream()
+    {
+        return Stream::factory(Utils::open($this->filename, $this->mode));
+    }
+}
@@ -0,0 +1,161 @@
+<?php
+namespace GuzzleHttp\Stream;
+
+use GuzzleHttp\Stream\Exception\SeekException;
+
+/**
+ * Decorator used to return only a subset of a stream
+ */
+class LimitStream implements StreamInterface
+{
+    use StreamDecoratorTrait;
+
+    /** @var int Offset to start reading from */
+    private $offset;
+
+    /** @var int Limit the number of bytes that can be read */
+    private $limit;
+
+    /**
+     * @param StreamInterface $stream Stream to wrap
+     * @param int             $limit  Total number of bytes to allow to be read
+     *                                from the stream. Pass -1 for no limit.
+     * @param int|null        $offset Position to seek to before reading (only
+     *                                works on seekable streams).
+     */
+    public function __construct(
+        StreamInterface $stream,
+        $limit = -1,
+        $offset = 0
+    ) {
+        $this->stream = $stream;
+        $this->setLimit($limit);
+        $this->setOffset($offset);
+    }
+
+    public function eof()
+    {
+        // Always return true if the underlying stream is EOF
+        if ($this->stream->eof()) {
+            return true;
+        }
+
+        // No limit and the underlying stream is not at EOF
+        if ($this->limit == -1) {
+            return false;
+        }
+
+        $tell = $this->stream->tell();
+        if ($tell === false) {
+            return false;
+        }
+
+        return $tell >= $this->offset + $this->limit;
+    }
+
+    /**
+     * Returns the size of the limited subset of data
+     * {@inheritdoc}
+     */
+    public function getSize()
+    {
+        if (null === ($length = $this->stream->getSize())) {
+            return null;
+        } elseif ($this->limit == -1) {
+            return $length - $this->offset;
+        } else {
+            return min($this->limit, $length - $this->offset);
+        }
+    }
+
+    /**
+     * Allow for a bounded seek on the read limited stream
+     * {@inheritdoc}
+     */
+    public function seek($offset, $whence = SEEK_SET)
+    {
+        if ($whence !== SEEK_SET || $offset < 0) {
+            return false;
+        }
+
+        $offset += $this->offset;
+
+        if ($this->limit !== -1) {
+            if ($offset > $this->offset + $this->limit) {
+                $offset = $this->offset + $this->limit;
+            }
+        }
+
+        return $this->stream->seek($offset);
+    }
+
+    /**
+     * Give a relative tell()
+     * {@inheritdoc}
+     */
+    public function tell()
+    {
+        return $this->stream->tell() - $this->offset;
+    }
+
+    /**
+     * Set the offset to start limiting from
+     *
+     * @param int $offset Offset to seek to and begin byte limiting from
+     *
+     * @return self
+     * @throws SeekException
+     */
+    public function setOffset($offset)
+    {
+        $current = $this->stream->tell();
+
+        if ($current !== $offset) {
+            // If the stream cannot seek to the offset position, then read to it
+            if (!$this->stream->seek($offset)) {
+                if ($current > $offset) {
+                    throw new SeekException($this, $offset);
+                } else {
+                    $this->stream->read($offset - $current);
+                }
+            }
+        }
+
+        $this->offset = $offset;
+
+        return $this;
+    }
+
+    /**
+     * Set the limit of bytes that the decorator allows to be read from the
+     * stream.
+     *
+     * @param int $limit Number of bytes to allow to be read from the stream.
+     *                   Use -1 for no limit.
+     * @return self
+     */
+    public function setLimit($limit)
+    {
+        $this->limit = $limit;
+
+        return $this;
+    }
+
+    public function read($length)
+    {
+        if ($this->limit == -1) {
+            return $this->stream->read($length);
+        }
+
+        // Check if the current position is less than the total allowed
+        // bytes + original offset
+        $remaining = ($this->offset + $this->limit) - $this->stream->tell();
+        if ($remaining > 0) {
+            // Only return the amount of requested data, ensuring that the byte
+            // limit is not exceeded
+            return $this->stream->read(min($remaining, $length));
+        } else {
+            return false;
+        }
+    }
+}
@@ -0,0 +1,11 @@
+<?php
+namespace GuzzleHttp\Stream;
+
+/**
+ * This interface is deprecated and should no longer be used. Just use
+ * StreamInterface now that the getMetadata method has been added to
+ * StreamInterface.
+ *
+ * @deprecated
+ */
+interface MetadataStreamInterface extends StreamInterface {}
@@ -0,0 +1,25 @@
+<?php
+namespace GuzzleHttp\Stream;
+
+/**
+ * Stream decorator that prevents a stream from being seeked
+ */
+class NoSeekStream implements StreamInterface
+{
+    use StreamDecoratorTrait;
+
+    public function seek($offset, $whence = SEEK_SET)
+    {
+        return false;
+    }
+
+    public function isSeekable()
+    {
+        return false;
+    }
+
+    public function attach($stream)
+    {
+        $this->stream->attach($stream);
+    }
+}
@@ -0,0 +1,79 @@
+<?php
+namespace GuzzleHttp\Stream;
+
+use GuzzleHttp\Stream\Exception\CannotAttachException;
+
+/**
+ * Does not store any data written to it.
+ */
+class NullStream implements StreamInterface
+{
+    public function __toString()
+    {
+        return '';
+    }
+
+    public function getContents()
+    {
+        return '';
+    }
+
+    public function close() {}
+
+    public function detach() {}
+
+    public function attach($stream)
+    {
+        throw new CannotAttachException();
+    }
+
+    public function getSize()
+    {
+        return 0;
+    }
+
+    public function isReadable()
+    {
+        return true;
+    }
+
+    public function isWritable()
+    {
+        return true;
+    }
+
+    public function isSeekable()
+    {
+        return true;
+    }
+
+    public function eof()
+    {
+        return true;
+    }
+
+    public function tell()
+    {
+        return 0;
+    }
+
+    public function seek($offset, $whence = SEEK_SET)
+    {
+        return false;
+    }
+
+    public function read($length)
+    {
+        return false;
+    }
+
+    public function write($string)
+    {
+        return strlen($string);
+    }
+
+    public function getMetadata($key = null)
+    {
+        return $key ? null : [];
+    }
+}
@@ -0,0 +1,161 @@
+<?php
+namespace GuzzleHttp\Stream;
+
+use GuzzleHttp\Stream\Exception\CannotAttachException;
+
+/**
+ * Provides a read only stream that pumps data from a PHP callable.
+ *
+ * When invoking the provided callable, the PumpStream will pass the amount of
+ * data requested to read to the callable. The callable can choose to ignore
+ * this value and return fewer or more bytes than requested. Any extra data
+ * returned by the provided callable is buffered internally until drained using
+ * the read() function of the PumpStream. The provided callable MUST return
+ * false when there is no more data to read.
+ */
+class PumpStream implements StreamInterface
+{
+    /** @var callable */
+    private $source;
+
+    /** @var int */
+    private $size;
+
+    /** @var int */
+    private $tellPos = 0;
+
+    /** @var array */
+    private $metadata;
+
+    /** @var BufferStream */
+    private $buffer;
+
+    /**
+     * @param callable $source Source of the stream data. The callable MAY
+     *                         accept an integer argument used to control the
+     *                         amount of data to return. The callable MUST
+     *                         return a string when called, or false on error
+     *                         or EOF.
+     * @param array $options   Stream options:
+     *                         - metadata: Hash of metadata to use with stream.
+     *                         - size: Size of the stream, if known.
+     */
+    public function __construct(callable $source, array $options = [])
+    {
+        $this->source = $source;
+        $this->size = isset($options['size']) ? $options['size'] : null;
+        $this->metadata = isset($options['metadata']) ? $options['metadata'] : [];
+        $this->buffer = new BufferStream();
+    }
+
+    public function __toString()
+    {
+        return Utils::copyToString($this);
+    }
+
+    public function close()
+    {
+        $this->detach();
+    }
+
+    public function detach()
+    {
+        $this->tellPos = false;
+        $this->source = null;
+    }
+
+    public function attach($stream)
+    {
+        throw new CannotAttachException();
+    }
+
+    public function getSize()
+    {
+        return $this->size;
+    }
+
+    public function tell()
+    {
+        return $this->tellPos;
+    }
+
+    public function eof()
+    {
+        return !$this->source;
+    }
+
+    public function isSeekable()
+    {
+        return false;
+    }
+
+    public function seek($offset, $whence = SEEK_SET)
+    {
+        return false;
+    }
+
+    public function isWritable()
+    {
+        return false;
+    }
+
+    public function write($string)
+    {
+        return false;
+    }
+
+    public function isReadable()
+    {
+        return true;
+    }
+
+    public function read($length)
+    {
+        $data = $this->buffer->read($length);
+        $readLen = strlen($data);
+        $this->tellPos += $readLen;
+        $remaining = $length - $readLen;
+
+        if ($remaining) {
+            $this->pump($remaining);
+            $data .= $this->buffer->read($remaining);
+            $this->tellPos += strlen($data) - $readLen;
+        }
+
+        return $data;
+    }
+
+    public function getContents()
+    {
+        $result = '';
+        while (!$this->eof()) {
+            $result .= $this->read(1000000);
+        }
+
+        return $result;
+    }
+
+    public function getMetadata($key = null)
+    {
+        if (!$key) {
+            return $this->metadata;
+        }
+
+        return isset($this->metadata[$key]) ? $this->metadata[$key] : null;
+    }
+
+    private function pump($length)
+    {
+        if ($this->source) {
+            do {
+                $data = call_user_func($this->source, $length);
+                if ($data === false || $data === null) {
+                    $this->source = null;
+                    return;
+                }
+                $this->buffer->write($data);
+                $length -= strlen($data);
+            } while ($length > 0);
+        }
+    }
+}
@@ -0,0 +1,261 @@
+<?php
+namespace GuzzleHttp\Stream;
+
+/**
+ * PHP stream implementation
+ */
+class Stream implements StreamInterface
+{
+    private $stream;
+    private $size;
+    private $seekable;
+    private $readable;
+    private $writable;
+    private $uri;
+    private $customMetadata;
+
+    /** @var array Hash of readable and writable stream types */
+    private static $readWriteHash = [
+        'read' => [
+            'r' => true, 'w+' => true, 'r+' => true, 'x+' => true, 'c+' => true,
+            'rb' => true, 'w+b' => true, 'r+b' => true, 'x+b' => true,
+            'c+b' => true, 'rt' => true, 'w+t' => true, 'r+t' => true,
+            'x+t' => true, 'c+t' => true, 'a+' => true,
+        ],
+        'write' => [
+            'w' => true, 'w+' => true, 'rw' => true, 'r+' => true, 'x+' => true,
+            'c+' => true, 'wb' => true, 'w+b' => true, 'r+b' => true,
+            'x+b' => true, 'c+b' => true, 'w+t' => true, 'r+t' => true,
+            'x+t' => true, 'c+t' => true, 'a' => true, 'a+' => true,
+        ],
+    ];
+
+    /**
+     * Create a new stream based on the input type.
+     *
+     * This factory accepts the same associative array of options as described
+     * in the constructor.
+     *
+     * @param resource|string|StreamInterface $resource Entity body data
+     * @param array                           $options  Additional options
+     *
+     * @return Stream
+     * @throws \InvalidArgumentException if the $resource arg is not valid.
+     */
+    public static function factory($resource = '', array $options = [])
+    {
+        $type = gettype($resource);
+
+        if ($type == 'string') {
+            $stream = fopen('php://temp', 'r+');
+            if ($resource !== '') {
+                fwrite($stream, $resource);
+                fseek($stream, 0);
+            }
+            return new self($stream, $options);
+        }
+
+        if ($type == 'resource') {
+            return new self($resource, $options);
+        }
+
+        if ($resource instanceof StreamInterface) {
+            return $resource;
+        }
+
+        if ($type == 'object' && method_exists($resource, '__toString')) {
+            return self::factory((string) $resource, $options);
+        }
+
+        if (is_callable($resource)) {
+            return new PumpStream($resource, $options);
+        }
+
+        if ($resource instanceof \Iterator) {
+            return new PumpStream(function () use ($resource) {
+                if (!$resource->valid()) {
+                    return false;
+                }
+                $result = $resource->current();
+                $resource->next();
+                return $result;
+            }, $options);
+        }
+
+        throw new \InvalidArgumentException('Invalid resource type: ' . $type);
+    }
+
+    /**
+     * This constructor accepts an associative array of options.
+     *
+     * - size: (int) If a read stream would otherwise have an indeterminate
+     *   size, but the size is known due to foreknownledge, then you can
+     *   provide that size, in bytes.
+     * - metadata: (array) Any additional metadata to return when the metadata
+     *   of the stream is accessed.
+     *
+     * @param resource $stream  Stream resource to wrap.
+     * @param array    $options Associative array of options.
+     *
+     * @throws \InvalidArgumentException if the stream is not a stream resource
+     */
+    public function __construct($stream, $options = [])
+    {
+        if (!is_resource($stream)) {
+            throw new \InvalidArgumentException('Stream must be a resource');
+        }
+
+        if (isset($options['size'])) {
+            $this->size = $options['size'];
+        }
+
+        $this->customMetadata = isset($options['metadata'])
+            ? $options['metadata']
+            : [];
+
+        $this->attach($stream);
+    }
+
+    /**
+     * Closes the stream when the destructed
+     */
+    public function __destruct()
+    {
+        $this->close();
+    }
+
+    public function __toString()
+    {
+        if (!$this->stream) {
+            return '';
+        }
+
+        $this->seek(0);
+
+        return (string) stream_get_contents($this->stream);
+    }
+
+    public function getContents()
+    {
+        return $this->stream ? stream_get_contents($this->stream) : '';
+    }
+
+    public function close()
+    {
+        if (is_resource($this->stream)) {
+            fclose($this->stream);
+        }
+
+        $this->detach();
+    }
+
+    public function detach()
+    {
+        $result = $this->stream;
+        $this->stream = $this->size = $this->uri = null;
+        $this->readable = $this->writable = $this->seekable = false;
+
+        return $result;
+    }
+
+    public function attach($stream)
+    {
+        $this->stream = $stream;
+        $meta = stream_get_meta_data($this->stream);
+        $this->seekable = $meta['seekable'];
+        $this->readable = isset(self::$readWriteHash['read'][$meta['mode']]);
+        $this->writable = isset(self::$readWriteHash['write'][$meta['mode']]);
+        $this->uri = $this->getMetadata('uri');
+    }
+
+    public function getSize()
+    {
+        if ($this->size !== null) {
+            return $this->size;
+        }
+
+        if (!$this->stream) {
+            return null;
+        }
+
+        // Clear the stat cache if the stream has a URI
+        if ($this->uri) {
+            clearstatcache(true, $this->uri);
+        }
+
+        $stats = fstat($this->stream);
+        if (isset($stats['size'])) {
+            $this->size = $stats['size'];
+            return $this->size;
+        }
+
+        return null;
+    }
+
+    public function isReadable()
+    {
+        return $this->readable;
+    }
+
+    public function isWritable()
+    {
+        return $this->writable;
+    }
+
+    public function isSeekable()
+    {
+        return $this->seekable;
+    }
+
+    public function eof()
+    {
+        return !$this->stream || feof($this->stream);
+    }
+
+    public function tell()
+    {
+        return $this->stream ? ftell($this->stream) : false;
+    }
+
+    public function setSize($size)
+    {
+        $this->size = $size;
+
+        return $this;
+    }
+
+    public function seek($offset, $whence = SEEK_SET)
+    {
+        return $this->seekable
+            ? fseek($this->stream, $offset, $whence) === 0
+            : false;
+    }
+
+    public function read($length)
+    {
+        return $this->readable ? fread($this->stream, $length) : false;
+    }
+
+    public function write($string)
+    {
+        // We can't know the size after writing anything
+        $this->size = null;
+
+        return $this->writable ? fwrite($this->stream, $string) : false;
+    }
+
+    public function getMetadata($key = null)
+    {
+        if (!$this->stream) {
+            return $key ? null : [];
+        } elseif (!$key) {
+            return $this->customMetadata + stream_get_meta_data($this->stream);
+        } elseif (isset($this->customMetadata[$key])) {
+            return $this->customMetadata[$key];
+        }
+
+        $meta = stream_get_meta_data($this->stream);
+
+        return isset($meta[$key]) ? $meta[$key] : null;
+    }
+}
@@ -0,0 +1,144 @@
+<?php
+namespace GuzzleHttp\Stream;
+
+use GuzzleHttp\Stream\Exception\CannotAttachException;
+
+/**
+ * Stream decorator trait
+ * @property StreamInterface stream
+ */
+trait StreamDecoratorTrait
+{
+    /**
+     * @param StreamInterface $stream Stream to decorate
+     */
+    public function __construct(StreamInterface $stream)
+    {
+        $this->stream = $stream;
+    }
+
+    /**
+     * Magic method used to create a new stream if streams are not added in
+     * the constructor of a decorator (e.g., LazyOpenStream).
+     */
+    public function __get($name)
+    {
+        if ($name == 'stream') {
+            $this->stream = $this->createStream();
+            return $this->stream;
+        }
+
+        throw new \UnexpectedValueException("$name not found on class");
+    }
+
+    public function __toString()
+    {
+        try {
+            $this->seek(0);
+            return $this->getContents();
+        } catch (\Exception $e) {
+            // Really, PHP? https://bugs.php.net/bug.php?id=53648
+            trigger_error('StreamDecorator::__toString exception: '
+                . (string) $e, E_USER_ERROR);
+            return '';
+        }
+    }
+
+    public function getContents()
+    {
+        return Utils::copyToString($this);
+    }
+
+    /**
+     * Allow decorators to implement custom methods
+     *
+     * @param string $method Missing method name
+     * @param array  $args   Method arguments
+     *
+     * @return mixed
+     */
+    public function __call($method, array $args)
+    {
+        $result = call_user_func_array(array($this->stream, $method), $args);
+
+        // Always return the wrapped object if the result is a return $this
+        return $result === $this->stream ? $this : $result;
+    }
+
+    public function close()
+    {
+        $this->stream->close();
+    }
+
+    public function getMetadata($key = null)
+    {
+        return $this->stream->getMetadata($key);
+    }
+
+    public function detach()
+    {
+        return $this->stream->detach();
+    }
+
+    public function attach($stream)
+    {
+        throw new CannotAttachException();
+    }
+
+    public function getSize()
+    {
+        return $this->stream->getSize();
+    }
+
+    public function eof()
+    {
+        return $this->stream->eof();
+    }
+
+    public function tell()
+    {
+        return $this->stream->tell();
+    }
+
+    public function isReadable()
+    {
+        return $this->stream->isReadable();
+    }
+
+    public function isWritable()
+    {
+        return $this->stream->isWritable();
+    }
+
+    public function isSeekable()
+    {
+        return $this->stream->isSeekable();
+    }
+
+    public function seek($offset, $whence = SEEK_SET)
+    {
+        return $this->stream->seek($offset, $whence);
+    }
+
+    public function read($length)
+    {
+        return $this->stream->read($length);
+    }
+
+    public function write($string)
+    {
+        return $this->stream->write($string);
+    }
+
+    /**
+     * Implement in subclasses to dynamically create streams when requested.
+     *
+     * @return StreamInterface
+     * @throws \BadMethodCallException
+     */
+    protected function createStream()
+    {
+        throw new \BadMethodCallException('createStream() not implemented in '
+            . get_class($this));
+    }
+}
@@ -0,0 +1,159 @@
+<?php
+namespace GuzzleHttp\Stream;
+
+/**
+ * Describes a stream instance.
+ */
+interface StreamInterface
+{
+    /**
+     * Attempts to seek to the beginning of the stream and reads all data into
+     * a string until the end of the stream is reached.
+     *
+     * Warning: This could attempt to load a large amount of data into memory.
+     *
+     * @return string
+     */
+    public function __toString();
+
+    /**
+     * Closes the stream and any underlying resources.
+     */
+    public function close();
+
+    /**
+     * Separates any underlying resources from the stream.
+     *
+     * After the underlying resource has been detached, the stream object is in
+     * an unusable state. If you wish to use a Stream object as a PHP stream
+     * but keep the Stream object in a consistent state, use
+     * {@see GuzzleHttp\Stream\GuzzleStreamWrapper::getResource}.
+     *
+     * @return resource|null Returns the underlying PHP stream resource or null
+     *                       if the Stream object did not utilize an underlying
+     *                       stream resource.
+     */
+    public function detach();
+
+    /**
+     * Replaces the underlying stream resource with the provided stream.
+     *
+     * Use this method to replace the underlying stream with another; as an
+     * example, in server-side code, if you decide to return a file, you
+     * would replace the original content-oriented stream with the file
+     * stream.
+     *
+     * Any internal state such as caching of cursor position should be reset
+     * when attach() is called, as the stream has changed.
+     *
+     * @param resource $stream
+     *
+     * @return void
+     */
+    public function attach($stream);
+
+    /**
+     * Get the size of the stream if known
+     *
+     * @return int|null Returns the size in bytes if known, or null if unknown
+     */
+    public function getSize();
+
+    /**
+     * Returns the current position of the file read/write pointer
+     *
+     * @return int|bool Returns the position of the file pointer or false on error
+     */
+    public function tell();
+
+    /**
+     * Returns true if the stream is at the end of the stream.
+     *
+     * @return bool
+     */
+    public function eof();
+
+    /**
+     * Returns whether or not the stream is seekable
+     *
+     * @return bool
+     */
+    public function isSeekable();
+
+    /**
+     * Seek to a position in the stream
+     *
+     * @param int $offset Stream offset
+     * @param int $whence Specifies how the cursor position will be calculated
+     *                    based on the seek offset. Valid values are identical
+     *                    to the built-in PHP $whence values for `fseek()`.
+     *                    SEEK_SET: Set position equal to offset bytes
+     *                    SEEK_CUR: Set position to current location plus offset
+     *                    SEEK_END: Set position to end-of-stream plus offset
+     *
+     * @return bool Returns true on success or false on failure
+     * @link   http://www.php.net/manual/en/function.fseek.php
+     */
+    public function seek($offset, $whence = SEEK_SET);
+
+    /**
+     * Returns whether or not the stream is writable
+     *
+     * @return bool
+     */
+    public function isWritable();
+
+    /**
+     * Write data to the stream
+     *
+     * @param string $string The string that is to be written.
+     *
+     * @return int|bool Returns the number of bytes written to the stream on
+     *                  success returns false on failure (e.g., broken pipe,
+     *                  writer needs to slow down, buffer is full, etc.)
+     */
+    public function write($string);
+
+    /**
+     * Returns whether or not the stream is readable
+     *
+     * @return bool
+     */
+    public function isReadable();
+
+    /**
+     * Read data from the stream
+     *
+     * @param int $length Read up to $length bytes from the object and return
+     *                    them. Fewer than $length bytes may be returned if
+     *                    underlying stream call returns fewer bytes.
+     *
+     * @return string     Returns the data read from the stream.
+     */
+    public function read($length);
+
+    /**
+     * Returns the remaining contents of the stream as a string.
+     *
+     * Note: this could potentially load a large amount of data into memory.
+     *
+     * @return string
+     */
+    public function getContents();
+
+    /**
+     * Get stream metadata as an associative array or retrieve a specific key.
+     *
+     * The keys returned are identical to the keys returned from PHP's
+     * stream_get_meta_data() function.
+     *
+     * @param string $key Specific metadata to retrieve.
+     *
+     * @return array|mixed|null Returns an associative array if no key is
+     *                          no key is provided. Returns a specific key
+     *                          value if a key is provided and the value is
+     *                          found, or null if the key is not found.
+     * @see http://php.net/manual/en/function.stream-get-meta-data.php
+     */
+    public function getMetadata($key = null);
+}
@@ -0,0 +1,198 @@
+<?php
+namespace GuzzleHttp\Stream;
+
+use GuzzleHttp\Stream\Exception\SeekException;
+
+/**
+ * Static utility class because PHP's autoloaders don't support the concept
+ * of namespaced function autoloading.
+ */
+class Utils
+{
+    /**
+     * Safely opens a PHP stream resource using a filename.
+     *
+     * When fopen fails, PHP normally raises a warning. This function adds an
+     * error handler that checks for errors and throws an exception instead.
+     *
+     * @param string $filename File to open
+     * @param string $mode     Mode used to open the file
+     *
+     * @return resource
+     * @throws \RuntimeException if the file cannot be opened
+     */
+    public static function open($filename, $mode)
+    {
+        $ex = null;
+        set_error_handler(function () use ($filename, $mode, &$ex) {
+            $ex = new \RuntimeException(sprintf(
+                'Unable to open %s using mode %s: %s',
+                $filename,
+                $mode,
+                func_get_args()[1]
+            ));
+        });
+
+        $handle = fopen($filename, $mode);
+        restore_error_handler();
+
+        if ($ex) {
+            /** @var $ex \RuntimeException */
+            throw $ex;
+        }
+
+        return $handle;
+    }
+
+    /**
+     * Copy the contents of a stream into a string until the given number of
+     * bytes have been read.
+     *
+     * @param StreamInterface $stream Stream to read
+     * @param int             $maxLen Maximum number of bytes to read. Pass -1
+     *                                to read the entire stream.
+     * @return string
+     */
+    public static function copyToString(StreamInterface $stream, $maxLen = -1)
+    {
+        $buffer = '';
+
+        if ($maxLen === -1) {
+            while (!$stream->eof()) {
+                $buf = $stream->read(1048576);
+                if ($buf === false) {
+                    break;
+                }
+                $buffer .= $buf;
+            }
+            return $buffer;
+        }
+
+        $len = 0;
+        while (!$stream->eof() && $len < $maxLen) {
+            $buf = $stream->read($maxLen - $len);
+            if ($buf === false) {
+                break;
+            }
+            $buffer .= $buf;
+            $len = strlen($buffer);
+        }
+
+        return $buffer;
+    }
+
+    /**
+     * Copy the contents of a stream into another stream until the given number
+     * of bytes have been read.
+     *
+     * @param StreamInterface $source Stream to read from
+     * @param StreamInterface $dest   Stream to write to
+     * @param int             $maxLen Maximum number of bytes to read. Pass -1
+     *                                to read the entire stream.
+     */
+    public static function copyToStream(
+        StreamInterface $source,
+        StreamInterface $dest,
+        $maxLen = -1
+    ) {
+        if ($maxLen === -1) {
+            while (!$source->eof()) {
+                if (!$dest->write($source->read(1048576))) {
+                    break;
+                }
+            }
+            return;
+        }
+
+        $bytes = 0;
+        while (!$source->eof()) {
+            $buf = $source->read($maxLen - $bytes);
+            if (!($len = strlen($buf))) {
+                break;
+            }
+            $bytes += $len;
+            $dest->write($buf);
+            if ($bytes == $maxLen) {
+                break;
+            }
+        }
+    }
+
+    /**
+     * Calculate a hash of a Stream
+     *
+     * @param StreamInterface $stream    Stream to calculate the hash for
+     * @param string          $algo      Hash algorithm (e.g. md5, crc32, etc)
+     * @param bool            $rawOutput Whether or not to use raw output
+     *
+     * @return string Returns the hash of the stream
+     * @throws SeekException
+     */
+    public static function hash(
+        StreamInterface $stream,
+        $algo,
+        $rawOutput = false
+    ) {
+        $pos = $stream->tell();
+
+        if ($pos > 0 && !$stream->seek(0)) {
+            throw new SeekException($stream);
+        }
+
+        $ctx = hash_init($algo);
+        while (!$stream->eof()) {
+            hash_update($ctx, $stream->read(1048576));
+        }
+
+        $out = hash_final($ctx, (bool) $rawOutput);
+        $stream->seek($pos);
+
+        return $out;
+    }
+
+    /**
+     * Read a line from the stream up to the maximum allowed buffer length
+     *
+     * @param StreamInterface $stream    Stream to read from
+     * @param int             $maxLength Maximum buffer length
+     * @param string          $eol       Line ending
+     *
+     * @return string|bool
+     */
+    public static function readline(StreamInterface $stream, $maxLength = null, $eol = PHP_EOL)
+    {
+        $buffer = '';
+        $size = 0;
+        $negEolLen = -strlen($eol);
+
+        while (!$stream->eof()) {
+            if (false === ($byte = $stream->read(1))) {
+                return $buffer;
+            }
+            $buffer .= $byte;
+            // Break when a new line is found or the max length - 1 is reached
+            if (++$size == $maxLength || substr($buffer, $negEolLen) === $eol) {
+                break;
+            }
+        }
+
+        return $buffer;
+    }
+
+    /**
+     * Alias of GuzzleHttp\Stream\Stream::factory.
+     *
+     * @param mixed $resource Resource to create
+     * @param array $options  Associative array of stream options defined in
+     *                        {@see \GuzzleHttp\Stream\Stream::__construct}
+     *
+     * @return StreamInterface
+     *
+     * @see GuzzleHttp\Stream\Stream::factory
+     * @see GuzzleHttp\Stream\Stream::__construct
+     */
+    public static function create($resource, array $options = [])
+    {
+        return Stream::factory($resource, $options);
+    }
+}
@@ -0,0 +1,12 @@
+root = true
+
+[*]
+charset = utf-8
+end_of_line = lf
+indent_size = 4
+indent_style = space
+insert_final_newline = true
+trim_trailing_whitespace = true
+
+[{Makefile,*.mk}]
+indent_style = tab
@@ -0,0 +1,44 @@
+name: PHP test
+
+on: [push, pull_request]
+
+jobs:
+  test:
+    name: Test
+    runs-on: ubuntu-latest
+
+    strategy:
+      matrix:
+        php-version: [7.4, 8.0, 8.1, 8.2, 8.3, 8.4, 8.5]
+        
+    steps:
+    - name: Checkout
+      uses: actions/checkout@v4
+
+    - name: Use PHP ${{ matrix.php-version }}
+      uses: shivammathur/setup-php@v2
+      with:
+        php-version: ${{ matrix.php-version }}
+        extensions: zip, curl
+      env:
+        COMPOSER_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+   
+    - name: Get composer cache directory
+      id: composercache
+      run: echo "dir=$(composer config cache-files-dir)" >> $GITHUB_OUTPUT
+      
+    - name: Cache dependencies
+      uses: actions/cache@v4
+      with:
+        path: ${{ steps.composercache.outputs.dir }}
+        key: ${{ runner.os }}-composer-${{ hashFiles('**/composer.json') }}
+        restore-keys: ${{ runner.os }}-composer-
+
+    - name: Install dependencies
+      run: |
+        composer install --prefer-dist
+
+    - name: Unit tests
+      run: |
+        composer run-script test
+        
@@ -0,0 +1,139 @@
+# Changelog
+
+
+All notable changes to this project will be documented in this file.
+
+The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
+and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
+
+## [1.2.1] - 2022-10-25
+
+- Fixed the support of PHP 8.1 with [#7](https://github.com/ezimuel/ringphp/pull/7) and [#8](https://github.com/ezimuel/ringphp/pull/8)
+- Fixed the integration with Symfony, adding explicit @return annotations to suppress User Deprecated notices [#5](https://github.com/ezimuel/ringphp/pull/5)
+
+## [1.2.0] - 2021-11-16
+
+### Added
+
+- Add attribute to avoid Deprecated notice for PHP 8.1 [#4](https://github.com/ezimuel/ringphp/pull/4)
+- Add replace guzzlehttp/ringphp in composer [#3](https://github.com/ezimuel/ringphp/pull/3)
+- Add .gitattributes file [#2](https://github.com/ezimuel/ringphp/pull/2)
+
+### Changed
+
+- Updated to PHPUnit 9 and fixed the unit tests for PHP 8
+  [d386b25](https://github.com/ezimuel/ringphp/commit/d386b2597389dafe3b437ab90d115eb092fff109)
+
+## [1.1.2] - 2020-02-15
+
+### Changed
+
+- Required guzzlestreams 3.0.1+ [0b78f89](https://github.com/ezimuel/ringphp/commit/0b78f89d8e0bb9e380046c31adfa40347e9f663b)
+
+
+## [1.1.1] - 2018-07-31
+
+### Fixed
+
+- `continue` keyword usage on PHP 7.3
+
+
+## [1.1.0] - 2015-05-19
+
+### Added
+
+- Added `CURL_HTTP_VERSION_2_0`
+
+### Changed
+
+- The PHP stream wrapper handler now sets `allow_self_signed` to `false` to
+  match the cURL handler when `verify` is set to `true` or a certificate file.
+- Ensuring that a directory exists before using the `save_to` option.
+- Response protocol version is now correctly extracted from a response.
+
+### Fixed
+
+- Fixed a bug in which the result of `CurlFactory::retryFailedRewind` did not
+  return an array.
+
+
+## [1.0.7] - 2015-03-29
+
+### Fixed
+
+- PHP 7 fixes.
+
+
+## [1.0.6] - 2015-02-26
+
+### Changed
+
+- The multi handle of the CurlMultiHandler is now created lazily.
+
+### Fixed
+
+- Bug fix: futures now extend from React's PromiseInterface to ensure that they
+  are properly forwarded down the promise chain.
+
+
+## [1.0.5] - 2014-12-10
+
+### Added
+
+- Adding more error information to PHP stream wrapper exceptions.
+- Added digest auth integration test support to test server.
+
+
+## [1.0.4] - 2014-12-01
+
+### Added
+
+- Added support for older versions of cURL that do not have CURLOPT_TIMEOUT_MS.
+- Added a fix to the StreamHandler to return a `FutureArrayInterface` when an
+
+### Changed
+
+- Setting debug to `false` does not enable debug output. error occurs.
+
+
+## [1.0.3] - 2014-11-03
+
+### Fixed
+
+- Setting the `header` stream option as a string to be compatible with GAE.
+- Header parsing now ensures that header order is maintained in the parsed
+  message.
+
+
+## [1.0.2] - 2014-10-28
+
+### Fixed
+
+- Now correctly honoring a `version` option is supplied in a request.
+  See https://github.com/guzzle/RingPHP/pull/8
+
+
+## [1.0.1] - 2014-10-26
+
+### Fixed
+
+- Fixed a header parsing issue with the `CurlHandler` and `CurlMultiHandler`
+  that caused cURL requests with multiple responses to merge repsonses together
+  (e.g., requests with digest authentication).
+
+
+## 1.0.0 - 2014-10-12
+
+- Initial release
+
+
+[Unreleased]: https://github.com/guzzle/RingPHP/compare/1.1.1...HEAD
+[1.1.1]: https://github.com/guzzle/RingPHP/compare/1.1.0...1.1.1
+[1.1.0]: https://github.com/guzzle/RingPHP/compare/1.0.7...1.1.0
+[1.0.7]: https://github.com/guzzle/RingPHP/compare/1.0.6...1.0.7
+[1.0.6]: https://github.com/guzzle/RingPHP/compare/1.0.5...1.0.6
+[1.0.5]: https://github.com/guzzle/RingPHP/compare/1.0.4...1.0.5
+[1.0.4]: https://github.com/guzzle/RingPHP/compare/1.0.3...1.0.4
+[1.0.3]: https://github.com/guzzle/RingPHP/compare/1.0.2...1.0.3
+[1.0.2]: https://github.com/guzzle/RingPHP/compare/1.0.1...1.0.2
+[1.0.1]: https://github.com/guzzle/RingPHP/compare/1.0.0...1.0.1
@@ -0,0 +1,19 @@
+Copyright (c) 2014 Michael Dowling, https://github.com/mtdowling <mtdowling@gmail.com>
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.
@@ -0,0 +1,46 @@
+all: clean coverage docs
+
+docs:
+	cd docs && make html
+
+view-docs:
+	open docs/_build/html/index.html
+
+start-server: stop-server
+	node tests/Client/server.js &> /dev/null &
+
+stop-server:
+	@PID=$(shell ps axo pid,command \
+	  | grep 'tests/Client/server.js' \
+	  | grep -v grep \
+	  | cut -f 1 -d " "\
+	) && [ -n "$$PID" ] && kill $$PID || true
+
+test: start-server
+	vendor/bin/phpunit $(TEST)
+	$(MAKE) stop-server
+
+coverage: start-server
+	vendor/bin/phpunit --coverage-html=build/artifacts/coverage $(TEST)
+	$(MAKE) stop-server
+
+view-coverage:
+	open build/artifacts/coverage/index.html
+
+clean:
+	rm -rf build/artifacts/*
+	cd docs && make clean
+
+tag:
+	$(if $(TAG),,$(error TAG is not defined. Pass via "make tag TAG=4.2.1"))
+	@echo Tagging $(TAG)
+	chag update -m '$(TAG) ()'
+	git add -A
+	git commit -m '$(TAG) release'
+	chag tag
+
+perf: start-server
+	php tests/perf.php
+	$(MAKE) stop-server
+
+.PHONY: docs
@@ -0,0 +1,48 @@
+RingPHP
+=======
+
+[![Build status](https://github.com/ezimuel/ringphp/workflows/PHP%20test/badge.svg)](https://github.com/ezimuel/ringphp/actions) [![Latest Stable Version](https://poser.pugx.org/ezimuel/ringphp/v/stable)](https://packagist.org/packages/ezimuel/ringphp)
+
+**Note:** this is a fork of the original project since it was abandoned.
+
+Provides a simple API and specification that abstracts away the details of HTTP
+into a single PHP function. RingPHP be used to power HTTP clients and servers
+through a PHP function that accepts a request hash and returns a response hash
+that is fulfilled using a [promise](https://github.com/reactphp/promise),
+allowing RingPHP to support both synchronous and asynchronous workflows.
+
+By abstracting the implementation details of different HTTP clients and
+servers, RingPHP allows you to utilize pluggable HTTP clients and servers
+without tying your application to a specific implementation.
+
+```php
+require 'vendor/autoload.php';
+
+use GuzzleHttp\Ring\Client\CurlHandler;
+
+$handler = new CurlHandler();
+$response = $handler([
+    'http_method' => 'GET',
+    'uri'         => '/',
+    'headers'     => [
+        'host'  => ['www.google.com'],
+        'x-foo' => ['baz']
+    ]
+]);
+
+$response->then(function (array $response) {
+    echo $response['status'];
+});
+
+$response->wait();
+```
+
+RingPHP is inspired by Clojure's [Ring](https://github.com/ring-clojure/ring),
+which, in turn, was inspired by Python's WSGI and Ruby's Rack. RingPHP is
+utilized as the handler layer in [Guzzle](https://guzzlephp.org) 5.0+ to send
+HTTP requests.
+
+Documentation
+-------------
+
+See https://ringphp.readthedocs.io/en/latest/ for the full online documentation.
@@ -0,0 +1,46 @@
+{
+    "name": "ezimuel/ringphp",
+    "description": "Fork of guzzle/RingPHP (abandoned) to be used with elasticsearch-php",
+    "license": "MIT",
+    "authors": [
+        {
+            "name": "Michael Dowling",
+            "email": "mtdowling@gmail.com",
+            "homepage": "https://github.com/mtdowling"
+        }
+    ],
+    "require": {
+        "php": ">=5.4.0",
+        "ezimuel/guzzlestreams": "^3.0.1 || ^4.0.0",
+        "react/promise": "^2.0 || ^3.0"
+    },
+    "require-dev": {
+        "ext-curl": "*",
+        "phpunit/phpunit": "~9.0"
+    },
+    "replace": {
+        "guzzlehttp/ringphp": "self.version"
+    },
+    "suggest": {
+        "ext-curl": "Guzzle will use specific adapters if cURL is present"
+    },
+    "autoload": {
+        "psr-4": {
+            "GuzzleHttp\\Ring\\": "src/"
+        }
+    },
+    "autoload-dev": {
+        "psr-4": {
+            "GuzzleHttp\\Tests\\Ring\\": "tests/"
+        }
+    },
+    "scripts": {
+        "test": "make test",
+        "test-ci": "make coverage"
+    },
+    "extra": {
+        "branch-alias": {
+            "dev-master": "1.1-dev"
+        }
+    }
+}
@@ -0,0 +1,9 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<phpunit bootstrap="./tests/bootstrap.php"
+         colors="true">
+    <testsuites>
+        <testsuite name="Unit tests">
+            <directory>tests</directory>
+        </testsuite>
+    </testsuites>
+</phpunit>
@@ -0,0 +1,74 @@
+<?php
+namespace GuzzleHttp\Ring\Client;
+
+/**
+ * Client specific utility functions.
+ */
+class ClientUtils
+{
+    /**
+     * Returns the default cacert bundle for the current system.
+     *
+     * First, the openssl.cafile and curl.cainfo php.ini settings are checked.
+     * If those settings are not configured, then the common locations for
+     * bundles found on Red Hat, CentOS, Fedora, Ubuntu, Debian, FreeBSD, OS X
+     * and Windows are checked. If any of these file locations are found on
+     * disk, they will be utilized.
+     *
+     * Note: the result of this function is cached for subsequent calls.
+     *
+     * @return string
+     * @throws \RuntimeException if no bundle can be found.
+     */
+    public static function getDefaultCaBundle()
+    {
+        static $cached = null;
+        static $cafiles = [
+            // Red Hat, CentOS, Fedora (provided by the ca-certificates package)
+            '/etc/pki/tls/certs/ca-bundle.crt',
+            // Ubuntu, Debian (provided by the ca-certificates package)
+            '/etc/ssl/certs/ca-certificates.crt',
+            // FreeBSD (provided by the ca_root_nss package)
+            '/usr/local/share/certs/ca-root-nss.crt',
+            // OS X provided by homebrew (using the default path)
+            '/usr/local/etc/openssl/cert.pem',
+            // Windows?
+            'C:\\windows\\system32\\curl-ca-bundle.crt',
+            'C:\\windows\\curl-ca-bundle.crt',
+        ];
+
+        if ($cached) {
+            return $cached;
+        }
+
+        if ($ca = ini_get('openssl.cafile')) {
+            return $cached = $ca;
+        }
+
+        if ($ca = ini_get('curl.cainfo')) {
+            return $cached = $ca;
+        }
+
+        foreach ($cafiles as $filename) {
+            if (file_exists($filename)) {
+                return $cached = $filename;
+            }
+        }
+
+        throw new \RuntimeException(self::CA_ERR);
+    }
+
+    const CA_ERR = "
+No system CA bundle could be found in any of the the common system locations.
+PHP versions earlier than 5.6 are not properly configured to use the system's
+CA bundle by default. In order to verify peer certificates, you will need to
+supply the path on disk to a certificate bundle to the 'verify' request
+option: http://docs.guzzlephp.org/en/5.3/clients.html#verify. If you do not
+need a specific certificate bundle, then Mozilla provides a commonly used CA
+bundle which can be downloaded here (provided by the maintainer of cURL):
+https://raw.githubusercontent.com/bagder/ca-bundle/master/ca-bundle.crt. Once
+you have a CA bundle available on disk, you can set the 'openssl.cafile' PHP
+ini setting to point to the path to the file, allowing you to omit the 'verify'
+request option. See http://curl.haxx.se/docs/sslcerts.html for more
+information.";
+}
@@ -0,0 +1,560 @@
+<?php
+namespace GuzzleHttp\Ring\Client;
+
+use GuzzleHttp\Ring\Core;
+use GuzzleHttp\Ring\Exception\ConnectException;
+use GuzzleHttp\Ring\Exception\RingException;
+use GuzzleHttp\Stream\LazyOpenStream;
+use GuzzleHttp\Stream\StreamInterface;
+
+/**
+ * Creates curl resources from a request
+ */
+class CurlFactory
+{
+    /**
+     * Creates a cURL handle, header resource, and body resource based on a
+     * transaction.
+     *
+     * @param array         $request Request hash
+     * @param null|resource $handle  Optionally provide a curl handle to modify
+     *
+     * @return array Returns an array of the curl handle, headers array, and
+     *               response body handle.
+     * @throws \RuntimeException when an option cannot be applied
+     */
+    public function __invoke(array $request, $handle = null)
+    {
+        $headers = [];
+        $options = $this->getDefaultOptions($request, $headers);
+        $this->applyMethod($request, $options);
+
+        if (isset($request['client'])) {
+            $this->applyHandlerOptions($request, $options);
+        }
+
+        $this->applyHeaders($request, $options);
+        unset($options['_headers']);
+
+        // Add handler options from the request's configuration options
+        if (isset($request['client']['curl'])) {
+            $options = $this->applyCustomCurlOptions(
+                $request['client']['curl'],
+                $options
+            );
+        }
+
+        if (!$handle) {
+            $handle = curl_init();
+        }
+
+        $body = $this->getOutputBody($request, $options);
+        curl_setopt_array($handle, $options);
+
+        return [$handle, &$headers, $body];
+    }
+
+    /**
+     * Creates a response hash from a cURL result.
+     *
+     * @param callable $handler  Handler that was used.
+     * @param array    $request  Request that sent.
+     * @param array    $response Response hash to update.
+     * @param array    $headers  Headers received during transfer.
+     * @param resource $body     Body fopen response.
+     *
+     * @return array
+     */
+    public static function createResponse(
+        callable $handler,
+        array $request,
+        array $response,
+        array $headers,
+        $body
+    ) {
+        if (isset($response['transfer_stats']['url'])) {
+            $response['effective_url'] = $response['transfer_stats']['url'];
+        }
+
+        if (!empty($headers)) {
+            $startLine = explode(' ', array_shift($headers), 3);
+            $headerList = Core::headersFromLines($headers);
+            $response['headers'] = $headerList;
+            $response['version'] = isset($startLine[0]) ? substr($startLine[0], 5) : null;
+            $response['status'] = isset($startLine[1]) ? (int) $startLine[1] : null;
+            $response['reason'] = isset($startLine[2]) ? $startLine[2] : null;
+            $response['body'] = $body;
+            Core::rewindBody($response);
+        }
+
+        return !empty($response['curl']['errno']) || !isset($response['status'])
+            ? self::createErrorResponse($handler, $request, $response)
+            : $response;
+    }
+
+    private static function createErrorResponse(
+        callable $handler,
+        array $request,
+        array $response
+    ) {
+        static $connectionErrors = [
+            CURLE_OPERATION_TIMEOUTED  => true,
+            CURLE_COULDNT_RESOLVE_HOST => true,
+            CURLE_COULDNT_CONNECT      => true,
+            CURLE_SSL_CONNECT_ERROR    => true,
+            CURLE_GOT_NOTHING          => true,
+        ];
+
+        // Retry when nothing is present or when curl failed to rewind.
+        if (!isset($response['err_message'])
+            && (empty($response['curl']['errno'])
+                || $response['curl']['errno'] == 65)
+        ) {
+            return self::retryFailedRewind($handler, $request, $response);
+        }
+
+        $message = isset($response['err_message'])
+            ? $response['err_message']
+            : sprintf('cURL error %s: %s',
+                $response['curl']['errno'],
+                isset($response['curl']['error'])
+                    ? $response['curl']['error']
+                    : 'See http://curl.haxx.se/libcurl/c/libcurl-errors.html');
+
+        $error = isset($response['curl']['errno'])
+            && isset($connectionErrors[$response['curl']['errno']])
+            ? new ConnectException($message)
+            : new RingException($message);
+
+        return $response + [
+            'status'  => null,
+            'reason'  => null,
+            'body'    => null,
+            'headers' => [],
+            'error'   => $error,
+        ];
+    }
+
+    private function getOutputBody(array $request, array &$options)
+    {
+        // Determine where the body of the response (if any) will be streamed.
+        if (isset($options[CURLOPT_WRITEFUNCTION])) {
+            return $request['client']['save_to'];
+        }
+
+        if (isset($options[CURLOPT_FILE])) {
+            return $options[CURLOPT_FILE];
+        }
+
+        if ($request['http_method'] != 'HEAD') {
+            // Create a default body if one was not provided
+            return $options[CURLOPT_FILE] = fopen('php://temp', 'w+');
+        }
+
+        return null;
+    }
+
+    private function getDefaultOptions(array $request, array &$headers)
+    {
+        $url = Core::url($request);
+        $startingResponse = false;
+
+        $options = [
+            '_headers'             => $request['headers'],
+            CURLOPT_CUSTOMREQUEST  => $request['http_method'],
+            CURLOPT_URL            => $url,
+            CURLOPT_RETURNTRANSFER => false,
+            CURLOPT_HEADER         => false,
+            CURLOPT_CONNECTTIMEOUT => 150,
+            CURLOPT_HEADERFUNCTION => function ($ch, $h) use (&$headers, &$startingResponse) {
+                $value = trim($h);
+                if ($value === '') {
+                    $startingResponse = true;
+                } elseif ($startingResponse) {
+                    $startingResponse = false;
+                    $headers = [$value];
+                } else {
+                    $headers[] = $value;
+                }
+                return strlen($h);
+            },
+        ];
+
+        if (isset($request['version'])) {
+            if ($request['version'] == 2.0) {
+                $options[CURLOPT_HTTP_VERSION] = CURL_HTTP_VERSION_2_0;
+            } else if ($request['version'] == 1.1) {
+                $options[CURLOPT_HTTP_VERSION] = CURL_HTTP_VERSION_1_1;
+            } else {
+                $options[CURLOPT_HTTP_VERSION] = CURL_HTTP_VERSION_1_0;
+            }
+        }
+
+        if (defined('CURLOPT_PROTOCOLS')) {
+            $options[CURLOPT_PROTOCOLS] = CURLPROTO_HTTP | CURLPROTO_HTTPS;
+        }
+
+        return $options;
+    }
+
+    private function applyMethod(array $request, array &$options)
+    {
+        if (isset($request['body'])) {
+            $this->applyBody($request, $options);
+            return;
+        }
+
+        switch ($request['http_method']) {
+            case 'PUT':
+            case 'POST':
+                // See http://tools.ietf.org/html/rfc7230#section-3.3.2
+                if (!Core::hasHeader($request, 'Content-Length')) {
+                    $options[CURLOPT_HTTPHEADER][] = 'Content-Length: 0';
+                }
+                break;
+            case 'HEAD':
+                $options[CURLOPT_NOBODY] = true;
+                unset(
+                    $options[CURLOPT_WRITEFUNCTION],
+                    $options[CURLOPT_READFUNCTION],
+                    $options[CURLOPT_FILE],
+                    $options[CURLOPT_INFILE]
+                );
+        }
+    }
+
+    private function applyBody(array $request, array &$options)
+    {
+        $contentLength = Core::firstHeader($request, 'Content-Length');
+        $size = $contentLength !== null ? (int) $contentLength : null;
+
+        // Send the body as a string if the size is less than 1MB OR if the
+        // [client][curl][body_as_string] request value is set.
+        if (($size !== null && $size < 1000000) ||
+            isset($request['client']['curl']['body_as_string']) ||
+            is_string($request['body'])
+        ) {
+            $options[CURLOPT_POSTFIELDS] = Core::body($request);
+            // Don't duplicate the Content-Length header
+            $this->removeHeader('Content-Length', $options);
+            $this->removeHeader('Transfer-Encoding', $options);
+        } else {
+            $options[CURLOPT_UPLOAD] = true;
+            if ($size !== null) {
+                // Let cURL handle setting the Content-Length header
+                $options[CURLOPT_INFILESIZE] = $size;
+                $this->removeHeader('Content-Length', $options);
+            }
+            $this->addStreamingBody($request, $options);
+        }
+
+        // If the Expect header is not present, prevent curl from adding it
+        if (!Core::hasHeader($request, 'Expect')) {
+            $options[CURLOPT_HTTPHEADER][] = 'Expect:';
+        }
+
+        // cURL sometimes adds a content-type by default. Prevent this.
+        if (!Core::hasHeader($request, 'Content-Type')) {
+            $options[CURLOPT_HTTPHEADER][] = 'Content-Type:';
+        }
+    }
+
+    private function addStreamingBody(array $request, array &$options)
+    {
+        $body = $request['body'];
+
+        if ($body instanceof StreamInterface) {
+            $options[CURLOPT_READFUNCTION] = function ($ch, $fd, $length) use ($body) {
+                return (string) $body->read($length);
+            };
+            if (!isset($options[CURLOPT_INFILESIZE])) {
+                if ($size = $body->getSize()) {
+                    $options[CURLOPT_INFILESIZE] = $size;
+                }
+            }
+        } elseif (is_resource($body)) {
+            $options[CURLOPT_INFILE] = $body;
+        } elseif ($body instanceof \Iterator) {
+            $buf = '';
+            $options[CURLOPT_READFUNCTION] = function ($ch, $fd, $length) use ($body, &$buf) {
+                if ($body->valid()) {
+                    $buf .= $body->current();
+                    $body->next();
+                }
+                $result = (string) substr($buf, 0, $length);
+                $buf = substr($buf, $length);
+                return $result;
+            };
+        } else {
+            throw new \InvalidArgumentException('Invalid request body provided');
+        }
+    }
+
+    private function applyHeaders(array $request, array &$options)
+    {
+        foreach ($options['_headers'] as $name => $values) {
+            foreach ($values as $value) {
+                $options[CURLOPT_HTTPHEADER][] = "$name: $value";
+            }
+        }
+
+        // Remove the Accept header if one was not set
+        if (!Core::hasHeader($request, 'Accept')) {
+            $options[CURLOPT_HTTPHEADER][] = 'Accept:';
+        }
+    }
+
+    /**
+     * Takes an array of curl options specified in the 'curl' option of a
+     * request's configuration array and maps them to CURLOPT_* options.
+     *
+     * This method is only called when a  request has a 'curl' config setting.
+     *
+     * @param array $config  Configuration array of custom curl option
+     * @param array $options Array of existing curl options
+     *
+     * @return array Returns a new array of curl options
+     */
+    private function applyCustomCurlOptions(array $config, array $options)
+    {
+        $curlOptions = [];
+        foreach ($config as $key => $value) {
+            if (is_int($key)) {
+                $curlOptions[$key] = $value;
+            }
+        }
+
+        return $curlOptions + $options;
+    }
+
+    /**
+     * Remove a header from the options array.
+     *
+     * @param string $name    Case-insensitive header to remove
+     * @param array  $options Array of options to modify
+     */
+    private function removeHeader($name, array &$options)
+    {
+        foreach (array_keys($options['_headers']) as $key) {
+            if (!strcasecmp($key, $name)) {
+                unset($options['_headers'][$key]);
+                return;
+            }
+        }
+    }
+
+    /**
+     * Applies an array of request client options to a the options array.
+     *
+     * This method uses a large switch rather than double-dispatch to save on
+     * high overhead of calling functions in PHP.
+     */
+    private function applyHandlerOptions(array $request, array &$options)
+    {
+        foreach ($request['client'] as $key => $value) {
+            switch ($key) {
+            // Violating PSR-4 to provide more room.
+            case 'verify':
+
+                if ($value === false) {
+                    unset($options[CURLOPT_CAINFO]);
+                    $options[CURLOPT_SSL_VERIFYHOST] = 0;
+                    $options[CURLOPT_SSL_VERIFYPEER] = false;
+                    continue 2;
+                }
+
+                $options[CURLOPT_SSL_VERIFYHOST] = 2;
+                $options[CURLOPT_SSL_VERIFYPEER] = true;
+
+                if (is_string($value)) {
+                    $options[CURLOPT_CAINFO] = $value;
+                    if (!file_exists($value)) {
+                        throw new \InvalidArgumentException(
+                            "SSL CA bundle not found: $value"
+                        );
+                    }
+                }
+                break;
+
+            case 'decode_content':
+
+                if ($value === false) {
+                    continue 2;
+                }
+
+                $accept = Core::firstHeader($request, 'Accept-Encoding');
+                if ($accept) {
+                    $options[CURLOPT_ENCODING] = $accept;
+                } else {
+                    $options[CURLOPT_ENCODING] = '';
+                    // Don't let curl send the header over the wire
+                    $options[CURLOPT_HTTPHEADER][] = 'Accept-Encoding:';
+                }
+                break;
+
+            case 'save_to':
+
+                if (is_string($value)) {
+                    if (!is_dir(dirname($value))) {
+                        throw new \RuntimeException(sprintf(
+                            'Directory %s does not exist for save_to value of %s',
+                            dirname($value),
+                            $value
+                        ));
+                    }
+                    $value = new LazyOpenStream($value, 'w+');
+                }
+
+                if ($value instanceof StreamInterface) {
+                    $options[CURLOPT_WRITEFUNCTION] =
+                        function ($ch, $write) use ($value) {
+                            return $value->write($write);
+                        };
+                } elseif (is_resource($value)) {
+                    $options[CURLOPT_FILE] = $value;
+                } else {
+                    throw new \InvalidArgumentException('save_to must be a '
+                        . 'GuzzleHttp\Stream\StreamInterface or resource');
+                }
+                break;
+
+            case 'timeout':
+
+                if (defined('CURLOPT_TIMEOUT_MS')) {
+                    $options[CURLOPT_TIMEOUT_MS] = $value * 1000;
+                } else {
+                    $options[CURLOPT_TIMEOUT] = $value;
+                }
+                break;
+
+            case 'connect_timeout':
+
+                if (defined('CURLOPT_CONNECTTIMEOUT_MS')) {
+                    $options[CURLOPT_CONNECTTIMEOUT_MS] = $value * 1000;
+                } else {
+                    $options[CURLOPT_CONNECTTIMEOUT] = $value;
+                }
+                break;
+
+            case 'proxy':
+
+                if (!is_array($value)) {
+                    $options[CURLOPT_PROXY] = $value;
+                } elseif (isset($request['scheme'])) {
+                    $scheme = $request['scheme'];
+                    if (isset($value[$scheme])) {
+                        $options[CURLOPT_PROXY] = $value[$scheme];
+                    }
+                }
+                break;
+
+            case 'cert':
+
+                if (is_array($value)) {
+                    $options[CURLOPT_SSLCERTPASSWD] = $value[1];
+                    $value = $value[0];
+                }
+
+                if (!file_exists($value)) {
+                    throw new \InvalidArgumentException(
+                        "SSL certificate not found: {$value}"
+                    );
+                }
+
+                $options[CURLOPT_SSLCERT] = $value;
+                break;
+
+            case 'ssl_key':
+
+                if (is_array($value)) {
+                    $options[CURLOPT_SSLKEYPASSWD] = $value[1];
+                    $value = $value[0];
+                }
+
+                if (!file_exists($value)) {
+                    throw new \InvalidArgumentException(
+                        "SSL private key not found: {$value}"
+                    );
+                }
+
+                $options[CURLOPT_SSLKEY] = $value;
+                break;
+
+            case 'progress':
+
+                if (!is_callable($value)) {
+                    throw new \InvalidArgumentException(
+                        'progress client option must be callable'
+                    );
+                }
+
+                $options[CURLOPT_NOPROGRESS] = false;
+                $options[CURLOPT_PROGRESSFUNCTION] =
+                    function () use ($value) {
+                        $args = func_get_args();
+                        // PHP 5.5 pushed the handle onto the start of the args
+                        if (is_resource($args[0])) {
+                            array_shift($args);
+                        }
+                        call_user_func_array($value, $args);
+                    };
+                break;
+
+            case 'debug':
+
+                if ($value) {
+                    $options[CURLOPT_STDERR] = Core::getDebugResource($value);
+                    $options[CURLOPT_VERBOSE] = true;
+                }
+                break;
+            }
+        }
+    }
+
+    /**
+     * This function ensures that a response was set on a transaction. If one
+     * was not set, then the request is retried if possible. This error
+     * typically means you are sending a payload, curl encountered a
+     * "Connection died, retrying a fresh connect" error, tried to rewind the
+     * stream, and then encountered a "necessary data rewind wasn't possible"
+     * error, causing the request to be sent through curl_multi_info_read()
+     * without an error status.
+     */
+    private static function retryFailedRewind(
+        callable $handler,
+        array $request,
+        array $response
+    ) {
+        // If there is no body, then there is some other kind of issue. This
+        // is weird and should probably never happen.
+        if (!isset($request['body'])) {
+            $response['err_message'] = 'No response was received for a request '
+                . 'with no body. This could mean that you are saturating your '
+                . 'network.';
+            return self::createErrorResponse($handler, $request, $response);
+        }
+
+        if (!Core::rewindBody($request)) {
+            $response['err_message'] = 'The connection unexpectedly failed '
+                . 'without providing an error. The request would have been '
+                . 'retried, but attempting to rewind the request body failed.';
+            return self::createErrorResponse($handler, $request, $response);
+        }
+
+        // Retry no more than 3 times before giving up.
+        if (!isset($request['curl']['retries'])) {
+            $request['curl']['retries'] = 1;
+        } elseif ($request['curl']['retries'] == 2) {
+            $response['err_message'] = 'The cURL request was retried 3 times '
+                . 'and did no succeed. cURL was unable to rewind the body of '
+                . 'the request and subsequent retries resulted in the same '
+                . 'error. Turn on the debug option to see what went wrong. '
+                . 'See https://bugs.php.net/bug.php?id=47204 for more information.';
+            return self::createErrorResponse($handler, $request, $response);
+        } else {
+            $request['curl']['retries']++;
+        }
+
+        return $handler($request);
+    }
+}
@@ -0,0 +1,140 @@
+<?php
+namespace GuzzleHttp\Ring\Client;
+
+use GuzzleHttp\Ring\Future\CompletedFutureArray;
+use GuzzleHttp\Ring\Core;
+
+/**
+ * HTTP handler that uses cURL easy handles as a transport layer.
+ *
+ * Requires PHP 5.5+
+ *
+ * When using the CurlHandler, custom curl options can be specified as an
+ * associative array of curl option constants mapping to values in the
+ * **curl** key of the "client" key of the request.
+ */
+class CurlHandler
+{
+    /** @var callable */
+    private $factory;
+
+    /** @var array Array of curl easy handles */
+    private $handles = [];
+
+    /** @var array Array of owned curl easy handles */
+    private $ownedHandles = [];
+
+    /** @var int Total number of idle handles to keep in cache */
+    private $maxHandles;
+
+    /**
+     * Accepts an associative array of options:
+     *
+     * - factory: Optional callable factory used to create cURL handles.
+     *   The callable is passed a request hash when invoked, and returns an
+     *   array of the curl handle, headers resource, and body resource.
+     * - max_handles: Maximum number of idle handles (defaults to 5).
+     *
+     * @param array $options Array of options to use with the handler
+     */
+    public function __construct(array $options = [])
+    {
+        $this->handles = $this->ownedHandles = [];
+        $this->factory = isset($options['handle_factory'])
+            ? $options['handle_factory']
+            : new CurlFactory();
+        $this->maxHandles = isset($options['max_handles'])
+            ? $options['max_handles']
+            : 5;
+    }
+
+    public function __destruct()
+    {
+        if (PHP_VERSION_ID >= 80000) {
+            return;
+        }
+        foreach ($this->handles as $handle) {
+            if (is_resource($handle)) {
+                curl_close($handle);
+            }
+        }
+    }
+
+    /**
+     * @param array $request
+     *
+     * @return CompletedFutureArray
+     */
+    public function __invoke(array $request)
+    {
+        return new CompletedFutureArray(
+            $this->_invokeAsArray($request)
+        );
+    }
+
+    /**
+     * @internal
+     *
+     * @param array $request
+     *
+     * @return array
+     */
+    public function _invokeAsArray(array $request)
+    {
+        $factory = $this->factory;
+
+        // Ensure headers are by reference. They're updated elsewhere.
+        $result = $factory($request, $this->checkoutEasyHandle());
+        $h = $result[0];
+        $hd =& $result[1];
+        $bd = $result[2];
+        Core::doSleep($request);
+        curl_exec($h);
+        $response = ['transfer_stats' => curl_getinfo($h)];
+        $response['curl']['error'] = curl_error($h);
+        $response['curl']['errno'] = curl_errno($h);
+        $response['transfer_stats'] = array_merge($response['transfer_stats'], $response['curl']);
+        $this->releaseEasyHandle($h);
+
+        return CurlFactory::createResponse([$this, '_invokeAsArray'], $request, $response, $hd, $bd);
+    }
+
+    private function checkoutEasyHandle()
+    {
+        // Find an unused handle in the cache
+        if (false !== ($key = array_search(false, $this->ownedHandles, true))) {
+            $this->ownedHandles[$key] = true;
+            return $this->handles[$key];
+        }
+
+        // Add a new handle
+        $handle = curl_init();
+        $id = (int) $handle;
+        $this->handles[$id] = $handle;
+        $this->ownedHandles[$id] = true;
+
+        return $handle;
+    }
+
+    private function releaseEasyHandle($handle)
+    {
+        $id = (int) $handle;
+        if (count($this->ownedHandles) > $this->maxHandles) {
+            if (PHP_VERSION_ID < 80000) {
+                curl_close($this->handles[$id]);
+            }
+            unset($this->handles[$id], $this->ownedHandles[$id]);
+        } else {
+            // curl_reset doesn't clear these out for some reason
+            static $unsetValues = [
+                CURLOPT_HEADERFUNCTION   => null,
+                CURLOPT_WRITEFUNCTION    => null,
+                CURLOPT_READFUNCTION     => null,
+                CURLOPT_PROGRESSFUNCTION => null,
+            ];
+            curl_setopt_array($handle, $unsetValues);
+            curl_reset($handle);
+            $this->ownedHandles[$id] = false;
+        }
+    }
+}
@@ -0,0 +1,252 @@
+<?php
+namespace GuzzleHttp\Ring\Client;
+
+use GuzzleHttp\Ring\Future\FutureArray;
+use React\Promise\Deferred;
+
+/**
+ * Returns an asynchronous response using curl_multi_* functions.
+ *
+ * This handler supports future responses and the "delay" request client
+ * option that can be used to delay before sending a request.
+ *
+ * When using the CurlMultiHandler, custom curl options can be specified as an
+ * associative array of curl option constants mapping to values in the
+ * **curl** key of the "client" key of the request.
+ *
+ * @property resource $_mh Internal use only. Lazy loaded multi-handle.
+ */
+#[\AllowDynamicProperties]
+class CurlMultiHandler
+{
+    /** @var callable */
+    private $factory;
+    private $selectTimeout;
+    private $active = 0;
+    private $handles = [];
+    private $delays = [];
+    private $maxHandles;
+
+    /**
+     * This handler accepts the following options:
+     *
+     * - mh: An optional curl_multi resource
+     * - handle_factory: An optional callable used to generate curl handle
+     *   resources. the callable accepts a request hash and returns an array
+     *   of the handle, headers file resource, and the body resource.
+     * - select_timeout: Optional timeout (in seconds) to block before timing
+     *   out while selecting curl handles. Defaults to 1 second.
+     * - max_handles: Optional integer representing the maximum number of
+     *   open requests. When this number is reached, the queued futures are
+     *   flushed.
+     *
+     * @param array $options
+     */
+    public function __construct(array $options = [])
+    {
+        if (isset($options['mh'])) {
+            $this->_mh = $options['mh'];
+        }
+        $this->factory = isset($options['handle_factory'])
+            ? $options['handle_factory'] : new CurlFactory();
+        $this->selectTimeout = isset($options['select_timeout'])
+            ? $options['select_timeout'] : 1;
+        $this->maxHandles = isset($options['max_handles'])
+            ? $options['max_handles'] : 100;
+    }
+
+    public function __get($name)
+    {
+        if ($name === '_mh') {
+            return $this->_mh = curl_multi_init();
+        }
+
+        throw new \BadMethodCallException();
+    }
+
+    public function __destruct()
+    {
+        // Finish any open connections before terminating the script.
+        if ($this->handles) {
+            $this->execute();
+        }
+
+        if (isset($this->_mh)) {
+            curl_multi_close($this->_mh);
+            unset($this->_mh);
+        }
+    }
+
+    public function __invoke(array $request)
+    {
+        $factory = $this->factory;
+        $result = $factory($request);
+        $entry = [
+            'request'  => $request,
+            'response' => [],
+            'handle'   => $result[0],
+            'headers'  => &$result[1],
+            'body'     => $result[2],
+            'deferred' => new Deferred(),
+        ];
+
+        $id = (int) $result[0];
+
+        $future = new FutureArray(
+            $entry['deferred']->promise(),
+            [$this, 'execute'],
+            function () use ($id) {
+                return $this->cancel($id);
+            }
+        );
+
+        $this->addRequest($entry);
+
+        // Transfer outstanding requests if there are too many open handles.
+        if (count($this->handles) >= $this->maxHandles) {
+            $this->execute();
+        }
+
+        return $future;
+    }
+
+    /**
+     * Runs until all outstanding connections have completed.
+     */
+    public function execute()
+    {
+        do {
+
+            if ($this->active &&
+                curl_multi_select($this->_mh, $this->selectTimeout) === -1
+            ) {
+                // Perform a usleep if a select returns -1.
+                // See: https://bugs.php.net/bug.php?id=61141
+                usleep(250);
+            }
+
+            // Add any delayed futures if needed.
+            if ($this->delays) {
+                $this->addDelays();
+            }
+
+            do {
+                $mrc = curl_multi_exec($this->_mh, $this->active);
+            } while ($mrc === CURLM_CALL_MULTI_PERFORM);
+
+            $this->processMessages();
+
+            // If there are delays but no transfers, then sleep for a bit.
+            if (!$this->active && $this->delays) {
+                usleep(500);
+            }
+
+        } while ($this->active || $this->handles);
+    }
+
+    private function addRequest(array &$entry)
+    {
+        $id = (int) $entry['handle'];
+        $this->handles[$id] = $entry;
+
+        // If the request is a delay, then add the reques to the curl multi
+        // pool only after the specified delay.
+        if (isset($entry['request']['client']['delay'])) {
+            $this->delays[$id] = microtime(true) + ($entry['request']['client']['delay'] / 1000);
+        } elseif (empty($entry['request']['future'])) {
+            curl_multi_add_handle($this->_mh, $entry['handle']);
+        } else {
+            curl_multi_add_handle($this->_mh, $entry['handle']);
+            // "lazy" futures are only sent once the pool has many requests.
+            if ($entry['request']['future'] !== 'lazy') {
+                do {
+                    $mrc = curl_multi_exec($this->_mh, $this->active);
+                } while ($mrc === CURLM_CALL_MULTI_PERFORM);
+                $this->processMessages();
+            }
+        }
+    }
+
+    private function removeProcessed($id)
+    {
+        if (isset($this->handles[$id])) {
+            curl_multi_remove_handle(
+                $this->_mh,
+                $this->handles[$id]['handle']
+            );
+            if (PHP_VERSION_ID < 80000) {
+                curl_close($this->handles[$id]['handle']);
+            }
+            unset($this->handles[$id], $this->delays[$id]);
+        }
+    }
+
+    /**
+     * Cancels a handle from sending and removes references to it.
+     *
+     * @param int $id Handle ID to cancel and remove.
+     *
+     * @return bool True on success, false on failure.
+     */
+    private function cancel($id)
+    {
+        // Cannot cancel if it has been processed.
+        if (!isset($this->handles[$id])) {
+            return false;
+        }
+
+        $handle = $this->handles[$id]['handle'];
+        unset($this->delays[$id], $this->handles[$id]);
+        curl_multi_remove_handle($this->_mh, $handle);
+        if (PHP_VERSION_ID < 80000) {
+            curl_close($handle);
+        }
+        return true;
+    }
+
+    private function addDelays()
+    {
+        $currentTime = microtime(true);
+
+        foreach ($this->delays as $id => $delay) {
+            if ($currentTime >= $delay) {
+                unset($this->delays[$id]);
+                curl_multi_add_handle(
+                    $this->_mh,
+                    $this->handles[$id]['handle']
+                );
+            }
+        }
+    }
+
+    private function processMessages()
+    {
+        while ($done = curl_multi_info_read($this->_mh)) {
+            $id = (int) $done['handle'];
+
+            if (!isset($this->handles[$id])) {
+                // Probably was cancelled.
+                continue;
+            }
+
+            $entry = $this->handles[$id];
+            $entry['response']['transfer_stats'] = curl_getinfo($done['handle']);
+
+            if ($done['result'] !== CURLM_OK) {
+                $entry['response']['curl']['errno'] = $done['result'];
+                $entry['response']['curl']['error'] = curl_error($done['handle']);
+            }
+
+            $result = CurlFactory::createResponse(
+                $this,
+                $entry['request'],
+                $entry['response'],
+                $entry['headers'],
+                $entry['body']
+            );
+
+            $this->removeProcessed($id);
+            $entry['deferred']->resolve($result);
+        }
+    }
+}
@@ -0,0 +1,58 @@
+<?php
+namespace GuzzleHttp\Ring\Client;
+
+/**
+ * Provides basic middleware wrappers.
+ *
+ * If a middleware is more complex than a few lines of code, then it should
+ * be implemented in a class rather than a static method.
+ */
+class Middleware
+{
+    /**
+     * Sends future requests to a future compatible handler while sending all
+     * other requests to a default handler.
+     *
+     * When the "future" option is not provided on a request, any future responses
+     * are automatically converted to synchronous responses and block.
+     *
+     * @param callable $default Handler used for non-streaming responses
+     * @param callable $future  Handler used for future responses
+     *
+     * @return callable Returns the composed handler.
+     */
+    public static function wrapFuture(
+        callable $default,
+        callable $future
+    ) {
+        return function (array $request) use ($default, $future) {
+            return empty($request['client']['future'])
+                ? $default($request)
+                : $future($request);
+        };
+    }
+
+    /**
+     * Sends streaming requests to a streaming compatible handler while sendin
+     * all other requests to a default handler.
+     *
+     * This, for example, could be useful for taking advantage of the
+     * performance benefits of curl while still supporting true streaming
+     * through the StreamHandler.
+     *
+     * @param callable $default   Handler used for non-streaming responses
+     * @param callable $streaming Handler used for streaming responses
+     *
+     * @return callable Returns the composed handler.
+     */
+    public static function wrapStreaming(
+        callable $default,
+        callable $streaming
+    ) {
+        return function (array $request) use ($default, $streaming) {
+            return empty($request['client']['stream'])
+                ? $default($request)
+                : $streaming($request);
+        };
+    }
+}
@@ -0,0 +1,52 @@
+<?php
+namespace GuzzleHttp\Ring\Client;
+
+use GuzzleHttp\Ring\Core;
+use GuzzleHttp\Ring\Future\CompletedFutureArray;
+use GuzzleHttp\Ring\Future\FutureArrayInterface;
+
+/**
+ * Ring handler that returns a canned response or evaluated function result.
+ */
+class MockHandler
+{
+    /** @var callable|array|FutureArrayInterface */
+    private $result;
+
+    /**
+     * Provide an array or future to always return the same value. Provide a
+     * callable that accepts a request object and returns an array or future
+     * to dynamically create a response.
+     *
+     * @param array|FutureArrayInterface|callable $result Mock return value.
+     */
+    public function __construct($result)
+    {
+        $this->result = $result;
+    }
+
+    public function __invoke(array $request)
+    {
+        Core::doSleep($request);
+        $response = is_callable($this->result)
+            ? call_user_func($this->result, $request)
+            : $this->result;
+
+        if (is_array($response)) {
+            $response = new CompletedFutureArray($response + [
+                'status'        => null,
+                'body'          => null,
+                'headers'       => [],
+                'reason'        => null,
+                'effective_url' => null,
+            ]);
+        } elseif (!$response instanceof FutureArrayInterface) {
+            throw new \InvalidArgumentException(
+                'Response must be an array or FutureArrayInterface. Found '
+                . Core::describeType($request)
+            );
+        }
+
+        return $response;
+    }
+}
@@ -0,0 +1,414 @@
+<?php
+namespace GuzzleHttp\Ring\Client;
+
+use GuzzleHttp\Ring\Core;
+use GuzzleHttp\Ring\Exception\ConnectException;
+use GuzzleHttp\Ring\Exception\RingException;
+use GuzzleHttp\Ring\Future\CompletedFutureArray;
+use GuzzleHttp\Stream\InflateStream;
+use GuzzleHttp\Stream\StreamInterface;
+use GuzzleHttp\Stream\Stream;
+use GuzzleHttp\Stream\Utils;
+
+/**
+ * RingPHP client handler that uses PHP's HTTP stream wrapper.
+ */
+class StreamHandler
+{
+    private $options;
+    private $lastHeaders;
+
+    public function __construct(array $options = [])
+    {
+        $this->options = $options;
+    }
+
+    public function __invoke(array $request)
+    {
+        $url = Core::url($request);
+        Core::doSleep($request);
+
+        try {
+            // Does not support the expect header.
+            $request = Core::removeHeader($request, 'Expect');
+            $stream = $this->createStream($url, $request);
+            return $this->createResponse($request, $url, $stream);
+        } catch (RingException $e) {
+            return $this->createErrorResponse($url, $e);
+        }
+    }
+
+    private function createResponse(array $request, $url, $stream)
+    {
+        $hdrs = $this->lastHeaders;
+        $this->lastHeaders = null;
+        $parts = explode(' ', array_shift($hdrs), 3);
+        $response = [
+            'version'        => substr($parts[0], 5),
+            'status'         => $parts[1],
+            'reason'         => isset($parts[2]) ? $parts[2] : null,
+            'headers'        => Core::headersFromLines($hdrs),
+            'effective_url'  => $url,
+        ];
+
+        $stream = $this->checkDecode($request, $response, $stream);
+
+        // If not streaming, then drain the response into a stream.
+        if (empty($request['client']['stream'])) {
+            $dest = isset($request['client']['save_to'])
+                ? $request['client']['save_to']
+                : fopen('php://temp', 'r+');
+            $stream = $this->drain($stream, $dest);
+        }
+
+        $response['body'] = $stream;
+
+        return new CompletedFutureArray($response);
+    }
+
+    private function checkDecode(array $request, array $response, $stream)
+    {
+        // Automatically decode responses when instructed.
+        if (!empty($request['client']['decode_content'])) {
+            switch (Core::firstHeader($response, 'Content-Encoding', true)) {
+                case 'gzip':
+                case 'deflate':
+                    $stream = new InflateStream(Stream::factory($stream));
+                    break;
+            }
+        }
+
+        return $stream;
+    }
+
+    /**
+     * Drains the stream into the "save_to" client option.
+     *
+     * @param resource                        $stream
+     * @param string|resource|StreamInterface $dest
+     *
+     * @return Stream
+     * @throws \RuntimeException when the save_to option is invalid.
+     */
+    private function drain($stream, $dest)
+    {
+        if (is_resource($stream)) {
+            if (!is_resource($dest)) {
+                $stream = Stream::factory($stream);
+            } else {
+                stream_copy_to_stream($stream, $dest);
+                fclose($stream);
+                rewind($dest);
+                return $dest;
+            }
+        }
+
+        // Stream the response into the destination stream
+        $dest = is_string($dest)
+            ? new Stream(Utils::open($dest, 'r+'))
+            : Stream::factory($dest);
+
+        Utils::copyToStream($stream, $dest);
+        $dest->seek(0);
+        $stream->close();
+
+        return $dest;
+    }
+
+    /**
+     * Creates an error response for the given stream.
+     *
+     * @param string        $url
+     * @param RingException $e
+     *
+     * @return array
+     */
+    private function createErrorResponse($url, RingException $e)
+    {
+        // Determine if the error was a networking error.
+        $message = $e->getMessage();
+
+        // This list can probably get more comprehensive.
+        if (strpos($message, 'getaddrinfo') // DNS lookup failed
+            || strpos($message, 'Connection refused')
+        ) {
+            $e = new ConnectException($e->getMessage(), 0, $e);
+        }
+
+        return new CompletedFutureArray([
+            'status'        => null,
+            'body'          => null,
+            'headers'       => [],
+            'effective_url' => $url,
+            'error'         => $e
+        ]);
+    }
+
+    /**
+     * Create a resource and check to ensure it was created successfully
+     *
+     * @param callable $callback Callable that returns stream resource
+     *
+     * @return resource
+     * @throws \RuntimeException on error
+     */
+    private function createResource(callable $callback)
+    {
+        $errors = null;
+        set_error_handler(function ($_, $msg, $file, $line) use (&$errors) {
+            $errors[] = [
+                'message' => $msg,
+                'file'    => $file,
+                'line'    => $line
+            ];
+            return true;
+        });
+
+        $resource = $callback();
+        restore_error_handler();
+
+        if (!$resource) {
+            $message = 'Error creating resource: ';
+            foreach ($errors as $err) {
+                foreach ($err as $key => $value) {
+                    $message .= "[$key] $value" . PHP_EOL;
+                }
+            }
+            throw new RingException(trim($message));
+        }
+
+        return $resource;
+    }
+
+    private function createStream($url, array $request)
+    {
+        static $methods;
+        if (!$methods) {
+            $methods = array_flip(get_class_methods(__CLASS__));
+        }
+
+        // HTTP/1.1 streams using the PHP stream wrapper require a
+        // Connection: close header
+        if ((!isset($request['version']) || $request['version'] == '1.1')
+            && !Core::hasHeader($request, 'Connection')
+        ) {
+            $request['headers']['Connection'] = ['close'];
+        }
+
+        // Ensure SSL is verified by default
+        if (!isset($request['client']['verify'])) {
+            $request['client']['verify'] = true;
+        }
+
+        $params = [];
+        $options = $this->getDefaultOptions($request);
+
+        if (isset($request['client'])) {
+            foreach ($request['client'] as $key => $value) {
+                $method = "add_{$key}";
+                if (isset($methods[$method])) {
+                    $this->{$method}($request, $options, $value, $params);
+                }
+            }
+        }
+
+        return $this->createStreamResource(
+            $url,
+            $request,
+            $options,
+            $this->createContext($request, $options, $params)
+        );
+    }
+
+    private function getDefaultOptions(array $request)
+    {
+        $headers = "";
+        foreach ($request['headers'] as $name => $value) {
+            foreach ((array) $value as $val) {
+                $headers .= "$name: $val\r\n";
+            }
+        }
+
+        $context = [
+            'http' => [
+                'method'           => $request['http_method'],
+                'header'           => $headers,
+                'protocol_version' => isset($request['version']) ? $request['version'] : 1.1,
+                'ignore_errors'    => true,
+                'follow_location'  => 0,
+            ],
+        ];
+
+        $body = Core::body($request);
+        if (isset($body)) {
+            $context['http']['content'] = $body;
+            // Prevent the HTTP handler from adding a Content-Type header.
+            if (!Core::hasHeader($request, 'Content-Type')) {
+                $context['http']['header'] .= "Content-Type:\r\n";
+            }
+        }
+
+        $context['http']['header'] = rtrim($context['http']['header']);
+
+        return $context;
+    }
+
+    private function add_proxy(array $request, &$options, $value, &$params)
+    {
+        if (!is_array($value)) {
+            $options['http']['proxy'] = $value;
+        } else {
+            $scheme = isset($request['scheme']) ? $request['scheme'] : 'http';
+            if (isset($value[$scheme])) {
+                $options['http']['proxy'] = $value[$scheme];
+            }
+        }
+    }
+
+    private function add_timeout(array $request, &$options, $value, &$params)
+    {
+        $options['http']['timeout'] = $value;
+    }
+
+    private function add_verify(array $request, &$options, $value, &$params)
+    {
+        if ($value === true) {
+            // PHP 5.6 or greater will find the system cert by default. When
+            // < 5.6, use the Guzzle bundled cacert.
+            if (PHP_VERSION_ID < 50600) {
+                $options['ssl']['cafile'] = ClientUtils::getDefaultCaBundle();
+            }
+        } elseif (is_string($value)) {
+            $options['ssl']['cafile'] = $value;
+            if (!file_exists($value)) {
+                throw new RingException("SSL CA bundle not found: $value");
+            }
+        } elseif ($value === false) {
+            $options['ssl']['verify_peer'] = false;
+            $options['ssl']['allow_self_signed'] = true;
+            return;
+        } else {
+            throw new RingException('Invalid verify request option');
+        }
+
+        $options['ssl']['verify_peer'] = true;
+        $options['ssl']['allow_self_signed'] = false;
+    }
+
+    private function add_cert(array $request, &$options, $value, &$params)
+    {
+        if (is_array($value)) {
+            $options['ssl']['passphrase'] = $value[1];
+            $value = $value[0];
+        }
+
+        if (!file_exists($value)) {
+            throw new RingException("SSL certificate not found: {$value}");
+        }
+
+        $options['ssl']['local_cert'] = $value;
+    }
+
+    private function add_progress(array $request, &$options, $value, &$params)
+    {
+        $fn = function ($code, $_1, $_2, $_3, $transferred, $total) use ($value) {
+            if ($code == STREAM_NOTIFY_PROGRESS) {
+                $value($total, $transferred, null, null);
+            }
+        };
+
+        // Wrap the existing function if needed.
+        $params['notification'] = isset($params['notification'])
+            ? Core::callArray([$params['notification'], $fn])
+            : $fn;
+    }
+
+    private function add_debug(array $request, &$options, $value, &$params)
+    {
+        if ($value === false) {
+            return;
+        }
+
+        static $map = [
+            STREAM_NOTIFY_CONNECT       => 'CONNECT',
+            STREAM_NOTIFY_AUTH_REQUIRED => 'AUTH_REQUIRED',
+            STREAM_NOTIFY_AUTH_RESULT   => 'AUTH_RESULT',
+            STREAM_NOTIFY_MIME_TYPE_IS  => 'MIME_TYPE_IS',
+            STREAM_NOTIFY_FILE_SIZE_IS  => 'FILE_SIZE_IS',
+            STREAM_NOTIFY_REDIRECTED    => 'REDIRECTED',
+            STREAM_NOTIFY_PROGRESS      => 'PROGRESS',
+            STREAM_NOTIFY_FAILURE       => 'FAILURE',
+            STREAM_NOTIFY_COMPLETED     => 'COMPLETED',
+            STREAM_NOTIFY_RESOLVE       => 'RESOLVE',
+        ];
+
+        static $args = ['severity', 'message', 'message_code',
+            'bytes_transferred', 'bytes_max'];
+
+        $value = Core::getDebugResource($value);
+        $ident = $request['http_method'] . ' ' . Core::url($request);
+        $fn = function () use ($ident, $value, $map, $args) {
+            $passed = func_get_args();
+            $code = array_shift($passed);
+            fprintf($value, '<%s> [%s] ', $ident, $map[$code]);
+            foreach (array_filter($passed) as $i => $v) {
+                fwrite($value, $args[$i] . ': "' . $v . '" ');
+            }
+            fwrite($value, "\n");
+        };
+
+        // Wrap the existing function if needed.
+        $params['notification'] = isset($params['notification'])
+            ? Core::callArray([$params['notification'], $fn])
+            : $fn;
+    }
+
+    private function applyCustomOptions(array $request, array &$options)
+    {
+        if (!isset($request['client']['stream_context'])) {
+            return;
+        }
+
+        if (!is_array($request['client']['stream_context'])) {
+            throw new RingException('stream_context must be an array');
+        }
+
+        $options = array_replace_recursive(
+            $options,
+            $request['client']['stream_context']
+        );
+    }
+
+    private function createContext(array $request, array $options, array $params)
+    {
+        $this->applyCustomOptions($request, $options);
+        return $this->createResource(
+            function () use ($request, $options, $params) {
+                return stream_context_create($options, $params);
+            },
+            $request,
+            $options
+        );
+    }
+
+    private function createStreamResource(
+        $url,
+        array $request,
+        array $options,
+        $context
+    ) {
+        return $this->createResource(
+            function () use ($url, $context) {
+                if (false === strpos($url, 'http')) {
+                    trigger_error("URL is invalid: {$url}", E_USER_WARNING);
+                    return null;
+                }
+                $resource = fopen($url, 'r', null, $context);
+                $this->lastHeaders = $http_response_header;
+                return $resource;
+            },
+            $request,
+            $options
+        );
+    }
+}
@@ -0,0 +1,364 @@
+<?php
+namespace GuzzleHttp\Ring;
+
+use GuzzleHttp\Stream\StreamInterface;
+use GuzzleHttp\Ring\Future\FutureArrayInterface;
+use GuzzleHttp\Ring\Future\FutureArray;
+
+/**
+ * Provides core functionality of Ring handlers and middleware.
+ */
+class Core
+{
+    /**
+     * Returns a function that calls all of the provided functions, in order,
+     * passing the arguments provided to the composed function to each function.
+     *
+     * @param callable[] $functions Array of functions to proxy to.
+     *
+     * @return callable
+     */
+    public static function callArray(array $functions)
+    {
+        return function () use ($functions) {
+            $args = func_get_args();
+            foreach ($functions as $fn) {
+                call_user_func_array($fn, $args);
+            }
+        };
+    }
+
+    /**
+     * Gets an array of header line values from a message for a specific header
+     *
+     * This method searches through the "headers" key of a message for a header
+     * using a case-insensitive search.
+     *
+     * @param array  $message Request or response hash.
+     * @param string $header  Header to retrieve
+     *
+     * @return array
+     */
+    public static function headerLines($message, $header)
+    {
+        $result = [];
+
+        if (!empty($message['headers'])) {
+            foreach ($message['headers'] as $name => $value) {
+                if (!strcasecmp($name, $header)) {
+                    $result = array_merge($result, $value);
+                }
+            }
+        }
+
+        return $result;
+    }
+
+    /**
+     * Gets a header value from a message as a string or null
+     *
+     * This method searches through the "headers" key of a message for a header
+     * using a case-insensitive search. The lines of the header are imploded
+     * using commas into a single string return value.
+     *
+     * @param array  $message Request or response hash.
+     * @param string $header  Header to retrieve
+     *
+     * @return string|null Returns the header string if found, or null if not.
+     */
+    public static function header($message, $header)
+    {
+        $match = self::headerLines($message, $header);
+        return $match ? implode(', ', $match) : null;
+    }
+
+    /**
+     * Returns the first header value from a message as a string or null. If
+     * a header line contains multiple values separated by a comma, then this
+     * function will return the first value in the list.
+     *
+     * @param array  $message Request or response hash.
+     * @param string $header  Header to retrieve
+     *
+     * @return string|null Returns the value as a string if found.
+     */
+    public static function firstHeader($message, $header)
+    {
+        if (!empty($message['headers'])) {
+            foreach ($message['headers'] as $name => $value) {
+                if (!strcasecmp($name, $header)) {
+                    // Return the match itself if it is a single value.
+                    $pos = strpos($value[0], ',');
+                    return $pos ? substr($value[0], 0, $pos) : $value[0];
+                }
+            }
+        }
+
+        return null;
+    }
+
+    /**
+     * Returns true if a message has the provided case-insensitive header.
+     *
+     * @param array  $message Request or response hash.
+     * @param string $header  Header to check
+     *
+     * @return bool
+     */
+    public static function hasHeader($message, $header)
+    {
+        if (!empty($message['headers'])) {
+            foreach ($message['headers'] as $name => $value) {
+                if (!strcasecmp($name, $header)) {
+                    return true;
+                }
+            }
+        }
+
+        return false;
+    }
+
+    /**
+     * Parses an array of header lines into an associative array of headers.
+     *
+     * @param array $lines Header lines array of strings in the following
+     *                     format: "Name: Value"
+     * @return array
+     */
+    public static function headersFromLines($lines)
+    {
+        $headers = [];
+
+        foreach ($lines as $line) {
+            $parts = explode(':', $line, 2);
+            $headers[trim($parts[0])][] = isset($parts[1])
+                ? trim($parts[1])
+                : null;
+        }
+
+        return $headers;
+    }
+
+    /**
+     * Removes a header from a message using a case-insensitive comparison.
+     *
+     * @param array  $message Message that contains 'headers'
+     * @param string $header  Header to remove
+     *
+     * @return array
+     */
+    public static function removeHeader(array $message, $header)
+    {
+        if (isset($message['headers'])) {
+            foreach (array_keys($message['headers']) as $key) {
+                if (!strcasecmp($header, $key)) {
+                    unset($message['headers'][$key]);
+                }
+            }
+        }
+
+        return $message;
+    }
+
+    /**
+     * Replaces any existing case insensitive headers with the given value.
+     *
+     * @param array  $message Message that contains 'headers'
+     * @param string $header  Header to set.
+     * @param array  $value   Value to set.
+     *
+     * @return array
+     */
+    public static function setHeader(array $message, $header, array $value)
+    {
+        $message = self::removeHeader($message, $header);
+        $message['headers'][$header] = $value;
+
+        return $message;
+    }
+
+    /**
+     * Creates a URL string from a request.
+     *
+     * If the "url" key is present on the request, it is returned, otherwise
+     * the url is built up based on the scheme, host, uri, and query_string
+     * request values.
+     *
+     * @param array $request Request to get the URL from
+     *
+     * @return string Returns the request URL as a string.
+     * @throws \InvalidArgumentException if no Host header is present.
+     */
+    public static function url(array $request)
+    {
+        if (isset($request['url'])) {
+            return $request['url'];
+        }
+
+        $uri = (isset($request['scheme'])
+                ? $request['scheme'] : 'http') . '://';
+
+        if ($host = self::header($request, 'host')) {
+            $uri .= $host;
+        } else {
+            throw new \InvalidArgumentException('No Host header was provided');
+        }
+
+        if (isset($request['uri'])) {
+            $uri .= $request['uri'];
+        }
+
+        if (isset($request['query_string'])) {
+            $uri .= '?' . $request['query_string'];
+        }
+
+        return $uri;
+    }
+
+    /**
+     * Reads the body of a message into a string.
+     *
+     * @param array|FutureArrayInterface $message Array containing a "body" key
+     *
+     * @return null|string Returns the body as a string or null if not set.
+     * @throws \InvalidArgumentException if a request body is invalid.
+     */
+    public static function body($message)
+    {
+        if (!isset($message['body'])) {
+            return null;
+        }
+
+        if ($message['body'] instanceof StreamInterface) {
+            return (string) $message['body'];
+        }
+
+        switch (gettype($message['body'])) {
+            case 'string':
+                return $message['body'];
+            case 'resource':
+                return stream_get_contents($message['body']);
+            case 'object':
+                if ($message['body'] instanceof \Iterator) {
+                    return implode('', iterator_to_array($message['body']));
+                } elseif (method_exists($message['body'], '__toString')) {
+                    return (string) $message['body'];
+                }
+            default:
+                throw new \InvalidArgumentException('Invalid request body: '
+                    . self::describeType($message['body']));
+        }
+    }
+
+    /**
+     * Rewind the body of the provided message if possible.
+     *
+     * @param array $message Message that contains a 'body' field.
+     *
+     * @return bool Returns true on success, false on failure
+     */
+    public static function rewindBody($message)
+    {
+        if ($message['body'] instanceof StreamInterface) {
+            return $message['body']->seek(0);
+        }
+
+        if ($message['body'] instanceof \Generator) {
+            return false;
+        }
+
+        if ($message['body'] instanceof \Iterator) {
+            $message['body']->rewind();
+            return true;
+        }
+
+        if (is_resource($message['body'])) {
+            return rewind($message['body']);
+        }
+
+        return is_string($message['body'])
+            || (is_object($message['body'])
+                && method_exists($message['body'], '__toString'));
+    }
+
+    /**
+     * Debug function used to describe the provided value type and class.
+     *
+     * @param mixed $input
+     *
+     * @return string Returns a string containing the type of the variable and
+     *                if a class is provided, the class name.
+     */
+    public static function describeType($input)
+    {
+        switch (gettype($input)) {
+            case 'object':
+                return 'object(' . get_class($input) . ')';
+            case 'array':
+                return 'array(' . count($input) . ')';
+            default:
+                ob_start();
+                var_dump($input);
+                // normalize float vs double
+                return str_replace('double(', 'float(', rtrim(ob_get_clean()));
+        }
+    }
+
+    /**
+     * Sleep for the specified amount of time specified in the request's
+     * ['client']['delay'] option if present.
+     *
+     * This function should only be used when a non-blocking sleep is not
+     * possible.
+     *
+     * @param array $request Request to sleep
+     */
+    public static function doSleep(array $request)
+    {
+        if (isset($request['client']['delay'])) {
+            usleep($request['client']['delay'] * 1000);
+        }
+    }
+
+    /**
+     * Returns a proxied future that modifies the dereferenced value of another
+     * future using a promise.
+     *
+     * @param FutureArrayInterface $future      Future to wrap with a new future
+     * @param callable    $onFulfilled Invoked when the future fulfilled
+     * @param callable    $onRejected  Invoked when the future rejected
+     * @param callable    $onProgress  Invoked when the future progresses
+     *
+     * @return FutureArray
+     */
+    public static function proxy(
+        FutureArrayInterface $future,
+        ?callable $onFulfilled = null,
+        ?callable $onRejected = null,
+        ?callable $onProgress = null
+    ) {
+        return new FutureArray(
+            $future->then($onFulfilled, $onRejected, $onProgress),
+            [$future, 'wait'],
+            [$future, 'cancel']
+        );
+    }
+
+    /**
+     * Returns a debug stream based on the provided variable.
+     *
+     * @param mixed $value Optional value
+     *
+     * @return resource
+     */
+    public static function getDebugResource($value = null)
+    {
+        if (is_resource($value)) {
+            return $value;
+        } elseif (defined('STDOUT')) {
+            return STDOUT;
+        } else {
+            return fopen('php://output', 'w');
+        }
+    }
+}
@@ -0,0 +1,7 @@
+<?php
+namespace GuzzleHttp\Ring\Exception;
+
+/**
+ * Marker interface for cancelled exceptions.
+ */
+interface CancelledException {}
@@ -0,0 +1,4 @@
+<?php
+namespace GuzzleHttp\Ring\Exception;
+
+class CancelledFutureAccessException extends RingException implements CancelledException {}
@@ -0,0 +1,7 @@
+<?php
+namespace GuzzleHttp\Ring\Exception;
+
+/**
+ * Occurs when the connection failed.
+ */
+class ConnectException extends RingException {}
@@ -0,0 +1,4 @@
+<?php
+namespace GuzzleHttp\Ring\Exception;
+
+class RingException extends \RuntimeException {};
@@ -0,0 +1,134 @@
+<?php
+namespace GuzzleHttp\Ring\Future;
+
+use GuzzleHttp\Ring\Exception\CancelledFutureAccessException;
+use GuzzleHttp\Ring\Exception\RingException;
+use React\Promise\PromiseInterface;
+
+/**
+ * Implements common future functionality built on top of promises.
+ */
+trait BaseFutureTrait
+{
+    /** @var callable */
+    private $waitfn;
+
+    /** @var callable */
+    private $cancelfn;
+
+    /** @var PromiseInterface */
+    private $wrappedPromise;
+
+    /** @var \Exception Error encountered. */
+    private $error;
+
+    /** @var mixed Result of the future */
+    private $result;
+
+    private $isRealized = false;
+
+    /**
+     * @param PromiseInterface $promise Promise to shadow with the future.
+     * @param callable         $wait    Function that blocks until the deferred
+     *                                  computation has been resolved. This
+     *                                  function MUST resolve the deferred value
+     *                                  associated with the supplied promise.
+     * @param callable         $cancel  If possible and reasonable, provide a
+     *                                  function that can be used to cancel the
+     *                                  future from completing.
+     */
+    public function __construct(
+        PromiseInterface $promise,
+        ?callable $wait = null,
+        ?callable $cancel = null
+    ) {
+        $this->wrappedPromise = $promise;
+        $this->waitfn = $wait;
+        $this->cancelfn = $cancel;
+    }
+
+    /**
+     * @return mixed
+     */
+    public function wait()
+    {
+        if (!$this->isRealized) {
+            $this->addShadow();
+            if (!$this->isRealized && $this->waitfn) {
+                $this->invokeWait();
+            }
+            if (!$this->isRealized) {
+                $this->error = new RingException('Waiting did not resolve future');
+            }
+        }
+
+        if ($this->error) {
+            throw $this->error;
+        }
+
+        return $this->result;
+    }
+
+    /**
+     * @return PromiseInterface
+     */
+    public function promise()
+    {
+        return $this->wrappedPromise;
+    }
+
+    /**
+     * @return PromiseInterface
+     */
+    public function then(
+        ?callable $onFulfilled = null,
+        ?callable $onRejected = null,
+        ?callable $onProgress = null
+    ) {
+        return $this->wrappedPromise->then($onFulfilled, $onRejected, $onProgress);
+    }
+
+    public function cancel(): void
+    {
+        if (!$this->isRealized) {
+            $cancelfn = $this->cancelfn;
+            $this->waitfn = $this->cancelfn = null;
+            $this->isRealized = true;
+            $this->error = new CancelledFutureAccessException();
+            if ($cancelfn) {
+                $cancelfn($this);
+            }
+        }
+    }
+
+    private function addShadow()
+    {
+        // Get the result and error when the promise is resolved. Note that
+        // calling this function might trigger the resolution immediately.
+        $this->wrappedPromise->then(
+            function ($value) {
+                $this->isRealized = true;
+                $this->result = $value;
+                $this->waitfn = $this->cancelfn = null;
+            },
+            function ($error) {
+                $this->isRealized = true;
+                $this->error = $error;
+                $this->waitfn = $this->cancelfn = null;
+            }
+        );
+    }
+
+    private function invokeWait()
+    {
+        try {
+            $wait = $this->waitfn;
+            $this->waitfn = null;
+            $wait();
+        } catch (\Exception $e) {
+            // Defer can throw to reject.
+            $this->error = $e;
+            $this->isRealized = true;
+        }
+    }
+}
@@ -0,0 +1,67 @@
+<?php
+namespace GuzzleHttp\Ring\Future;
+
+/**
+ * Represents a future array that has been completed successfully.
+ */
+class CompletedFutureArray extends CompletedFutureValue implements FutureArrayInterface
+{
+    public function __construct(array $result)
+    {
+        parent::__construct($result);
+    }
+
+    #[\ReturnTypeWillChange]
+    /**
+     * @return bool 
+     */
+    public function offsetExists($offset)
+    {
+        return isset($this->result[$offset]);
+    }
+
+    #[\ReturnTypeWillChange]
+    /**
+     * @return mixed 
+     */
+    public function offsetGet($offset)
+    {
+        return $this->result[$offset];
+    }
+
+    #[\ReturnTypeWillChange]
+    /**
+     * @return void 
+     */
+    public function offsetSet($offset, $value)
+    {
+        $this->result[$offset] = $value;
+    }
+
+    #[\ReturnTypeWillChange]
+    /**
+     * @return void 
+     */
+    public function offsetUnset($offset)
+    {
+        unset($this->result[$offset]);
+    }
+
+    #[\ReturnTypeWillChange]
+    /**
+     * @return int 
+     */
+    public function count()
+    {
+        return count($this->result);
+    }
+
+    #[\ReturnTypeWillChange]
+    /**
+     * @return \ArrayIterator 
+     */
+    public function getIterator()
+    {
+        return new \ArrayIterator($this->result);
+    }
+}
@@ -0,0 +1,68 @@
+<?php
+namespace GuzzleHttp\Ring\Future;
+
+use React\Promise\PromiseInterface;
+
+use function React\Promise\reject;
+use function React\Promise\resolve;
+
+/**
+ * Represents a future value that has been resolved or rejected.
+ */
+class CompletedFutureValue implements FutureInterface
+{
+    protected $result;
+    protected $error;
+
+    private $cachedPromise;
+
+    /**
+     * @param mixed      $result Resolved result
+     * @param \Exception $e      Error. Pass a GuzzleHttp\Ring\Exception\CancelledFutureAccessException
+     *                           to mark the future as cancelled.
+     */
+    public function __construct($result, ?\Exception $e = null)
+    {
+        $this->result = $result;
+        $this->error = $e;
+    }
+
+    /**
+     * @return mixed
+     */
+    public function wait()
+    {
+        if ($this->error) {
+            throw $this->error;
+        }
+
+        return $this->result;
+    }
+
+    public function cancel(): void
+    {}
+
+    /**
+     * @return PromiseInterface
+     */
+    public function promise()
+    {
+        if (!$this->cachedPromise) {
+            $this->cachedPromise = $this->error
+                ? reject($this->error)
+                : resolve($this->result);
+        }
+
+        return $this->cachedPromise;
+    }
+
+    /**
+     * @return PromiseInterface
+     */
+    public function then(
+        ?callable $onFulfilled = null,
+        ?callable $onRejected = null
+    ) {
+        return $this->promise()->then($onFulfilled, $onRejected);
+    }
+}
@@ -0,0 +1,65 @@
+<?php
+namespace GuzzleHttp\Ring\Future;
+
+/**
+ * Represents a future array value that when dereferenced returns an array.
+ */
+#[\AllowDynamicProperties]
+class FutureArray implements FutureArrayInterface
+{
+    use MagicFutureTrait;
+
+    #[\ReturnTypeWillChange]
+    /** 
+     * @return bool 
+     */
+    public function offsetExists($offset)
+    {
+        return isset($this->_value[$offset]);
+    }
+
+    #[\ReturnTypeWillChange]
+    /** 
+     * @return mixed 
+     */
+    public function offsetGet($offset)
+    {
+        return $this->_value[$offset];
+    }
+
+    #[\ReturnTypeWillChange]
+    /** 
+     * @return void 
+     */
+    public function offsetSet($offset, $value)
+    {
+        $this->_value[$offset] = $value;
+    }
+
+    #[\ReturnTypeWillChange]
+    /** 
+     * @return void 
+     */
+    public function offsetUnset($offset)
+    {
+        unset($this->_value[$offset]);
+    }
+
+    #[\ReturnTypeWillChange]
+    /** 
+     * @return int 
+     */
+    public function count()
+    {
+        return count($this->_value);
+    }
+
+    #[\ReturnTypeWillChange]
+    /** 
+     * @return \ArrayIterator 
+     */
+    public function getIterator()
+    {
+        return new \ArrayIterator($this->_value);
+    }
+}
@@ -0,0 +1,11 @@
+<?php
+namespace GuzzleHttp\Ring\Future;
+
+/**
+ * Future that provides array-like access.
+ */
+interface FutureArrayInterface extends
+    FutureInterface,
+    \ArrayAccess,
+    \Countable,
+    \IteratorAggregate {};
@@ -0,0 +1,80 @@
+<?php
+namespace GuzzleHttp\Ring\Future;
+
+use React\Promise\PromiseInterface;
+
+/**
+ * Represents the result of a computation that may not have completed yet.
+ *
+ * You can use the future in a blocking manner using the wait() function, or
+ * you can use a promise from the future to receive the result when the future
+ * has been resolved.
+ *
+ * When the future is dereferenced using wait(), the result of the computation
+ * is cached and returned for subsequent calls to wait(). If the result of the
+ * computation has not yet completed when wait() is called, the call to wait()
+ * will block until the future has completed.
+ */
+interface FutureInterface
+{
+    /**
+     * Returns the result of the future either from cache or by blocking until
+     * it is complete.
+     *
+     * This method must block until the future has a result or is cancelled.
+     * Throwing an exception in the wait() method will mark the future as
+     * realized and will throw the exception each time wait() is called.
+     * Throwing an instance of GuzzleHttp\Ring\CancelledException will mark
+     * the future as realized, will not throw immediately, but will throw the
+     * exception if the future's wait() method is called again.
+     *
+     * @return mixed
+     */
+    public function wait();
+
+    /**
+     * Cancels the future, if possible.
+     */
+    public function cancel();
+
+    /**
+     * Transforms a promise's value by applying a function to the promise's fulfillment
+     * or rejection value. Returns a new promise for the transformed result.
+     *
+     * The `then()` method registers new fulfilled and rejection handlers with a promise
+     * (all parameters are optional):
+     *
+     *  * `$onFulfilled` will be invoked once the promise is fulfilled and passed
+     *     the result as the first argument.
+     *  * `$onRejected` will be invoked once the promise is rejected and passed the
+     *     reason as the first argument.
+     *  * `$onProgress` (deprecated) will be invoked whenever the producer of the promise
+     *     triggers progress notifications and passed a single argument (whatever it
+     *     wants) to indicate progress.
+     *
+     * It returns a new promise that will fulfill with the return value of either
+     * `$onFulfilled` or `$onRejected`, whichever is called, or will reject with
+     * the thrown exception if either throws.
+     *
+     * A promise makes the following guarantees about handlers registered in
+     * the same call to `then()`:
+     *
+     *  1. Only one of `$onFulfilled` or `$onRejected` will be called,
+     *      never both.
+     *  2. `$onFulfilled` and `$onRejected` will never be called more
+     *      than once.
+     *  3. `$onProgress` (deprecated) may be called multiple times.
+     *
+     * @param callable|null $onFulfilled
+     * @param callable|null $onRejected
+     * @return PromiseInterface
+     */
+    public function then(?callable $onFulfilled = null, ?callable $onRejected = null);
+
+    /**
+     * Returns the promise of the deferred.
+     *
+     * @return PromiseInterface
+     */
+    public function promise();
+}
@@ -0,0 +1,12 @@
+<?php
+namespace GuzzleHttp\Ring\Future;
+
+/**
+ * Represents a future value that responds to wait() to retrieve the promised
+ * value, but can also return promises that are delivered the value when it is
+ * available.
+ */
+class FutureValue implements FutureInterface
+{
+    use BaseFutureTrait;
+}
@@ -0,0 +1,32 @@
+<?php
+namespace GuzzleHttp\Ring\Future;
+
+/**
+ * Implements common future functionality that is triggered when the result
+ * property is accessed via a magic __get method.
+ *
+ * @property mixed $_value Actual data used by the future. Accessing this
+ *     property will cause the future to block if needed.
+ */
+trait MagicFutureTrait
+{
+    use BaseFutureTrait;
+
+    /**
+     * This function handles retrieving the dereferenced result when requested.
+     *
+     * @param string $name Should always be "data" or an exception is thrown.
+     *
+     * @return mixed Returns the dereferenced data.
+     * @throws \RuntimeException
+     * @throws \GuzzleHttp\Ring\Exception\CancelledException
+     */
+    public function __get($name)
+    {
+        if ($name !== '_value') {
+            throw new \RuntimeException("Class has no {$name} property");
+        }
+
+        return $this->_value = $this->wait();
+    }
+}