*/ abstract class AbstractContext implements Context { use ForbidCloning; use ForbidSerialization; /** @var Future>|null */ private ?Future $result = null; protected function __construct( private readonly Channel $ipcChannel, private readonly Channel $resultChannel, ) { } public function receive(?Cancellation $cancellation = null): mixed { try { $data = $this->ipcChannel->receive($cancellation); } catch (ChannelException $exception) { $this->ipcChannel->close(); throw new ContextException( "The context stopped responding, potentially due to a fatal error or calling exit", previous: $exception, ); } if (!$data instanceof ContextMessage) { if ($data instanceof ExitResult) { $data = $data->getResult(); throw new ContextException(\sprintf( 'Context unexpectedly exited when waiting to receive data with result: %s', flattenArgument($data), )); } throw new ContextException(\sprintf( 'Unexpected data type from context: %s', flattenArgument($data), )); } return $data->getMessage(); } public function send(mixed $data): void { try { $this->ipcChannel->send($data); } catch (ChannelException $exception) { $this->ipcChannel->close(); throw new ContextException( "The context stopped responding, potentially due to a fatal error or calling exit", previous: $exception, ); } } public function close(): void { $this->ipcChannel->close(); } public function isClosed(): bool { return $this->ipcChannel->isClosed(); } public function onClose(\Closure $onClose): void { $this->ipcChannel->onClose($onClose); } protected function receiveExitResult(?Cancellation $cancellation = null): ExitResult { while ($this->result) { try { $this->result->await($cancellation); } catch (CancelledException) { // Ignore cancellation from a prior join request, throw only if this request was cancelled. $cancellation?->throwIfRequested(); } } $this->result = async(function () use ($cancellation): ExitResult { try { $data = $this->resultChannel->receive($cancellation); $this->resultChannel->close(); } catch (CancelledException $exception) { throw $exception; } catch (\Throwable $exception) { $this->resultChannel->close(); throw new ContextException("Failed to receive result from context", previous: $exception); } if (!$data instanceof ExitResult) { throw new ContextException(\sprintf( "The context sent data instead of exiting: %s", flattenArgument($data), )); } return $data; }); try { return $this->result->await(); } catch (CancelledException $exception) { $this->result = null; throw $exception; } } }