-
Notifications
You must be signed in to change notification settings - Fork 3
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #31 from roadrunner-php/feature/1728
Dynamic Workers Scaling via RPC
- Loading branch information
Showing
6 changed files
with
377 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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); | ||
} | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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()); | ||
} | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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.'); | ||
} | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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); | ||
} | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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()); | ||
} | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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'); | ||
} | ||
} |