[HttpClient] add ResponseMonitor to trigger callbacks on response completion

nicolas-grekas · nicolas-grekas · commit 3419652ed25e · 2019-06-25T18:26:38.000+02:00
diff --git a/src/Symfony/Component/HttpClient/ResponseSetMonitor.php b/src/Symfony/Component/HttpClient/ResponseSetMonitor.php
@@ -0,0 +1,84 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\HttpClient;
+
+use Symfony\Contracts\HttpClient\Exception\ExceptionInterface;
+use Symfony\Contracts\HttpClient\HttpClientInterface;
+use Symfony\Contracts\HttpClient\ResponseInterface;
+
+/**
+ * Monitors a set of responses and triggers callbacks as they complete.
+ *
+ * @author Nicolas Grekas <p@tchwork.com>
+ */
+final class ResponseSetMonitor
+{
+    private $client;
+    private $responses;
+
+    public function __construct(HttpClientInterface $client)
+    {
+        $this->client = $client;
+        $this->responses = new \SplObjectStorage();
+    }
+
+    public function add(ResponseInterface $response, callable $onHeaders = null, callable $onBody = null, callable $onError = null): void
+    {
+        $this->responses[$response] = [$onHeaders, $onBody, $onError];
+        $this->tick();
+    }
+
+    public function tick(?float $timeout = 0.0, bool $errorOnTimeout = false): int
+    {
+        foreach ($this->client->stream($this->responses, $timeout) as $response => $chunk) {
+            try {
+                if ($chunk->isTimeout() && !$errorOnTimeout) {
+                    continue;
+                }
+
+                if (!$chunk->isFirst() && !$chunk->isLast()) {
+                    continue;
+                }
+
+                [$onHeaders, $onBody] = $this->responses[$response];
+
+                if (null !== $onHeaders && $chunk->isFirst()) {
+                    $onHeaders($response);
+                }
+
+                if (null !== $onBody && $chunk->isLast()) {
+                    $onBody($response);
+                }
+
+                if (null === $onBody || $chunk->isLast()) {
+                    unset($this->responses[$response]);
+                }
+            } catch (ExceptionInterface $e) {
+                [, , $onError] = $this->responses[$response];
+                unset($this->responses[$response]);
+
+                if (null === $onError) {
+                    throw $e;
+                }
+
+                $onError($e);
+            }
+        }
+
+        return \count($this->responses);
+    }
+
+    public function __destruct()
+    {
+        $this->tick(null, true);
+    }
+}
