*/ private array $waitingByKey = []; /** @var \Closure(): void */ private readonly \Closure $accept; private bool $queued = false; /** @var LocalCache */ private LocalCache $clientsByKey; /** * @param float $keyReceiveTimeout Timeout to receive the key on accepted connections. * @param positive-int $keyLength Length of the random key exchanged on the IPC channel when connecting. */ public function __construct( private readonly Socket\ServerSocket $server, float $keyReceiveTimeout = self::DEFAULT_KEY_RECEIVE_TIMEOUT, private readonly int $keyLength = self::DEFAULT_KEY_LENGTH, ) { $address = $this->server->getAddress(); $this->uri = match ($address->getType()) { SocketAddressType::Unix => 'unix://' . $address->toString(), SocketAddressType::Internet => 'tcp://' . $address->toString(), }; $this->clientsByKey = new LocalCache(1024, $keyReceiveTimeout); $queued = &$this->queued; $waitingByKey = &$this->waitingByKey; $clientsByKey = &$this->clientsByKey; $this->accept = static function () use ( &$queued, &$waitingByKey, &$clientsByKey, $server, $keyReceiveTimeout, $keyLength, ): void { while ($waitingByKey) { $client = $server->accept(); if (!$client) { $queued = false; $exception = new Socket\SocketException('IPC socket closed before the client connected'); foreach ($waitingByKey as $deferred) { $deferred->error($exception); } return; } try { $received = readKey($client, new TimeoutCancellation($keyReceiveTimeout), $keyLength); } catch (\Throwable) { $client->close(); continue; // Ignore possible foreign connection attempt. } if (isset($waitingByKey[$received])) { $waitingByKey[$received]->complete($client); unset($waitingByKey[$received]); } else { $clientsByKey->set($received, $client); } } $queued = false; }; } public function __destruct() { $this->close(); } public function isClosed(): bool { return $this->server->isClosed(); } public function close(): void { $this->server->close(); if (!$this->waitingByKey) { return; } $exception = new Socket\SocketException('IPC socket closed before the client connected'); foreach ($this->waitingByKey as $deferred) { $deferred->error($exception); } } public function onClose(\Closure $onClose): void { $this->server->onClose($onClose); } public function getUri(): string { return $this->uri; } public function generateKey(): string { return \random_bytes($this->keyLength); } /** * @param string $key A key generated by {@see generateKey()}. */ public function accept(string $key, ?Cancellation $cancellation = null): ResourceSocket { if ($this->server->isClosed()) { throw new Socket\SocketException('The IPC server has been closed'); } if (\strlen($key) !== $this->keyLength) { throw new \ValueError(\sprintf( "Key provided is of length %d, expected %d", \strlen($key), $this->keyLength, )); } if (isset($this->waitingByKey[$key])) { throw new \Error("An accept is already pending for the given key"); } $client = $this->clientsByKey->get($key); if ($client !== null) { $this->clientsByKey->delete($key); return $client; } if (!$this->queued) { EventLoop::queue($this->accept); $this->queued = true; } $cancellation ??= new NullCancellation(); $this->waitingByKey[$key] = $deferred = new DeferredFuture(); $waitingByKey = &$this->waitingByKey; $cancellationId = $cancellation->subscribe(static function () use (&$waitingByKey, $key): void { unset($waitingByKey[$key]); }); try { $client = $deferred->getFuture()->await($cancellation); } finally { $cancellation->unsubscribe($cancellationId); } return $client; } }