Make HTTP stream stoppable; add phpunit; up PHP version to 8.1

Author: roxblnfk

.gitignore

@@ -2,4 +2,5 @@
 # vendor/
 .idea
 vendor
+.phpunit.result.cache
 composer.lock

composer.json

@@ -14,33 +14,32 @@
         }
     ],
     "require": {
-        "php": ">=7.4",
+        "php": ">=8.1",
         "ext-json": "*",
         "spiral/roadrunner-worker": "^2.2.0",
         "psr/http-factory": "^1.0.1",
         "psr/http-message": "^1.0.1"
     },
-    "autoload": {
-        "psr-4": {
-            "Spiral\\RoadRunner\\Http\\": "src"
-        }
-    },
     "require-dev": {
         "nyholm/psr7": "^1.3",
-        "phpstan/phpstan": "~0.12",
-        "phpunit/phpunit": "~8.0",
+        "phpunit/phpunit": "^9.5",
         "jetbrains/phpstorm-attributes": "^1.0",
         "vimeo/psalm": "^4.22",
         "symfony/var-dumper": "^5.1"
     },
-    "scripts": {
-        "analyze": "psalm"
+    "autoload": {
+        "psr-4": {
+            "Spiral\\RoadRunner\\Http\\": "src"
+        }
     },
-    "extra": {
-        "branch-alias": {
-            "dev-master": "2.2.x-dev"
+    "autoload-dev": {
+        "psr-4": {
+            "Spiral\\RoadRunner\\Tests\\Http\\": "tests"
         }
     },
+    "scripts": {
+        "analyze": "psalm"
+    },
     "suggest": {
         "spiral/roadrunner-cli": "Provides RoadRunner installation and management CLI tools"
     },

phpunit.xml

@@ -0,0 +1,28 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<phpunit xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:noNamespaceSchemaLocation="./vendor/phpunit/phpunit/phpunit.xsd"
+         backupGlobals="false"
+         backupStaticAttributes="false"
+         colors="true"
+         verbose="true"
+         convertErrorsToExceptions="true"
+         convertNoticesToExceptions="true"
+         convertWarningsToExceptions="true"
+         processIsolation="false"
+         executionOrder="random"
+         stopOnFailure="false"
+         stopOnError="false"
+         stderr="true"
+>
+    <testsuites>
+        <testsuite name="RR Worker Tests">
+            <directory>tests</directory>
+        </testsuite>
+    </testsuites>
+
+    <coverage>
+        <include>
+            <directory>src</directory>
+        </include>
+    </coverage>
+</phpunit>

src/HttpWorker.php

@@ -12,6 +12,7 @@
 namespace Spiral\RoadRunner\Http;
 
 use Generator;
+use Spiral\RoadRunner\Message\Command\StreamStop;
 use Spiral\RoadRunner\Payload;
 use Spiral\RoadRunner\WorkerInterface;
 use Stringable;
@@ -114,6 +115,10 @@ public function respondStream(int $status, Generator $body, array $headers = [])
                 break;
             }
             $content = (string)$body->current();
+            if ($this->worker->getPayload(StreamStop::class) !== null) {
+                $body->throw(new \RuntimeException('Stream has been stopped by the client.'));
+                return;
+            }
             $this->worker->respond(new Payload($content, $head, false));
             $body->next();
             $head = null;

tests/Unit/StreamResponseTest.php

@@ -0,0 +1,91 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Spiral\RoadRunner\Tests\Http\Unit;
+
+use PHPUnit\Framework\TestCase;
+use Spiral\RoadRunner\Http\HttpWorker;
+use Spiral\RoadRunner\Payload;
+use Spiral\RoadRunner\Tests\Http\Unit\Stub\TestRelay;
+use Spiral\RoadRunner\Worker;
+
+class StreamResponseTest extends TestCase
+{
+    private TestRelay $relay;
+    private Worker $worker;
+
+    protected function tearDown(): void
+    {
+        unset($this->relay, $this->worker);
+        parent::tearDown();
+    }
+
+    /**
+     * Regular case
+     */
+    public function testRegularCase(): void
+    {
+        $worker = $this->getWorker();
+        $this->getRelay()
+            ->addFrame(status: 200, body: 'Hello, World!', headers: ['Content-Type' => 'text/plain'], stream: true);
+
+        self::assertTrue($worker->hasPayload());
+        self::assertInstanceOf(Payload::class, $payload = $worker->waitPayload());
+        self::assertSame('Hello, World!', $payload->body);
+    }
+
+    /**
+     * Test stream response with multiple frames
+     */
+    public function testStreamResponseWithMultipleFrames(): void
+    {
+        $httpWorker = $this->makeHttpWorker();
+
+        $httpWorker->respondStream(200, (function () {
+            yield 'Hel';
+            yield 'lo,';
+            yield ' Wo';
+            yield 'rld';
+            yield '!';
+        })());
+
+        self::assertFalse($this->worker->hasPayload());
+        self::assertSame('Hello, World!', $this->getRelay()->getReceivedBody());
+    }
+
+    public function testStopStreamResponse(): void
+    {
+        $httpWorker = $this->makeHttpWorker();
+
+        $httpWorker->respondStream(200, (function () {
+            yield 'Hel';
+            yield 'lo,';
+            $this->getRelay()->addStopStreamFrame();
+            try {
+                yield ' Wo';
+            } catch (\Throwable $e) {
+                return;
+            }
+            yield 'rld';
+            yield '!';
+        })());
+
+        self::assertSame('Hello,', $this->getRelay()->getReceivedBody());
+    }
+
+    private function getRelay(): TestRelay
+    {
+        return $this->relay ??= new TestRelay();
+    }
+
+    private function getWorker(): Worker
+    {
+        return $this->worker ??= new Worker($this->getRelay(), false);
+    }
+
+    private function makeHttpWorker(): HttpWorker
+    {
+        return new HttpWorker($this->getWorker());
+    }
+}

tests/Unit/Stub/TestRelay.php

@@ -0,0 +1,75 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Spiral\RoadRunner\Tests\Http\Unit\Stub;
+
+use Spiral\Goridge\Frame;
+use Spiral\Goridge\Relay;
+
+final class TestRelay extends Relay
+{
+    /** @var Frame[] */
+    private array $frames = [];
+
+    /** @var Frame[] */
+    private array $received = [];
+
+    public function addFrames(Frame ...$frames): self
+    {
+        $this->frames = [...$this->frames, ...\array_values($frames)];
+        return $this;
+    }
+
+    public function addFrame(
+        int $status = 200,
+        string $body = '',
+        array $headers = [],
+        bool $stream = false,
+        bool $stopStream = false,
+    ): self {
+        $head = (string)\json_encode([
+            'status'  => $status,
+            'headers' => $headers,
+        ], \JSON_THROW_ON_ERROR);
+        $frame = new Frame($head .$body, [\strlen($head)]);
+        $frame->byte10 |= $stream ? Frame::BYTE10_STREAM : 0;
+        $frame->byte10 |= $stopStream ? Frame::BYTE10_STOP : 0;
+        return $this->addFrames($frame);
+    }
+
+    public function addStopStreamFrame(): self
+    {
+        return $this->addFrame(stopStream: true);
+    }
+
+    public function getReceived(): array
+    {
+        return $this->received;
+    }
+
+    public function getReceivedBody(): string
+    {
+        return \implode('', \array_map(static fn (Frame $frame)
+            => \substr($frame->payload, $frame->options[0]), $this->received));
+    }
+
+    public function waitFrame(): Frame
+    {
+        if ($this->frames === []) {
+            throw new \RuntimeException('There are no frames to return.');
+        }
+
+        return \array_shift($this->frames);
+    }
+
+    public function send(Frame $frame): void
+    {
+        $this->received[] = $frame;
+    }
+
+    public function hasFrame(): bool
+    {
+        return $this->frames !== [];
+    }
+}