Cleaning up Server

sirn-se · sirn-se · commit 5b55be7cad53 · 2021-03-07T15:28:23.000+01:00
diff --git a/README.md b/README.md
@@ -10,9 +10,9 @@ It does not include convenience operations such as listeners and implicit error
 
 ## Documentation
 
-- [Client](docs/Client.md)
-- [Server](docs/Server.md)
-- [Message](docs/Message.md)
+- [Client overwiew](docs/Client.md)
+- [Server overview](docs/Server.md)
+- [Classes](docs/Classes/Classes.md)
 - [Examples](docs/Examples.md)
 - [Changelog](docs/Changelog.md)
 - [Contributing](docs/Contributing.md)
@@ -25,8 +25,8 @@ composer require textalk/websocket
 ```
 
 * Current version support PHP versions `^7.2|8.0`.
-* For PHP `7.1` support use version `1.4`.
-* For PHP `^5.4` and `7.0` support use version `1.3`.
+* For PHP `7.1` support use version [`1.4`](https://github.com/Textalk/websocket-php/tree/1.4.0).
+* For PHP `^5.4` and `7.0` support use version [`1.3`](https://github.com/Textalk/websocket-php/tree/1.3.0).
 
 ## Client
 
@@ -42,18 +42,17 @@ $client->close();
 
 ## Server
 
-The library contains a rudimentary single stream/single thread [server](docs/Server.md).
+The library contains a websocket [server](docs/Server.md).
 It internally supports Upgrade handshake and implicit close and ping/pong operations.
-
-Note that it does **not** support threading or automatic association ot continuous client requests.
-If you require this kind of server behavior, you need to build it on top of provided server implementation.
+Preferred operation is using the listener function, but optional operations exist.
 
 ```php
 $server = new WebSocket\Server();
