Added __invoke() to SignalTrap and fixed phpunit xml

README.md

@@ -5,13 +5,14 @@ This is a one-size-fits-all IPC library to facilitate communication between
 threads and processes. 
 
 For complete examples, see the `examples` directory in the source tree.
+
 ## Signals
 
 Asynchronous signals are automatically enabled if supported. Otherwise, the
 `pcntl_signal_dispatch()` method must be frequently called from your main loop.
-You can test for this using the `ASYNC_SIGNALS` constant:
+You can test for this using the `SIGNALS_ASYNC` constant:
 
-    if (!ASYNC_SIGNALS) {
+    if (!SIGNALS_ASYNC) {
         pcntl_signal_dispatch();
     }
 
@@ -85,8 +86,31 @@ this identifier, and thus point to another segment, just clone it.
 
 ### Semaphores
 
+Semaphores are created with a key and a max count.
+
+    $key = new FileKey(__FILE__);
+    $sem = new Semaphore($key, 2);
+
+    $sem->allocate(); // -> true
+    $sem->allocate(); // -> true
+    $sem->allocate(); // -> false
+    $sem->release();
+
+    $mutex->destroy();
+
 ### Mutexes
 
+A mutex is a semaphore with a max count of 1.
+
+    $key = new FileKey(__FILE__);
+    $mutex = new Mutex($key);
+
+    $mutex->allocate(); // -> true
+    $mutex->allocate(); // -> false
+    $mutex->release();
+
+    $mutex->destroy();
+
 ### Message Queues
 
     $key = new FileKey(__FILE__);
@@ -131,3 +155,15 @@ or through the `createPair()` factory method.
     [ $ch1, $ch2 ] = StreamChannel::createPair();
     $ch1->send($data);
     $rcvd = $ch2->receive();
+
+### MultiChannels
+
+The `MultiStreamChannel` lets you connect multiple clients to a single master channel.
+
+    $master = new MultiStreamChannel();
+
+    $ch1 = $master->createClient();
+    $ch2 = $master->createClient();
+    $ch1->send($data);
+    $rcvd = $ch2->receive();
+

examples/multichannels.php

@@ -0,0 +1,39 @@
+<?php
+
+require_once __DIR__."/../vendor/autoload.php";
+
+use NoccyLabs\Ipc\Interop\Channel\MultiStreamChannel;
+
+// It really is this simple
+$channel = new MultiStreamChannel();
+
+// Create a child to report back in 3 seconds
+$channel1 = $channel->createClient();
+$pid1 = pcntl_fork();
+if ($pid1 === 0) {
+    sleep(1);
+    echo "child 1 received ".$channel1->receive()."\n";
+    $channel1->send("Hello from child 1");
+    sleep(5);
+    exit;
+}
+
+// Create another child to report back in 4 seconds
+$channel2 = $channel->createClient();
+$pid2 = pcntl_fork();
+if ($pid2 === 0) {
+    sleep(1);
+    echo "child 2 received ".$channel2->receive()."\n";
+    $channel2->send("Hello from child 2");
+    sleep(5);
+    exit;
+}
+
+// Writing to the master works whether any clients have been added
+$channel->send("Hello from master");
+
+// Wait for 5 seconds and dump whatever messages are waiting
+sleep(5);
+while ($msg = $channel->receive()) {
+    echo "master received ".$msg."\n";
+}

examples/semaphores.php

@@ -0,0 +1,21 @@
+<?php
+
+require_once __DIR__."/../vendor/autoload.php";
+
+use NoccyLabs\Ipc\Key\FileKey;
+use NoccyLabs\Ipc\Sem\Semaphore;
+
+$key = new FileKey(__FILE__);
+
+// Create semaphore with max count of 2
+$sem1 = new Semaphore($key, 2); 
+$sem2 = new Semaphore($key, 2); 
+$sem3 = new Semaphore($key, 2); 
+
+printf("sem1 acquire: %d\n", $sem1->acquire());
+printf("sem2 acquire: %d\n", $sem2->acquire());
+printf("sem3 acquire: %d\n", $sem3->acquire());
+printf("sem2 release: %d\n", $sem2->release());
+printf("sem3 acquire: %d\n", $sem3->acquire());
+
+$sem1->destroy();

examples/signaltraps.php

@@ -9,8 +9,15 @@ use NoccyLabs\Ipc\Signal\SignalTrap;
 $trap = new SignalTrap(SIGINT);
 echo "Press ctrl-c...\n";
 
-while (!$trap->isTrapped()) {
+while (!$trap->isTrapped(true)) {
     usleep(10000);
 }
 
-echo "Thanks!\n";
+echo "Thanks!\n";
+echo "And once more...\n";
+
+while (!$trap(true)) {
+    usleep(10000);
+}
+
+echo "Thanks!\n";

phpunit.xml

@@ -8,7 +8,7 @@
          beStrictAboutTestsThatDoNotTestAnything="true"
          beStrictAboutTodoAnnotatedTests="true"
          verbose="true">
-    <testsuite>
+    <testsuite name="default">
         <directory suffix="Test.php">tests</directory>
     </testsuite>
 

src/Interop/Async/Timer.php

@@ -23,7 +23,6 @@ class Timer
     public function __construct(callable $callback)
     {
         $this->callback = $callback;
-        $this->seconds = $seconds;
         self::registerTimer($this->callback);
     }
 

src/Interop/Channel/ChannelInterface.php

@@ -5,9 +5,25 @@ namespace NoccyLabs\Ipc\Interop\Channel;
 
 interface ChannelInterface
 {
+    /**
+     * Check if the channel is open
+     *
+     * @return boolean
+     */
     public function isOpen():bool;
 
+    /**
+     * Receive (and unserialize) a frame of data.
+     *
+     * @return mixed
+     */
     public function receive();
 
-    public function send($data);
+    /**
+     * Send a frame of data, returns true on success
+     *
+     * @param mixed $data
+     * @return bool
+     */
+    public function send($data):bool;
 }

src/Interop/Channel/MultiStreamChannel.php

@@ -0,0 +1,70 @@
+<?php
+
+namespace NoccyLabs\Ipc\Interop\Channel;
+
+/**
+ * The MultiChannel is used to create a single channel and allow other threads
+ * to each grab an end of it. Think for example a web server that could write
+ * log output to a channel (as it´the MultiStreamChannel is always open); one
+ * thread could grab a channel to dump it to the console, and another thread
+ * could write to a logfile.
+ */
+class MultiStreamChannel implements ChannelInterface
+{
+
+    protected $clients = [];
+
+    /**
+     * Create a new client to receive data sent to the channel
+     *
+     * @return StreamChannel
+     */
+    public function createClient(): StreamChannel
+    {
+        [ $a, $b ] = StreamChannel::createPair();
+        $this->clients[] = $a;
+        return $b;
+    }
+
+    /**
+     * {@inheritDoc}
+     *
+     * @return boolean
+     */
+    public function isOpen(): bool
+    {
+        return true;
+    }
+
+    /**
+     * {@inheritDoc}
+     *
+     * @param mixed $data
+     * @return bool
+     */
+    public function send($data):bool
+    {
+        foreach ($this->clients as $i=>$client) {
+            if (false === $client->send($data)) {
+                unset($this->clients[$i]);
+            }
+        }
+        return true;
+    }
+
+    /**
+     * {@inheritDoc}
+     *
+     * @return mixed
+     */
+    public function receive()
+    {
+        foreach ($this->clients as $client) {
+            if ($read = $client->receive()) {
+                return $read;
+            }
+        }
+        return false;
+    }
+
+}

src/Interop/Channel/StreamChannel.php

@@ -2,11 +2,19 @@
 
 namespace NoccyLabs\Ipc\Interop\Channel;
 
-
+/**
+ * Channel based on streams
+ * 
+ */
 class StreamChannel implements ChannelInterface
 {
     protected $stream;
 
+    /**
+     * Constructor
+     *
+     * @param resource|string $stream
+     */
     public function __construct($stream)
     {
         if (!is_resource($stream)) {
@@ -21,26 +29,55 @@ class StreamChannel implements ChannelInterface
         
         $this->stream = $stream;
 
-        $this->isOpen();
+        stream_set_blocking($stream, false);
     }
     
-    public function send($data)
+    /**
+     * {@inheritDoc}
+     *
+     * @param mixed $data
+     * @return bool
+     */
+    public function send($data):bool
     {
-        fwrite($this->stream, json_encode($data)."\0");
+        $json = json_encode($data) . "\0";
+        $ret = @fwrite($this->stream, $json, strlen($json));
+        if (false === $ret) {
+            return false;
+        }
+        return true;
     }
 
+    /**
+     * {@inheritDoc}
+     *
+     * @return mixed
+     */
     public function receive()
     {
         $buf = fread($this->stream, 8192);
+        if (!$buf) {
+            return false;
+        }
 
-        return json_decode(rtrim($buf, "\0"));
+        return json_decode(rtrim($buf, "\0"), true);
     }
 
+    /**
+     * {@inheritDoc}
+     *
+     * @return boolean
+     */
     public function isOpen(): bool
     {
         return is_resource($this->stream);
     }
 
+    /**
+     * Create a pair of connected StreamChannel instances
+     *
+     * @return StreamChannel[]
+     */
     public static function createPair()
     {
         $fd = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);

src/Key/FileKey.php

@@ -15,20 +15,10 @@ class FileKey implements KeyInterface
      * @param string $proj
      * @param boolean $create
      */
-    public function __construct(string $pathname, $proj="\0", bool $create=false)
+    public function __construct(string $pathname, $proj="\0", bool $create = false)
     {
-        if (!file_exists($pathname)) {
-            if (!$create) {
-                throw new \RuntimeException("Path does not exist: {$pathname}");
-            }
-            if (!is_dir(dirname($pathname))) {
-                throw new \RuntimeException("The directory ".dirname($pathname)." does not exist");
-            }
-            touch($pathname);
-        }
-
-        $this->pathname = $pathname;
-        $this->proj = substr($proj,0,1);
+        $this->setPathname($pathname, $create);
+        $this->setProjectIdentifier($proj);
     }
 
     /**
@@ -41,6 +31,20 @@ class FileKey implements KeyInterface
         return $this->pathname;
     }
 
+    public function setPathname(string $pathname, bool $create = false)
+    {
+        if (!file_exists($pathname)) {
+            if (!$create) {
+                throw new \RuntimeException("Path does not exist: {$pathname}");
+            }
+            if (!is_dir(dirname($pathname))) {
+                throw new \RuntimeException("The directory ".dirname($pathname)." does not exist");
+            }
+            touch($pathname);
+        }
+        $this->pathname = $pathname;        
+    }
+
     /**
      * Get the project identifier. Not guaranteed to be printable
      *
@@ -51,6 +55,17 @@ class FileKey implements KeyInterface
         return $this->proj;
     }
 
+    /**
+     * Set the project identifier
+     *
+     * @param string $proj
+     * @return void
+     */
+    public function setProjectIdentifier(string $proj)
+    {
+        $this->proj = substr($proj, 0, 1);
+    }
+
     /**
      * Get the key value
      *

src/Key/KeyInterface.php

@@ -4,6 +4,10 @@ namespace NoccyLabs\Ipc\Key;
 
 interface KeyInterface
 {
+    /**
+     * Get the integer key used for SysV ipc operations
+     *
+     * @return integer
+     */
     public function getKey():int;
-    public function __toString();
 }

src/Msg/Queue.php

@@ -32,6 +32,11 @@ class Queue
 
     }
 
+    /**
+     * Destroy the queue
+     *
+     * @return void
+     */
     public function destroy()
     {
         if ($this->resource) {

src/Sem/Semaphore.php

@@ -20,9 +20,9 @@ class Semaphore
         sem_remove($this->resource);
     }
 
-    public function acquire(float $timeout = 0):bool
+    public function acquire(bool $wait = false):bool
     {
-        return sem_acquire($this->resource, true);
+        return sem_acquire($this->resource, !$wait);
     }
 
     public function release():bool

src/Signal/Signal.php

@@ -7,17 +7,34 @@ class Signal
 {
     private $signo;
 
+    /**
+     * Constructor
+     *
+     * @param integer $signo
+     */
     public function __construct(int $signo)
     {
         $this->signo = $signo;
     }
 
+    /**
+     * Set a signal handler, overwriting any previous handler
+     *
+     * @param callable $handler
+     * @return void
+     */
     public function setHandler(callable $handler):void
     {
         pcntl_signal($this->signo, $handler);
     }
 
-    public function dispatch($pid=null):bool
+    /**
+     * Dispatch the signal to the specified pid. Default is own pid
+     *
+     * @param int $pid
+     * @return boolean
+     */ 
+    public function dispatch(int $pid=null):bool
     {
         return posix_kill($pid?:posix_getpid(), $this->signo);
     }

src/Signal/SignalHandler.php

@@ -9,6 +9,13 @@ class SignalHandler
 
     private $callbacks;
 
+    /**
+     * Constructor, should only be called once for each signal but can have
+     * multiple handlers attached.
+     *
+     * @param integer $signo
+     * @param array $callbacks
+     */
     public function __construct(int $signo, array $callbacks=[])
     {
         $this->signo = $signo;
@@ -21,11 +28,23 @@ class SignalHandler
 
     }
 
+    /**
+     * Append a handler to the signal
+     *
+     * @param callable $handler
+     * @return void
+     */
     public function addHandler(callable $handler):void
     {
         $this->callbacks[] = $handler;
     }
 
+    /**
+     * Callback for signals
+     *
+     * @param int $signo
+     * @return void
+     */
     public function onSignal($signo)
     {
         foreach ($this->callbacks as $callback) {

src/Signal/SignalTrap.php

@@ -9,17 +9,33 @@ class SignalTrap
 
     protected $trapped = false;
 
+    /**
+     * Constructor
+     *
+     * @param integer $signo
+     */
     public function __construct(int $signo)
     {
         $this->signal = new Signal($signo);
         $this->signal->setHandler([ $this, "onSignal" ]);
     }
 
+    /**
+     * Signal handler callback
+     *
+     * @return void
+     */
     public function onSignal()
     {
         $this->trapped = true;
     }
 
+    /**
+     * Check if the signal has been received
+     *
+     * @param boolean $reset
+     * @return boolean
+     */
     public function isTrapped($reset=true):bool
     {
         if ($this->trapped) {
@@ -29,4 +45,9 @@ class SignalTrap
         return false;
     }
 
-}
+    public function __invoke($reset=true):bool
+    {
+        return $this->isTrapped($reset);
+    }
+
+}

tests/Interop/Channel/MultiStreamChannelTest.php

@@ -0,0 +1,44 @@
+<?php
+
+namespace NoccyLabs\Ipc\Interop\Channel;
+
+
+
+class MultiStreamChannelTest extends \PhpUnit\Framework\TestCase
+{
+
+    public function testCreatingMultipleClients()
+    {
+        $master = new MultiStreamChannel();
+
+        $client1 = $master->createClient();
+        $client2 = $master->createClient();
+        $client3 = $master->createClient();
+
+        $master->send("Hello World");
+
+        $this->assertEquals("Hello World", $client1->receive());
+        $this->assertEquals("Hello World", $client2->receive());
+        $this->assertEquals("Hello World", $client3->receive());
+
+    }
+
+    public function testReadingFromMultipleClients()
+    {
+        $master = new MultiStreamChannel();
+
+        $client1 = $master->createClient();
+        $client2 = $master->createClient();
+        $client3 = $master->createClient();
+
+        $client1->send("a");
+        $client2->send("b");
+        $client3->send("c");
+
+        $this->assertEquals("a", $master->receive());
+        $this->assertEquals("b", $master->receive());
+        $this->assertEquals("c", $master->receive());
+
+    }
+
+}