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;
+
+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');
+ }
+}