Skip to content

Commit c6aef2a

Browse files
roxblnfkroxblnfk
authored
Merge pull request #18: better streaming response supporting
Add ability to send separated stream frames
2 parents 30d89df + f785dec commit c6aef2a

File tree

4 files changed

+85
-15
lines changed

4 files changed

+85
-15
lines changed

‎composer.json‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -41,14 +41,16 @@
4141
"ext-json": "*",
4242
"psr/http-factory": "^1.0.1",
4343
"psr/http-message": "^1.0.1 || ^2.0",
44-
"spiral/roadrunner": "^2023.1",
45-
"spiral/roadrunner-worker": "^3.0"
44+
"spiral/roadrunner": "^2023.3",
45+
"spiral/roadrunner-worker": "^3.1.0"
4646
},
4747
"require-dev": {
48+
"buggregator/trap": "^1.0",
4849
"jetbrains/phpstorm-attributes": "^1.0",
4950
"nyholm/psr7": "^1.3",
5051
"phpunit/phpunit": "^10.0",
5152
"symfony/process": "^6.2",
53+
"symfony/var-dumper": "^6.3",
5254
"vimeo/psalm": "^5.9"
5355
},
5456
"autoload": {

‎src/HttpWorker.php‎

Lines changed: 36 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88
use Spiral\RoadRunner\Http\Exception\StreamStoppedException;
99
use Spiral\RoadRunner\Message\Command\StreamStop;
1010
use Spiral\RoadRunner\Payload;
11+
use Spiral\RoadRunner\StreamWorkerInterface;
1112
use Spiral\RoadRunner\WorkerInterface;
1213

1314
/**
@@ -64,10 +65,14 @@ public function waitRequest(): ?Request
6465
/**
6566
* @throws \JsonException
6667
*/
67-
public function respond(int $status, string|Generator $body, array $headers = []): void
68+
public function respond(int $status, string|Generator $body = '', array $headers = [], bool $endOfStream = true): void
6869
{
70+
if ($status < 200 && $status >= 100 && $body !== '') {
71+
throw new \InvalidArgumentException('Unable to send a body with informational status code.');
72+
}
73+
6974
if ($body instanceof Generator) {
70-
$this->respondStream($status, $body, $headers);
75+
$this->respondStream($status, $body, $headers, $endOfStream);
7176
return;
7277
}
7378

@@ -76,30 +81,53 @@ public function respond(int $status, string|Generator $body, array $headers = []
7681
'headers' => $headers ?: (object)[],
7782
], \JSON_THROW_ON_ERROR);
7883

79-
$this->worker->respond(new Payload($body, $head));
84+
$this->worker->respond(new Payload($body, $head, $endOfStream));
8085
}
8186

