Package split: EMessenger

Author: cursedcoder

.travis.yml

@@ -10,3 +10,5 @@ install:
 cache:
   directories:
     - vendor
+script:
+  - php examples/simple.php

EProcess/Adapter/BaseAdapter.php

@@ -2,6 +2,7 @@
 
 namespace EProcess\Adapter;
 
+use EMessenger\Transport\UnixTransport;
 use React\EventLoop\LoopInterface;
 
 abstract class BaseAdapter
@@ -20,7 +21,12 @@ public function getUnixSocketFile()
         return sprintf('%s/%s.sock', EPROCESS_SOCKET_DIR, $this->node);
     }
 
-    protected function createUnixSocket()
+    public function getUnixSocketAddress()
+    {
+        return sprintf('unix://%s', $this->getUnixSocketFile());
+    }
+
+    protected function createUnixTransport()
     {
         if (!defined('EPROCESS_SOCKET_DIR')) {
             throw new \RuntimeException('EPROCESS_SOCKET_DIR is not defined.');
@@ -34,9 +40,10 @@ protected function createUnixSocket()
             throw new \RuntimeException(sprintf('Cannot write to "%s".', EPROCESS_SOCKET_DIR));
         }
 
-        $unix = sprintf('unix://%s', $this->getUnixSocketFile());
+        echo $this->getUnixSocketFile();
+        echo PHP_EOL;
 
-        return $unix;
+        return new UnixTransport($this->loop, $this->getUnixSocketAddress());
     }
 
     abstract public function create($class, array $data = []);

EProcess/Adapter/ChildProcess.php

@@ -2,15 +2,15 @@
 
 namespace EProcess\Adapter;
 
