[WebSocket] Separating responsibilities

Separated HandshakeNegotiator into HttpRequestParser and VersionManager
Moved WsConnection to Version specific Connection

HandshakeNegotiator.php

@@ -1,130 +0,0 @@
-<?php
-namespace Ratchet\WebSocket;
-use Ratchet\MessageInterface;
-use Ratchet\ConnectionInterface;
-use Ratchet\WebSocket\Guzzle\Http\Message\RequestFactory;
-use Ratchet\WebSocket\Version\VersionInterface;
-use Guzzle\Http\Message\RequestInterface;
-use Guzzle\Http\Message\Response;
-
-class HandshakeNegotiator implements MessageInterface {
-    const EOM = "\r\n\r\n";
-
-    /**
-     * The maximum number of bytes the request can be
-     * This is a security measure to prevent attacks
-     * @var int
-     */
-    public $maxSize = 4096;
-
-    private $versionString = '';
-
-    protected $versions = array();
-
-    /**
-     * @param WsConnection
-     */
-    public function onOpen(ConnectionInterface $conn) {
-        $conn->WebSocket->handshakeBuffer = '';
-    }
-
-    /**
-     * @param WsConnection
-     * @param string Data stream to buffer
-     * @return Guzzle\Http\Message\Response|null Response object if it's done parsing, null if there's more to be buffered
-     * @throws HttpException
-     */
-    public function onMessage(ConnectionInterface $conn, $data) {
-        $conn->WebSocket->handshakeBuffer .= $data;
-
-        if (strlen($conn->WebSocket->handshakeBuffer) >= (int)$this->maxSize) {
-            return new Response(413, array('X-Powered-By' => \Ratchet\VERSION));
-        }
-
-        if ($this->isEom($conn->WebSocket->handshakeBuffer)) {
-            $conn->WebSocket->request = RequestFactory::getInstance()->fromMessage($conn->WebSocket->handshakeBuffer);
-
-            if (null === ($version = $this->getVersion($conn->WebSocket->request))) {
-                return new Response(400, array(
-                    'Sec-WebSocket-Version' => $this->getSupportedVersionString()
-                  , 'X-Powered-By'          => \Ratchet\VERSION
-                ));
-            }
-
-            // TODO: confirm message is buffered
-            // Hixie requires the body to complete the handshake (6 characters long) - is that 6 ASCII or UTF-8 characters?
-            // Update VersionInterface to check for this, ::canHandshake() maybe
-            // return if can't, continue buffering
-
-            $response = $version->handshake($conn->WebSocket->request);
-            $response->setHeader('X-Powered-By', \Ratchet\VERSION);
-
-            // This needs to be decoupled
-            $conn->WebSocket->version = $version;
-            unset($conn->WebSocket->handshakeBuffer);
-
-            return $response;
-        }
-    }
-
-    /**
-     * Determine if the message has been buffered as per the HTTP specification
-     * @param string
-     * @return boolean
-     */
-    public function isEom($message) {
-        //return (static::EOM === substr($message, 0 - strlen(static::EOM)));
-        return (boolean)strpos($message, static::EOM);
-    }
-
-    /**
-     * Get the protocol negotiator for the request, if supported
-     * @param Guzzle\Http\Message\RequestInterface
-     * @return Ratchet\WebSocket\Version\VersionInterface
-     */
-    public function getVersion(RequestInterface $request) {
-        foreach ($this->versions as $version) {
-            if ($version->isProtocol($request)) {
-                return $version;
-            }
-        }
-    }
-
-    /**
-     * Enable support for a specific version of the WebSocket protocol
-     * @param Ratchet\WebSocket\Vesion\VersionInterface
-     * @return HandshakeNegotiator
-     */
-    public function enableVersion(VersionInterface $version) {
-        $this->versions[$version->getVersionNumber()] = $version;
-
-        if (empty($this->versionString)) {
-            $this->versionString = (string)$version->getVersionNumber();
-        } else {
-            $this->versionString .= ", {$version->getVersionNumber()}";
-        }
-
-        return $this;
-    }
-
-    /**
-     * Disable support for a specific WebSocket protocol version
-     * @param int The version ID to un-support
-     * @return HandshakeNegotiator
-     */
-    public function disableVersion($versionId) {
-        unset($this->versions[$versionId]);
-
-        $this->versionString = implode(',', array_keys($this->versions));
-
-        return $this;
-    }
-
-    /**
-     * Get a string of version numbers supported (comma delimited)
-     * @return string
-     */
-    public function getSupportedVersionString() {
-        return $this->versionString;
-    }
-}

