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

Update to latest react/socket and drop react/http for tests

Browse Source
This commit is contained in:
Matt Bonneau 2019-12-10 19:22:57 -05:00
parent 9eae01044a
commit 8944361dbe
3 changed files with 89 additions and 75 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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.*"
} }
} }

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();