-use EProcess\Behaviour\UniversalSerializer;
-use EProcess\MessengerFactory;
+use UniversalSerializer\UniversalSerializerTrait;
+use EMessenger\MessengerFactory;
 use EProcess\Stream\FullDrainStream;
 use React\ChildProcess\Process;
 use Symfony\Component\Process\PhpExecutableFinder;
 
 class ChildProcess extends BaseAdapter
 {
-    use UniversalSerializer;
+    use UniversalSerializerTrait;
 
     private $script = <<<PHP
 <?php
@@ -19,15 +19,15 @@ class ChildProcess extends BaseAdapter
 
 set_time_limit(0);
 
-use EProcess\MessengerFactory;
+use EMessenger\MessengerFactory;
+use EMessenger\Transport\UnixTransport;
 use EProcess\Application\ApplicationFactory;
 use React\EventLoop\Factory;
 
 \$loop = Factory::create();
 
 \$messenger = MessengerFactory::client(
-    '%s',
-    \$loop
+    new UnixTransport(\$loop, '%s')
 );
 
 \$application = ApplicationFactory::create('%s');
@@ -36,7 +36,7 @@ class ChildProcess extends BaseAdapter
 \$application->loop(\$loop);
 \$application->data(\$application->unserialize(base64_decode('%s')));
 
-\$messenger->emit('initialized', true);
+\$messenger->send('initialized', true);
 
 try {
     \$application->run();
@@ -56,13 +56,13 @@ public function create($class, array $data = [])
             throw new \RuntimeException('Unable to find the PHP executable.');
         }
 
-        $unix = $this->createUnixSocket();
-        $messenger = MessengerFactory::server($unix, $this->loop);
+        $transport = $this->createUnixTransport();
+        $messenger = MessengerFactory::server($transport);
 
         $script = sprintf(
             $this->script,
             EPROCESS_AUTOLOAD,
-            $unix,
+            $this->getUnixSocketAddress(),
             $class,
             base64_encode($this->serialize($data))
         );
@@ -77,15 +77,15 @@ public function create($class, array $data = [])
 
         $this->process->stdin->write($script);
 
-        $this->process->stdin->on('full-drain', function() {
+        $this->process->stdin->on('full-drain', function () {
             $this->process->stdin->close();
         });
 
-        $this->process->stdout->on('data', function($data) {
+        $this->process->stdout->on('data', function ($data) {
             echo $data;
         });
 
-        $this->process->stderr->on('data', function($data) {
+        $this->process->stderr->on('data', function ($data) {
             echo $data;
         });
 

EProcess/Adapter/PThreads.php

@@ -2,18 +2,18 @@
 
 namespace EProcess\Adapter;
 
-use EProcess\MessengerFactory;
+use EMessenger\MessengerFactory;
 
 class PThreads extends BaseAdapter
 {
     private $process;
 
     public function create($class, array $data = [])
     {
-        $unix = $this->createUnixSocket();
-        $messenger = MessengerFactory::server($unix, $this->loop);
+        $transport = $this->createUnixTransport();
+        $messenger = MessengerFactory::server($transport);
 
-        $this->process = new Thread($unix, $class, $data);
+        $this->process = new Thread($transport, $class, $data);
         $this->process->start(PTHREADS_INHERIT_NONE);
 
         return $messenger;

EProcess/Adapter/SymfonyProcess.php

@@ -2,13 +2,13 @@
 
 namespace EProcess\Adapter;
 
-use EProcess\Behaviour\UniversalSerializer;
 use Symfony\Component\Process\PhpProcess;
-use EProcess\MessengerFactory;
+use EMessenger\MessengerFactory;
+use UniversalSerializer\UniversalSerializerTrait;
 
 class SymfonyProcess extends BaseAdapter
 {
-    use UniversalSerializer;
+    use UniversalSerializerTrait;
 
     private $script = <<<PHP
 <?php
@@ -34,7 +34,7 @@ class SymfonyProcess extends BaseAdapter
 \$application->loop(\$loop);
 \$application->data(\$application->unserialize(base64_decode('%s')));
 
-\$messenger->emit('initialized', true);
+\$messenger->send('initialized', true);
 
 try {
     \$application->run();
@@ -48,10 +48,15 @@ class SymfonyProcess extends BaseAdapter
 
     public function create($class, array $data = [])
     {
-        $unix = $this->createUnixSocket();
-        $messenger = MessengerFactory::server($unix, $this->loop);
-
-        $script = sprintf($this->script, EPROCESS_AUTOLOAD, $unix, $class, base64_encode($this->serialize($data)));
+        $transport = $this->createUnixTransport();
+        $messenger = MessengerFactory::server($transport);
+
+        $script = sprintf(
+            $this->script,
+            EPROCESS_AUTOLOAD,
+            $transport, $class,
+            base64_encode($this->serialize($data))
+        );
 
         $this->process = new PhpProcess($script);
         $this->process->start();

EProcess/Adapter/Thread.php

@@ -32,7 +32,7 @@ public function run()
         $application->loop($loop);
         $application->data($this->data);
 
-        $messenger->emit('initialized', true);
+        $messenger->send('initialized', true);
 
         try {
             $application->run();

EProcess/Application/Application.php

@@ -2,21 +2,19 @@
 
 namespace EProcess\Application;
 
-use EProcess\Behaviour\UniversalSerializer;
+use EMessenger\Message;
+use EMessenger\Messenger;
 use EProcess\Behaviour\Workable;
-use EProcess\Message;
-use EProcess\Messenger;
 use EProcess\Worker;
 use Evenement\EventEmitterTrait;
 use MKraemer\ReactPCNTL\PCNTL;
 use React\EventLoop\LoopInterface;
+use UniversalSerializer\UniversalSerializerTrait;
 
 abstract class Application
 {
-    use EventEmitterTrait {
-        EventEmitterTrait::emit as emitterEmit;
-    }
-    use UniversalSerializer;
+    use EventEmitterTrait;
+    use UniversalSerializerTrait;
     use Workable;
 
     private $loop;
@@ -33,7 +31,7 @@ public function addWorker(Worker $worker)
     public function cleanWorkers()
     {
         foreach ($this->workers as $worker) {
-            $worker->emit('shutdown');
+            $worker->send('shutdown');
             unlink($worker->adapter()->getUnixSocketFile());
         }
     }
@@ -60,7 +58,7 @@ public function messenger(Messenger $messenger = null)
     {
         if ($messenger) {
             $messenger->on('message', function (Message $message) {
-                $this->emitterEmit($message->getEvent(), [$message->getContent()]);
+                $this->emit($message->getEvent(), [$message->getContent()]);
             });
 
             $this->messenger = $messenger;
@@ -78,9 +76,9 @@ public function data(array $data = null)
         return $this->data;
     }
 
-    public function emit($event, $data = '')
+    public function send($event, $data = '')
     {
-        $this->messenger->emit($event, $data);
+        $this->messenger->send($event, $data);
     }
 
     abstract public function run();

EProcess/Behaviour/UniversalSerializer.php

@@ -15,7 +15,7 @@ public function createWorker($fqcn, array $data = [])
             $data
         );
 
-        $this->emitterEmit('worker.created', [$worker]);
+        $this->emit('worker.created', [$worker]);
 
         return $worker;
     }