faeez/rfc6455
1
0
Fork 0
Code Issues Pull Requests Actions Packages Repo. Project Releases Wiki Activity

Merge pull request #36 from mbonneau/devdeps

Browse Source 
Update development deps and speed up frame processing
This commit is contained in:
Matt Bonneau 2019-12-11 10:16:05 -05:00 committed by GitHub
commit cb8b04beba
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 134 additions and 86 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.travis.yml
View File

@ -1,12 +1,12 @@
language: php language: php
php: php:
- 5.4
- 5.5
- 5.6 - 5.6
- 7.0 - 7.0
- 7.1 - 7.1
- 7.2 - 7.2
- 7.3
- 7.4
before_install: before_install:
- export PATH=$HOME/.local/bin:$PATH - export PATH=$HOME/.local/bin:$PATH

5
composer.json
View File

@ -25,8 +25,7 @@
"guzzlehttp/psr7": "^1.0" "guzzlehttp/psr7": "^1.0"
}, },
"require-dev": { "require-dev": {
"react/http": "^0.4.1", "phpunit/phpunit": "4.8.*",
"react/socket-client": "^0.4.3", "react/socket": "^1.3"
"phpunit/phpunit": "4.8.*"
} }
} }

50
src/Messaging/MessageBuffer.php
View File

