Prevent blocking operation when loading sitemap files by using promises

composer.json

@@ -12,6 +12,7 @@
     "ext-simplexml": "*",
     "clue/reactphp-flux": "^1.3",
     "psr/log": "^1.1",
+    "react/filesystem": "^0.1",
     "react/http": "^1.2",
     "symfony/console": "^5.2",
     "teapot/status-code": "^1.1"

src/Config.php

@@ -89,6 +89,8 @@ class Config
 
     /**
      * An array of some additional response headers to count.
+     *
+     * @var array<int, string>
      */
     public array $additionalResponseHeadersToCount = [];
 

src/Processor.php

@@ -6,19 +6,18 @@
 
 use Clue\React\Flux\Transformer;
 use Exception;
-use Octopus\TargetManager\TargetManagerFactory;
+use Octopus\TargetManager\StreamTargetManager;
 use Psr\Http\Message\ResponseInterface;
 use Psr\Log\LoggerInterface;
 use Psr\Log\NullLogger;
 use React\EventLoop\Factory as EventLoopFactory;
 use React\EventLoop\LoopInterface;
+use React\Filesystem\Filesystem;
 use React\Http\Browser;
 use React\Http\Message\ResponseException;
 use React\Promise\ExtendedPromiseInterface;
 use React\Promise\PromiseInterface;
 use React\Promise\Timer\TimeoutException;
-use React\Stream\ReadableResourceStream;
-use React\Stream\ReadableStreamInterface;
 use RuntimeException;
 use Teapot\StatusCode\Http;
 
@@ -53,7 +52,7 @@ class Processor
     private Browser $browser;
     private LoggerInterface $logger;
     private LoopInterface $loop;
-    private TargetManager$targetManager;
+    private StreamTargetManager $targetManager;
     private Transformer $transformer;
     private bool $followRedirects = false;
     private Presenter $presenter;
@@ -122,21 +121,34 @@ private function getLoop(): LoopInterface
         return $this->loop ??= EventLoopFactory::create();
     }
 
-    private function getTargetManager(): TargetManager
+    private function getTargetManager(): StreamTargetManager
     {
-        return $this->targetManager ??= TargetManagerFactory::getInstance($this->getStream(), $this->logger);
+        return $this->targetManager ??= new StreamTargetManager($this->getStream(), $this->logger);
     }
 
-    private function getStream(): ?ReadableStreamInterface
+    private function getStream(): PromiseInterface
     {
-        $handle = @\fopen($this->config->targetFile, 'r');
-        if (!$handle) {
-            $this->logger->error(\sprintf('Could not open target file "%s"', $this->config->targetFile));
+        return \filter_var($this->config->targetFile, \FILTER_VALIDATE_URL)
+            ? $this->getStreamForUrl($this->config->targetFile)
+            : $this->getStreamForLocalFile($this->config->targetFile);
+    }
 
-            return null;
-        }
+    private function getStreamForUrl(string $url): PromiseInterface
+    {
+        $this->logger->info('acquire stream for URL '.$url);
+        $browser = new Browser($this->getLoop());
+
+        return $browser->requestStreaming('GET', $url);
+    }
+
+    private function getStreamForLocalFile(string $filename): PromiseInterface
+    {
+        $this->logger->info('acquire stream for local file '.$filename);
 
-        return new ReadableResourceStream($handle, $this->getLoop());
+        $filesystem = Filesystem::create($this->getLoop());
+        $file = $filesystem->file($filename);
+
+        return $file->getContents();
     }
 
     private function isCompleted(): bool
@@ -230,10 +242,10 @@ private function getOnFulfilledCallback(string $url): callable
 
             /*
             if ($this->saveEnabled) {
-                $path = $this->savePath.$this->makeFilename($url);
-                if (\file_put_contents($path, $response->getBody(), FILE_APPEND) === false) {
-                    throw new Exception("Cannot write file: $path");
-                }
+            $path = $this->savePath.$this->makeFilename($url);
+            if (\file_put_contents($path, $response->getBody(), FILE_APPEND) === false) {
+            throw new Exception("Cannot write file: $path");
+            }
             }
              */
 

src/Result.php

@@ -6,60 +6,53 @@
 
 class Result
 {
-    /**
-     * @var Config
-     */
-    public $config;
+    public Config $config;
 
     /**
-     * @var array
+     * @var array<int, string>
      */
-    private $additionalResponseHeadersToCount = [];
+    private array $additionalResponseHeadersToCount = [];
 
     /**
      * URLs that could not be loaded.
      *
-     * @var array
+     * @var array<string, int|string>
      */
-    private $brokenUrls = [];
+    private array $brokenUrls = [];
 
     /**
-     * @var array
+     * @var array<int, string>
      */
-    private $finishedUrls = [];
+    private array $finishedUrls = [];
 
     /**
      * URLs that were redirected to another location.
      *
-     * @var array
+     * @var array<string, string>
      */
-    private $redirectedUrls = [];
+    private array $redirectedUrls = [];
 
     /**
      * Timestamp to track execution time.
-     *
-     * @var float
      */
-    private $started;
+    private float $started;
 
-    /**
-     * @var array
-     */
-    private $statusCodes = [];
+    private array $statusCodes = [];
 
     /**
      * Total amount of processed data.
-     *
-     * @var int
      */
-    private $totalData = 0;
+    private int $totalData = 0;
 
     public function __construct(Config $config)
     {
         $this->config = $config;
         $this->started = \microtime(true);
     }
 
+    /**
+     * @param array<int, string> $additionalResponseHeadersToCount
+     */
     public function setAdditionalResponseHeadersToCount(array $additionalResponseHeadersToCount): void
     {
         $this->additionalResponseHeadersToCount = $additionalResponseHeadersToCount;