Skip to content

Commit 3cd8de8

Browse files
mbonneaumbonneau
committed
Merge branch 'update-to-react-http-v1'
# Conflicts: # composer.json # test/ab/clientRunner.php # test/ab/fuzzingclient.json # test/ab/run_ab_tests.sh
2 parents 50ee890 + d708235 commit 3cd8de8

File tree

8 files changed

+125
-149
lines changed

8 files changed

+125
-149
lines changed

‎composer.json‎

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -37,13 +37,12 @@
3737
]
3838
},
3939
"require": {
40-
"react/http": "^0.7.3 | ^0.8",
41-
"react/http-client": "^0.5.3",
40+
"react/http": "^1",
4241
"voryx/event-loop": "^3.0 || ^2.0.2",
43-
"ratchet/rfc6455": "^0.2.2 || ^0.3",
42+
"ratchet/rfc6455": "^0.3",
4443
"reactivex/rxphp": "^2.0.1"
4544
},
4645
"require-dev": {
47-
"phpunit/phpunit": "~5.7.0"
46+
"phpunit/phpunit": "^9"
4847
}
4948
}

‎phpunit.xml‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,6 @@
88
convertWarningsToExceptions="true"
99
processIsolation="false"
1010
stopOnFailure="false"
11-
syntaxCheck="false"
1211
bootstrap="test/bootstrap.php">
1312
<testsuites>
1413
<testsuite name="RxWebsocket Test">

‎src/Client.php‎

Lines changed: 15 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -5,11 +5,11 @@
55
use GuzzleHttp\Psr7\Request as Psr7Request;
66
use GuzzleHttp\Psr7\Response as Psr7Response;
77
use GuzzleHttp\Psr7\Uri;
8+
use Psr\Http\Message\ResponseInterface;
89
use Ratchet\RFC6455\Handshake\ClientNegotiator;
910
use React\EventLoop\LoopInterface;
10-
use React\HttpClient\Client as HttpClient;
11-
use React\HttpClient\Request;
12-
use React\HttpClient\Response;
11+
use React\Http\Client\Client as HttpClient;
12+
use React\Socket\ConnectionInterface;
1313
use React\Socket\ConnectorInterface;
1414
use Rx\Disposable\CallbackDisposable;
1515
use Rx\DisposableInterface;
@@ -78,17 +78,17 @@ public function _subscribe(ObserverInterface $clientObserver): DisposableInterfa
7878
$clientObserver->onError($error);
7979
});
8080