HttpRequestParser.php

@@ -0,0 +1,57 @@
+<?php
+namespace Ratchet\WebSocket;
+use Ratchet\MessageInterface;
+use Ratchet\ConnectionInterface;
+use Ratchet\WebSocket\Guzzle\Http\Message\RequestFactory;
+use Ratchet\WebSocket\Version\VersionInterface;
+use Guzzle\Http\Message\RequestInterface;
+use Guzzle\Http\Message\Response;
+
+class HttpRequestParser implements MessageInterface {
+    const EOM = "\r\n\r\n";
+
+    /**
+     * The maximum number of bytes the request can be
+     * This is a security measure to prevent attacks
+     * @var int
+     */
+    public $maxSize = 4096;
+
+    /**
+     * @param StdClass
+     * @param string Data stream to buffer
+     * @return Guzzle\Http\Message\Response|null Response object if it's done parsing, null if there's more to be buffered
+     * @throws OverflowException
+     */
+    public function onMessage(ConnectionInterface $context, $data) {
+        if (!isset($context->httpBuffer)) {
+            $context->httpBuffer = '';
+        }
+
+        $context->httpBuffer .= $data;
+
+        if (strlen($context->httpBuffer) > (int)$this->maxSize) {
+            throw new \OverflowException("Maximum buffer size of {$this->maxSize} exceeded parsing HTTP header");
+
+            //return new Response(413, array('X-Powered-By' => \Ratchet\VERSION));
+        }
+
+        if ($this->isEom($context->httpBuffer)) {
+            $request = RequestFactory::getInstance()->fromMessage($context->httpBuffer);
+
+            unset($context->httpBuffer);
+
+            return $request;
+        }
+    }
+
+    /**
+     * Determine if the message has been buffered as per the HTTP specification
+     * @param string
+     * @return boolean
+     */
+    public function isEom($message) {
+        //return (static::EOM === substr($message, 0 - strlen(static::EOM)));
+        return (boolean)strpos($message, static::EOM);
+    }
+}

Version/RFC6455.php

@@ -5,6 +5,7 @@ use Ratchet\MessageInterface;
 use Ratchet\WebSocket\Version\RFC6455\HandshakeVerifier;
 use Ratchet\WebSocket\Version\RFC6455\Message;
 use Ratchet\WebSocket\Version\RFC6455\Frame;
+use Ratchet\WebSocket\Version\RFC6455\Connection;
 use Guzzle\Http\Message\RequestInterface;
 use Guzzle\Http\Message\Response;
 
