Add option to measure client latency

composer.json

@@ -28,8 +28,7 @@
         }
     }
   , "require": {
-        "php": "^7.2"
-      , "ratchet/rfc6455": "^0.2"
+      "ratchet/rfc6455": "^0.3"
       , "react/socket": "^1.0 || ^0.8 || ^0.7 || ^0.6 || ^0.5"
       , "guzzlehttp/psr7": "^1.0"
       , "symfony/http-foundation": "^2.6|^3.0|^4.0|^5.0"

src/Ratchet/Session/SessionProvider.php

@@ -46,24 +46,20 @@ class SessionProvider implements HttpServerInterface {
      * @param array<string, mixed> $options
      * @throws \RuntimeException
      */
-    public function __construct(HttpServerInterface $app, \SessionHandlerInterface $handler, array $options = array(), HandlerInterface $serializer = null, ?OptionsHandlerInterface $optionsHandler = null) {
+    public function __construct(HttpServerInterface $app, \SessionHandlerInterface $handler, array $options = array(), ?HandlerInterface $serializer = null, ?OptionsHandlerInterface $optionsHandler = null) {
         $this->_app           = $app;
         $this->_handler       = $handler;
         $this->_null          = new NullSessionHandler;
-        $this->optionsHandler = $optionsHandler;
+        $this->optionsHandler = $optionsHandler ?? new IniOptionsHandler();
 
-        if($optionsHandler === null){
-            $optionsHandler = new IniOptionsHandler();
-        }
-
-        $optionsHandler->set('session.auto_start', 0);
-        $optionsHandler->set('session.cache_limiter', '');
-        $optionsHandler->set('session.use_cookies', 0);
+        $this->optionsHandler->set('session.auto_start', 0);
+        $this->optionsHandler->set('session.cache_limiter', '');
+        $this->optionsHandler->set('session.use_cookies', 0);
 
         $this->setOptions($options);
 
         if (null === $serializer) {
-            $serialClass = __NAMESPACE__ . "\\Serialize\\{$this->toClassCase($optionsHandler->get('session.serialize_handler'))}Handler"; // awesome/terrible hack, eh?
+            $serialClass = __NAMESPACE__ . "\\Serialize\\{$this->toClassCase($this->optionsHandler->get('session.serialize_handler'))}Handler"; // awesome/terrible hack, eh?
             if (!class_exists($serialClass)) {
                 throw new \RuntimeException('Unable to parse session serialize handler');
             }
@@ -118,7 +114,7 @@ function onMessage(ConnectionInterface $from, $msg) {
      */
     function onClose(ConnectionInterface $conn) {
         // "close" session for Connection
-
+        $this->_handler->close();
         return $this->_app->onClose($conn);
     }
 

src/Ratchet/WebSocket/WsConnection.php

@@ -9,6 +9,21 @@
  * @property \stdClass $WebSocket
  */
 class WsConnection extends AbstractConnectionDecorator {
+    /**
+     * @var int
+     */
+    public $pingSendTime = 0;
+
+    /**
+     * @var sting
+     */
+    public $lastPingPayload = '';
+
+    /**
+     * @var float
+     */
+    public $latency = NAN;
+
     /**
      * {@inheritdoc}
      */

src/Ratchet/WebSocket/WsServer.php

@@ -56,6 +56,11 @@ class WsServer implements HttpServerInterface {
      */
     private $pongReceiver;
 
+    /**
+     * @var \Closure
+     */
+    private $pongReceiver2;
+
     /**
      * @var \Closure
      */
@@ -94,6 +99,7 @@ public function __construct(ComponentInterface $component) {
         }
 
         $this->pongReceiver = function() {};
+        $this->pongReceiver2 = function() {};
 
         $reusableUnderflowException = new \UnderflowException;
         $this->ueFlowFactory = function() use ($reusableUnderflowException) {
@@ -127,12 +133,12 @@ public function onOpen(ConnectionInterface $conn, RequestInterface $request = nu
         $streamer = new MessageBuffer(
             $this->closeFrameChecker,
             function(MessageInterface $msg) use ($wsConn) {
-                $cb = $this->msgCb;
-                $cb($wsConn, $msg);
-            },
+            $cb = $this->msgCb;
+            $cb($wsConn, $msg);
+        },
             function(FrameInterface $frame) use ($wsConn) {
-                $this->onControlFrame($frame, $wsConn);
-            },
+            $this->onControlFrame($frame, $wsConn);
+        },
             true,
             $this->ueFlowFactory
         );
@@ -187,7 +193,9 @@ public function onControlFrame(FrameInterface $frame, WsConnection $conn) {
             case Frame::OP_PONG:
                 $pongReceiver = $this->pongReceiver;
                 $pongReceiver($frame, $conn);
-            break;
+                $pongReceiver2 = $this->pongReceiver2;
+                $pongReceiver2($frame, $conn);
+                break;
         }
     }
 
@@ -221,5 +229,30 @@ public function enableKeepAlive(LoopInterface $loop, $interval = 30) {
                 $pingedConnections->attach($wsConn);
             }
         });
-   }
-}
+    }
+
+    /**
+     * sends pings periodically to each open connection (therefore also keeping the connections alive) and measures
+     * the response time to calculate the clients latency. Latency defaults to NAN while no pong is received.
+     * @param LoopInterface $loop
+     * @param type $updateInterval number of seconds between latency updates
+     * @param type $numBytes number of bytes to send with each ping
+     */
+    public function enableLatencyMeasuring(LoopInterface $loop, $updateInterval = 5, $numBytes = 8)
+    {
+        $this->pongReceiver2 = function(FrameInterface $frame, $wsConn) {
+            if ($frame->getPayload() === $wsConn->lastPingPayload) {
+                $wsConn->latency = microtime(true) - $wsConn->pingSendTime;
+            }
+        };
+        $loop->addPeriodicTimer((int)$updateInterval, function() use($numBytes) {
+            foreach ($this->connections as $conn) {
+                $wsConn = $this->connections[$conn]->connection;
+                $wsConn->pingSendTime = microtime(true);
+                $pingPayload = random_bytes($numBytes);
+                $wsConn->lastPingPayload = $pingPayload;
+                $wsConn->send(new Frame($pingPayload, true, Frame::OP_PING));
+            }
+        });
+    }
+}