82-
private function respondStream(int $status, Generator $body, array $headers = []): void
87+
private function respondStream(int $status, Generator $body, array $headers = [], bool $endOfStream = true): void
8388
{
8489
$head = \json_encode([
8590
'status' => $status,
8691
'headers' => $headers ?: (object)[],
8792
], \JSON_THROW_ON_ERROR);
8893

94+
$worker = $this->worker instanceof StreamWorkerInterface
95+
? $this->worker->withStreamMode()
96+
: $this->worker;
97+
8998
do {
9099
if (!$body->valid()) {
100+
// End of generator
91101
$content = (string)$body->getReturn();
92-
$this->worker->respond(new Payload($content, $head, true));
102+
if ($endOfStream === false && $content === '') {
103+
// We don't need to send an empty frame if the stream is not ended
104+
return;
105+
}
106+
$worker->respond(new Payload($content, $head, $endOfStream));
93107
break;
94108
}
109+
95110
$content = (string)$body->current();
96-
if ($this->worker->getPayload(StreamStop::class) !== null) {
111+
if ($worker->getPayload(StreamStop::class) !== null) {
97112
$body->throw(new StreamStoppedException());
113+
114+
// RoadRunner is waiting for a Stream Stop Frame to confirm that the stream is closed
115+
// and the worker doesn't hang
116+
$worker->respond(new Payload(''));
98117
return;
99118
}
100-
$this->worker->respond(new Payload($content, $head, false));
101-
$body->next();
119+
120+
// Send a chunk of data
121+
$worker->respond(new Payload($content, $head, false));
102122
$head = null;
123+
124+
try {
125+
$body->next();
126+
} catch (\Throwable) {
127+
// Stop the stream if an exception is thrown from the generator
128+
$worker->respond(new Payload(''));
129+
return;
130+
}
103131
} while (true);
104132
}
105133

‎src/HttpWorkerInterface.php‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -25,8 +25,7 @@ public function waitRequest(): ?Request;
2525
* @param Generator<mixed, scalar|Stringable, mixed, Stringable|scalar|null>|string $body Body of response.
2626
* If the body is a generator, then each yielded value will be sent as a separated stream chunk.
2727
* Returned value will be sent as a last stream package.
28-
* Note: Stream response is experimental feature and isn't supported by RoadRunner yet.
29-
* But you can try to use RoadRunner 2.9-alpha to test it.
28+
* Note: Stream response is supported by RoadRunner since version 2023.3
3029
* @param HeadersList|array $headers An associative array of the message's headers. Each key MUST be a header name,
3130
* and each value MUST be an array of strings for that header.
3231
*/

‎tests/Feature/StreamResponseTest.php‎

Lines changed: 44 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -4,10 +4,12 @@
44

55
namespace Spiral\RoadRunner\Tests\Http\Feature;
66

7+
use Exception;
78
use PHPUnit\Framework\TestCase;
89
use Spiral\Goridge\SocketRelay;
910
use Spiral\RoadRunner\Http\Exception\StreamStoppedException;
1011
use Spiral\RoadRunner\Http\HttpWorker;
12+
use Spiral\RoadRunner\Message\Command\GetProcessId;
1113
use Spiral\RoadRunner\Payload;
1214
use Spiral\RoadRunner\Tests\Http\Server\Command\BaseCommand;
1315
use Spiral\RoadRunner\Tests\Http\Server\Command\StreamStop;
@@ -94,10 +96,46 @@ public function testStopStreamResponse(): void
9496
self::assertSame(\implode("\n", ['Hel', 'lo,']), \trim(ServerRunner::getBuffer()));
9597
}
9698

99+
public function testSend1xxWithBody(): void
100+
{
101+
$httpWorker = $this->makeHttpWorker();
102+
103+
$this->expectExceptionMessage('Unable to send a body with informational status code');
104+
105+
$httpWorker->respond(
106+
103,
107+
(function () {
108+
yield 'Hel';
109+
yield 'lo,';
110+
})(),
111+
);
112+
}
113+
114+
public function testExceptionInGenerator(): void
115+
{
116+
$httpWorker = $this->makeHttpWorker();
117+
118+
// Flush buffer
119+
ServerRunner::getBuffer();
120+
121+
$httpWorker->respond(
122+
200,
123+
(function () {
124+
yield 'Hel';
125+
yield 'lo,';
126+
throw new Exception('test');
127+
})(),
128+
);
129+
130+
131+
\usleep(100_000);
132+
self::assertSame(\implode("\n", ['Hel', 'lo,']), \trim(ServerRunner::getBuffer()));
133+
}
134+
97135
/**
98136
* StopStream should be ignored if stream is already ended.
99137
* Commented because doesn't pass in CI
100-
* todo: check after RoadRunner Stream Response release
138+
*/
101139
public function testStopStreamAfterStreamEnd(): void
102140
{
103141
$httpWorker = $this->makeHttpWorker();
@@ -116,11 +154,14 @@ public function testStopStreamAfterStreamEnd(): void
116154
$this->assertFalse($this->getWorker()->hasPayload(\Spiral\RoadRunner\Message\Command\StreamStop::class));
117155
$this->sendCommand(new StreamStop());
118156
\usleep(200_000);
119-
self::assertSame(\implode("\n", ['Hello', 'World!']), \trim(ServerRunner::getBuffer()));
157+
$this->assertSame(\implode("\n", ['Hello', 'World!']), \trim(ServerRunner::getBuffer()));
120158
$this->assertTrue($this->getWorker()->hasPayload(\Spiral\RoadRunner\Message\Command\StreamStop::class));
159+
160+
$this->getWorker()->getPayload(\Spiral\RoadRunner\Message\Command\StreamStop::class);
161+
$this->getWorker()->getPayload(GetProcessId::class);
162+
121163
$this->assertFalse($this->getWorker()->hasPayload());
122164
}
123-
*/
124165

125166
private function getRelay(): SocketRelay
126167
{

0 commit comments

Comments
 (0)