Hello - asynchronously wait for it... World!

.gitignore

@@ -0,0 +1,4 @@
+phpunit.xml
+reports
+vendor
+composer.lock

LICENSE

@@ -0,0 +1,19 @@
+Copyright (c) 2011-2014 Chris Boden
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is furnished
+to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.

README.md

@@ -0,0 +1,27 @@
+#Pawl
+
+An asynchronous PHP WebSocket client. (early alpha)
+
+---
+
+```php
+<?php
+
+    require __DIR__ . '/vendor/autoload.php';
+
+    $loop = React\EventLoop\Factory::create();
+    $connector = new Ratchet\Client\Factory($loop);
+
+    $connector('ws://127.0.0.1:8080')->then(function(Ratchet\Client\WebSocket $conn) {
+        $conn->on('message', function($msg) {
+            echo "Received: {$msg}\n";
+        });
+
+        $conn->send('Hello World!');
+    }, function($e) use ($loop) {
+        echo "Could not connect: {$e->getMessage()}\n";
+        $loop->stop();
+    });
+
+    $loop->run();
+```

composer.json

@@ -0,0 +1,17 @@
+{
+    "name": "ratchet/pawl"
+  , "description": "Asynchronous WebSocket client"
+  , "keywords": ["WebSocket", "client", "Ratchet", "async"]
+  , "license": "MIT"
+  , "autoload": {
+        "psr-4": {
+            "Ratchet\\Client\\": "src"
+        }
+    }
+  , "require": {
+        "php": ">=5.4"
+      , "cboden/ratchet": "0.3.*"
+      , "react/socket-client": "0.4.*"
+    }
+
+}

src/Factory.php

@@ -0,0 +1,135 @@
+<?php
+namespace Ratchet\Client;
+use React\EventLoop\LoopInterface;
+use React\Stream\Stream;
+use React\SocketClient\Connector;
+use React\SocketClient\SecureConnector;
+use React\Dns\Resolver\Resolver;
+use React\Dns\Resolver\Factory as DnsFactory;
+use React\Promise\Deferred;
+use React\Promise\RejectedPromise;
+use Guzzle\Http\Message\Response;
+use Guzzle\Http\Message\Request;
+use Ratchet\WebSocket\Version\RFC6455;
+
+class Factory {
+    protected $_loop;
+    protected $_connector;
+    protected $_secureConnector;
+
+    public $defaultHeaders = [
+        'Connection'            => 'Upgrade'
+      , 'Cache-Control'         => 'no-cache'
+      , 'Pragma'                => 'no-cache'
+      , 'Upgrade'               => 'websocket'
+      , 'Sec-WebSocket-Version' => 13
+      , 'User-Agent'            => "Ratchet->Pawl/0.0.1"
+    ];
+
+    public function __construct(LoopInterface $loop, Resolver $resolver = null) {
+        if (null === $resolver) {
+            $factory  = new DnsFactory();
+            $resolver = $factory->create('8.8.8.8', $loop);
+        }
+
+        $this->_loop            = $loop;
+        $this->_connector       = new Connector($loop, $resolver);
+        $this->_secureConnector = new SecureConnector($this->_connector, $loop);
+    }
+
+    public function __invoke($url, array $subProtocols = [], array $headers = []) {
+        try {
+            $request   = $this->generateRequest($url, $subProtocols, $headers);
+        } catch (\Exception $e) {
+            return new RejectedPromise($e);
+        }
+        $connector = 'wss' === substr($url, 0, 3) ? $this->_secureConnector : $this->_connector;
+
+        return $connector->create($request->getHost(), $request->getPort())->then(function(Stream $stream) use ($request) {
+            $futureWsConn = new Deferred;
+
+            $buffer = '';
+            $headerParser = function($data, Stream $stream) use (&$headerParser, &$buffer, $futureWsConn, $request) {
+                $buffer .= $data;
+                if (false == strpos($buffer, "\r\n\r\n")) {
+                    return;
+                }
+
+                $stream->removeListener('data', $headerParser);
+
+                $response = Response::fromMessage($buffer);
+
+                if (101 !== $response->getStatusCode()) {
+                    $futureWsConn->reject($response);
+                    $stream->close();
+
+                    return;
+                }
+
+                $acceptCheck = base64_encode(pack('H*', sha1($request->getHeader('Sec-WebSocket-Key') . RFC6455::GUID)));
+                if ((string)$response->getHeader('Sec-WebSocket-Accept') !== $acceptCheck) {
+                    $futureWsConn->reject(new \DomainException("Could not verify Accept Key during WebSocket handshake"));
+                    $stream->close();
+
+                    return;
+                }
+
+                // verify sub-protocol headers if any
+
+                $futureWsConn->resolve(new WebSocket($stream, $response, $request));
+
+                $extraBody = trim($response->getBody());
+                if (strlen($extraBody) > 0) {
+                    $futureWsConn->promise()->then(function(WebSocket $conn) use ($stream, $extraBody) {
+                        $stream->emit('data', [$extraBody, $stream]);
+                    });
+                }
+            };
+
+            $stream->on('data', $headerParser);
+            $stream->write($request);
+
+            return $futureWsConn->promise();
+        });
+    }
+
+    protected function generateRequest($url, array $subProtocols, array $headers) {
+        $headers = array_merge($this->defaultHeaders, $headers);
+        $headers['Sec-WebSocket-Key'] = $this->generateKey();
+
+        $request = new Request('GET', $url, $headers);
+
+        $scheme = strtolower($request->getScheme());
+        if (!in_array($scheme, ['ws', 'wss'])) {
+            throw new \InvalidArgumentsException(sprintf('Cannot connect to invalid URL (%s)', $url));
+        }
+
+        $request->setScheme('HTTP');
+
+        if (!$request->getPort()) {
+            $request->setPort('wss' === $scheme ? 443 : 80);
+        } else {
+            $request->setHeader('Host', $request->getHeader('Host') . ":{$request->getPort()}");
+        }
+
+        if (!$request->getHeader('Origin')) {
+            $request->setHeader('Origin', str_replace('ws', 'http', $scheme) . '://' . $request->getHost());
+        }
+
+        // do protocol headers
+
+        return $request;
+    }
+
+    protected function generateKey() {
+        $chars     = 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwzyz1234567890+/=';
+        $charRange = strlen($chars) - 1;
+        $key       = '';
+
+        for ($i = 0;$i < 16;$i++) {
+            $key .= $chars[mt_rand(0, $charRange)];
+        }
+
+        return base64_encode($key);
+    }
+}

