Move IPC hub to public API

amphp · Feb 15, 2020 · f8a0790 · f8a0790
1 parent 8b74bb3
commit f8a0790
Show file tree

Hide file tree

Showing 7 changed files with 250 additions and 191 deletions.
diff --git a/composer.json b/composer.json
@@ -26,6 +26,7 @@
         "amphp/byte-stream": "^1.6.1",
         "amphp/parser": "^1",
         "amphp/process": "^1",
+        "amphp/socket": "^1",
         "amphp/sync": "^1.0.1"
     },
     "require-dev": {

diff --git a/lib/Context/Internal/ParallelHub.php b/lib/Context/Internal/ParallelHub.php
@@ -3,62 +3,63 @@
 namespace Amp\Parallel\Context\Internal;
 
 use Amp\Loop;
-use Amp\Parallel\Sync\ChannelledSocket;
+use Amp\Parallel\Context\IpcHub;
+use Amp\Socket\Socket;
 use parallel\Events;
 use parallel\Future;
 
-class ParallelHub extends ProcessHub
+class ParallelHub extends IpcHub
 {
     const EXIT_CHECK_FREQUENCY = 250;
 
-    /** @var ChannelledSocket[] */
-    private $channels;
+    /** @var Socket[] */
+    private $sockets;
 
     /** @var string */
     private $watcher;
 
     /** @var Events */
     private $events;
 
-    public function __construct()
+    public function __construct(int $keyLength = 32)
     {
-        parent::__construct();
+        parent::__construct($keyLength);
 
         $events = $this->events = new Events;
         $this->events->setBlocking(false);
 
-        $channels = &$this->channels;
-        $this->watcher = Loop::repeat(self::EXIT_CHECK_FREQUENCY, static function () use (&$channels, $events): void {
+        $sockets = &$this->sockets;
+        $this->watcher = Loop::repeat(self::EXIT_CHECK_FREQUENCY, static function () use (&$sockets, $events): void {
             while ($event = $events->poll()) {
                 $id = (int) $event->source;
-                \assert(isset($channels[$id]), 'Channel for context ID not found');
-                $channel = $channels[$id];
-                unset($channels[$id]);
-                $channel->close();
+                \assert(isset($sockets[$id]), 'Channel for context ID not found');
+                $socket = $sockets[$id];
+                unset($sockets[$id]);
+                $socket->close();
             }
         });
         Loop::disable($this->watcher);
         Loop::unreference($this->watcher);
     }
 
-    public function add(int $id, ChannelledSocket $channel, Future $future): void
+    final public function add(int $id, Socket $socket, Future $future): void
     {
-        $this->channels[$id] = $channel;
+        $this->sockets[$id] = $socket;
         $this->events->addFuture((string) $id, $future);
 
         Loop::enable($this->watcher);
     }
 
-    public function remove(int $id): void
+    final public function remove(int $id): void
     {
-        if (!isset($this->channels[$id])) {
+        if (!isset($this->sockets[$id])) {
             return;
         }
 
-        unset($this->channels[$id]);
+        unset($this->sockets[$id]);
         $this->events->remove((string) $id);
 
-        if (empty($this->channels)) {
+        if (empty($this->sockets)) {
             Loop::disable($this->watcher);
         }
     }

diff --git a/lib/Context/Internal/ProcessHub.php b/lib/Context/Internal/ProcessHub.php
diff --git a/lib/Context/Internal/process-runner.php b/lib/Context/Internal/process-runner.php
@@ -2,6 +2,7 @@
 
 namespace Amp\Parallel\Context\Internal;
 
+use Amp\Parallel\Context\IpcHub;
 use Amp\Parallel\Context\Process;
 use Amp\Parallel\Sync;
 use Amp\Promise;
@@ -61,20 +62,15 @@
         $key .= $chunk;
     } while (\strlen($key) < Process::KEY_LENGTH);
 
-    if (!$socket = \stream_socket_client($uri, $errno, $errstr, 5, \STREAM_CLIENT_CONNECT)) {
-        \trigger_error("Could not connect to IPC socket", E_USER_ERROR);
-        exit(1);
-    }
-
-    $channel = new Sync\ChannelledSocket($socket, $socket);
-
     try {
-        Promise\wait($channel->send($key));
+        $socket = Promise\wait(IpcHub::connect($uri, $key));
     } catch (\Throwable $exception) {
         \trigger_error("Could not send key to parent", E_USER_ERROR);
         exit(1);
     }
 
+    $channel = new Sync\ChannelledStream($socket, $socket);
+
     try {
         if (!isset($argv[0])) {
             throw new \Error("No script path given");