@@ -20,14 +21,8 @@ class RFC6455 implements VersionInterface {
      */
     protected $_verifier;
 
-    /**
+    public function __construct() {
-     * @var Ratchet\MessageInterface
+        $this->_verifier = new HandshakeVerifier;
-     */
-    protected $coalescedCallback;
-
-    public function __construct(MessageInterface $coalescedCallback = null) {
-        $this->_verifier         = new HandshakeVerifier;
-        $this->coalescedCallback = $coalescedCallback;
     }
 
     /**
@@ -52,6 +47,8 @@ class RFC6455 implements VersionInterface {
      */
     public function handshake(RequestInterface $request) {
         if (true !== $this->_verifier->verifyAll($request)) {
+            // new header with 4xx error message
+
             throw new \InvalidArgumentException('Invalid HTTP header');
         }
 
@@ -59,13 +56,31 @@ class RFC6455 implements VersionInterface {
             'Upgrade'              => 'websocket'
           , 'Connection'           => 'Upgrade'
           , 'Sec-WebSocket-Accept' => $this->sign($request->getHeader('Sec-WebSocket-Key'))
+          , 'X-Powered-By'         => \Ratchet\VERSION
         );
 
         return new Response(101, $headers);
     }
 
     /**
-     * {@inheritdoc}
+     * @param Ratchet\ConnectionInterface
+     * @return Ratchet\WebSocket\Version\RFC6455\Connection
+     */
+    public function upgradeConnection(ConnectionInterface $conn, MessageInterface $coalescedCallback) {
+        $upgraded = new Connection($conn);
+
+        if (!isset($upgraded->WebSocket)) {
+            $upgraded->WebSocket = new \StdClass;
+        }
+
+        $upgraded->WebSocket->coalescedCallback = $coalescedCallback;
+
+        return $upgraded;
+    }
+
+    /**
+     * @param Ratchet\WebSocket\Version\RFC6455\Connection
+     * @param string
      */
     public function onMessage(ConnectionInterface $from, $data) {
         $overflow = '';
@@ -134,7 +149,7 @@ class RFC6455 implements VersionInterface {
             $parsed = $from->WebSocket->message->getPayload();
             unset($from->WebSocket->message);
 
-            $this->coalescedCallback->onMessage($from, $parsed);
+            $from->WebSocket->coalescedCallback->onMessage($from, $parsed);
         }
 
         if (strlen($overflow) > 0) {

Version/RFC6455/Connection.php

@@ -0,0 +1,37 @@
+<?php
+namespace Ratchet\WebSocket\Version\RFC6455;
+use Ratchet\ConnectionInterface;
+use Ratchet\AbstractConnectionDecorator;
+use Ratchet\WebSocket\Version\VersionInterface;
+use Ratchet\WebSocket\Version\FrameInterface;
+
+/**
+ * {@inheritdoc}
+ */
+class Connection extends AbstractConnectionDecorator {
+    public function __construct(ConnectionInterface $conn) {
+        parent::__construct($conn);
+    }
+
+    public function send($msg) {
+        if ($msg instanceof FrameInterface) {
+            $data = $msg->data;
+        } else {
+            $frame = new Frame($msg);
+            $data  = $frame->data;
+        }
+
+        $this->getConnection()->send($data);
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function close($code = 1000) {
+        $frame = new Frame($code, true, Frame::OP_CLOSE);
+
+        $this->send($frame->data);
+
+        $this->getConnection()->close();
+    }
+}

VersionManager.php

@@ -0,0 +1,77 @@
+<?php
+namespace Ratchet\WebSocket;
+use Ratchet\WebSocket\Version\VersionInterface;
+use Guzzle\Http\Message\RequestInterface;
+
+class VersionManager {
+    private $versionString = '';
+
+    protected $versions = array();
+
+    /**
+     * Get the protocol negotiator for the request, if supported
+     * @param Guzzle\Http\Message\RequestInterface
+     * @return Ratchet\WebSocket\Version\VersionInterface
+     */
+    public function getVersion(RequestInterface $request) {
+        foreach ($this->versions as $version) {
+            if ($version->isProtocol($request)) {
+                return $version;
+            }
+        }
+
+        throw new \InvalidArgumentException("Version not found");
+    }
+
+    /**
+     * @param Guzzle\Http\Message\RequestInterface
+     * @return bool
+     */
+    public function isVersionEnabled(RequestInterface $request) {
+        foreach ($this->versions as $version) {
+            if ($version->isProtocol($request)) {
+                return true;
+            }
+        }
+
+        return false;
+    }
+
+    /**
+     * Enable support for a specific version of the WebSocket protocol
+     * @param Ratchet\WebSocket\Vesion\VersionInterface
+     * @return HandshakeNegotiator
+     */
+    public function enableVersion(VersionInterface $version) {
+        $this->versions[$version->getVersionNumber()] = $version;
+
+        if (empty($this->versionString)) {
+            $this->versionString = (string)$version->getVersionNumber();
+        } else {
+            $this->versionString .= ", {$version->getVersionNumber()}";
+        }
+
+        return $this;
+    }
+
+    /**
+     * Disable support for a specific WebSocket protocol version
+     * @param int The version ID to un-support
+     * @return HandshakeNegotiator
+     */
+    public function disableVersion($versionId) {
+        unset($this->versions[$versionId]);
+
+        $this->versionString = implode(',', array_keys($this->versions));
+
+        return $this;
+    }
+
+    /**
+     * Get a string of version numbers supported (comma delimited)
+     * @return string
+     */
+    public function getSupportedVersionString() {
+        return $this->versionString;
+    }
+}

WsConnection.php

@@ -1,46 +0,0 @@
-<?php
-namespace Ratchet\WebSocket;
-use Ratchet\ConnectionInterface;
-use Ratchet\AbstractConnectionDecorator;
-use Ratchet\WebSocket\Version\VersionInterface;
-use Ratchet\WebSocket\Version\FrameInterface;
-
-use Ratchet\WebSocket\Version\RFC6455\Frame;
-
-/**
- * {@inheritdoc}
- * @property stdClass $WebSocket
- */
-class WsConnection extends AbstractConnectionDecorator {
-    public function __construct(ConnectionInterface $conn) {
-        parent::__construct($conn);
-
-        $this->WebSocket = new \StdClass;
-    }
-
-    public function send($data) {
-        if ($data instanceof FrameInterface) {
-            $data = $data->data;
-        } elseif (isset($this->WebSocket->version)) {
-            // need frame caching
-            $data = $this->WebSocket->version->frame($data, false);
-        }
-
-        $this->getConnection()->send($data);
-    }
-
-    /**
-     * {@inheritdoc}
-     * @todo If code is 1000 send close frame - false is close w/o frame...?
-     */
-    public function close($code = 1000) {
-        $this->send(Frame::create($code, true, Frame::OP_CLOSE));
-        // send close frame with code 1000
-
-        // ???
-
-        // profit
-
-        $this->getConnection()->close(); // temporary
-    }
-}

WsServer.php

@@ -3,8 +3,7 @@ namespace Ratchet\WebSocket;
 use Ratchet\MessageComponentInterface;
 use Ratchet\ConnectionInterface;
 use Ratchet\WebSocket\Version;
-use Guzzle\Http\Message\RequestInterface;
+use Guzzle\Http\Message\Response;
-use Ratchet\WebSocket\Guzzle\Http\Message\RequestFactory;
 
 /**
  * The adapter to handle WebSocket requests/responses
@@ -15,12 +14,17 @@ use Ratchet\WebSocket\Guzzle\Http\Message\RequestFactory;
  */
 class WsServer implements MessageComponentInterface {
     /**
-     * Negotiates upgrading the HTTP connection to a WebSocket connection
+     * Buffers incoming HTTP requests returning a Guzzle Request when coalesced
-     * It contains useful configuration properties and methods
+     * @var HttpRequestParser
-     * @var HandshakeNegotiator
      * @note May not expose this in the future, may do through facade methods
      */
-    public $handshaker;
+    public $reqParser;
+
+    /**
+     * Manage the various WebSocket versions to support
+     * @var VersionManager
+     */
+    protected $versioner;
 
     /**
      * Decorated component
@@ -52,9 +56,10 @@ class WsServer implements MessageComponentInterface {
     public function __construct(MessageComponentInterface $component) {
         //mb_internal_encoding('UTF-8');
 
-        $this->handshaker = new HandshakeNegotiator();
+        $this->reqParser = new HttpRequestParser;
+        $this->versioner = new VersionManager;
 
-        $this->handshaker
+        $this->versioner
             ->enableVersion(new Version\RFC6455($component))
             ->enableVersion(new Version\HyBi10($component))
             //->enableVersion(new Version\Hixie76)
@@ -68,12 +73,13 @@ class WsServer implements MessageComponentInterface {
      * {@inheritdoc}
      */
     public function onOpen(ConnectionInterface $conn) {
-        $wsConn = new WsConnection($conn);
+        //$wsConn = new WsConnection($conn);
 
-        $this->connections->attach($conn, $wsConn);
+        //$this->connections->attach($conn, $wsConn);
 
-        $this->handshaker->onOpen($wsConn);
+        //$this->reqParser->onOpen($wsConn);
 
+        $conn->WebSocket = new \StdClass;
         $conn->WebSocket->established = false;
     }
 
@@ -81,15 +87,28 @@ class WsServer implements MessageComponentInterface {
      * {@inheritdoc}
      */
     public function onMessage(ConnectionInterface $from, $msg) {
-        $conn = $this->connections[$from];
+        if (true !== $from->WebSocket->established) {
-
+            if (null === ($request = $this->reqParser->onMessage($from, $msg))) {
-        if (true !== $conn->WebSocket->established) {
-            if (null === ($response = $this->handshaker->onMessage($conn, $msg))) {
                 return;
             }
 
+            if (!$this->versioner->isVersionEnabled($request)) {
+                $response = new Response(400, array(
+                    'Sec-WebSocket-Version' => $this->versioner->getSupportedVersionString()
+                  , 'X-Powered-By'          => \Ratchet\VERSION
+                ));
+
+                $from->send((string)$response);
+                $from->close();
+
+                return;
+            }
+
+            $from->WebSocket->version = $this->versioner->getVersion($request);
+            $response = $from->WebSocket->version->handshake($request);
+
             // This needs to be refactored later on, incorporated with routing
-            if ('' !== ($agreedSubProtocols = $this->getSubProtocolString($from->WebSocket->request->getTokenizedHeader('Sec-WebSocket-Protocol', ',')))) {
+            if ('' !== ($agreedSubProtocols = $this->getSubProtocolString($request->getTokenizedHeader('Sec-WebSocket-Protocol', ',')))) {
                 $response->setHeader('Sec-WebSocket-Protocol', $agreedSubProtocols);
             }
 
@@ -99,24 +118,30 @@ class WsServer implements MessageComponentInterface {
                 return $from->close();
             }
 
-            $conn->WebSocket->established = true;
+            $upgraded = $from->WebSocket->version->upgradeConnection($from, $this->_decorating);
 
-            return $this->_decorating->onOpen($conn);
+            $this->connections->attach($from, $upgraded);
+
+            $upgraded->WebSocket->established = true;
+
+            return $this->_decorating->onOpen($upgraded);
         }
 
-        $conn->WebSocket->version->onMessage($conn, $msg);
+        $from->WebSocket->version->onMessage($this->connections[$from], $msg);
     }
 
     /**
      * {@inheritdoc}
      */
     public function onClose(ConnectionInterface $conn) {
-        $decor = $this->connections[$conn];
+        if ($this->connections->contains($conn)) {
-        $this->connections->detach($conn);
+            $decor = $this->connections[$conn];
+            $this->connections->detach($conn);
+        }
 
         // WS::onOpen is not called when the socket connects, it's call when the handshake is done
         // The socket could close before WS calls onOpen, so we need to check if we've "opened" it for the developer yet
-        if ($decor->WebSocket->established) {
+        if (isset($decor)) {
             $this->_decorating->onClose($decor);
         }
     }