rcsofttech85 · rcsofttech85 · Sep 10, 2023 · Sep 10, 2023
diff --git a/src/Stream.php b/src/Stream.php
diff --git a/src/StreamHandler.php b/src/StreamHandler.php
@@ -0,0 +1,104 @@
+<?php
+
+namespace rcsofttech85\FileHandler;
+
+use Fiber;
+use rcsofttech85\FileHandler\Exception\StreamException;
+use Throwable;
+
+class StreamHandler
+{
+    private array $fibers = [];
+
+
+    /**
+     * @throws StreamException
+     */
+    public function __construct(public readonly array $streamUrls, public readonly int $chunk = 100)
+    {
+        if (!$this->streamUrls) {
+            throw new StreamException('No stream URLs provided.');
+        }
+    }
+
+    /**
+     */
+    private function stream(string $streamUrl, string $outputFilename): Fiber
+    {
+        return new Fiber(function () use ($streamUrl, $outputFilename) {
+            $stream = fopen($streamUrl, 'r');
+            if (!$stream) {
+                throw new StreamException("Failed to open stream: $streamUrl");
+            }
+            stream_set_blocking($stream, false);
+
+            $outputFile = fopen($outputFilename, 'w');
+
+            try {
+                while (!feof($stream)) {
+                    $contents = fread($stream, $this->chunk);
+                    fwrite($outputFile, $contents);
+                    Fiber::suspend();
+                }
+            } catch (Throwable $e) {
+                throw new StreamException();
+            } finally {
+                fclose($stream);
+                fclose($outputFile);
+            }
+        });
+    }
+
+    /**
+     */
+    public function initiateConcurrentStreams(): self
+    {
+        foreach ($this->streamUrls as $outputFile => $streamUrl) {
+            $fiber = $this->stream($streamUrl, $outputFile);
+
+            $this->fibers[] = $fiber;
+        }
+
+        return $this;
+    }
+
+    /**
+     * @throws StreamException
+     * @throws Throwable
+     */
+    public function start(): self
+    {
+        if (!$this->fibers) {
+            throw new StreamException("No fibers available to start");
+        }
+
+        /** @var Fiber $fiber */
+        foreach ($this->fibers as $fiber) {
+            $fiber->start();
+        }
+
+        return $this;
+    }
+
+    /**
+     * @throws Throwable
+     */
+    public function resume(bool $resumeOnce = false): void
+    {
+        if (!$this->fibers) {
+            throw new StreamException("No fibers are currently running");
+        }
+
+        /** @var Fiber $fiber */
+        foreach ($this->fibers as $fiber) {
+            while (!$fiber->isTerminated()) {
+                $fiber->resume();
+                if ($resumeOnce) {
+                    break;
+                }
+            }
+        }
+
+        $this->fibers = [];
+    }
+}
diff --git a/tests/Integration/StreamHandlerTest.php b/tests/Integration/StreamHandlerTest.php
@@ -0,0 +1,81 @@
+<?php
+
+namespace Integration;
+
+use PHPUnit\Framework\Attributes\DataProvider;
+use PHPUnit\Framework\Attributes\Group;
+use PHPUnit\Framework\Attributes\Test;
+use PHPUnit\Framework\TestCase;
+use rcsofttech85\FileHandler\Exception\StreamException;
+use rcsofttech85\FileHandler\StreamHandler;
+
+#[Group("integration")]
+class StreamHandlerTest extends TestCase
+{
+    public static function tearDownAfterClass(): void
+    {
+        parent::tearDownAfterClass();
+
+        $files = ["output.html", "output1.html", "output2.html"];
+
+        foreach ($files as $file) {
+            if (file_exists($file)) {
+                unlink($file);
+            }
+        }
+    }
+
+    #[Test]
+    #[DataProvider('streamDataProvider')]
+    public function streamAndWriteToFile($urls)
+    {
+        $stream = new StreamHandler($urls);
+        $stream->initiateConcurrentStreams()->start()->resume();
+
+        foreach ($urls as $file => $url) {
+            $this->assertGreaterThan(0, filesize($file));
+            $this->assertStringContainsString('<!DOCTYPE html>', file_get_contents($file));
+            $this->assertStringContainsString('</html>', file_get_contents($file));
+        }
+    }
+
+    #[Test]
+    #[DataProvider('wrongStreamDataProvider')]
+    public function throwExceptionIfUrlIsInvalid($outputFile, $url)
+    {
+        $stream = new StreamHandler([$outputFile => $url]);
+
+
+        $this->expectException(StreamException::class);
+        $stream->initiateConcurrentStreams()->start()->resume();
+    }
+
+    #[Test]
+    public function throwExceptionIfEmptyDataProvided()
+    {
+        $this->expectException(StreamException::class);
+        $this->expectExceptionMessage('No stream URLs provided.');
+        new StreamHandler([]);
+    }
+
+    public static function wrongStreamDataProvider(): iterable
+    {
+        yield ["output.html", "https://gist.github"];
+    }
+
+
+    public static function streamDataProvider(): iterable
+    {
+        yield [
+            [
+                "output.html" =>
+                    "https://gist.github.com/rcsofttech85/629b37d483c4796db7bdcb3704067631#file-gistfile1-txt",
+
+                "output1.html" => "https://gist.github.com/rcsofttech85/f71f2454b1fc40a077cda14ef3097385#file-gistfile1-txt",
+
+
+                "output2.html" => "https://gist.github.com/rcsofttech85/79ab19f1502e72c95cfa97d5205fa47d#file-gistfile1-txt"
+            ]
+        ];
+    }
+}
diff --git a/tests/Integration/StreamTest.php b/tests/Integration/StreamTest.php