Add advanced UnixServer

Author: andig

README.md

@@ -34,6 +34,7 @@ handle multiple concurrent connections without blocking.
   * [Advanced server usage](#advanced-server-usage)
     * [TcpServer](#tcpserver)
     * [SecureServer](#secureserver)
+    * [UnixServer](#unixserver)
     * [LimitingServer](#limitingserver)
       * [getConnections()](#getconnections)
 * [Client usage](#client-usage)
@@ -255,7 +256,8 @@ If the address can not be determined or is unknown at this time (such as
 after the socket has been closed), it MAY return a `NULL` value instead.
 
 Otherwise, it will return the full address (URI) as a string value, such
-as `tcp://127.0.0.1:8080`, `tcp://[::1]:80` or `tls://127.0.0.1:443`.
+as `tcp://127.0.0.1:8080`, `tcp://[::1]:80`, `tls://127.0.0.1:443`
+`unix://example.sock` or `unix:///path/to/example.sock`.
 Note that individual URI components are application specific and depend
 on the underlying transport protocol.
 
@@ -648,6 +650,43 @@ If you use a custom `ServerInterface` and its `connection` event does not
 meet this requirement, the `SecureServer` will emit an `error` event and
 then close the underlying connection.
 
+#### UnixServer
+
+The `UnixServer` class implements the [`ServerInterface`](#serverinterface) and
+is responsible for accepting connections on Unix domain sockets (UDS).
+
+```php
+$server = new UnixServer('/tmp/server.sock', $loop);
+```
+
+As above, the `$uri` parameter can consist of only a socket path or socket path
+prefixed by the `unix://` scheme.
+
+If the given URI appears to be valid, but listening on it fails (such as if the
+socket is already in use or the file not accessible etc.), it will throw a
+`RuntimeException`:
+
+```php
+$first = new UnixServer('/tmp/same.sock', $loop);
+
+// throws RuntimeException because socket is already in use
+$second = new UnixServer('/tmp/same.sock', $loop);
+```
+
+Whenever a client connects, it will emit a `connection` event with a connection
+instance implementing [`ConnectionInterface`](#connectioninterface):
+
+```php
+$server->on('connection', function (ConnectionInterface $connection) {
+    echo 'New connection' . PHP_EOL;
+
+    $connection->write('hello there!' . PHP_EOL);
+    …
+});
+```
+
+See also the [`ServerInterface`](#serverinterface) for more details.
+
 #### LimitingServer
 
 The `LimitingServer` decorator wraps a given `ServerInterface` and is responsible

src/UnixServer.php

@@ -0,0 +1,130 @@
+<?php
+
+namespace React\Socket;
+
+use Evenement\EventEmitter;
+use React\EventLoop\LoopInterface;
+use InvalidArgumentException;
+use RuntimeException;
+
+/**
+ * The `UnixServer` class implements the `ServerInterface` and
+ * is responsible for accepting plaintext connections on unix domain sockets.
+ *
+ * ```php
+ * $server = new UnixServer('unix:///tmp/app.sock', $loop);
+ * ```
+ *
+ * See also the `ServerInterface` for more details.
+ *
+ * @see ServerInterface
+ * @see ConnectionInterface
+ */
+final class UnixServer extends EventEmitter implements ServerInterface
+{
+    private $master;
+    private $loop;
+    private $listening = false;
+
+    /**
+     * Creates a plaintext socket server and starts listening on the given unix socket
+     *
+     * This starts accepting new incoming connections on the given address.
+     * See also the `connection event` documented in the `ServerInterface`
+     * for more details.
+     *
+     * ```php
+     * $server = new UnixServer('unix:///tmp/app.sock', $loop);
+     * ```
+     *
+     * @param string        $path
+     * @param LoopInterface $loop
+     * @param array         $context
+     * @throws InvalidArgumentException if the listening address is invalid
+     * @throws RuntimeException if listening on this address fails (already in use etc.)
+     */
+    public function __construct($path, LoopInterface $loop, array $context = array())
+    {
+        $this->loop = $loop;
+
+        if (strpos($path, '://') === false) {
+            $path = 'unix://' . $path;
+        } elseif (substr($path, 0, 7) !== 'unix://') {
+            throw new \InvalidArgumentException('Given URI "' . $path . '" is invalid');
+        }
+
+        $this->master = @stream_socket_server(
+            $path,
+            $errno,
+            $errstr,
+            STREAM_SERVER_BIND | STREAM_SERVER_LISTEN,
+            stream_context_create(array('socket' => $context))
+        );
+        if (false === $this->master) {
+            throw new RuntimeException('Failed to listen on unix domain socket "' . $path . '": ' . $errstr, $errno);
+        }
+        stream_set_blocking($this->master, 0);
+
+        $this->resume();
+    }
+
+    public function getAddress()
+    {
+        if (!is_resource($this->master)) {
+            return null;
+        }
+
+        return 'unix://' . stream_socket_get_name($this->master, false);
+    }
+
+    public function pause()
+    {
+        if (!$this->listening) {
+            return;
+        }
+
+        $this->loop->removeReadStream($this->master);
+        $this->listening = false;
+    }
+
+    public function resume()
+    {
+        if ($this->listening || !is_resource($this->master)) {
+            return;
+        }
+
+        $that = $this;
+        $this->loop->addReadStream($this->master, function ($master) use ($that) {
+            $newSocket = @stream_socket_accept($master);
+            if (false === $newSocket) {
+                $that->emit('error', array(new \RuntimeException('Error accepting new connection')));
+
+                return;
+            }
+            $that->handleConnection($newSocket);
+        });
+        $this->listening = true;
+    }
+
+    public function close()
+    {
+        if (!is_resource($this->master)) {
+            return;
+        }
+
+        $this->pause();
+        fclose($this->master);
+        $this->removeAllListeners();
+    }
+
+    /** @internal */
+    public function handleConnection($socket)
+    {
+        $connection = new Connection($socket, $this->loop);
+        $connection->unix = true;
+
+        $this->emit('connection', array(
+            $connection
+        ));
+    }
+}