Dynamic Workers Scaling via RPC

src/WorkerPool.php

@@ -0,0 +1,39 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Spiral\RoadRunner;
+
+use Spiral\Goridge\RPC\Codec\JsonCodec;
+use Spiral\Goridge\RPC\RPCInterface;
+
+final class WorkerPool
+{
+    private readonly RPCInterface $rpc;
+
+    public function __construct(
+        RPCInterface $rpc,
+    ) {
+        $this->rpc = $rpc->withCodec(new JsonCodec());
+    }
+
+    /**
+     * Add worker to the pool.
+     *
+     * @param non-empty-string $plugin
+     */
+    public function addWorker(string $plugin): void
+    {
+        $this->rpc->call('informer.AddWorker', $plugin);
+    }
+
+    /**
+     * Remove worker from the pool.
+     *
+     * @param non-empty-string $plugin
+     */
+    public function removeWorker(string $plugin): void
+    {
+        $this->rpc->call('informer.RemoveWorker', $plugin);
+    }
+}

tests/Unit/EnvironmentTest.php

@@ -0,0 +1,59 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Spiral\RoadRunner\Tests\Worker\Unit;
+
+use PHPUnit\Framework\TestCase;
+use Spiral\RoadRunner\Environment;
+
+final class EnvironmentTest extends TestCase
+{
+    public function testGetModeWithDefault(): void
+    {
+        $env = new Environment();
+        $this->assertEquals('', $env->getMode());
+    }
+
+    public function testGetModeWithValue(): void
+    {
+        $env = new Environment(['RR_MODE' => 'mode_value']);
+        $this->assertEquals('mode_value', $env->getMode());
+    }
+
+    public function testGetRelayAddressWithDefault(): void
+    {
+        $env = new Environment();
+        $this->assertEquals('pipes', $env->getRelayAddress());
+    }
+
+    public function testGetRelayAddressWithValue(): void
+    {
+        $env = new Environment(['RR_RELAY' => 'relay_value']);
+        $this->assertEquals('relay_value', $env->getRelayAddress());
+    }
+
+    public function testGetRPCAddressWithDefault(): void
+    {
+        $env = new Environment();
+        $this->assertEquals('tcp://127.0.0.1:6001', $env->getRPCAddress());
+    }
+
+    public function testGetRPCAddressWithValue(): void
+    {
+        $env = new Environment(['RR_RPC' => 'rpc_value']);
+        $this->assertEquals('rpc_value', $env->getRPCAddress());
+    }
+
+    public function testFromGlobals(): void
+    {
+        $_ENV['RR_MODE'] = 'global_mode';
+        $_SERVER['RR_RELAY'] = 'global_relay';
+
+        $env = Environment::fromGlobals();
+
+        $this->assertEquals('global_mode', $env->getMode());
+        $this->assertEquals('global_relay', $env->getRelayAddress());
+        $this->assertEquals('tcp://127.0.0.1:6001', $env->getRPCAddress());
+    }
+}

tests/Unit/PayloadFactoryTest.php

@@ -0,0 +1,90 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Spiral\RoadRunner\Tests\Worker\Unit;
+
+use PHPUnit\Framework\TestCase;
+use Spiral\Goridge\Frame;
+use Spiral\RoadRunner\Exception\RoadRunnerException;
+use Spiral\RoadRunner\Message\Command\GetProcessId;
+use Spiral\RoadRunner\Message\Command\Pong;
+use Spiral\RoadRunner\Message\Command\StreamStop;
+use Spiral\RoadRunner\Message\Command\WorkerStop;
+use Spiral\RoadRunner\PayloadFactory;
+
+final class PayloadFactoryTest extends TestCase
+{
+    public function testFromFrameWithStopFlag(): void
+    {
+        $frame = new Frame("{}", []);
+        $frame->byte10 = Frame::BYTE10_STOP;
+        $payload = PayloadFactory::fromFrame($frame);
+
+        $this->assertInstanceOf(StreamStop::class, $payload);
+    }
+
+    public function testFromFrameWithPongFlag(): void
+    {
+        $frame = new Frame("{}", []);
+        $frame->byte10 = Frame::BYTE10_PONG;
+        $payload = PayloadFactory::fromFrame($frame);
+
+        $this->assertInstanceOf(Pong::class, $payload);
+    }
+
+    public function testFromFrameWithoutSpecificFlags(): void
+    {
+        $frame = new Frame("test", [0]);
+        $payload = PayloadFactory::fromFrame($frame);
+
+        $this->assertNotNull($payload);
+        $this->assertSame("test", $payload->body);
+        $this->assertSame("", $payload->header);
+    }
+
+    public function testMakeControlWithWorkerStop(): void
+    {
+        $json = \json_encode(['stop' => true]);
+        $frame = new Frame($json);
+        $frame->setFlag(Frame::CONTROL);
+
+        $payload = PayloadFactory::fromFrame($frame);
+        $this->assertInstanceOf(WorkerStop::class, $payload);
+    }
+
+    public function testMakeControlWithGetProcessId(): void
+    {
+        $json = \json_encode(['pid' => true]);
+        $frame = new Frame($json);
+        $frame->setFlag(Frame::CONTROL);
+
+        $payload = PayloadFactory::fromFrame($frame);
+        $this->assertInstanceOf(GetProcessId::class, $payload);
+    }
+
+    public function testFromFrameWithControlFlag(): void
+    {
+        $frame = new Frame(null, [], Frame::CONTROL);
+
+        $this->expectException(RoadRunnerException::class);
+        $this->expectExceptionMessage('Invalid task header, JSON payload is expected: Syntax error');
+        PayloadFactory::fromFrame($frame);
+    }
+
+    public function testMakeControlWithException(): void
+    {
+        $this->expectException(RoadRunnerException::class);
+        $this->expectExceptionMessage('Invalid task header, undefined control package');
+        $json = json_encode([]);
+        $frame = new Frame($json);
+        $frame->setFlag(Frame::CONTROL);
+
+        PayloadFactory::fromFrame($frame);
+    }
+
+    public function testMakePayload(): void
+    {
+        $this->markTestIncomplete('Not implemented yet.');
+    }
+}

