laysense/laylink
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add KCP support

Browse Source
This commit is contained in:
Ziki Shay
2026-05-29 01:06:18 +08:00
parent 6d9299eeb0
commit 661ed401da
31 changed files with 3829 additions and 174 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Agent/AgentClient.php
+170 -74
View File
@@ -5,11 +5,10 @@ declare(strict_types=1);
namespace LayLink\Agent;
use LayLink\Protocol\Frame;
use LayLink\Protocol\FrameCodec;
use LayLink\Protocol\FrameParser;
use LayLink\Protocol\FrameType;
use LayLink\Transport\FrameClientTransport;
use LayLink\Transport\FrameClientTransportFactory;
use LayLink\Util\Uuid;
use Workerman\Connection\AsyncTcpConnection;
use Workerman\Connection\TcpConnection;
use Workerman\Connection\UdpConnection;
use Workerman\Timer;
@@ -17,8 +16,13 @@ use Workerman\Worker;
final class AgentClient
{
private ?AsyncTcpConnection $pop = null;
private ?FrameParser $parser = null;
/** @var array<int, FrameClientTransport> */
private array $pops = [];
/** @var array<int, true> */
private array $authenticatedPops = [];
/** @var array<string, FrameClientTransport> */
private array $sessionTransports = [];
private int $nextTransportCursor = 0;
private bool $authenticated = false;
/** @var array<int, string> */
private array $initialBuffers = [];
@@ -66,6 +70,7 @@ final class AgentClient
private readonly int $maxSendBuffer = 64 * 1024 * 1024,
private readonly int $backpressureHighWatermark = 32 * 1024 * 1024,
private readonly int $dataChunkSize = 1024 * 1024,
private readonly int $popConnectionCount = 1,
) {
}
@@ -75,9 +80,6 @@ final class AgentClient
$worker->name = $workerName;
$worker->count = 1;
$worker->onWorkerStart = function (): void {
if ($this->transportProtocol !== 'tcp') {
throw new \RuntimeException("Agent transport '{$this->transportProtocol}' is configured but not implemented yet.");
}
$this->connect();
Timer::add(10, fn () => $this->heartbeat());
};
@@ -95,59 +97,53 @@ final class AgentClient
private function connect(): void
{
$this->parser = new FrameParser();
$this->authenticated = false;
$connection = new AsyncTcpConnection($this->popAddress);
$connection->maxSendBufferSize = $this->maxSendBuffer;
$this->pop = $connection;
$connection->onConnect = function (AsyncTcpConnection $connection): void {
$this->send(new Frame(FrameType::AUTH, null, [
'node_id' => $this->nodeId,
'node_type' => $this->nodeType,
'node_zone' => $this->nodeZone,
'node_token' => $this->nodeToken,
'transport_protocol' => $this->transportProtocol,
'supported_protocols' => ['tcp'],
'supported_transports' => ['tcp', 'udp', 'kcp'],
]));
};
$connection->onMessage = function (AsyncTcpConnection $connection, string $data): void {
try {
foreach ($this->parser?->push($data) ?? [] as $frame) {
$this->handleFrame($frame);
}
} catch (\Throwable $e) {
$connection->close();
}
};
$connection->onBufferDrain = fn (AsyncTcpConnection $connection) => $this->resumeClientsForPop();
$connection->onClose = function (): void {
$this->authenticated = false;
foreach ($this->clients as $client) {
$client->close();
}
$this->initialBuffers = [];
$this->connectionSessionIds = [];
$this->clients = [];
$this->sessionStates = [];
$this->pendingData = [];
$this->connectionStages = [];
$this->sessionIngressProtocols = [];
$this->pausedClientsForPop = [];
$this->clientsPausingPop = [];
$this->pendingClientCloses = [];
$this->suppressClientCloseFrames = [];
Timer::add(3, fn () => $this->connect(), [], false);
};
$connection->connect();
$this->pops = [];
$this->authenticatedPops = [];
for ($i = 0; $i < $this->popConnectionCount; $i++) {
$this->connectOne();
}
}
private function handleFrame(Frame $frame): void
private function connectOne(): void
{
$transport = (new FrameClientTransportFactory())->create(
$this->transportProtocol,
$this->popAddress,
$this->maxSendBuffer,
function (FrameClientTransport $transport): void {
$transport->send(new Frame(FrameType::AUTH, null, [
'node_id' => $this->nodeId,
'node_type' => $this->nodeType,
'node_zone' => $this->nodeZone,
'node_token' => $this->nodeToken,
'transport_protocol' => $this->transportProtocol,
'supported_protocols' => ['tcp'],
'supported_transports' => ['tcp', 'udp', 'kcp'],
]));
},
function (FrameClientTransport $transport, Frame $frame): void {
$this->handleFrame($transport, $frame);
},
function (FrameClientTransport $transport): void {
$this->handlePopClose($transport);
Timer::add(3, fn () => $this->connectOne(), [], false);
},
function (FrameClientTransport $transport): void {
$this->resumeClientsForPop($transport);
},
function (\Throwable $e): void {
},
);
$this->pops[spl_object_id($transport)] = $transport;
$transport->connect();
}
private function handleFrame(FrameClientTransport $transport, Frame $frame): void
{
match ($frame->type) {
FrameType::AUTH_OK => $this->authenticated = true,
FrameType::AUTH_FAIL => $this->pop?->close(),
FrameType::AUTH_OK => $this->markPopAuthenticated($transport),
FrameType::AUTH_FAIL => $transport->close(),
FrameType::PONG => null,
FrameType::OPEN_OK => $this->openClientSession($frame),
FrameType::OPEN_FAIL => $this->failClientSession($frame),
@@ -184,7 +180,8 @@ final class AgentClient
$this->closeClient($sessionId);
return;
}
if ($this->pop !== null && $this->pop->getSendBufferQueueSize() >= $this->backpressureHighWatermark) {
$transport = $this->sessionTransports[$sessionId] ?? null;
if ($transport !== null && $transport->getSendBufferQueueSize() >= $this->backpressureHighWatermark) {
$connection->pauseRecv();
$this->pausedClientsForPop[$connection->id] = $connection;
}
@@ -447,7 +444,8 @@ final class AgentClient
private function startPopSession(TcpConnection $connection, array $request, string $payloadBytes, string $ingressProtocol): void
{
if (!$this->authenticated || $this->pop === null) {
$transport = $this->selectTransport();
if (!$this->authenticated || $transport === null) {
$this->failOpeningLocalClient($connection, $ingressProtocol, 'pop_not_connected');
return;
}
@@ -455,13 +453,14 @@ final class AgentClient
$sessionId = Uuid::v4();
$this->connectionSessionIds[$connection->id] = $sessionId;
$this->clients[$sessionId] = $connection;
$this->sessionTransports[$sessionId] = $transport;
$this->sessionStates[$sessionId] = 'opening';
$this->sessionIngressProtocols[$sessionId] = $ingressProtocol;
if ($payloadBytes !== '') {
$this->pendingData[$sessionId] = $payloadBytes;
}
$this->send(new Frame(FrameType::OPEN, $sessionId, [
$transport->send(new Frame(FrameType::OPEN, $sessionId, [
'auth_token' => (string)($request['auth_token'] ?? ''),
'user_id' => (string)($request['user_id'] ?? ''),
'target_host' => (string)($request['target_host'] ?? ''),
@@ -486,7 +485,7 @@ final class AgentClient
private function onSocks5UdpMessage(UdpConnection $connection, string $data): void
{
if (!$this->authenticated || $this->pop === null || strlen($data) < 10) {
if (!$this->authenticated || $this->selectTransport() === null || strlen($data) < 10) {
return;
}
@@ -638,7 +637,7 @@ final class AgentClient
$sessionId = $this->connectionSessionIds[$connection->id];
unset($this->pendingClientCloses[$sessionId]);
unset($this->connectionSessionIds[$connection->id]);
unset($this->clients[$sessionId], $this->sessionStates[$sessionId], $this->pendingData[$sessionId], $this->sessionIngressProtocols[$sessionId]);
unset($this->clients[$sessionId], $this->sessionStates[$sessionId], $this->pendingData[$sessionId], $this->sessionIngressProtocols[$sessionId], $this->sessionTransports[$sessionId]);
if ($suppressCloseFrame) {
return;
}
@@ -668,7 +667,7 @@ final class AgentClient
}
$connection = $this->clients[$sessionId];
unset($this->clients[$sessionId], $this->sessionStates[$sessionId], $this->pendingData[$sessionId], $this->sessionIngressProtocols[$sessionId]);
unset($this->clients[$sessionId], $this->sessionStates[$sessionId], $this->pendingData[$sessionId], $this->sessionIngressProtocols[$sessionId], $this->sessionTransports[$sessionId]);
unset($this->connectionSessionIds[$connection->id]);
unset($this->pausedClientsForPop[$connection->id], $this->clientsPausingPop[$connection->id], $this->pendingClientCloses[$sessionId]);
$this->suppressClientCloseFrames[$connection->id] = true;
@@ -812,21 +811,23 @@ final class AgentClient
private function heartbeat(): void
{
if (!$this->authenticated || $this->pop === null) {
if (!$this->authenticated) {
return;
}
$this->send(new Frame(FrameType::PING, null, [
'node_id' => $this->nodeId,
'active_sessions' => count($this->clients),
'load' => sys_getloadavg()[0] ?? 0.0,
'timestamp' => time(),
]));
foreach ($this->authenticatedTransports() as $transport) {
$transport->send(new Frame(FrameType::PING, null, [
'node_id' => $this->nodeId,
'active_sessions' => $this->activeSessionsForTransport($transport),
'load' => sys_getloadavg()[0] ?? 0.0,
'timestamp' => time(),
]));
}
}
private function send(Frame $frame): bool|null
{
return $this->pop?->send(FrameCodec::encode($frame));
return $this->transportForSession($frame->sessionId)?->send($frame);
}
private function sendData(string $sessionId, string $data): bool
@@ -844,9 +845,12 @@ final class AgentClient
return true;
}
private function resumeClientsForPop(): void
private function resumeClientsForPop(FrameClientTransport $transport): void
{
foreach ($this->pausedClientsForPop as $connectionId => $client) {
if ($this->transportForConnection($client) !== $transport) {
continue;
}
$client->resumeRecv();
unset($this->pausedClientsForPop[$connectionId]);
}
@@ -854,15 +858,20 @@ final class AgentClient
private function pausePopForClient(TcpConnection $connection): void
{
$transport = $this->transportForConnection($connection);
if ($transport === null) {
return;
}
$this->clientsPausingPop[$connection->id] = true;
$this->pop?->pauseRecv();
$transport->pauseRecv();
}
private function resumePopForClient(TcpConnection $connection): void
{
unset($this->clientsPausingPop[$connection->id]);
if ($this->clientsPausingPop === []) {
$this->pop?->resumeRecv();
$transport = $this->transportForConnection($connection);
if ($transport !== null && !$this->hasClientsPausingTransport($transport)) {
$transport->resumeRecv();
}
}
@@ -874,4 +883,91 @@ final class AgentClient
$this->finalizeClientClose($sessionId);
}
}
private function markPopAuthenticated(FrameClientTransport $transport): void
{
$this->authenticatedPops[spl_object_id($transport)] = true;
$this->authenticated = true;
}
private function handlePopClose(FrameClientTransport $transport): void
{
$transportId = spl_object_id($transport);
unset($this->pops[$transportId], $this->authenticatedPops[$transportId]);
$this->authenticated = $this->authenticatedPops !== [];
foreach ($this->sessionTransports as $sessionId => $sessionTransport) {
if ($sessionTransport !== $transport || !isset($this->clients[$sessionId])) {
continue;
}
$client = $this->clients[$sessionId];
$this->suppressClientCloseFrames[$client->id] = true;
$client->close();
}
}
private function selectTransport(): ?FrameClientTransport
{
$transports = $this->authenticatedTransports();
if ($transports === []) {
return null;
}
$transport = $transports[$this->nextTransportCursor % count($transports)];
$this->nextTransportCursor++;
return $transport;
}
/**
* @return FrameClientTransport[]
*/
private function authenticatedTransports(): array
{
$transports = [];
foreach ($this->authenticatedPops as $transportId => $_) {
if (isset($this->pops[$transportId])) {
$transports[] = $this->pops[$transportId];
}
}
return $transports;
}
private function transportForSession(?string $sessionId): ?FrameClientTransport
{
if ($sessionId !== null && isset($this->sessionTransports[$sessionId])) {
return $this->sessionTransports[$sessionId];
}
return $this->selectTransport();
}
private function transportForConnection(TcpConnection $connection): ?FrameClientTransport
{
$sessionId = $this->connectionSessionIds[$connection->id] ?? null;
return is_string($sessionId) ? ($this->sessionTransports[$sessionId] ?? null) : null;
}
private function hasClientsPausingTransport(FrameClientTransport $transport): bool
{
foreach ($this->clientsPausingPop as $connectionId => $_) {
$sessionId = $this->connectionSessionIds[$connectionId] ?? null;
if (is_string($sessionId) && ($this->sessionTransports[$sessionId] ?? null) === $transport) {
return true;
}
}
return false;
}
private function activeSessionsForTransport(FrameClientTransport $transport): int
{
$count = 0;
foreach ($this->sessionTransports as $sessionId => $sessionTransport) {
if ($sessionTransport === $transport && isset($this->clients[$sessionId])) {
$count++;
}
}
return $count;
}
}
src/Node/NodeConnection.php
+2 -2
View File
@@ -4,7 +4,7 @@ declare(strict_types=1);
namespace LayLink\Node;
use Workerman\Connection\TcpConnection;
use LayLink\Transport\FrameServerConnection;
final class NodeConnection
{
@@ -15,7 +15,7 @@ final class NodeConnection
public readonly string $nodeId,
public readonly string $nodeType,
public readonly string $nodeZone,
public readonly TcpConnection $connection,
public readonly FrameServerConnection $connection,
) {
$this->lastHeartbeat = time();
}
src/Node/NodeRegistry.php
+38 -10
View File
@@ -4,23 +4,30 @@ declare(strict_types=1);
namespace LayLink\Node;
use Workerman\Connection\TcpConnection;
use LayLink\Transport\FrameServerConnection;
final class NodeRegistry
{
/** @var array<string, NodeConnection> */
/** @var array<string, array<int, NodeConnection>> */
private array $nodes = [];
public function register(NodeConnection $node): void
{
$this->nodes[$node->nodeId] = $node;
$this->nodes[$node->nodeId][$node->connection->id()] = $node;
}
public function unregisterByConnection(TcpConnection $connection): ?NodeConnection
public function unregisterByConnection(FrameServerConnection $connection): ?NodeConnection
{
foreach ($this->nodes as $nodeId => $node) {
if ($node->connection === $connection) {
unset($this->nodes[$nodeId]);
foreach ($this->nodes as $nodeId => $connections) {
foreach ($connections as $connectionId => $node) {
if ($node->connection !== $connection) {
continue;
}
unset($this->nodes[$nodeId][$connectionId]);
if ($this->nodes[$nodeId] === []) {
unset($this->nodes[$nodeId]);
}
return $node;
}
}
@@ -30,12 +37,26 @@ final class NodeRegistry
public function get(string $nodeId): ?NodeConnection
{
return $this->nodes[$nodeId] ?? null;
$connections = $this->nodes[$nodeId] ?? [];
return $connections === [] ? null : array_values($connections)[0];
}
public function getByConnection(FrameServerConnection $connection): ?NodeConnection
{
foreach ($this->nodes as $connections) {
foreach ($connections as $node) {
if ($node->connection === $connection) {
return $node;
}
}
}
return null;
}
public function isOnline(string $nodeId): bool
{
return isset($this->nodes[$nodeId]);
return ($this->nodes[$nodeId] ?? []) !== [];
}
/**
@@ -43,6 +64,13 @@ final class NodeRegistry
*/
public function all(): array
{
return array_values($this->nodes);
$nodes = [];
foreach ($this->nodes as $connections) {
foreach ($connections as $node) {
$nodes[] = $node;
}
}
return $nodes;
}
}
src/Server/AgentListener.php
+50 -49
View File
@@ -10,22 +10,19 @@ use LayLink\Auth\NodeAuthenticator;
use LayLink\Node\NodeConnection;
use LayLink\Node\NodeRegistry;
use LayLink\Protocol\Frame;
use LayLink\Protocol\FrameCodec;
use LayLink\Protocol\FrameParser;
use LayLink\Protocol\FrameType;
use LayLink\Route\RouteResolver;
use LayLink\Session\SessionManager;
use LayLink\Session\TunnelSession;
use LayLink\Transport\FrameServerConnection;
use LayLink\Transport\FrameServerListenerFactory;
use Workerman\Connection\AsyncTcpConnection;
use Workerman\Connection\AsyncUdpConnection;
use Workerman\Connection\TcpConnection;
use Workerman\Timer;
use Workerman\Worker;
final class AgentListener
{
/** @var array<int, FrameParser> */
private array $parsers = [];
/** @var array<int, string> */
private array $connectionNodeIds = [];
/** @var array<int, array<string, AsyncTcpConnection>> */
@@ -42,33 +39,38 @@ final class AgentListener
private readonly int $maxSendBuffer = 64 * 1024 * 1024,
private readonly int $backpressureHighWatermark = 32 * 1024 * 1024,
private readonly int $dataChunkSize = 1024 * 1024,
private readonly string $agentTransportProtocol = 'tcp',
) {
$worker->onConnect = fn (TcpConnection $connection) => $this->onConnect($connection);
$worker->onMessage = fn (TcpConnection $connection, string $data) => $this->onMessage($connection, $data);
$worker->onClose = fn (TcpConnection $connection) => $this->onClose($connection);
$worker->onWorkerStart = fn () => Timer::add(10, fn () => $this->sweepHeartbeats());
}
private function onConnect(TcpConnection $connection): void
{
$connection->maxSendBufferSize = $this->maxSendBuffer;
$connection->onBufferDrain = fn (TcpConnection $connection) => $this->resumeTargetsForAgent($connection);
$this->parsers[$connection->id] = new FrameParser();
}
private function onMessage(TcpConnection $connection, string $data): void
{
try {
foreach ($this->parsers[$connection->id]->push($data) as $frame) {
$this->handleFrame($connection, $frame);
(new FrameServerListenerFactory())->attach(
$this->agentTransportProtocol,
$worker,
$this->maxSendBuffer,
fn (FrameServerConnection $connection) => $this->onConnect($connection),
fn (FrameServerConnection $connection, Frame $frame) => $this->handleFrame($connection, $frame),
fn (FrameServerConnection $connection) => $this->onClose($connection),
fn (FrameServerConnection $connection) => $this->resumeTargetsForAgent($connection),
fn (FrameServerConnection $connection, \Throwable $e) => $this->onInvalidFrame($connection),
);
$transportWorkerStart = $worker->onWorkerStart;
$worker->onWorkerStart = function () use ($transportWorkerStart): void {
if ($transportWorkerStart !== null) {
$transportWorkerStart();
}
} catch (\Throwable $e) {
$this->send($connection, new Frame(FrameType::ERROR, null, ['reason' => 'invalid_frame']));
$connection->close();
}
Timer::add(10, fn () => $this->sweepHeartbeats());
};
}
private function handleFrame(TcpConnection $connection, Frame $frame): void
private function onConnect(FrameServerConnection $connection): void
{
}
private function onInvalidFrame(FrameServerConnection $connection): void
{
$this->send($connection, new Frame(FrameType::ERROR, null, ['reason' => 'invalid_frame']));
$connection->close();
}
private function handleFrame(FrameServerConnection $connection, Frame $frame): void
{
if ($frame->type === FrameType::AUTH) {
$result = $this->authenticator->authenticate($frame->payload);
@@ -86,7 +88,7 @@ final class AgentListener
$connection,
);
$this->nodes->register($node);
$this->connectionNodeIds[$connection->id] = $nodeId;
$this->connectionNodeIds[$connection->id()] = $nodeId;
$this->send($connection, new Frame(FrameType::AUTH_OK, null, [
'node_id' => $nodeId,
'heartbeat_interval' => 10,
@@ -95,8 +97,8 @@ final class AgentListener
return;
}
$nodeId = $this->connectionNodeIds[$connection->id] ?? null;
$node = is_string($nodeId) ? $this->nodes->get($nodeId) : null;
$nodeId = $this->connectionNodeIds[$connection->id()] ?? null;
$node = is_string($nodeId) ? $this->nodes->getByConnection($connection) : null;
if (!is_string($nodeId) || $node === null) {
$this->send($connection, new Frame(FrameType::ERROR, $frame->sessionId, ['reason' => 'invalid_auth']));
$connection->close();
@@ -133,7 +135,7 @@ final class AgentListener
};
}
private function forwardUdpDatagram(TcpConnection $agentConnection, string $nodeId, Frame $frame): void
private function forwardUdpDatagram(FrameServerConnection $agentConnection, string $nodeId, Frame $frame): void
{
if ($frame->sessionId === null) {
$this->send($agentConnection, new Frame(FrameType::ERROR, null, ['reason' => 'invalid_frame']));
@@ -177,7 +179,7 @@ final class AgentListener
$target->connect();
}
private function openTargetForAgent(TcpConnection $agentConnection, string $nodeId, Frame $frame): void
private function openTargetForAgent(FrameServerConnection $agentConnection, string $nodeId, Frame $frame): void
{
if ($frame->sessionId === null || $this->sessions->get($frame->sessionId) !== null) {
$this->send($agentConnection, new Frame(FrameType::OPEN_FAIL, $frame->sessionId, ['reason' => 'invalid_frame']));
@@ -248,7 +250,7 @@ final class AgentListener
}
if ($agentConnection->getSendBufferQueueSize() >= $this->backpressureHighWatermark) {
$target->pauseRecv();
$this->pausedTargetsByAgent[$agentConnection->id][$session->sessionId] = $target;
$this->pausedTargetsByAgent[$agentConnection->id()][$session->sessionId] = $target;
}
};
$target->onClose = fn () => $this->closeSession($session, 'closed', null);
@@ -282,7 +284,7 @@ final class AgentListener
}
}
private function failOpenSession(TcpConnection $agentConnection, TunnelSession $session, string $reason): void
private function failOpenSession(FrameServerConnection $agentConnection, TunnelSession $session, string $reason): void
{
if ($session->state === TunnelSession::OPENING) {
$this->send($agentConnection, new Frame(FrameType::OPEN_FAIL, $session->sessionId, ['reason' => $reason]));
@@ -299,7 +301,7 @@ final class AgentListener
return base64_decode((string)($frame->payload['data'] ?? ''), true);
}
private function rejectOpen(TcpConnection $agentConnection, Frame $frame, string $reason, string $userId, string $nodeId, ?string $policyId = null): void
private function rejectOpen(FrameServerConnection $agentConnection, Frame $frame, string $reason, string $userId, string $nodeId, ?string $policyId = null): void
{
$this->send($agentConnection, new Frame(FrameType::OPEN_FAIL, $frame->sessionId, ['reason' => $reason]));
$this->audit->write([
@@ -330,7 +332,7 @@ final class AgentListener
$session->state = TunnelSession::CLOSED;
$this->sessions->remove($session->sessionId);
if ($session->agent !== null) {
unset($this->pausedTargetsByAgent[$session->agent->id][$session->sessionId]);
unset($this->pausedTargetsByAgent[$session->agent->id()][$session->sessionId]);
}
if ($session->agent !== null) {
@@ -359,11 +361,10 @@ final class AgentListener
]);
}
private function onClose(TcpConnection $connection): void
private function onClose(FrameServerConnection $connection): void
{
unset($this->parsers[$connection->id]);
unset($this->connectionNodeIds[$connection->id]);
unset($this->pausedTargetsByAgent[$connection->id]);
unset($this->connectionNodeIds[$connection->id()]);
unset($this->pausedTargetsByAgent[$connection->id()]);
$node = $this->nodes->unregisterByConnection($connection);
foreach ($this->sessions->all() as $session) {
if ($session->agent === $connection) {
@@ -386,12 +387,12 @@ final class AgentListener
}
}
private function send(TcpConnection $connection, Frame $frame): bool|null
private function send(FrameServerConnection $connection, Frame $frame): bool|null
{
return $connection->send(FrameCodec::encode($frame));
return $connection->send($frame);
}
private function sendData(TcpConnection $connection, string $sessionId, string $data): bool
private function sendData(FrameServerConnection $connection, string $sessionId, string $data): bool
{
$length = strlen($data);
for ($offset = 0; $offset < $length; $offset += $this->dataChunkSize) {
@@ -406,14 +407,14 @@ final class AgentListener
return true;
}
private function resumeTargetsForAgent(TcpConnection $connection): void
private function resumeTargetsForAgent(FrameServerConnection $connection): void
{
foreach ($this->pausedTargetsByAgent[$connection->id] ?? [] as $sessionId => $target) {
foreach ($this->pausedTargetsByAgent[$connection->id()] ?? [] as $sessionId => $target) {
$target->resumeRecv();
unset($this->pausedTargetsByAgent[$connection->id][$sessionId]);
unset($this->pausedTargetsByAgent[$connection->id()][$sessionId]);
}
if (($this->pausedTargetsByAgent[$connection->id] ?? []) === []) {
unset($this->pausedTargetsByAgent[$connection->id]);
if (($this->pausedTargetsByAgent[$connection->id()] ?? []) === []) {
unset($this->pausedTargetsByAgent[$connection->id()]);
}
}
src/Server/PopServer.php
+24 -15
View File
@@ -36,20 +36,29 @@ final class PopServer
public function boot(): void
{
$agentWorker = new Worker('tcp://' . $this->agentListen);
$agentWorker->name = 'laylink-pop-agent-listener';
$agentWorker->count = 1;
new AgentListener(
$agentWorker,
new NodeAuthenticator($this->nodeConfig, $this->allowedAgentTransports),
new ClientAuthenticator(),
new RouteResolver(new PolicyChecker($this->policies), $this->nodes),
$this->nodes,
$this->sessions,
$this->audit,
$this->maxSendBuffer,
$this->backpressureHighWatermark,
$this->dataChunkSize,
);
$implementedTransports = array_values(array_intersect($this->allowedAgentTransports, ['tcp', 'kcp']));
if ($implementedTransports === []) {
throw new \RuntimeException('no_implemented_pop_transport_enabled');
}
foreach ($implementedTransports as $transport) {
$scheme = $transport === 'kcp' ? 'udp' : 'tcp';
$agentWorker = new Worker($scheme . '://' . $this->agentListen);
$agentWorker->name = 'laylink-pop-agent-listener-' . $transport;
$agentWorker->count = 1;
new AgentListener(
$agentWorker,
new NodeAuthenticator($this->nodeConfig, $this->allowedAgentTransports),
new ClientAuthenticator(),
new RouteResolver(new PolicyChecker($this->policies), $this->nodes),
$this->nodes,
$this->sessions,
$this->audit,
$this->maxSendBuffer,
$this->backpressureHighWatermark,
$this->dataChunkSize,
$transport,
);
}
}
}
src/Session/TunnelSession.php
+2 -1
View File
@@ -4,6 +4,7 @@ declare(strict_types=1);
namespace LayLink\Session;
use LayLink\Transport\FrameServerConnection;
use Workerman\Connection\AsyncTcpConnection;
use Workerman\Connection\TcpConnection;
@@ -18,7 +19,7 @@ final class TunnelSession
public string $state = self::NEW;
public ?TcpConnection $client = null;
public ?TcpConnection $agent = null;
public ?FrameServerConnection $agent = null;
public ?AsyncTcpConnection $target = null;
public ?string $nodeId = null;
public string $startTime;
src/Transport/FrameClientTransport.php
+22
View File
@@ -0,0 +1,22 @@
<?php
declare(strict_types=1);
namespace LayLink\Transport;
use LayLink\Protocol\Frame;
interface FrameClientTransport
{
public function connect(): void;
public function send(Frame $frame): bool|null;
public function close(): void;
public function pauseRecv(): void;
public function resumeRecv(): void;
public function getSendBufferQueueSize(): int;
}
src/Transport/FrameClientTransportFactory.php
+44
View File
@@ -0,0 +1,44 @@
<?php
declare(strict_types=1);
namespace LayLink\Transport;
use InvalidArgumentException;
final class FrameClientTransportFactory
{
public function create(
string $protocol,
string $address,
int $maxSendBuffer,
\Closure $onConnect,
\Closure $onFrame,
\Closure $onClose,
\Closure $onBufferDrain,
\Closure $onInvalidFrame,
): FrameClientTransport {
return match (strtolower(trim($protocol))) {
'tcp' => new TcpFrameClientTransport(
$address,
$maxSendBuffer,
$onConnect,
$onFrame,
$onClose,
$onBufferDrain,
$onInvalidFrame,
),
'kcp' => new KcpFrameClientTransport(
$address,
$maxSendBuffer,
$onConnect,
$onFrame,
$onClose,
$onBufferDrain,
$onInvalidFrame,
),
'udp' => throw new InvalidArgumentException('agent_transport_not_implemented'),
default => throw new InvalidArgumentException('unsupported_agent_transport'),
};
}
}
src/Transport/FrameServerConnection.php
+24
View File
@@ -0,0 +1,24 @@
<?php
declare(strict_types=1);
namespace LayLink\Transport;
use LayLink\Protocol\Frame;
interface FrameServerConnection
{
public function id(): int;
public function send(Frame $frame): bool|null;
public function close(): void;
public function pauseRecv(): void;
public function resumeRecv(): void;
public function getSendBufferQueueSize(): int;
public function getRemoteIp(): string;
}
src/Transport/FrameServerListenerFactory.php
+45
View File
@@ -0,0 +1,45 @@
<?php
declare(strict_types=1);
namespace LayLink\Transport;
use InvalidArgumentException;
use Workerman\Worker;
final class FrameServerListenerFactory
{
public function attach(
string $protocol,
Worker $worker,
int $maxSendBuffer,
\Closure $onConnect,
\Closure $onFrame,
\Closure $onClose,
\Closure $onBufferDrain,
\Closure $onInvalidFrame,
): void {
match (strtolower(trim($protocol))) {
'tcp' => new TcpFrameServerListener(
$worker,
$maxSendBuffer,
$onConnect,
$onFrame,
$onClose,
$onBufferDrain,
$onInvalidFrame,
),
'kcp' => new KcpFrameServerListener(
$worker,
$maxSendBuffer,
$onConnect,
$onFrame,
$onClose,
$onBufferDrain,
$onInvalidFrame,
),
'udp' => throw new InvalidArgumentException('pop_transport_not_implemented'),
default => throw new InvalidArgumentException('unsupported_pop_transport'),
};
}
}
src/Transport/KcpFrameClientTransport.php
+176
View File
@@ -0,0 +1,176 @@
<?php
declare(strict_types=1);
namespace LayLink\Transport;
use LayLink\Protocol\Frame;
use Workerman\Connection\AsyncUdpConnection;
use Workerman\Timer;
final class KcpFrameClientTransport implements FrameClientTransport
{
private ?AsyncUdpConnection $connection = null;
private KcpReliableSession|NativeKcpSession|null $session = null;
private ?int $timerId = null;
private int $conv;
private bool $connected = false;
public function __construct(
private readonly string $address,
private readonly int $maxSendBuffer,
private readonly \Closure $onConnect,
private readonly \Closure $onFrame,
private readonly \Closure $onClose,
private readonly \Closure $onBufferDrain,
private readonly \Closure $onInvalidFrame,
) {
$this->conv = random_int(1, 0x7fffffff);
}
public function connect(): void
{
$this->connected = false;
$this->session = null;
$connection = new AsyncUdpConnection($this->normalizeAddress($this->address));
$this->connection = $connection;
$connection->onConnect = function () use ($connection): void {
$connection->send(KcpPacketCodec::encode([
'type' => KcpPacketCodec::SYN,
'conv' => $this->conv,
]));
};
$connection->onMessage = function (AsyncUdpConnection $connection, string $data): void {
$this->handlePacket($data);
};
$connection->onClose = function () use ($connection): void {
if ($this->connection === $connection) {
$this->connection = null;
}
$this->stopTimer();
($this->onClose)($this);
};
$connection->connect();
$this->timerId = Timer::add(0.02, fn () => $this->tick());
}
public function send(Frame $frame): bool|null
{
return $this->session?->sendFrame($frame, $this->maxSendBuffer) ?? false;
}
public function close(): void
{
$this->session?->close();
$this->connection?->close();
}
public function pauseRecv(): void
{
$this->session?->pause();
}
public function resumeRecv(): void
{
$this->session?->resume();
}
public function getSendBufferQueueSize(): int
{
return $this->session?->getSendBufferQueueSize() ?? 0;
}
private function handlePacket(string $data): void
{
$packet = KcpPacketCodec::decode($data);
if ($packet === null) {
$this->session?->receive($data);
return;
}
if ($packet['conv'] !== $this->conv) {
return;
}
if ($packet['type'] === KcpPacketCodec::SYN_ACK && !$this->connected) {
$this->connected = true;
$this->session = $this->createSession();
($this->onConnect)($this);
return;
}
if ($packet['type'] === KcpPacketCodec::CLOSE) {
$this->connection?->close();
return;
}
$this->session?->receive($packet);
if ($this->session?->isClosed()) {
$this->connection?->close();
}
}
private function tick(): void
{
if ($this->connection === null) {
$this->stopTimer();
return;
}
if (!$this->connected) {
$this->connection->send(KcpPacketCodec::encode([
'type' => KcpPacketCodec::SYN,
'conv' => $this->conv,
]));
return;
}
$before = $this->getSendBufferQueueSize();
$this->session?->tick();
if ($before >= $this->maxSendBuffer && $this->getSendBufferQueueSize() < $this->maxSendBuffer) {
($this->onBufferDrain)($this);
}
}
private function createSession(): KcpReliableSession|NativeKcpSession
{
$backend = strtolower(trim((string)(getenv('LAYLINK_KCP_BACKEND') ?: 'ffi')));
$libraryPath = (string)(getenv('LAYLINK_KCP_FFI_LIB') ?: dirname(__DIR__, 2) . '/native/kcp/liblaylink_kcp.so');
$args = [
$this->conv,
fn (string $packet): bool|null => $this->connection?->send($packet),
fn (Frame $frame) => ($this->onFrame)($this, $frame),
fn (\Throwable $e) => ($this->onInvalidFrame)($e),
];
if ($backend === 'php') {
return new KcpReliableSession(...$args);
}
return new NativeKcpSession(...[...$args, $libraryPath]);
}
private function stopTimer(): void
{
if ($this->timerId === null) {
return;
}
Timer::del($this->timerId);
$this->timerId = null;
}
private function normalizeAddress(string $address): string
{
if (str_starts_with($address, 'udp://')) {
return $address;
}
if (str_starts_with($address, 'tcp://')) {
return 'udp://' . substr($address, 6);
}
return 'udp://' . $address;
}
}
src/Transport/KcpFrameServerConnection.php
+75
View File
@@ -0,0 +1,75 @@
<?php
declare(strict_types=1);
namespace LayLink\Transport;
use LayLink\Protocol\Frame;
use Workerman\Connection\UdpConnection;
final class KcpFrameServerConnection implements FrameServerConnection
{
private bool $closed = false;
public function __construct(
private readonly int $id,
private UdpConnection $connection,
private readonly KcpReliableSession|NativeKcpSession $session,
private readonly int $maxSendBuffer,
) {
}
public function id(): int
{
return $this->id;
}
public function send(Frame $frame): bool|null
{
if ($this->closed) {
return false;
}
return $this->session->sendFrame($frame, $this->maxSendBuffer);
}
public function close(): void
{
if ($this->closed) {
return;
}
$this->closed = true;
$this->session->close();
}
public function pauseRecv(): void
{
$this->session->pause();
}
public function resumeRecv(): void
{
$this->session->resume();
}
public function getSendBufferQueueSize(): int
{
return $this->session->getSendBufferQueueSize();
}
public function getRemoteIp(): string
{
return $this->connection->getRemoteIp();
}
public function updateConnection(UdpConnection $connection): void
{
$this->connection = $connection;
}
public function isClosed(): bool
{
return $this->closed || $this->session->isClosed();
}
}
src/Transport/KcpFrameServerListener.php
+177
View File
@@ -0,0 +1,177 @@
<?php
declare(strict_types=1);
namespace LayLink\Transport;
use LayLink\Protocol\Frame;
use Workerman\Connection\UdpConnection;
use Workerman\Timer;
use Workerman\Worker;
final class KcpFrameServerListener
{
/** @var array<string, KcpFrameServerConnection> */
private array $connections = [];
/** @var array<string, KcpReliableSession|NativeKcpSession> */
private array $sessions = [];
private ?int $timerId = null;
private int $nextConnectionId = 1_000_000;
public function __construct(
Worker $worker,
private readonly int $maxSendBuffer,
private readonly \Closure $onConnect,
private readonly \Closure $onFrame,
private readonly \Closure $onClose,
private readonly \Closure $onBufferDrain,
private readonly \Closure $onInvalidFrame,
) {
$worker->onMessage = fn (UdpConnection $connection, string $data) => $this->handleMessage($connection, $data);
$worker->onWorkerStart = function (): void {
$this->timerId = Timer::add(0.02, fn () => $this->tick());
};
$worker->onWorkerStop = function (): void {
if ($this->timerId !== null) {
Timer::del($this->timerId);
$this->timerId = null;
}
};
}
private function handleMessage(UdpConnection $connection, string $data): void
{
$packet = KcpPacketCodec::decode($data);
if ($packet === null) {
$conv = KcpPacketCodec::rawKcpConv($data);
if ($conv === null) {
return;
}
$key = $this->key($connection, $conv);
$wrapped = $this->connections[$key] ?? null;
$session = $this->sessions[$key] ?? null;
if ($wrapped === null || $session === null) {
return;
}
$wrapped->updateConnection($connection);
$wasFull = $session->getSendBufferQueueSize() >= $this->maxSendBuffer;
$session->receive($data);
if ($session->isClosed()) {
$this->closeConnection($key);
return;
}
if ($wasFull && $session->getSendBufferQueueSize() < $this->maxSendBuffer) {
($this->onBufferDrain)($wrapped);
}
return;
}
$key = $this->key($connection, $packet['conv']);
if ($packet['type'] === KcpPacketCodec::SYN) {
$this->handleSyn($connection, $key, $packet['conv']);
return;
}
$wrapped = $this->connections[$key] ?? null;
$session = $this->sessions[$key] ?? null;
if ($wrapped === null || $session === null) {
return;
}
if ($packet['type'] === KcpPacketCodec::CLOSE) {
$this->closeConnection($key);
return;
}
$wrapped->updateConnection($connection);
$wasFull = $session->getSendBufferQueueSize() >= $this->maxSendBuffer;
$session->receive($packet);
if ($session->isClosed()) {
$this->closeConnection($key);
return;
}
if ($wasFull && $session->getSendBufferQueueSize() < $this->maxSendBuffer) {
($this->onBufferDrain)($wrapped);
}
}
private function handleSyn(UdpConnection $connection, string $key, int $conv): void
{
if (isset($this->connections[$key])) {
$connection->send(KcpPacketCodec::encode([
'type' => KcpPacketCodec::SYN_ACK,
'conv' => $conv,
]));
return;
}
$session = $this->createSession(
$conv,
fn (string $packet): bool|null => $connection->send($packet),
function (Frame $frame) use ($key): void {
$wrapped = $this->connections[$key] ?? null;
if ($wrapped !== null) {
($this->onFrame)($wrapped, $frame);
}
},
function (\Throwable $e) use ($key): void {
$wrapped = $this->connections[$key] ?? null;
if ($wrapped !== null) {
($this->onInvalidFrame)($wrapped, $e);
}
},
);
$wrapped = new KcpFrameServerConnection($this->nextConnectionId++, $connection, $session, $this->maxSendBuffer);
$this->sessions[$key] = $session;
$this->connections[$key] = $wrapped;
$connection->send(KcpPacketCodec::encode([
'type' => KcpPacketCodec::SYN_ACK,
'conv' => $conv,
]));
($this->onConnect)($wrapped);
}
private function tick(): void
{
foreach ($this->sessions as $key => $session) {
$wrapped = $this->connections[$key] ?? null;
if ($wrapped === null || $wrapped->isClosed()) {
$this->closeConnection($key);
continue;
}
$before = $session->getSendBufferQueueSize();
$session->tick();
if ($before >= $this->maxSendBuffer && $session->getSendBufferQueueSize() < $this->maxSendBuffer) {
($this->onBufferDrain)($wrapped);
}
}
}
private function closeConnection(string $key): void
{
$wrapped = $this->connections[$key] ?? null;
unset($this->connections[$key], $this->sessions[$key]);
if ($wrapped !== null) {
($this->onClose)($wrapped);
}
}
private function createSession(int $conv, \Closure $sendPacket, \Closure $onFrame, \Closure $onInvalidFrame): KcpReliableSession|NativeKcpSession
{
$backend = strtolower(trim((string)(getenv('LAYLINK_KCP_BACKEND') ?: 'ffi')));
if ($backend === 'php') {
return new KcpReliableSession($conv, $sendPacket, $onFrame, $onInvalidFrame);
}
$libraryPath = (string)(getenv('LAYLINK_KCP_FFI_LIB') ?: dirname(__DIR__, 2) . '/native/kcp/liblaylink_kcp.so');
return new NativeKcpSession($conv, $sendPacket, $onFrame, $onInvalidFrame, $libraryPath);
}
private function key(UdpConnection $connection, int $conv): string
{
return $connection->getRemoteAddress() . '#' . $conv;
}
}
src/Transport/KcpPacketCodec.php
+71
View File
@@ -0,0 +1,71 @@
<?php
declare(strict_types=1);
namespace LayLink\Transport;
final class KcpPacketCodec
{
public const MAGIC = "LLK1";
public const SYN = 1;
public const SYN_ACK = 2;
public const DATA = 3;
public const ACK = 4;
public const CLOSE = 5;
private const HEADER_LENGTH = 25;
/**
* @param array{type:int, conv:int, seq?:int, ack?:int, message_id?:int, fragment_index?:int, fragment_count?:int, payload?:string} $packet
*/
public static function encode(array $packet): string
{
return pack(
'a4CNNNNnn',
self::MAGIC,
$packet['type'],
$packet['conv'],
$packet['seq'] ?? 0,
$packet['ack'] ?? 0,
$packet['message_id'] ?? 0,
$packet['fragment_index'] ?? 0,
$packet['fragment_count'] ?? 0,
) . ($packet['payload'] ?? '');
}
/**
* @return array{type:int, conv:int, seq:int, ack:int, message_id:int, fragment_index:int, fragment_count:int, payload:string}|null
*/
public static function decode(string $bytes): ?array
{
if (strlen($bytes) < self::HEADER_LENGTH || substr($bytes, 0, 4) !== self::MAGIC) {
return null;
}
$header = unpack('a4magic/Ctype/Nconv/Nseq/Nack/Nmessage_id/nfragment_index/nfragment_count', substr($bytes, 0, self::HEADER_LENGTH));
if ($header === false) {
return null;
}
return [
'type' => (int)$header['type'],
'conv' => (int)$header['conv'],
'seq' => (int)$header['seq'],
'ack' => (int)$header['ack'],
'message_id' => (int)$header['message_id'],
'fragment_index' => (int)$header['fragment_index'],
'fragment_count' => (int)$header['fragment_count'],
'payload' => substr($bytes, self::HEADER_LENGTH),
];
}
public static function rawKcpConv(string $bytes): ?int
{
if (strlen($bytes) < 4 || substr($bytes, 0, 4) === self::MAGIC) {
return null;
}
$decoded = unpack('Vconv', substr($bytes, 0, 4));
return $decoded === false ? null : (int)$decoded['conv'];
}
}
src/Transport/KcpReliableSession.php
+235
View File
@@ -0,0 +1,235 @@
<?php
declare(strict_types=1);
namespace LayLink\Transport;
use LayLink\Protocol\Frame;
use LayLink\Protocol\FrameCodec;
use LayLink\Protocol\FrameParser;
final class KcpReliableSession
{
private const MTU = 1200;
private const RESEND_AFTER_MS = 120;
private FrameParser $parser;
private int $nextSeq = 1;
private int $expectedSeq = 1;
private int $nextMessageId = 1;
/** @var array<int, array{packet:string, sent_at:float, bytes:int}> */
private array $unacked = [];
/** @var array<int, array{message_id:int, fragment_index:int, fragment_count:int, payload:string}> */
private array $pendingSegments = [];
/** @var array<int, array<int, string>> */
private array $messageFragments = [];
/** @var array<int, int> */
private array $messageFragmentCounts = [];
/** @var Frame[] */
private array $pausedFrames = [];
private bool $paused = false;
private bool $closed = false;
public function __construct(
private readonly int $conv,
private readonly \Closure $sendPacket,
private readonly \Closure $onFrame,
private readonly \Closure $onInvalidFrame,
) {
$this->parser = new FrameParser();
}
public function conv(): int
{
return $this->conv;
}
public function sendFrame(Frame $frame, int $maxSendBuffer): bool|null
{
if ($this->closed || $this->getSendBufferQueueSize() >= $maxSendBuffer) {
return false;
}
$bytes = FrameCodec::encode($frame);
$payloadSize = self::MTU - 25;
$fragments = str_split($bytes, $payloadSize);
$messageId = $this->nextMessageId++;
$fragmentCount = count($fragments);
foreach ($fragments as $fragmentIndex => $payload) {
$seq = $this->nextSeq++;
$packet = KcpPacketCodec::encode([
'type' => KcpPacketCodec::DATA,
'conv' => $this->conv,
'seq' => $seq,
'message_id' => $messageId,
'fragment_index' => $fragmentIndex,
'fragment_count' => $fragmentCount,
'payload' => $payload,
]);
$this->unacked[$seq] = [
'packet' => $packet,
'sent_at' => $this->nowMs(),
'bytes' => strlen($packet),
];
($this->sendPacket)($packet);
}
return true;
}
/**
* @param array{type:int, conv:int, seq:int, ack:int, message_id:int, fragment_index:int, fragment_count:int, payload:string}|string $packet
*/
public function receive(array|string $packet): void
{
if ($this->closed || is_string($packet)) {
return;
}
if ($packet['type'] === KcpPacketCodec::ACK) {
unset($this->unacked[$packet['ack']]);
return;
}
if ($packet['type'] === KcpPacketCodec::CLOSE) {
$this->closed = true;
return;
}
if ($packet['type'] !== KcpPacketCodec::DATA) {
return;
}
$this->ack($packet['seq']);
if ($packet['seq'] < $this->expectedSeq) {
return;
}
$this->pendingSegments[$packet['seq']] = [
'message_id' => $packet['message_id'],
'fragment_index' => $packet['fragment_index'],
'fragment_count' => $packet['fragment_count'],
'payload' => $packet['payload'],
];
$this->drainOrderedSegments();
}
public function tick(): void
{
if ($this->closed) {
return;
}
$now = $this->nowMs();
foreach ($this->unacked as $seq => $pending) {
if ($now - $pending['sent_at'] < self::RESEND_AFTER_MS) {
continue;
}
$this->unacked[$seq]['sent_at'] = $now;
($this->sendPacket)($pending['packet']);
}
}
public function pause(): void
{
$this->paused = true;
}
public function resume(): void
{
$this->paused = false;
while (!$this->paused && $this->pausedFrames !== []) {
$frame = array_shift($this->pausedFrames);
($this->onFrame)($frame);
}
}
public function close(): void
{
if ($this->closed) {
return;
}
$this->closed = true;
($this->sendPacket)(KcpPacketCodec::encode([
'type' => KcpPacketCodec::CLOSE,
'conv' => $this->conv,
]));
}
public function isClosed(): bool
{
return $this->closed;
}
public function getSendBufferQueueSize(): int
{
$bytes = 0;
foreach ($this->unacked as $pending) {
$bytes += $pending['bytes'];
}
return $bytes;
}
/**
* @param array{message_id:int, fragment_index:int, fragment_count:int, payload:string} $segment
*/
private function receiveOrderedSegment(array $segment): void
{
$messageId = $segment['message_id'];
$fragmentIndex = $segment['fragment_index'];
$fragmentCount = $segment['fragment_count'];
if ($fragmentCount < 1 || $fragmentIndex < 0 || $fragmentIndex >= $fragmentCount) {
return;
}
$this->messageFragmentCounts[$messageId] = $fragmentCount;
$this->messageFragments[$messageId][$fragmentIndex] = $segment['payload'];
if (count($this->messageFragments[$messageId]) !== $fragmentCount) {
return;
}
ksort($this->messageFragments[$messageId]);
$bytes = implode('', $this->messageFragments[$messageId]);
unset($this->messageFragments[$messageId], $this->messageFragmentCounts[$messageId]);
try {
foreach ($this->parser->push($bytes) as $frame) {
if ($this->paused) {
$this->pausedFrames[] = $frame;
continue;
}
($this->onFrame)($frame);
}
} catch (\Throwable $e) {
($this->onInvalidFrame)($e);
}
}
private function drainOrderedSegments(): void
{
while (isset($this->pendingSegments[$this->expectedSeq])) {
$segment = $this->pendingSegments[$this->expectedSeq];
unset($this->pendingSegments[$this->expectedSeq]);
$this->expectedSeq++;
$this->receiveOrderedSegment($segment);
}
}
private function ack(int $seq): void
{
($this->sendPacket)(KcpPacketCodec::encode([
'type' => KcpPacketCodec::ACK,
'conv' => $this->conv,
'ack' => $seq,
]));
}
private function nowMs(): float
{
return microtime(true) * 1000;
}
}
src/Transport/NativeKcpLibrary.php
+52
View File
@@ -0,0 +1,52 @@
<?php
declare(strict_types=1);
namespace LayLink\Transport;
use FFI;
final class NativeKcpLibrary
{
private const CDEF = <<<'CDEF'
typedef struct laylink_kcp laylink_kcp;
laylink_kcp* laylink_kcp_create(unsigned int conv);
void laylink_kcp_release(laylink_kcp* session);
int laylink_kcp_nodelay(laylink_kcp* session, int nodelay, int interval, int resend, int nc);
int laylink_kcp_wndsize(laylink_kcp* session, int sndwnd, int rcvwnd);
int laylink_kcp_setmtu(laylink_kcp* session, int mtu);
int laylink_kcp_send(laylink_kcp* session, const char* buffer, int len);
int laylink_kcp_input(laylink_kcp* session, const char* buffer, long size);
void laylink_kcp_update(laylink_kcp* session, unsigned int current);
unsigned int laylink_kcp_check(laylink_kcp* session, unsigned int current);
int laylink_kcp_peeksize(laylink_kcp* session);
int laylink_kcp_recv(laylink_kcp* session, char* buffer, int len);
void laylink_kcp_flush(laylink_kcp* session);
int laylink_kcp_pending_output_size(laylink_kcp* session);
int laylink_kcp_pop_output(laylink_kcp* session, char* buffer, int len);
CDEF;
private static ?FFI $ffi = null;
public static function load(?string $libraryPath = null): FFI
{
if (self::$ffi !== null) {
return self::$ffi;
}
if (!class_exists(FFI::class)) {
throw new \RuntimeException('ffi_extension_not_loaded');
}
$libraryPath ??= dirname(__DIR__, 2) . '/native/kcp/liblaylink_kcp.so';
if (!str_starts_with($libraryPath, '/')) {
$libraryPath = dirname(__DIR__, 2) . '/' . $libraryPath;
}
if (!is_file($libraryPath)) {
throw new \RuntimeException('kcp_ffi_library_not_found');
}
self::$ffi = FFI::cdef(self::CDEF, $libraryPath);
return self::$ffi;
}
}
src/Transport/NativeKcpSession.php
+188
View File
@@ -0,0 +1,188 @@
<?php
declare(strict_types=1);
namespace LayLink\Transport;
use FFI;
use LayLink\Protocol\Frame;
use LayLink\Protocol\FrameCodec;
use LayLink\Protocol\FrameParser;
final class NativeKcpSession
{
private const RECEIVE_BUFFER_BYTES = 8 * 1024 * 1024;
private \FFI $ffi;
private mixed $session;
private FrameParser $parser;
private bool $paused = false;
private bool $closed = false;
/** @var Frame[] */
private array $pausedFrames = [];
private int $queuedBytes = 0;
public function __construct(
private readonly int $conv,
private readonly \Closure $sendPacket,
private readonly \Closure $onFrame,
private readonly \Closure $onInvalidFrame,
private readonly ?string $libraryPath = null,
) {
$this->ffi = NativeKcpLibrary::load($this->libraryPath);
$this->session = $this->ffi->laylink_kcp_create($conv);
if ($this->session === null) {
throw new \RuntimeException('kcp_create_failed');
}
$this->parser = new FrameParser();
}
public function __destruct()
{
if (!$this->closed) {
$this->closed = true;
$this->ffi->laylink_kcp_release($this->session);
}
}
public function conv(): int
{
return $this->conv;
}
public function sendFrame(Frame $frame, int $maxSendBuffer): bool|null
{
if ($this->closed || $this->queuedBytes >= $maxSendBuffer) {
return false;
}
$bytes = FrameCodec::encode($frame);
$result = $this->ffi->laylink_kcp_send($this->session, $bytes, strlen($bytes));
if ($result < 0) {
return false;
}
$this->queuedBytes += strlen($bytes);
$this->ffi->laylink_kcp_flush($this->session);
$this->drainOutput();
return true;
}
public function receive(string|array $packet): void
{
if ($this->closed || !is_string($packet)) {
return;
}
if ($this->ffi->laylink_kcp_input($this->session, $packet, strlen($packet)) < 0) {
return;
}
$this->drainFrames();
$this->drainOutput();
}
public function tick(): void
{
if ($this->closed) {
return;
}
$this->ffi->laylink_kcp_update($this->session, $this->nowMs());
$this->drainFrames();
$this->drainOutput();
}
public function pause(): void
{
$this->paused = true;
}
public function resume(): void
{
$this->paused = false;
while (!$this->paused && $this->pausedFrames !== []) {
$frame = array_shift($this->pausedFrames);
($this->onFrame)($frame);
}
}
public function close(): void
{
if ($this->closed) {
return;
}
$this->closed = true;
($this->sendPacket)(KcpPacketCodec::encode([
'type' => KcpPacketCodec::CLOSE,
'conv' => $this->conv,
]));
$this->ffi->laylink_kcp_release($this->session);
}
public function isClosed(): bool
{
return $this->closed;
}
public function getSendBufferQueueSize(): int
{
return max($this->queuedBytes, $this->pendingOutputBytes());
}
private function drainFrames(): void
{
while (($size = $this->ffi->laylink_kcp_peeksize($this->session)) > 0) {
if ($size > self::RECEIVE_BUFFER_BYTES) {
($this->onInvalidFrame)(new \RuntimeException('kcp_message_too_large'));
$this->close();
return;
}
$buffer = $this->ffi->new("char[$size]");
$read = $this->ffi->laylink_kcp_recv($this->session, $buffer, $size);
if ($read <= 0) {
return;
}
$this->queuedBytes = max(0, $this->queuedBytes - $read);
$bytes = FFI::string($buffer, $read);
try {
foreach ($this->parser->push($bytes) as $frame) {
if ($this->paused) {
$this->pausedFrames[] = $frame;
continue;
}
($this->onFrame)($frame);
}
} catch (\Throwable $e) {
($this->onInvalidFrame)($e);
}
}
}
private function drainOutput(): void
{
while (($size = $this->ffi->laylink_kcp_pending_output_size($this->session)) > 0) {
$buffer = $this->ffi->new("char[$size]");
$read = $this->ffi->laylink_kcp_pop_output($this->session, $buffer, $size);
if ($read <= 0) {
return;
}
($this->sendPacket)(FFI::string($buffer, $read));
}
}
private function pendingOutputBytes(): int
{
$size = $this->ffi->laylink_kcp_pending_output_size($this->session);
return $size > 0 ? $size : 0;
}
private function nowMs(): int
{
return (int)floor(microtime(true) * 1000);
}
}
src/Transport/TcpFrameClientTransport.php
+85
View File
@@ -0,0 +1,85 @@
<?php
declare(strict_types=1);
namespace LayLink\Transport;
use LayLink\Protocol\Frame;
use LayLink\Protocol\FrameCodec;
use LayLink\Protocol\FrameParser;
use Workerman\Connection\AsyncTcpConnection;
final class TcpFrameClientTransport implements FrameClientTransport
{
private ?AsyncTcpConnection $connection = null;
private FrameParser $parser;
public function __construct(
private readonly string $address,
private readonly int $maxSendBuffer,
private readonly \Closure $onConnect,
private readonly \Closure $onFrame,
private readonly \Closure $onClose,
private readonly \Closure $onBufferDrain,
private readonly \Closure $onInvalidFrame,
) {
$this->parser = new FrameParser();
}
public function connect(): void
{
$this->parser = new FrameParser();
$connection = new AsyncTcpConnection($this->address);
$connection->maxSendBufferSize = $this->maxSendBuffer;
$this->connection = $connection;
$connection->onConnect = function () use ($connection): void {
($this->onConnect)($this);
};
$connection->onMessage = function (AsyncTcpConnection $connection, string $data): void {
try {
foreach ($this->parser->push($data) as $frame) {
($this->onFrame)($this, $frame);
}
} catch (\Throwable $e) {
($this->onInvalidFrame)($e);
$connection->close();
}
};
$connection->onBufferDrain = function () use ($connection): void {
($this->onBufferDrain)($this);
};
$connection->onClose = function () use ($connection): void {
if ($this->connection === $connection) {
$this->connection = null;
}
($this->onClose)($this);
};
$connection->connect();
}
public function send(Frame $frame): bool|null
{
return $this->connection?->send(FrameCodec::encode($frame));
}
public function close(): void
{
$this->connection?->close();
}
public function pauseRecv(): void
{
$this->connection?->pauseRecv();
}
public function resumeRecv(): void
{
$this->connection?->resumeRecv();
}
public function getSendBufferQueueSize(): int
{
return $this->connection?->getSendBufferQueueSize() ?? 0;
}
}
src/Transport/TcpFrameServerConnection.php
+63
View File
@@ -0,0 +1,63 @@
<?php
declare(strict_types=1);
namespace LayLink\Transport;
use LayLink\Protocol\Frame;
use LayLink\Protocol\FrameCodec;
use LayLink\Protocol\FrameParser;
use Workerman\Connection\TcpConnection;
final class TcpFrameServerConnection implements FrameServerConnection
{
private FrameParser $parser;
public function __construct(private readonly TcpConnection $connection)
{
$this->parser = new FrameParser();
}
public function id(): int
{
return $this->connection->id;
}
/**
* @return Frame[]
*/
public function push(string $data): array
{
return $this->parser->push($data);
}
public function send(Frame $frame): bool|null
{
return $this->connection->send(FrameCodec::encode($frame));
}
public function close(): void
{
$this->connection->close();
}
public function pauseRecv(): void
{
$this->connection->pauseRecv();
}
public function resumeRecv(): void
{
$this->connection->resumeRecv();
}
public function getSendBufferQueueSize(): int
{
return $this->connection->getSendBufferQueueSize();
}
public function getRemoteIp(): string
{
return $this->connection->getRemoteIp();
}
}
src/Transport/TcpFrameServerListener.php
+66
View File
@@ -0,0 +1,66 @@
<?php
declare(strict_types=1);
namespace LayLink\Transport;
use Workerman\Connection\TcpConnection;
use Workerman\Worker;
final class TcpFrameServerListener
{
/** @var array<int, TcpFrameServerConnection> */
private array $connections = [];
public function __construct(
Worker $worker,
private readonly int $maxSendBuffer,
private readonly \Closure $onConnect,
private readonly \Closure $onFrame,
private readonly \Closure $onClose,
private readonly \Closure $onBufferDrain,
private readonly \Closure $onInvalidFrame,
) {
$worker->onConnect = fn (TcpConnection $connection) => $this->handleConnect($connection);
$worker->onMessage = fn (TcpConnection $connection, string $data) => $this->handleMessage($connection, $data);
$worker->onClose = fn (TcpConnection $connection) => $this->handleClose($connection);
}
private function handleConnect(TcpConnection $connection): void
{
$connection->maxSendBufferSize = $this->maxSendBuffer;
$wrapped = new TcpFrameServerConnection($connection);
$this->connections[$connection->id] = $wrapped;
$connection->onBufferDrain = function () use ($wrapped): void {
($this->onBufferDrain)($wrapped);
};
($this->onConnect)($wrapped);
}
private function handleMessage(TcpConnection $connection, string $data): void
{
$wrapped = $this->connections[$connection->id] ?? null;
if ($wrapped === null) {
return;
}
try {
foreach ($wrapped->push($data) as $frame) {
($this->onFrame)($wrapped, $frame);
}
} catch (\Throwable $e) {
($this->onInvalidFrame)($wrapped, $e);
}
}
private function handleClose(TcpConnection $connection): void
{
$wrapped = $this->connections[$connection->id] ?? null;
unset($this->connections[$connection->id]);
if ($wrapped === null) {
return;
}
($this->onClose)($wrapped);
}
}
src/Util/Env.php
+11
View File
@@ -74,6 +74,17 @@ final class Env
return $default;
}
$value = trim($value);
if (str_starts_with($value, '[')) {
$decoded = json_decode($value, true);
if (is_array($decoded)) {
return array_values(array_filter(array_map(
static fn (mixed $item): string => strtolower(trim((string)$item)),
$decoded,
)));
}
}
return array_values(array_filter(array_map(
static fn (string $item): string => strtolower(trim($item)),
explode(',', $value),