Polish adapter, add cleanup of unix sockets

Author: cursedcoder

EProcess/Adapter/BaseAdapter.php

@@ -0,0 +1,41 @@
+<?php
+
+namespace EProcess\Adapter;
+
+use React\EventLoop\LoopInterface;
+
+abstract class BaseAdapter
+{
+    protected $loop;
+    protected $node;
+
+    public function __construct(LoopInterface $loop)
+    {
+        $this->loop = $loop;
+        $this->node = uniqid('thread_');
+    }
+
+    protected function createUnixSocket()
+    {
+        $unixFile = sprintf('tmp/%s.sock', $this->node);
+        $unix = sprintf('unix://%s', $unixFile);
+
+        $cleanup = function() use ($unixFile) {
+            $this->loop->stop();
+            @unlink($unixFile);
+        };
+
+        register_shutdown_function($cleanup);
+        pcntl_signal(SIGINT, $cleanup);
+
+        return $unix;
+    }
+
+    protected function getNode()
+    {
+        
+    }
+
+    abstract public function create($class, array $data = []);
+    abstract public function kill();
+}

EProcess/Adapter/ChildProcess.php

@@ -6,10 +6,9 @@
 use EProcess\MessengerFactory;
 
 use React\ChildProcess\Process;
-use React\EventLoop\LoopInterface;
 use Symfony\Component\Process\PhpExecutableFinder;
 
-class ChildProcess
+class ChildProcess extends BaseAdapter
 {
     use UniversalSerializer;
 
@@ -49,26 +48,19 @@ class ChildProcess
 
     private $loop;
     private $process;
-    private $executableFinder;
-
-    public function __construct(LoopInterface $loop)
-    {
-        $this->loop = $loop;
-        $this->executableFinder = new PhpExecutableFinder();
-    }
 
     public function create($class, array $data = [])
     {
-        if (false === $php = $this->executableFinder->find()) {
+        $executableFinder = new PhpExecutableFinder();
+
+        if (false === $php = $executableFinder->find()) {
             throw new \RuntimeException('Unable to find the PHP executable.');
         }
 
-        $node = uniqid('thread_');
-        $unix = sprintf('unix://tmp/%s.sock', $node);
-
+        $unix = $this->createUnixSocket();
         $messenger = MessengerFactory::server($unix, $this->loop);
 
-        $file = sprintf(__DIR__ . '/../../tmp/%s.php', $node);
+        $file = sprintf(__DIR__ . '/../../tmp/%s.php', $this->node);
 
         file_put_contents($file, sprintf(
             $this->script,
@@ -81,7 +73,7 @@ public function create($class, array $data = [])
         $this->process = new Process(sprintf('exec %s %s', $php, realpath($file)));
         $this->process->start($this->loop);
 
-        $this->loop->addTimer(5, function() use ($file) {
+        $this->loop->addTimer(3, function() use ($file) {
             unlink($file);
         });
 
@@ -93,10 +85,6 @@ public function create($class, array $data = [])
             echo $data;
         });
 
-        register_shutdown_function(function() use ($unix) {
-            unlink($unix);
-        });
-
         return $messenger;
     }
 

EProcess/Adapter/PThreads.php

@@ -3,27 +3,14 @@
 namespace EProcess\Adapter;
 
 use EProcess\MessengerFactory;
-use React\EventLoop\LoopInterface;
 
-class PThreads
+class PThreads extends BaseAdapter
 {
-    private $loop;
     private $process;
 
-    public function __construct(LoopInterface $loop)
-    {
-        $this->loop = $loop;
-    }
-
     public function create($class, array $data = [])
     {
-        $node = uniqid('thread_');
-        $unix = sprintf('unix://tmp/%s.sock', $node);
-
-        register_shutdown_function(function() use ($unix) {
-            unlink($unix);
-        });
-
+        $unix = $this->createUnixSocket();
         $messenger = MessengerFactory::server($unix, $this->loop);
 
         $this->process = new Thread($unix, $class, $data);

EProcess/Adapter/SymfonyProcess.php

@@ -4,10 +4,9 @@
 
 use EProcess\Behaviour\UniversalSerializer;
 use Symfony\Component\Process\PhpProcess;
-use React\EventLoop\LoopInterface;
 use EProcess\MessengerFactory;
 
-class SymfonyProcess
+class SymfonyProcess extends BaseAdapter
 {
     use UniversalSerializer;
 
@@ -48,20 +47,9 @@ class SymfonyProcess
     private $loop;
     private $process;
 
-    public function __construct(LoopInterface $loop)
-    {
-        $this->loop = $loop;
-    }
-
     public function create($class, array $data = [])
     {
-        $node = uniqid('thread_');
-        $unix = sprintf('unix://app/cache/%s.sock', $node);
-
-        register_shutdown_function(function() use ($unix) {
-            unlink($unix);
-        });
-
+        $unix = $this->createUnixSocket();
         $messenger = MessengerFactory::server($unix, $this->loop);
 
         $script = sprintf($this->script, EPROCESS_AUTOLOAD, $unix, $class, base64_encode($this->serialize($data)));

examples/autoload.php

@@ -1,5 +1,7 @@
 <?php
 
+declare(ticks = 1);
+
 define('EPROCESS_AUTOLOAD', __FILE__);
 
 $loader = require __DIR__ . '/../vendor/autoload.php';