tests/Unit/PayloadTest.php

@@ -0,0 +1,47 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Spiral\RoadRunner\Tests\Worker\Unit;
+
+use PHPUnit\Framework\TestCase;
+use Spiral\RoadRunner\Payload;
+
+final class PayloadTest extends TestCase
+{
+    public function testPayloadConstructionWithValues(): void
+    {
+        $payload = new Payload('body_content', 'header_content', false);
+
+        $this->assertEquals('body_content', $payload->body);
+        $this->assertEquals('header_content', $payload->header);
+        $this->assertFalse($payload->eos);
+    }
+
+    public function testPayloadConstructionWithDefaultValues(): void
+    {
+        $payload = new Payload(null, null);
+
+        $this->assertEquals('', $payload->body);
+        $this->assertEquals('', $payload->header);
+        $this->assertTrue($payload->eos);
+    }
+
+    public function testPayloadConstructionWithPartialValues(): void
+    {
+        $payload = new Payload('body_content');
+
+        $this->assertEquals('body_content', $payload->body);
+        $this->assertEquals('', $payload->header);
+        $this->assertTrue($payload->eos);
+    }
+
+    public function testPayloadConstructionWithEosFalse(): void
+    {
+        $payload = new Payload(null, null, false);
+
+        $this->assertEquals('', $payload->body);
+        $this->assertEquals('', $payload->header);
+        $this->assertFalse($payload->eos);
+    }
+}

tests/Unit/VersionTest.php

@@ -0,0 +1,98 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Spiral\RoadRunner\Tests\Worker\Unit;
+
+use Composer\InstalledVersions;
+use PHPUnit\Framework\Attributes\DataProvider;
+use PHPUnit\Framework\TestCase;
+use Spiral\RoadRunner\Version;
+
+final class VersionTest extends TestCase
+{
+    public static function provideVersions(): iterable
+    {
+        yield [
+            [
+                'spiral/roadrunner' => [
+                    'pretty_version' => 'v1.9.0',
+                ],
+                'spiral/roadrunner-worker' => [
+                    'pretty_version' => 'v1.8.0',
+                ],
+            ],
+            '1.9.0',
+            '1.*'
+        ];
+
+
+        yield [
+            [
+                'spiral/roadrunner' => [
+                    'pretty_version' => '2.1.0',
+                ],
+            ],
+            '2.1.0',
+            '2.*'
+        ];
+
+        yield [
+            [
+                'spiral/roadrunner-worker' => [
+                    'pretty_version' => 'v1.8.0',
+                ],
+                'spiral/roadrunner' => [
+                    'pretty_version' => 'v1.9.0',
+                ],
+            ],
+            '1.9.0',
+            '1.*'
+        ];
+
+        yield [
+            [
+                'spiral/roadrunner-worker' => [
+                    'pretty_version' => 'v1.8.0',
+                ],
+            ],
+            '1.8.0',
+            '1.*'
+        ];
+
+        yield [
+            [
+                'spiral/roadrunner-http' => [
+                    'pretty_version' => 'v1.8.0',
+                ],
+            ],
+            Version::VERSION_FALLBACK,
+            '*'
+        ];
+
+        yield [
+            [],
+            Version::VERSION_FALLBACK,
+            '*'
+        ];
+    }
+
+    protected function setUp(): void
+    {
+        parent::setUp();
+
+        $ref = new \ReflectionClass(InstalledVersions::class);
+        $ref->setStaticPropertyValue('canGetVendors', false);
+    }
+
+    #[DataProvider('provideVersions')]
+    public function testGetVersion(array $versions, string $expectedVersion, string $expectedConstraint): void
+    {
+        InstalledVersions::reload([
+            'versions' => $versions,
+        ]);
+
+        $this->assertSame($expectedVersion, Version::current());
+        $this->assertSame($expectedConstraint, Version::constraint());
+    }
+}

tests/Unit/WorkerPoolTest.php

@@ -0,0 +1,44 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Spiral\RoadRunner\Tests\Worker\Unit;
+
+use PHPUnit\Framework\MockObject\Exception;
+use PHPUnit\Framework\TestCase;
+use Spiral\Goridge\RPC\Codec\JsonCodec;
+use Spiral\Goridge\RPC\RPCInterface;
+use Spiral\RoadRunner\WorkerPool;
+
+final class WorkerPoolTest extends TestCase
+{
+    private \PHPUnit\Framework\MockObject\MockObject|RPCInterface $rpc;
+    private WorkerPool $workerPool;
+
+    /**
+     * @throws Exception
+     */
+    protected function setUp(): void
+    {
+        parent::setUp();
+
+        $this->rpc = $this->createMock(RPCInterface::class);
+        $this->rpc->expects($this->once())->method('withCodec')->with($this->isInstanceOf(JsonCodec::class))->willReturnSelf();
+
+        $this->workerPool = new WorkerPool($this->rpc);
+    }
+
+    public function testAddWorker(): void
+    {
+        $this->rpc->expects($this->once())->method('call')->with('informer.AddWorker', 'test');
+
+        $this->workerPool->addWorker('test');
+    }
+
+    public function testRemoveWorker(): void
+    {
+        $this->rpc->expects($this->once())->method('call')->with('informer.RemoveWorker', 'test');
+
+        $this->workerPool->removeWorker('test');
+    }
+}