Add always seekable body plugin

spec/Plugin/AlwaysSeekableBodyPluginSpec.php

@@ -0,0 +1,98 @@
+<?php
+
+namespace spec\Http\Client\Common\Plugin;
+
+use Http\Client\Common\Plugin;
+use Http\Client\Common\Plugin\AlwaysSeekableBodyPlugin;
+use Http\Client\Promise\HttpFulfilledPromise;
+use Http\Message\Stream\BufferedStream;
+use PhpSpec\ObjectBehavior;
+use Prophecy\Argument;
+use Psr\Http\Message\RequestInterface;
+use Psr\Http\Message\ResponseInterface;
+use Psr\Http\Message\StreamInterface;
+
+class AlwaysSeekableBodyPluginSpec extends ObjectBehavior
+{
+    public function it_is_initializable()
+    {
+        $this->shouldHaveType(AlwaysSeekableBodyPlugin::class);
+    }
+
+    public function it_is_a_plugin()
+    {
+        $this->shouldImplement(Plugin::class);
+    }
+
+    public function it_decorate_response_body_if_not_seekable(RequestInterface $request, ResponseInterface $response, StreamInterface $responseStream, StreamInterface $requestStream)
+    {
+        $next = function () use ($response) {
+            return new HttpFulfilledPromise($response->getWrappedObject());
+        };
+
+        $request->getBody()->shouldBeCalled()->willReturn($requestStream);
+        $requestStream->isSeekable()->shouldBeCalled()->willReturn(true);
+
+        $response->getBody()->shouldBeCalled()->willReturn($responseStream);
+        $responseStream->isSeekable()->shouldBeCalled()->willReturn(false);
+        $responseStream->getSize()->willReturn(null);
+
+        $response->withBody(Argument::type(BufferedStream::class))->shouldBeCalled()->willReturn($response);
+
+        $this->handleRequest($request, $next, function () {});
+    }
+
+    public function it_does_not_decorate_response_body_if_seekable(RequestInterface $request, ResponseInterface $response, StreamInterface $responseStream, StreamInterface $requestStream)
+    {
+        $next = function () use ($response) {
+            return new HttpFulfilledPromise($response->getWrappedObject());
+        };
+
+        $request->getBody()->shouldBeCalled()->willReturn($requestStream);
+        $requestStream->isSeekable()->shouldBeCalled()->willReturn(true);
+
+        $response->getBody()->shouldBeCalled()->willReturn($responseStream);
+        $responseStream->isSeekable()->shouldBeCalled()->willReturn(true);
+        $responseStream->getSize()->willReturn(null);
+
+        $response->withBody(Argument::type(BufferedStream::class))->shouldNotBeCalled();
+
+        $this->handleRequest($request, $next, function () {});
+    }
+
+    public function it_decorate_request_body_if_not_seekable(RequestInterface $request, ResponseInterface $response, StreamInterface $responseStream, StreamInterface $requestStream)
+    {
+        $next = function () use ($response) {
+            return new HttpFulfilledPromise($response->getWrappedObject());
+        };
+
+        $response->getBody()->willReturn($responseStream);
+        $responseStream->isSeekable()->willReturn(true);
+
+        $request->getBody()->shouldBeCalled()->willReturn($requestStream);
+        $requestStream->isSeekable()->shouldBeCalled()->willReturn(false);
+        $requestStream->getSize()->willReturn(null);
+
+        $request->withBody(Argument::type(BufferedStream::class))->shouldBeCalled();
+
+        $this->handleRequest($request, $next, function () {});
+    }
+
+    public function it_does_not_decorate_request_body_if_seekable(RequestInterface $request, ResponseInterface $response, StreamInterface $responseStream, StreamInterface $requestStream)
+    {
+        $next = function () use ($response) {
+            return new HttpFulfilledPromise($response->getWrappedObject());
+        };
+
+        $response->getBody()->willReturn($responseStream);
+        $responseStream->isSeekable()->willReturn(true);
+
+        $request->getBody()->shouldBeCalled()->willReturn($requestStream);
+        $requestStream->isSeekable()->shouldBeCalled()->willReturn(true);
+        $requestStream->getSize()->willReturn(null);
+
+        $request->withBody(Argument::type(BufferedStream::class))->shouldNotBeCalled();
+
+        $this->handleRequest($request, $next, function () {});
+    }
+}

src/Plugin/AlwaysSeekableBodyPlugin.php

@@ -0,0 +1,65 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Http\Client\Common\Plugin;
+
+use Http\Client\Common\Plugin;
+use Http\Message\Stream\BufferedStream;
+use Http\Promise\Promise;
+use Psr\Http\Message\RequestInterface;
+use Psr\Http\Message\ResponseInterface;
+use Symfony\Component\OptionsResolver\OptionsResolver;
+
+/**
+ * Decorate the body of the request and the response if it's not seekable by using Http\Message\Stream\BufferedStream.
+ *
+ * @author Joel Wurtz <[email protected]>
+ */
+final class AlwaysSeekableBodyPlugin implements Plugin
+{
+    private $useFileBuffer;
+
+    private $memoryBufferSize;
+
+    /**
+     * @param array $config {
+     *
+     *    @var bool $use_file_buffer    Whether this plugin should use a file as a buffer if the stream is too big, defaults to true
+     *    @var int  $memory_buffer_size Max memory size in bytes to use for the buffer before it use a file, defaults to 2097152 (2 mb)
+     * }
+     */
+    public function __construct(array $config = [])
+    {
+        $resolver = new OptionsResolver();
+        $resolver->setDefaults([
+            'use_file_buffer' => true,
+            'memory_buffer_size' => 2097152,
+        ]);
+        $resolver->setAllowedTypes('use_file_buffer', 'bool');
+        $resolver->setAllowedTypes('memory_buffer_size', 'int');
+
+        $options = $resolver->resolve($config);
+
+        $this->useFileBuffer = $options['use_file_buffer'];
+        $this->memoryBufferSize = $options['memory_buffer_size'];
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function handleRequest(RequestInterface $request, callable $next, callable $first): Promise
+    {
+        if (!$request->getBody()->isSeekable()) {
+            $request = $request->withBody(new BufferedStream($request->getBody(), $this->useFileBuffer, $this->memoryBufferSize));
+        }
+
+        return $next($request)->then(function (ResponseInterface $response) {
+            if ($response->getBody()->isSeekable()) {
+                return $response;
+            }
+
+            return $response->withBody(new BufferedStream($response->getBody(), $this->useFileBuffer, $this->memoryBufferSize));
+        });
+    }
+}