-$server->accept();
-$message = $server->receive();
-$server->text($message);
-$server->close();
+$server->listen(function ($message, $connection = null) {
+    echo "Got {$message->getContent()}\n";
+    if (!$connection) return; // Current connection is closed
+    $connection->text('Sending message to client');
+});
 ```
 
 ### License and Contributors
diff --git a/docs/Changelog.md b/docs/Changelog.md
@@ -2,6 +2,16 @@
 
 # Websocket: Changelog
 
+## `v1.6`
+
+ > PHP version `^7.2`
+
+### `1.6.0`
+
+ * Listener functions (@sirn-se)
+ * Multi connection server (@sirn-se)
+ * Major refactoring, using Connections (@sirn-se)
+
 ## `v1.5`
 
  > PHP version `^7.2`
diff --git a/examples/echoserver.php b/examples/echoserver.php
@@ -32,7 +32,7 @@
     echo "> Using logger\n";
 }
 
-// Setting timeout to 200 seconds to make time for all tests and manual runs.
+// Initiate server
 try {
     $server = new Server($options);
 } catch (ConnectionException $e) {
@@ -42,21 +42,21 @@
 
 echo "> Listening to port {$server->getPort()}\n";
 
-$server->listen(function ($message, $connection) use ($server) {
+$server->listen(function ($message, $connection = null) use ($server) {
     $content = $message->getContent();
     $opcode = $message->getOpcode();
     $peer = $connection ? $connection->getPeer() : '(closed)';
     echo "> Got '{$content}' [opcode: {$opcode}, peer: {$peer}]\n";
 
     // Connection closed, can't respond
     if (!$connection) {
-        return null; // Continue listening
+        return; // Continue listening
     }
 
     if (in_array($opcode, ['ping', 'pong'])) {
         $connection->text($content);
         echo "< Sent '{$content}' [opcode: text, peer: {$peer}]\n";
-        return null; // Continue listening
+        return; // Continue listening
     }
 
     // Allow certain string to trigger server action
@@ -71,7 +71,7 @@
             echo "< Sent '{$content}' [opcode: close, peer: {$peer}]\n";
             break;
         case 'exit':
-            echo "> Client told me to quit.  Bye bye.\n";
+            echo "> Client told me to quit.\n";
             $server->close();
             return true; // Stop listener
         case 'headers':
@@ -87,6 +87,10 @@
             $connection->pong($content);
             echo "< Sent '{$content}' [opcode: pong, peer: {$peer}]\n";
             break;
+        case 'stop':
+            $server->stop();
+            echo "> Client told me to stop listening.\n";
+            break;
         default:
             $connection->text($content);
             echo "< Sent '{$content}' [opcode: text, peer: {$peer}]\n";
diff --git a/lib/Server.php b/lib/Server.php
@@ -10,6 +10,7 @@
 namespace WebSocket;
 
 use Closure;
+use Psr\Log\NullLogger;
 use Throwable;
 
 class Server extends Base
@@ -24,23 +25,31 @@ class Server extends Base
       'timeout'       => null,
     ];
 
-    protected $addr;
     protected $port;
     protected $listening;
     protected $request;
     protected $request_path;
-    private $connectors = [];
+    private $connections = [];
+    private $listen = false;
+
+
+    /* ---------- Construct & Destruct ----------------------------------------------- */
 
     /**
      * @param array $options
      *   Associative array containing:
-     *   - timeout:       Set the socket timeout in seconds.
+     *   - filter:        Array of opcodes to handle. Default: ['text', 'binary'].
      *   - fragment_size: Set framgemnt size.  Default: 4096
-     *   - port:          Chose port for listening. Default 8000.
+     *   - logger:        PSR-3 compatible logger.  Default NullLogger.
+     *   - port:          Chose port for listening.  Default 8000.
+     *   - return_obj:    If receive() function return Message instance.  Default false.
+     *   - timeout:       Set the socket timeout in seconds.
      */
     public function __construct(array $options = [])
     {
-        $this->options = array_merge(self::$default_options, $options);
+        $this->options = array_merge(self::$default_options, [
+            'logger' => new NullLogger(),
+        ], $options);
         $this->port = $this->options['port'];
         $this->setLogger($this->options['logger']);
 
@@ -65,57 +74,40 @@ public function __construct(array $options = [])
         $this->logger->info("Server listening to port {$this->port}");
     }
 
+    /**
+     * Disconnect streams on shutdown.
+     */
     public function __destruct()
     {
+/*
         if ($this->connection && $this->connection->isConnected()) {
             $this->connection->disconnect();
         }
         $this->connection = null;
-    }
-
-    public function getPort(): int
-    {
-        return $this->port;
-    }
-
-    public function getPath(): string
-    {
-        return $this->request_path;
-    }
-
-    public function getRequest(): array
-    {
-        return $this->request;
-    }
-
-    public function getHeader($header): ?string
-    {
-        foreach ($this->request as $row) {
-            if (stripos($row, $header) !== false) {
-                list($headername, $headervalue) = explode(":", $row);
-                return trim($headervalue);
+*/
+        foreach ($this->connections as $connection) {
+            if ($connection->isConnected()) {
+                $connection->disconnect();
             }
         }
-        return null;
+        $this->connections = [];
     }
 
-    public function accept(): bool
-    {
-        $this->connection = null;
-        return (bool)$this->listening;
-    }
+
+    /* ---------- Server operations -------------------------------------------------- */
 
     /**
      * Set server to listen to incoming requests.
      * @param Closure A callback function that will be called when server receives message.
      *   function (Message $message, Connection $connection = null)
-     *   If callback function returns not null value, the listener will halt and return that value.
+     *   If callback function returns non-null value, the listener will halt and return that value.
      *   Otherwise it will continue listening and propagating messages.
-     * @return Returns any not null value returned by callback function.
+     * @return mixed Returns any non-null value returned by callback function.
      */
     public function listen(Closure $callback)
     {
-        while (true) {
+        $this->listen = true;
+        while ($this->listen) {
             // Server accept
             if ($stream = @stream_socket_accept($this->listening, 0)) {
                 $peer = stream_socket_get_name($stream, true);
@@ -126,18 +118,18 @@ public function listen(Closure $callback)
                     $connection->setTimeout($this->options['timeout']);
                 }
                 $this->performHandshake($connection);
-                $this->connectors[$peer] = $connection;
+                $this->connections[$peer] = $connection;
             }
 
             // Collect streams to listen to
             $streams = array_filter(array_map(function ($connection, $peer) {
                 $stream = $connection->getStream();
                 if (is_null($stream)) {
                     $this->logger->debug("[server] Remove {$peer} from listener stack");
-                    unset($this->connectors[$peer]);
+                    unset($this->connections[$peer]);
                 }
                 return $stream;
-            }, $this->connectors, array_keys($this->connectors)));
+            }, $this->connections, array_keys($this->connections)));
 
             // Handle incoming
             if (!empty($streams)) {
@@ -149,11 +141,15 @@ public function listen(Closure $callback)
                         try {
                             $result = null;
                             $peer = stream_socket_get_name($stream, true);
-                            $connection = $this->connectors[$peer];
+                            if (empty($peer)) {
+                                $this->logger->warning("[server] Got detached stream '{$peer}'");
+                                continue;
+                            }
+                            $connection = $this->connections[$peer];
                             $this->logger->debug("[server] Handling {$peer}");
                             $message = $connection->pullMessage();
                             if (!$connection->isConnected()) {
-                                unset($this->connectors[$peer]);
+                                unset($this->connections[$peer]);
                                 $connection = null;
                             }
                             // Trigger callback according to filter
@@ -175,14 +171,102 @@ public function listen(Closure $callback)
         }
     }
 
+    /**
+     * Tell server to stop listening to incoming requests.
+     * Active connections are still available when restarting listening.
+     */
+    public function stop(): void
+    {
+        $this->listen = false;
+    }
+
+    /**
+     * Accept an incoming request.
+     * Note that this operation will block accepting additional requests.
+     * @return bool True if listening
+     * @deprecated Will be removed in future version
+     */
+    public function accept(): bool
+    {
+        $this->connection = null;
+        return (bool)$this->listening;
+    }
+
+
+    /* ---------- Server option functions -------------------------------------------- */
+
+    /**
+     * Get current port.
+     * @return int port
+     */
+    public function getPort(): int
+    {
+        return $this->port;
+    }
+
+    // Inherited from Base:
+    // - setLogger
+    // - setTimeout
+    // - setFragmentSize
+    // - getFragmentSize
+
+
+    /* ---------- Connection broadcast operations ------------------------------------ */
+
+    /**
+     * Close all connections.
+     * @param int Close status, default: 1000
+     * @param string Close message, default: 'ttfn'
+     */
     public function close(int $status = 1000, string $message = 'ttfn'): void
     {
-        parent::close($status, $message);
-        foreach ($this->connectors as $connection) {
-            $connection->close($status, $message);
+        foreach ($this->connections as $connection) {
+            if ($connection->isConnected()) {
+                $connection->close($status, $message);
+            }
+        }
+    }
+
+    // Inherited from Base:
+    // - receive
+    // - send
+    // - text, binary, ping, pong
+
+
+    /* ---------- Connection functions (all deprecated) ------------------------------ */
+
+    public function getPath(): string
+    {
+        return $this->request_path;
+    }
+
+    public function getRequest(): array
+    {
+        return $this->request;
+    }
+
+    public function getHeader($header): ?string
+    {
+        foreach ($this->request as $row) {
+            if (stripos($row, $header) !== false) {
+                list($headername, $headervalue) = explode(":", $row);
+                return trim($headervalue);
+            }
         }
+        return null;
     }
 
+    // Inherited from Base:
+    // - getLastOpcode
+    // - getCloseStatus
+    // - isConnected
+    // - disconnect
+    // - getName, getPeer, getPier
+
+
+    /* ---------- Helper functions --------------------------------------------------- */
+
+    // Connect when read/write operation is performed.
     protected function connect(): void
     {
         $error = null;
@@ -215,8 +299,10 @@ protected function connect(): void
             'pier' => $this->connection->getPeer(),
         ]);
         $this->performHandshake($this->connection);
+        $this->connections = ['*' => $this->connection];
     }
 
+    // Perform upgrade handshake on new connections.
     protected function performHandshake(Connection $connection): void
     {
         $request = '';