@ -37,6 +37,11 @@ class MessageBuffer {
*/ */
private $checkForMask; private $checkForMask;
/**
* @var string
*/
private $leftovers;
function __construct( function __construct(
CloseFrameChecker $frameChecker, CloseFrameChecker $frameChecker,
callable $onMessage, callable $onMessage,
@ -53,12 +58,47 @@ class MessageBuffer {
$this->onMessage = $onMessage; $this->onMessage = $onMessage;
$this->onControl = $onControl ?: function() {}; $this->onControl = $onControl ?: function() {};
$this->leftovers = '';
} }
public function onData($data) { public function onData($data) {
while (strlen($data) > 0) { $data = $this->leftovers . $data;
$data = $this->processData($data); $dataLen = strlen($data);
if ($dataLen < 2) {
$this->leftovers = $data;
return;
} }
$frameStart = 0;
while ($frameStart + 2 <= $dataLen) {
$headerSize = 2;
$payload_length = unpack('C', $data[$frameStart + 1] & "\x7f")[1];
$isMasked = ($data[$frameStart + 1] & "\x80") === "\x80";
$headerSize += $isMasked ? 4 : 0;
if ($payload_length > 125 && ($dataLen - $frameStart < $headerSize + 125)) {
// no point of checking - this frame is going to be bigger than the buffer is right now
break;
}
if ($payload_length > 125) {
$payloadLenBytes = $payload_length === 126 ? 2 : 8;
$headerSize += $payloadLenBytes;
$bytesToUpack = substr($data, $frameStart + 2, $payloadLenBytes);
$payload_length = $payload_length === 126
? unpack('n', $bytesToUpack)[1]
: unpack('J', $bytesToUpack)[1];
}
$isCoalesced = $dataLen - $frameStart >= $payload_length + $headerSize;
if (!$isCoalesced) {
break;
}
$this->processData(substr($data, $frameStart, $payload_length + $headerSize));
$frameStart = $frameStart + $payload_length + $headerSize;
}
$this->leftovers = substr($data, $frameStart);
} }
/** /**
@ -70,16 +110,12 @@ class MessageBuffer {
$this->frameBuffer ?: $this->frameBuffer = $this->newFrame(); $this->frameBuffer ?: $this->frameBuffer = $this->newFrame();
$this->frameBuffer->addBuffer($data); $this->frameBuffer->addBuffer($data);
if (!$this->frameBuffer->isCoalesced()) {
return '';
}
$onMessage = $this->onMessage; $onMessage = $this->onMessage;
$onControl = $this->onControl; $onControl = $this->onControl;
$this->frameBuffer = $this->frameCheck($this->frameBuffer); $this->frameBuffer = $this->frameCheck($this->frameBuffer);
$overflow = $this->frameBuffer->extractOverflow();
$this->frameBuffer->unMaskPayload(); $this->frameBuffer->unMaskPayload();
$opcode = $this->frameBuffer->getOpcode(); $opcode = $this->frameBuffer->getOpcode();
@ -108,8 +144,6 @@ class MessageBuffer {
$onMessage($msgBuffer); $onMessage($msgBuffer);
} }
} }
return $overflow;
} }
/** /**

78
tests/ab/clientRunner.php
View File

@ -1,7 +1,14 @@
<?php <?php
use GuzzleHttp\Psr7\Uri; use GuzzleHttp\Psr7\Uri;
use Ratchet\RFC6455\Handshake\ClientNegotiator;
use Ratchet\RFC6455\Messaging\CloseFrameChecker;
use Ratchet\RFC6455\Messaging\FrameInterface;
use Ratchet\RFC6455\Messaging\MessageBuffer;
use Ratchet\RFC6455\Messaging\MessageInterface;
use React\Promise\Deferred; use React\Promise\Deferred;
use Ratchet\RFC6455\Messaging\Frame; use Ratchet\RFC6455\Messaging\Frame;
use React\Socket\ConnectionInterface;
use React\Socket\Connector;
require __DIR__ . '/../bootstrap.php'; require __DIR__ . '/../bootstrap.php';
@ -11,23 +18,20 @@ $testServer = "127.0.0.1";
$loop = React\EventLoop\Factory::create(); $loop = React\EventLoop\Factory::create();
$dnsResolverFactory = new React\Dns\Resolver\Factory(); $connector = new Connector($loop);
$dnsResolver = $dnsResolverFactory->createCached('8.8.8.8', $loop);
$factory = new \React\SocketClient\Connector($loop, $dnsResolver);
function echoStreamerFactory($conn) function echoStreamerFactory($conn)
{ {
return new \Ratchet\RFC6455\Messaging\MessageBuffer( return new MessageBuffer(
new \Ratchet\RFC6455\Messaging\CloseFrameChecker, new CloseFrameChecker,
function (\Ratchet\RFC6455\Messaging\MessageInterface $msg) use ($conn) { function (MessageInterface $msg) use ($conn) {
/** @var Frame $frame */ /** @var Frame $frame */
foreach ($msg as $frame) { foreach ($msg as $frame) {
$frame->maskPayload(); $frame->maskPayload();
} }
$conn->write($msg->getContents()); $conn->write($msg->getContents());
}, },
function (\Ratchet\RFC6455\Messaging\FrameInterface $frame) use ($conn) { function (FrameInterface $frame) use ($conn) {
switch ($frame->getOpcode()) { switch ($frame->getOpcode()) {
case Frame::OP_PING: case Frame::OP_PING:
return $conn->write((new Frame($frame->getPayload(), true, Frame::OP_PONG))->maskPayload()->getContents()); return $conn->write((new Frame($frame->getPayload(), true, Frame::OP_PONG))->maskPayload()->getContents());
@ -42,22 +46,22 @@ function echoStreamerFactory($conn)
} }
function getTestCases() { function getTestCases() {
global $factory;
global $testServer; global $testServer;
global $connector;
$deferred = new Deferred(); $deferred = new Deferred();
$factory->create($testServer, 9001)->then(function (\React\Stream\Stream $stream) use ($deferred) { $connector->connect($testServer . ':9001')->then(function (ConnectionInterface $connection) use ($deferred) {
$cn = new \Ratchet\RFC6455\Handshake\ClientNegotiator(); $cn = new ClientNegotiator();
$cnRequest = $cn->generateRequest(new Uri('ws://127.0.0.1:9001/getCaseCount')); $cnRequest = $cn->generateRequest(new Uri('ws://127.0.0.1:9001/getCaseCount'));
$rawResponse = ""; $rawResponse = "";
$response = null; $response = null;
/** @var \Ratchet\RFC6455\Messaging\Streaming\MessageBuffer $ms */ /** @var MessageBuffer $ms */
$ms = null; $ms = null;
$stream->on('data', function ($data) use ($stream, &$rawResponse, &$response, &$ms, $cn, $deferred, &$context, $cnRequest) { $connection->on('data', function ($data) use ($connection, &$rawResponse, &$response, &$ms, $cn, $deferred, &$context, $cnRequest) {
if ($response === null) { if ($response === null) {
$rawResponse .= $data; $rawResponse .= $data;
$pos = strpos($rawResponse, "\r\n\r\n"); $pos = strpos($rawResponse, "\r\n\r\n");
@ -67,14 +71,14 @@ function getTestCases() {
$response = \GuzzleHttp\Psr7\parse_response($rawResponse); $response = \GuzzleHttp\Psr7\parse_response($rawResponse);
if (!$cn->validateResponse($cnRequest, $response)) { if (!$cn->validateResponse($cnRequest, $response)) {
$stream->end(); $connection->end();
$deferred->reject(); $deferred->reject();
} else { } else {
$ms = new \Ratchet\RFC6455\Messaging\MessageBuffer( $ms = new MessageBuffer(
new \Ratchet\RFC6455\Messaging\CloseFrameChecker, new CloseFrameChecker,
function (\Ratchet\RFC6455\Messaging\MessageInterface $msg) use ($deferred, $stream) { function (MessageInterface $msg) use ($deferred, $connection) {
$deferred->resolve($msg->getPayload()); $deferred->resolve($msg->getPayload());
$stream->close(); $connection->close();
}, },
null, null,
false false
@ -89,7 +93,7 @@ function getTestCases() {
} }
}); });
$stream->write(\GuzzleHttp\Psr7\str($cnRequest)); $connection->write(\GuzzleHttp\Psr7\str($cnRequest));
}); });
return $deferred->promise(); return $deferred->promise();
@ -97,15 +101,15 @@ function getTestCases() {
function runTest($case) function runTest($case)
{ {
global $factory; global $connector;
global $testServer; global $testServer;
$casePath = "/runCase?case={$case}&agent=" . AGENT; $casePath = "/runCase?case={$case}&agent=" . AGENT;
$deferred = new Deferred(); $deferred = new Deferred();
$factory->create($testServer, 9001)->then(function (\React\Stream\Stream $stream) use ($deferred, $casePath, $case) { $connector->connect($testServer . ':9001')->then(function (ConnectionInterface $connection) use ($deferred, $casePath, $case) {
$cn = new \Ratchet\RFC6455\Handshake\ClientNegotiator(); $cn = new ClientNegotiator();
$cnRequest = $cn->generateRequest(new Uri('ws://127.0.0.1:9001' . $casePath)); $cnRequest = $cn->generateRequest(new Uri('ws://127.0.0.1:9001' . $casePath));
$rawResponse = ""; $rawResponse = "";
@ -113,7 +117,7 @@ function runTest($case)
$ms = null; $ms = null;
$stream->on('data', function ($data) use ($stream, &$rawResponse, &$response, &$ms, $cn, $deferred, &$context, $cnRequest) { $connection->on('data', function ($data) use ($connection, &$rawResponse, &$response, &$ms, $cn, $deferred, &$context, $cnRequest) {
if ($response === null) { if ($response === null) {
$rawResponse .= $data; $rawResponse .= $data;
$pos = strpos($rawResponse, "\r\n\r\n"); $pos = strpos($rawResponse, "\r\n\r\n");
@ -123,10 +127,10 @@ function runTest($case)
$response = \GuzzleHttp\Psr7\parse_response($rawResponse); $response = \GuzzleHttp\Psr7\parse_response($rawResponse);
if (!$cn->validateResponse($cnRequest, $response)) { if (!$cn->validateResponse($cnRequest, $response)) {
$stream->end(); $connection->end();
$deferred->reject(); $deferred->reject();
} else { } else {
$ms = echoStreamerFactory($stream); $ms = echoStreamerFactory($connection);
} }
} }
} }
@ -137,34 +141,34 @@ function runTest($case)
} }
}); });
$stream->on('close', function () use ($deferred) { $connection->on('close', function () use ($deferred) {
$deferred->resolve(); $deferred->resolve();
}); });
$stream->write(\GuzzleHttp\Psr7\str($cnRequest)); $connection->write(\GuzzleHttp\Psr7\str($cnRequest));
}); });
return $deferred->promise(); return $deferred->promise();
} }
function createReport() { function createReport() {
global $factory; global $connector;
global $testServer; global $testServer;
$deferred = new Deferred(); $deferred = new Deferred();
$factory->create($testServer, 9001)->then(function (\React\Stream\Stream $stream) use ($deferred) { $connector->connect($testServer . ':9001')->then(function (ConnectionInterface $connection) use ($deferred) {
$reportPath = "/updateReports?agent=" . AGENT . "&shutdownOnComplete=true"; $reportPath = "/updateReports?agent=" . AGENT . "&shutdownOnComplete=true";
$cn = new \Ratchet\RFC6455\Handshake\ClientNegotiator(); $cn = new ClientNegotiator();
$cnRequest = $cn->generateRequest(new Uri('ws://127.0.0.1:9001' . $reportPath)); $cnRequest = $cn->generateRequest(new Uri('ws://127.0.0.1:9001' . $reportPath));
$rawResponse = ""; $rawResponse = "";
$response = null; $response = null;
/** @var \Ratchet\RFC6455\Messaging\MessageBuffer $ms */ /** @var MessageBuffer $ms */
$ms = null; $ms = null;
$stream->on('data', function ($data) use ($stream, &$rawResponse, &$response, &$ms, $cn, $deferred, &$context, $cnRequest) { $connection->on('data', function ($data) use ($connection, &$rawResponse, &$response, &$ms, $cn, $deferred, &$context, $cnRequest) {
if ($response === null) { if ($response === null) {
$rawResponse .= $data; $rawResponse .= $data;
$pos = strpos($rawResponse, "\r\n\r\n"); $pos = strpos($rawResponse, "\r\n\r\n");
@ -174,12 +178,12 @@ function createReport() {
$response = \GuzzleHttp\Psr7\parse_response($rawResponse); $response = \GuzzleHttp\Psr7\parse_response($rawResponse);
if (!$cn->validateResponse($cnRequest, $response)) { if (!$cn->validateResponse($cnRequest, $response)) {
$stream->end(); $connection->end();
$deferred->reject(); $deferred->reject();
} else { } else {
$ms = new \Ratchet\RFC6455\Messaging\MessageBuffer( $ms = new MessageBuffer(
new \Ratchet\RFC6455\Messaging\CloseFrameChecker, new CloseFrameChecker,
function (\Ratchet\RFC6455\Messaging\MessageInterface $msg) use ($deferred, $stream) { function (MessageInterface $msg) use ($deferred, $stream) {
$deferred->resolve($msg->getPayload()); $deferred->resolve($msg->getPayload());
$stream->close(); $stream->close();
}, },
@ -196,7 +200,7 @@ function createReport() {
} }
}); });
$stream->write(\GuzzleHttp\Psr7\str($cnRequest)); $connection->write(\GuzzleHttp\Psr7\str($cnRequest));
}); });
return $deferred->promise(); return $deferred->promise();

61
tests/ab/startServer.php
View File

@ -7,49 +7,60 @@ require_once __DIR__ . "/../bootstrap.php";
$loop = \React\EventLoop\Factory::create(); $loop = \React\EventLoop\Factory::create();
$socket = new \React\Socket\Server($loop); $socket = new \React\Socket\Server('127.0.0.1:9001', $loop);
$server = new \React\Http\Server($socket);
$closeFrameChecker = new \Ratchet\RFC6455\Messaging\CloseFrameChecker; $closeFrameChecker = new \Ratchet\RFC6455\Messaging\CloseFrameChecker;
$negotiator = new \Ratchet\RFC6455\Handshake\ServerNegotiator(new \Ratchet\RFC6455\Handshake\RequestVerifier); $negotiator = new \Ratchet\RFC6455\Handshake\ServerNegotiator(new \Ratchet\RFC6455\Handshake\RequestVerifier);
$uException = new \UnderflowException; $uException = new \UnderflowException;
$server->on('request', function (\React\Http\Request $request, \React\Http\Response $response) use ($negotiator, $closeFrameChecker, $uException) { $socket->on('connection', function (React\Socket\ConnectionInterface $connection) use ($negotiator, $closeFrameChecker, $uException) {
$psrRequest = new \GuzzleHttp\Psr7\Request($request->getMethod(), $request->getPath(), $request->getHeaders()); $headerComplete = false;
$buffer = '';
$negotiatorResponse = $negotiator->handshake($psrRequest); $parser = null;
$connection->on('data', function ($data) use ($connection, &$parser, &$headerComplete, &$buffer, $negotiator, $closeFrameChecker, $uException) {
$response->writeHead( if ($headerComplete) {
$negotiatorResponse->getStatusCode(), $parser->onData($data);
array_merge(
$negotiatorResponse->getHeaders(),
["Content-Length" => "0"]
)
);
if ($negotiatorResponse->getStatusCode() !== 101) {
$response->end();
return; return;
} }
$parser = new \Ratchet\RFC6455\Messaging\MessageBuffer($closeFrameChecker, function(MessageInterface $message) use ($response) { $buffer .= $data;
$response->write($message->getContents()); $parts = explode("\r\n\r\n", $buffer);
}, function(FrameInterface $frame) use ($response, &$parser) { if (count($parts) < 2) {
return;
}
$headerComplete = true;
$psrRequest = \GuzzleHttp\Psr7\parse_request($parts[0] . "\r\n\r\n");
$negotiatorResponse = $negotiator->handshake($psrRequest);
$negotiatorResponse = $negotiatorResponse->withAddedHeader("Content-Length", "0");
$connection->write(\GuzzleHttp\Psr7\str($negotiatorResponse));
if ($negotiatorResponse->getStatusCode() !== 101) {
$connection->end();
return;
}
$parser = new \Ratchet\RFC6455\Messaging\MessageBuffer($closeFrameChecker,
function (MessageInterface $message) use ($connection) {
$connection->write($message->getContents());
}, function (FrameInterface $frame) use ($connection, &$parser) {
switch ($frame->getOpCode()) { switch ($frame->getOpCode()) {
case Frame::OP_CLOSE: case Frame::OP_CLOSE:
$response->end($frame->getContents()); $connection->end($frame->getContents());
break; break;
case Frame::OP_PING: case Frame::OP_PING:
$response->write($parser->newFrame($frame->getPayload(), true, Frame::OP_PONG)->getContents()); $connection->write($parser->newFrame($frame->getPayload(), true, Frame::OP_PONG)->getContents());
break; break;
} }
}, true, function() use ($uException) { }, true, function () use ($uException) {
return $uException; return $uException;
}); });
$request->on('data', [$parser, 'onData']); array_shift($parts);
$parser->onData(implode("\r\n\r\n", $parts));
});
}); });
$socket->listen(9001, '0.0.0.0');
$loop->run(); $loop->run();