Skip to content

Commit 4d1423f

Browse files
roxblnfkroxblnfk
committed
Add test server; add feature tests
1 parent 912fa99 commit 4d1423f

File tree

8 files changed

+484
-2
lines changed

8 files changed

+484
-2
lines changed

‎composer.json‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -16,9 +16,10 @@
1616
"require": {
1717
"php": ">=8.1",
1818
"ext-json": "*",
19-
"spiral/roadrunner-worker": "^2.2.0",
2019
"psr/http-factory": "^1.0.1",
21-
"psr/http-message": "^1.0.1"
20+
"psr/http-message": "^1.0.1",
21+
"spiral/roadrunner-worker": "^2.2.0",
22+
"symfony/process": "^6.2"
2223
},
2324
"require-dev": {
2425
"nyholm/psr7": "^1.3",

‎tests/Feature/StreamResponseTest.php‎

Lines changed: 117 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,117 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace Spiral\RoadRunner\Tests\Http\Feature;
6+
7+
use PHPUnit\Framework\TestCase;
8+
use Spiral\Goridge\Frame;
9+
use Spiral\Goridge\SocketRelay;
10+
use Spiral\RoadRunner\Http\HttpWorker;
11+
use Spiral\RoadRunner\Payload;
12+
use Spiral\RoadRunner\Tests\Http\Server\Command\BaseCommand;
13+
use Spiral\RoadRunner\Tests\Http\Server\Command\StreamStop;
14+
use Spiral\RoadRunner\Tests\Http\Server\ServerRunner;
15+
use Spiral\RoadRunner\Worker;
16+
17+
class StreamResponseTest extends TestCase
18+
{
19+
private SocketRelay $relay;
20+
private Worker $worker;
21+
private $serverAddress = 'tcp://127.0.0.1:6002';
22+
23+
protected function setUp(): void
24+
{
25+
parent::setUp();
26+
ServerRunner::start();
27+
ServerRunner::getBuffer();
28+
}
29+
30+
protected function tearDown(): void
31+
{
32+
unset($this->relay, $this->worker);
33+
ServerRunner::stop();
34+
parent::tearDown();
35+
}
36+
37+
/**
38+
* Regular case
39+
*/
40+
public function testRegularCase(): void
41+
{
42+
$worker = $this->getWorker();
43+
$worker->respond(new Payload('Hello, World!'));
44+
45+
\usleep(100_000);
46+
self::assertSame('Hello, World!', \trim(ServerRunner::getBuffer()));
47+
}
48+
49+
/**
50+
* Test stream response with multiple frames
51+
*/
52+
public function testStreamResponseWithMultipleFrames(): void
53+
{
54+
$httpWorker = $this->makeHttpWorker();
55+
56+
$chunks = ['Hel', 'lo,', ' Wo', 'rld', '!'];
57+
ServerRunner::getBuffer();
58+
$httpWorker->respondStream(
59+
200,
60+
(function () use ($chunks) {
61+
yield from $chunks;
62+
})(),
63+
);
64+
65+
\usleep(100_000);
66+
self::assertSame(\implode("\n", $chunks), \trim(ServerRunner::getBuffer()));
67+
}
68+
69+
public function testStopStreamResponse(): void
70+
{
71+
$httpWorker = $this->makeHttpWorker();
72+
73+
// Flush buffer
74+
ServerRunner::getBuffer();
75+
76+
$httpWorker->respondStream(
77+
200,
78+
(function () {
79+
yield 'Hel';
80+
yield 'lo,';
81+
$this->sendCommand(new StreamStop());
82+
try {
83+
yield ' Wo';
84+
} catch (\Throwable $e) {
85+
return;
86+
}
87+
yield 'rld';
88+
yield '!';
89+
})(),
90+
);
91+
92+
93+
\usleep(100_000);
94+
self::assertSame(\implode("\n", ['Hel', 'lo,']), \trim(ServerRunner::getBuffer()));
95+
}
96+
97+
private function getRelay(): SocketRelay
98+
{
99+
return $this->relay ??= SocketRelay::create($this->serverAddress);
100+
}
101+
102+
private function getWorker(): Worker
103+
{
104+
return $this->worker ??= new Worker($this->getRelay(), false);
105+
}
106+
107+
private function makeHttpWorker(): HttpWorker
108+
{
109+
return new HttpWorker($this->getWorker());
110+
}
111+
112+
private function sendCommand(BaseCommand $command)
113+
{
114+
$this->getRelay()->send($command->getRequestFrame());
115+
\usleep(500_000);
116+
}
117+
}

‎tests/Server/Client.php‎

Lines changed: 173 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,173 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace Spiral\RoadRunner\Tests\Http\Server;
6+
7+
use Fiber;
8+
use Spiral\Goridge\Frame;
9+
use Spiral\RoadRunner\Tests\Http\Server\Command\BaseCommand;
10+
use Spiral\RoadRunner\Tests\Http\Server\Command\StreamStop;
11+
12+
/**
13+
* Client state on the server side.
14+
*/
15+
class Client
16+
{
17+
private \Socket $socket;
18+
19+
/** @var string[] */
20+
private array $writeQueue = [];
21+
22+
/** @var string */
23+
private string $readBuffer = '';
24+
25+
public function __construct(
26+
\Socket $socket,
27+
) {
28+
$this->socket = $socket;
29+
\socket_set_nonblock($this->socket);
30+
}
31+
32+
public function __destruct()
33+
{
34+
\socket_close($this->socket);
35+
}
36+
37+
public static function init(\Socket $socket): self
38+
{
39+
return new self($socket);
40+
}
41+
42+
public function process(): void
43+
{
44+
$this->onInit();
45+
46+
do {
47+
$read = [$this->socket];
48+
$write = [$this->socket];
49+
$except = [$this->socket];
50+
if (\socket_select($read, $write, $except, 0, 0) === false) {
51+
throw new \RuntimeException('Socket select failed.');
52+
}
53+
54+
if ($read !== []) {
55+
$this->readMessage();
56+
}
57+
58+
if ($write !== [] && $this->writeQueue !== []) {
59+
$this->writeQueue();
60+
}
61+
62+
Fiber::suspend();
63+
} while (true);
64+
}
65+
66+
private function onInit()
67+
{
68+
$this->writeQueue[] = Frame::packFrame(new Frame('{"pid":true}', [], Frame::CONTROL));
69+
}
70+
71+
private function onFrame(Frame $frame): void
72+
{
73+
$command = $this->getCommand($frame);
74+
75+
if ($command === null) {
76+
echo \substr($frame->payload, $frame->options[0]) . "\n";
77+
return;
78+
}
79+
80+
$this->onCommand($command);
81+
}
82+
83+
private function writeQueue(): void
84+
{
85+
foreach ($this->writeQueue as $data) {
86+
\socket_write($this->socket, $data);
87+
}
88+
socket_set_nonblock($this->socket);
89+
90+
$this->writeQueue = [];
91+
}
92+
93+
/**
94+
* @see \Spiral\Goridge\SocketRelay::waitFrame()
95+
*/
96+
private function readMessage(): void
97+
{
98+
$header = $this->readNBytes(12);
99+
100+
$parts = Frame::readHeader($header);
101+
// total payload length
102+
$length = $parts[1] * 4 + $parts[2];
103+
104+
if ($length >= 8 * 1024 * 1024) {
105+
throw new \RuntimeException('Frame payload is too large.');
106+
}
107+
$payload = $this->readNBytes($length);
108+
109+
$frame = Frame::initFrame($parts, $payload);
110+
111+
$this->onFrame($frame);
112+
}
113+
114+
/**
115+
* @param positive-int $bytes
116+
*
117+
* @return non-empty-string
118+
*/
119+
private function readNBytes(int $bytes, bool $canBeLess = false): string
120+
{
121+
while (($left = $bytes - \strlen($this->readBuffer)) > 0) {
122+
$data = @\socket_read($this->socket, $left, \PHP_BINARY_READ);
123+
if ($data === false) {
124+
$errNo = \socket_last_error($this->socket);
125+
throw new \RuntimeException('Socket read failed [' . $errNo . ']: ' . \socket_strerror($errNo));
126+
}
127+
128+
if ($canBeLess) {
129+
return $data;
130+
}
131+
132+
if ($data === '') {
133+
Fiber::suspend();
134+
continue;
135+
}
136+
137+
$this->readBuffer .= $data;
138+
}
139+
140+
$result = \substr($this->readBuffer, 0, $bytes);
141+
$this->readBuffer = \substr($this->readBuffer, $bytes);
142+
143+
return $result;
144+
}
145+
146+
private function getCommand(Frame $frame): ?BaseCommand
147+
{
148+
$payload = $frame->payload;
149+
try {
150+
$data = \json_decode($payload, true, 3, \JSON_THROW_ON_ERROR);
151+
} catch (\JsonException) {
152+
return null;
153+
}
154+
155+
return match (false) {
156+
\is_array($data),
157+
\array_key_exists(BaseCommand::COMMAND_KEY, $data),
158+
\is_string($data[BaseCommand::COMMAND_KEY]),
159+
\class_exists($data[BaseCommand::COMMAND_KEY]),
160+
\is_a($data[BaseCommand::COMMAND_KEY], BaseCommand::class, true) => null,
161+
default => new ($data[BaseCommand::COMMAND_KEY])(),
162+
};
163+
}
164+
165+
private function onCommand(BaseCommand $command): void
166+
{
167+
switch ($command::class) {
168+
case StreamStop::class:
169+
$this->writeQueue[] = $command->getResponse();
170+
break;
171+
}
172+
}
173+
}

‎tests/Server/Command/BaseCommand.php‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,29 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace Spiral\RoadRunner\Tests\Http\Server\Command;
6+
7+
use Spiral\Goridge\Frame;
8+
9+
abstract class BaseCommand
10+
{
11+
public const COMMAND_KEY = 'test-command';
12+
protected Frame $frame;
13+
14+
public function __construct() {
15+
$this->frame = new Frame(\json_encode([self::COMMAND_KEY => static::class]));
16+
}
17+
18+
public function getRequestFrame(): Frame
19+
{
20+
return $this->frame;
21+
}
22+
23+
public function getResponse(): string
24+
{
25+
return Frame::packFrame($this->getResponseFrame());
26+
}
27+
28+
public abstract function getResponseFrame(): Frame;
29+
}

‎tests/Server/Command/StreamStop.php‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace Spiral\RoadRunner\Tests\Http\Server\Command;
6+
7+
use Spiral\Goridge\Frame;
8+
9+
final class StreamStop extends BaseCommand
10+
{
11+
public function getResponseFrame(): Frame
12+
{
13+
$frame = new Frame('', [0]);
14+
$frame->byte10 |= Frame::BYTE10_STOP;
15+
16+
return $frame;
17+
}
18+
}

0 commit comments

Comments
 (0)