81-
$request->on('response', function (Response $response, Request $request) use ($flatHeaders, $cNegotiator, $nRequest, $clientObserver) {
82-
if ($response->getCode() !== 101) {
83-
$clientObserver->onError(new \Exception('Unexpected response code ' . $response->getCode()));
81+
$request->on('response', function (ResponseInterface $response, ConnectionInterface $request) use ($flatHeaders, $cNegotiator, $nRequest, $clientObserver) {
82+
if ($response->getStatusCode() !== 101) {
83+
$clientObserver->onError(new \Exception('Unexpected response code ' . $response->getStatusCode()));
8484
return;
8585
}
8686

8787
$psr7Response = new Psr7Response(
88-
$response->getCode(),
88+
$response->getStatusCode(),
8989
$response->getHeaders(),
9090
null,
91-
$response->getVersion()
91+
$response->getProtocolVersion()
9292
);
9393

9494
$psr7Request = new Psr7Request('GET', $this->url, $flatHeaders);
@@ -103,19 +103,22 @@ public function _subscribe(ObserverInterface $clientObserver): DisposableInterfa
103103
$clientObserver->onNext(new MessageSubject(
104104
new AnonymousObservable(function (ObserverInterface $observer) use ($response, $request, $clientObserver) {
105105

106-
$response->on('data', function ($data) use ($observer) {
106+
$request->on('data', function ($data) use ($observer) {
107107
$observer->onNext($data);
108108
});
109109

110-
$response->on('error', function ($e) use ($observer) {
110+
$request->on('error', function ($e) use ($observer) {
111111
$observer->onError($e);
112112
});
113113

114-
$response->on('close', function () use ($observer) {
114+
$request->on('close', function () use ($observer, $clientObserver) {
115115
$observer->onCompleted();
116+
117+
// complete the parent observer - we only do 1 connection
118+
$clientObserver->onCompleted();
116119
});
117120

118-
$response->on('end', function () use ($observer, $clientObserver) {
121+
$request->on('end', function () use ($observer, $clientObserver) {
119122
$observer->onCompleted();
120123

121124
// complete the parent observer - we only do 1 connection

‎src/Server.php‎

Lines changed: 87 additions & 92 deletions
Original file line numberDiff line numberDiff line change
@@ -2,23 +2,24 @@
22

33
namespace Rx\Websocket;
44

5-
use GuzzleHttp\Psr7\Request;
65
use Psr\Http\Message\ServerRequestInterface;
76
use Ratchet\RFC6455\Handshake\RequestVerifier;
87
use Ratchet\RFC6455\Handshake\ServerNegotiator;
98
use React\EventLoop\LoopInterface;
10-
use React\Http\Response;
11-
use React\Http\Server as HttpServer;
12-
use React\Socket\Server as SocketServer;
9+
use React\Http\Message\Response;
10+
use React\Http\Middleware\StreamingRequestMiddleware as StreamingRequestMiddlewareAlias;
11+
use React\Http\HttpServer;
12+
use React\Socket\SocketServer;
1313
use React\Stream\CompositeStream;
14-
use React\Stream\ReadableStreamInterface;
1514
use React\Stream\ThroughStream;
1615
use Rx\Disposable\CallbackDisposable;
16+
use Rx\Disposable\EmptyDisposable;
1717
use Rx\DisposableInterface;
1818
use Rx\Observable;
1919
use Rx\Observable\AnonymousObservable;
2020
use Rx\Observer\CallbackObserver;
2121
use Rx\ObserverInterface;
22+
use function RingCentral\Psr7\str;
2223

2324
class Server extends Observable
2425
{
@@ -40,100 +41,94 @@ public function __construct(string $bindAddressOrPort, bool $useMessageObject =
4041

4142
public function _subscribe(ObserverInterface $observer): DisposableInterface
4243
{
43-
$socket = new SocketServer($this->bindAddress, $this->loop);
44+
$socket = new SocketServer($this->bindAddress, [], $this->loop);
4445

4546
$negotiator = new ServerNegotiator(new RequestVerifier());
4647
if (!empty($this->subProtocols)) {
4748
$negotiator->setSupportedSubProtocols($this->subProtocols);
4849
}
4950

50-
$http = new HttpServer(function (ServerRequestInterface $request) use ($negotiator, $observer) {
51-
$uri = $request->getUri();
52-
53-
$psrRequest = new Request(
54-
$request->getMethod(),
55-
$uri,
56-
$request->getHeaders()
57-
);
58-
59-
// cram the remote address into the header in our own X- header so
60-
// the user will have access to it
61-
$psrRequest = $psrRequest->withAddedHeader('X-RxWebsocket-Remote-Address', $request->getServerParams()['REMOTE_ADDR'] ?? '');
62-
63-
$negotiatorResponse = $negotiator->handshake($psrRequest);
64-
65-
/** @var ReadableStreamInterface $requestStream */
66-
$requestStream = new ThroughStream();
67-
$responseStream = new ThroughStream();
68-
69-
$response = new Response(
70-
$negotiatorResponse->getStatusCode(),
71-
array_merge(
72-
$negotiatorResponse->getHeaders()
73-
),
74-
new CompositeStream(
75-
$responseStream,
76-
$requestStream
77-
)
78-
);
79-
80-
81-
if ($negotiatorResponse->getStatusCode() !== 101) {
82-
$responseStream->end();
83-
return;
84-
}
85-
86-
$subProtocol = "";
87-
if (count($negotiatorResponse->getHeader('Sec-WebSocket-Protocol')) > 0) {
88-
$subProtocol = $negotiatorResponse->getHeader('Sec-WebSocket-Protocol')[0];
51+
$http = new HttpServer(
52+
$this->loop,
53+
new StreamingRequestMiddlewareAlias(),
54+
function (ServerRequestInterface $request) use ($negotiator, $observer) {
55+
// cram the remote address into the header in our own X- header so
56+
// the user will have access to it
57+
$request = $request->withAddedHeader('X-RxWebsocket-Remote-Address', $request->getServerParams()['REMOTE_ADDR'] ?? '');
58+
59+
$negotiatorResponse = $negotiator->handshake($request);
60+
61+
$requestStream = new ThroughStream();
62+
$responseStream = new ThroughStream();
63+
64+
$response = new Response(
65+
$negotiatorResponse->getStatusCode(),
66+
array_merge(
67+
$negotiatorResponse->getHeaders()
68+
),
69+
new CompositeStream(
70+
$responseStream,
71+
$requestStream
72+
)
73+
);
74+
75+
if ($negotiatorResponse->getStatusCode() !== 101) {
76+
$responseStream->end(str($negotiatorResponse));
77+
return new EmptyDisposable();
78+
}
79+
80+
$subProtocol = "";
81+
if (count($negotiatorResponse->getHeader('Sec-WebSocket-Protocol')) > 0) {
82+
$subProtocol = $negotiatorResponse->getHeader('Sec-WebSocket-Protocol')[0];
83+
}
84+
85+
$messageSubject = new MessageSubject(
86+
new AnonymousObservable(
87+
function (ObserverInterface $observer) use ($requestStream) {
88+
$requestStream->on('data', function ($data) use ($observer) {
89+
$observer->onNext($data);
90+
});
91+
$requestStream->on('error', function ($error) use ($observer) {
92+
$observer->onError($error);
93+
});
94+
$requestStream->on('close', function () use ($observer) {
95+
$observer->onCompleted();
96+
});
97+
$requestStream->on('end', function () use ($observer) {
98+
$observer->onCompleted();
99+
});
100+
101+
return new CallbackDisposable(
102+
function () use ($requestStream) {
103+
$requestStream->close();
104+
}
105+
);
106+
}
107+
),
108+
new CallbackObserver(
109+
function ($x) use ($responseStream) {
110+
$responseStream->write($x);
111+
},
112+
function ($error) use ($responseStream) {
113+
$responseStream->close();
114+
},
115+
function () use ($responseStream) {
116+
$responseStream->end();
117+
}
118+
),
119+
false,
120+
$this->useMessageObject,
121+
$subProtocol,
122+
$request,
123+
$negotiatorResponse,
124+
$this->keepAlive
125+
);
126+
127+
$observer->onNext($messageSubject);
128+
129+
return $response;
89130
}
90-
91-
$messageSubject = new MessageSubject(
92-
new AnonymousObservable(
93-
function (ObserverInterface $observer) use ($requestStream) {
94-
$requestStream->on('data', function ($data) use ($observer) {
95-
$observer->onNext($data);
96-
});
97-
$requestStream->on('error', function ($error) use ($observer) {
98-
$observer->onError($error);
99-
});
100-
$requestStream->on('close', function () use ($observer) {
101-
$observer->onCompleted();
102-
});
103-
$requestStream->on('end', function () use ($observer) {
104-
$observer->onCompleted();
105-
});
106-
107-
return new CallbackDisposable(
108-
function () use ($requestStream) {
109-
$requestStream->close();
110-
}
111-
);
112-
}
113-
),
114-
new CallbackObserver(
115-
function ($x) use ($responseStream) {
116-
$responseStream->write($x);
117-
},
118-
function ($error) use ($responseStream) {
119-
$responseStream->close();
120-
},
121-
function () use ($responseStream) {
122-
$responseStream->end();
123-
}
124-
),
125-
false,
126-
$this->useMessageObject,
127-
$subProtocol,
128-
$psrRequest,
129-
$negotiatorResponse,
130-
$this->keepAlive
131-
);
132-
133-
$observer->onNext($messageSubject);
134-
135-
return $response;
136-
});
131+
);
137132

138133
$http->listen($socket);
139134

‎test/ABResultsTest.php‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@
22

33
namespace Rx\Websocket\Test;
44

5-
class ABResultsTest extends \PHPUnit_Framework_TestCase
5+
class ABResultsTest extends \PHPUnit\Framework\TestCase
66
{
77
private function verifyAutobahnResults($fileName)
88
{

‎test/ClientTest.php‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@
44

55
use React\EventLoop\Factory;
66

7-
class ClientTest extends \PHPUnit_Framework_TestCase
7+
class ClientTest extends \PHPUnit\Framework\TestCase
88
{
99
public function testErrorBeforeRequest()
1010
{

‎test/ab/fuzzingclient.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@
44

55
"servers": [
66
{"agent": "RxWebsocketServer/0.0.0",
7-
"url": "ws://host.ratchet.internal:9001",
7+
"url": "ws://host.docker.internal:9001",
88
"options": {"version": 18}}
99
],
1010
"cases": ["*"],

‎test/ab/run_ab_tests.sh‎

Lines changed: 17 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -1,52 +1,32 @@
1-
cd test/ab || exit
1+
cd test/ab
22

33
docker run --rm \
4-
-d \
5-
-v ${PWD}:/config \
6-
-v ${PWD}/reports:/reports \
7-
-p 9001:9001 \
8-
--name fuzzingserver \
9-
crossbario/autobahn-testsuite wstest -m fuzzingserver -s /config/fuzzingserver.json
10-
sleep 5
11-
4+
-d \
5+
-v ${PWD}:/config \
6+
-v ${PWD}/reports:/reports \
7+
-p 9001:9001 \
8+
--name rxwsfuzzingserver \
9+
crossbario/autobahn-testsuite wstest -m fuzzingserver -s /config/fuzzingserver.json
10+
sleep 10
1211
php -d memory_limit=256M clientRunner.php
1312

14-
docker ps -a
13+
docker stop rxwsfuzzingserver
1514

16-
docker logs fuzzingserver
1715

18-
docker stop fuzzingserver
19-
20-
sleep 2
16+
sleep 5
2117

2218

2319
php -d memory_limit=256M testServer.php &
2420
SERVER_PID=$!
2521
sleep 3
2622

27-
if [ "$OSTYPE" = "linux-gnu" ]; then
28-
IPADDR=`hostname -I | cut -f 1 -d ' '`
29-
else
30-
IPADDR=`ifconfig | grep "inet " | grep -Fv 127.0.0.1 | awk '{print $2}' | head -1 | tr -d 'adr:'`
31-
fi
32-
3323
docker run --rm \
34-
-it \
35-
-v ${PWD}:/config \
36-
-v ${PWD}/reports:/reports \
37-
--name fuzzingclient \
38-
crossbario/autobahn-testsuite /bin/sh -c "sh /config/docker_bootstrap.sh $IPADDR; wstest -m fuzzingclient -s /config/fuzzingclient.json"
39-
sleep 1
24+
\
25+
-v ${PWD}:/config \
26+
-v ${PWD}/reports:/reports \
27+
-p 9001:9001 \
28+
--name rxwsfuzzingclient \
29+
crossbario/autobahn-testsuite wstest -m fuzzingclient -s /config/fuzzingclient.json
30+
sleep 12
4031

4132
kill $SERVER_PID
42-
43-
#wstest -m fuzzingserver -s fuzzingserver.json &
44-
#sleep 5
45-
#php clientRunner.php
46-
#
47-
#sleep 2
48-
49-
#php testServer.php 600 &
50-
#sleep 3
51-
#wstest -m fuzzingclient -s fuzzingclient.json
52-
#sleep 12

0 commit comments

Comments
 (0)