Added sink for guzzle. (#1871)

CHANGELOG-2.0.md

@@ -42,6 +42,7 @@
 - [#1819](https://github.com/hyperf/hyperf/pull/1819) Added `hyperf/signal` component.
 - [#1844](https://github.com/hyperf/hyperf/pull/1844) Support type `\DateInterval` for `ttl` in `model-cache`.
 - [#1855](https://github.com/hyperf/hyperf/pull/1855) Added `ConstantFrequency` to flush one connection, when it is idle connection for the interval of time.
+- [#1871](https://github.com/hyperf/hyperf/pull/1871) Added `sink` for guzzle.
 
 ## Fixed
 

src/guzzle/src/CoroutineHandler.php

@@ -14,10 +14,12 @@ namespace Hyperf\Guzzle;
 use GuzzleHttp\Exception\ConnectException;
 use GuzzleHttp\Exception\RequestException;
 use GuzzleHttp\Promise\FulfilledPromise;
+use GuzzleHttp\Psr7;
 use GuzzleHttp\Psr7\Uri;
 use GuzzleHttp\RequestOptions;
 use GuzzleHttp\TransferStats;
 use Psr\Http\Message\RequestInterface;
+use Psr\Http\Message\StreamInterface;
 use Swoole\Coroutine;
 use Swoole\Coroutine\Http\Client;
 use function GuzzleHttp\is_host_in_noproxy;
@@ -179,10 +181,15 @@ class CoroutineHandler
             $client->headers['set-cookie'] = $client->set_cookie_headers;
         }
 
-        $response = new \GuzzleHttp\Psr7\Response(
+        $body = $client->body;
+        if (isset($options['sink']) && is_string($options['sink'])) {
+            $body = $this->createSink($body, $options['sink']);
+        }
+
+        $response = new Psr7\Response(
             $client->statusCode,
             isset($client->headers) ? $client->headers : [],
-            $client->body
+            $body
         );
 
         if ($callback = $options[RequestOptions::ON_STATS] ?? null) {
@@ -200,6 +207,26 @@ class CoroutineHandler
         return $response;
     }
 
+    protected function createStream(string $body): StreamInterface
+    {
+        return Psr7\stream_for($body);
+    }
+
+    protected function createSink(string $body, string $sink)
+    {
+        if (! empty($options['stream'])) {
+            return $body;
+        }
+
+        $stream = fopen($sink, 'w+');
+        if ($body !== '') {
+            fwrite($stream, $body);
+            fseek($stream, 0);
+        }
+
+        return $stream;
+    }
+
     protected function checkStatusCode(Client $client, $request)
     {
         $statusCode = $client->statusCode;

src/guzzle/tests/Cases/CoroutineHandlerTest.php

@@ -311,6 +311,16 @@ class CoroutineHandlerTest extends TestCase
         $this->assertTrue($bool);
     }
 
+    public function testSink()
+    {
+        $dir = BASE_PATH . '/runtime/guzzle/';
+        @mkdir($dir, 0755, true);
+
+        $handler = new CoroutineHandlerStub();
+        $handler->createSink($body = uniqid(), $sink = $dir . uniqid());
+        $this->assertSame($body, file_get_contents($sink));
+    }
+
     protected function getHandler($options = [])
     {
         return new CoroutineHandler($options);

src/guzzle/tests/Stub/CoroutineHandlerStub.php

@@ -30,6 +30,11 @@ class CoroutineHandlerStub extends CoroutineHandler
         return parent::checkStatusCode($client, $request);
     }
 
+    public function createSink(string $body, string $sink)
+    {
+        return parent::createSink($body, $sink);
+    }
+
     protected function execute(Client $client, $path)
     {
         $client->body = json_encode([