src/WebSocket.php

@@ -0,0 +1,123 @@
+<?php
+namespace Ratchet\Client;
+use Evenement\EventEmitterTrait;
+use Evenement\EventEmitterInterface;
+use Ratchet\ConnectionInterface;
+use React\Stream\Stream;
+use Guzzle\Http\Message\Request;
+use Guzzle\Http\Message\Response;
+use Ratchet\WebSocket\Version\RFC6455\Message;
+use Ratchet\WebSocket\Version\RFC6455\Frame;
+
+class WebSocket implements EventEmitterInterface, ConnectionInterface {
+    use EventEmitterTrait;
+
+    /**
+     * The request headers sent to establish the connection
+     * @var \Guzzle\Http\Message\Request
+     */
+    public $request;
+
+    /**
+     * The response headers received from the server to establish the connection
+     * @var \Guzzle\Http\Message\Response
+     */
+    public $response;
+
+    /**
+     * @var \React\Stream\Stream
+     */
+    protected $_stream;
+
+    /**
+     * @var Ratchet\WebSocket\Version\RFC6455\Message
+     */
+    private $_message;
+
+    /**
+     * @var Ratchet\WebSocket\Version\RFC6455\Frame
+     */
+    private $_frame;
+
+    public function __construct(Stream $stream, Response $response, Request $request) {
+        $this->_stream  = $stream;
+        $this->response = $response;
+        $this->request  = $request;
+
+        $stream->on('data', function($data) {
+            $this->handleData($data);
+        });
+
+        $stream->on('end', function(Stream $stream) {
+            if (is_resource($stream->stream)) {
+                stream_socket_shutdown($stream->stream, STREAM_SHUT_RDWR);
+                stream_set_blocking($stream->stream, false);
+            }
+        });
+
+        $stream->on('close', function() {
+            $this->emit('close', [$this]);
+        });
+
+        $stream->on('error', function($error) {
+            $this->emit('error', [$error, $this]);
+        });
+    }
+
+    public function send($msg) {
+        $frame = new Frame($msg);
+        $frame->maskPayload($frame->generateMaskingKey());
+
+        $this->_stream->write($frame->getContents());
+    }
+
+    public function close($code = 1000) {
+        $frame = new Frame(pack('n', $code), true, Frame::OP_CLOSE);
+
+        $this->_stream->write($frame->getContents());
+        $this->_stream->end();
+    }
+
+    private function handleData($data) {
+        if (0 === strlen($data)) {
+            return;
+        }
+
+        if (!$this->_message) {
+            $this->_message = new Message;
+        }
+        if (!$this->_frame) {
+            $this->_frame = new Frame;
+            $this->_message->addFrame($this->_frame);
+        }
+
+        $this->_frame->addBuffer($data);
+
+        if ($this->_frame->isCoalesced()) {
+            if (Frame::OP_CLOSE === $this->_frame->getOpcode()) {
+                $this->close($this->_frame->getPayload());
+
+                return;
+            }
+
+            $overflow = $this->_frame->extractOverflow();
+            $this->_frame = null;
+        }
+
+        if (!$this->_message->isCoalesced()) {
+            if (isset($overflow)) {
+                $this->handleData($overflow);
+            }
+
+            return;
+        }
+
+        $message  = $this->_message->getPayload();
+
+        $this->_frame = $this->_message = null;
+
+        $this->emit('message', [$message, $this]);
+
+        $this->handleData($overflow);
+    }
+}

tests/autobahn/fuzzingserver.json

@@ -0,0 +1,10 @@
+{
+    "url": "ws://127.0.0.1:9001"
+  , "options": {
+        "failByDrop": false
+    }
+  , "outdir": "reports/ab"
+  , "cases": ["*"]
+  , "exclude-cases": ["9.8.6"]
+  , "exclude-agent-cases": {}
+}