1
|
|
|
<?php |
2
|
|
|
|
3
|
|
|
namespace WSSC; |
4
|
|
|
|
5
|
|
|
use WSSC\Components\Connection; |
6
|
|
|
use WSSC\Contracts\CommonsContract; |
7
|
|
|
use WSSC\Contracts\WebSocketMessageContract; |
8
|
|
|
use WSSC\Contracts\WebSocketServerContract; |
9
|
|
|
use WSSC\Exceptions\WebSocketException; |
10
|
|
|
|
11
|
|
|
/** |
12
|
|
|
* Create by Arthur Kushman |
13
|
|
|
*/ |
14
|
|
|
class WebSocketServer implements WebSocketServerContract, CommonsContract |
15
|
|
|
{ |
16
|
|
|
|
17
|
|
|
private $clients = []; |
18
|
|
|
// set any template You need ex.: GET /subscription/messenger/token |
19
|
|
|
private $pathParams = []; |
20
|
|
|
private $config; |
21
|
|
|
private $handshakes = []; |
22
|
|
|
private $headersUpgrade = []; |
23
|
|
|
private $totalClients = 0; |
24
|
|
|
private $maxClients = 1; |
25
|
|
|
private $handler; |
26
|
|
|
private $connImpl; |
27
|
|
|
private $cureentConn; |
28
|
|
|
// for the very 1st time must be true |
29
|
|
|
private $stepRecursion = true; |
30
|
|
|
|
31
|
|
|
const MAX_BYTES_READ = 8192; |
32
|
|
|
const HEADER_BYTES_READ = 1024; |
33
|
|
|
// must be the time for interaction between each client |
34
|
|
|
const STREAM_SELECT_TIMEOUT = 3600; |
35
|
|
|
// stream non-blocking |
36
|
|
|
const NON_BLOCK = 0; |
37
|
|
|
// max clients to fork another process |
38
|
|
|
const MAX_CLIENTS_REMAINDER_FORK = 1000; |
39
|
|
|
const PROC_TITLE = 'php-wss'; |
40
|
|
|
|
41
|
|
|
/** |
42
|
|
|
* WebSocketServer constructor. |
43
|
|
|
* @param WebSocketMessageContract $handler |
44
|
|
|
* @param array $config |
45
|
|
|
*/ |
46
|
|
|
public function __construct( |
47
|
|
|
WebSocketMessageContract $handler, |
48
|
|
|
$config = [ |
49
|
|
|
'host' => self::DEFAULT_HOST, |
50
|
|
|
'port' => self::DEFAULT_PORT, |
51
|
|
|
] |
52
|
|
|
) |
53
|
|
|
{ |
54
|
|
|
ini_set('default_socket_timeout', 5); // this should be >= 5 sec, otherwise there will be broken pipe - tested |
55
|
|
|
$this->handler = $handler; |
56
|
|
|
$this->config = $config; |
57
|
|
|
$this->connImpl = new Connection(); |
58
|
|
|
} |
59
|
|
|
|
60
|
|
|
/** |
61
|
|
|
* Runs main process - Anscestor with server socket on TCP |
62
|
|
|
*/ |
63
|
|
|
public function run() |
64
|
|
|
{ |
65
|
|
|
$errno = NULL; |
66
|
|
|
$errorMessage = ''; |
67
|
|
|
|
68
|
|
|
$server = stream_socket_server("tcp://{$this->config['host']}:{$this->config['port']}", $errno, $errorMessage); |
69
|
|
|
if ($server === false) { |
70
|
|
|
die('Could not bind to socket: ' . $errno . ' - ' . $errorMessage . PHP_EOL); |
71
|
|
|
} |
72
|
|
|
@cli_set_process_title(self::PROC_TITLE); |
|
|
|
|
73
|
|
|
$this->eventLoop($server); |
74
|
|
|
} |
75
|
|
|
|
76
|
|
|
/** |
77
|
|
|
* Recursive event loop that input intu recusion by remainder = 0 - thus when N users, |
78
|
|
|
* and when forks equals true which prevents it from infinite recursive iterations |
79
|
|
|
* |
80
|
|
|
* @param resource $server server connection |
81
|
|
|
* @param bool $fork flag to fork or run event loop |
82
|
|
|
*/ |
83
|
|
|
private function eventLoop($server, bool $fork = false) |
84
|
|
|
{ |
85
|
|
|
if ($fork === true) { |
86
|
|
|
$pid = pcntl_fork(); |
87
|
|
|
|
88
|
|
|
if ($pid) { // run eventLoop in parent |
89
|
|
|
@cli_set_process_title(self::PROC_TITLE); |
|
|
|
|
90
|
|
|
$this->eventLoop($server); |
91
|
|
|
} |
92
|
|
|
} else { |
93
|
|
|
$this->looping($server); |
94
|
|
|
} |
95
|
|
|
} |
96
|
|
|
|
97
|
|
|
/** |
98
|
|
|
* @param resource $server |
99
|
|
|
*/ |
100
|
|
|
private function looping($server) |
101
|
|
|
{ |
102
|
|
|
while (true) { |
103
|
|
|
$this->totalClients = count($this->clients) + 1; |
104
|
|
|
|
105
|
|
|
// maxClients prevents process fork on count down |
106
|
|
|
if ($this->totalClients > $this->maxClients) { |
107
|
|
|
$this->maxClients = $this->totalClients; |
108
|
|
|
} |
109
|
|
|
|
110
|
|
|
if ($this->totalClients !== 0 // avoid 0 process creation |
111
|
|
|
&& $this->totalClients % self::MAX_CLIENTS_REMAINDER_FORK === 0 // only when N is there |
112
|
|
|
&& true === $this->stepRecursion // only once |
113
|
|
|
&& $this->maxClients === $this->totalClients // only if stack grows |
114
|
|
|
) { |
115
|
|
|
$this->stepRecursion = false; |
116
|
|
|
$this->eventLoop($server, true); |
117
|
|
|
} |
118
|
|
|
|
119
|
|
|
if ($this->totalClients !== 0 && $this->totalClients % self::MAX_CLIENTS_REMAINDER_FORK === 0 && $this->maxClients > $this->totalClients) { // there is less connection for amount of processes at this moment |
120
|
|
|
exit(1); |
121
|
|
|
} |
122
|
|
|
|
123
|
|
|
//prepare readable sockets |
124
|
|
|
$readSocks = $this->clients; |
125
|
|
|
$readSocks[] = $server; |
126
|
|
|
|
127
|
|
|
//start reading and use a large timeout |
128
|
|
|
if (!stream_select($readSocks, $write, $except, self::STREAM_SELECT_TIMEOUT)) { |
129
|
|
|
die('something went wrong while selecting'); |
130
|
|
|
} |
131
|
|
|
|
132
|
|
|
//new client |
133
|
|
|
if (in_array($server, $readSocks, false)) { |
134
|
|
|
$this->acceptNewClient($server, $readSocks); |
135
|
|
|
} |
136
|
|
|
|
137
|
|
|
//message from existing client |
138
|
|
|
$this->messagesWorker($readSocks); |
139
|
|
|
} |
140
|
|
|
} |
141
|
|
|
|
142
|
|
|
/** |
143
|
|
|
* @param resource $server |
144
|
|
|
* @param array $readSocks |
145
|
|
|
*/ |
146
|
|
|
private function acceptNewClient($server, array &$readSocks) : void |
147
|
|
|
{ |
148
|
|
|
$newClient = stream_socket_accept($server, 0); // must be 0 to non-block |
149
|
|
|
if ($newClient) { |
150
|
|
|
// print remote client information, ip and port number |
151
|
|
|
// $socketName = stream_socket_get_name($newClient, true); |
152
|
|
|
// important to read from headers here coz later client will change and there will be only msgs on pipe |
153
|
|
|
$headers = fread($newClient, self::HEADER_BYTES_READ); |
154
|
|
|
if (empty($this->handler->pathParams[0]) === false) { |
|
|
|
|
155
|
|
|
$this->setPathParams($headers); |
156
|
|
|
} |
157
|
|
|
$this->clients[] = $newClient; |
158
|
|
|
$this->stepRecursion = true; // set on new client coz of remainder % is always 0 |
159
|
|
|
// trigger OPEN event |
160
|
|
|
$this->handler->onOpen($this->connImpl->getConnection($newClient)); |
161
|
|
|
$this->handshake($newClient, $headers); |
162
|
|
|
} |
163
|
|
|
//delete the server socket from the read sockets |
164
|
|
|
unset($readSocks[array_search($server, $readSocks, false)]); |
165
|
|
|
} |
166
|
|
|
|
167
|
|
|
/** |
168
|
|
|
* @uses onMessage |
169
|
|
|
* @uses onPing |
170
|
|
|
* @uses onPong |
171
|
|
|
* @param array $readSocks |
172
|
|
|
*/ |
173
|
|
|
private function messagesWorker(array $readSocks) : void |
174
|
|
|
{ |
175
|
|
|
foreach ($readSocks as $kSock => $sock) { |
176
|
|
|
$data = $this->decode(fread($sock, self::MAX_BYTES_READ)); |
177
|
|
|
$dataType = $data['type']; |
178
|
|
|
$dataPayload = $data['payload']; |
179
|
|
|
// to manipulate connection through send/close methods via handler, specified in IConnection |
180
|
|
|
$this->cureentConn = $this->connImpl->getConnection($sock); |
181
|
|
|
if (empty($data) || $dataType === self::EVENT_TYPE_CLOSE) { // close event triggered from client - browser tab or close socket event |
182
|
|
|
// trigger CLOSE event |
183
|
|
|
try { |
184
|
|
|
$this->handler->onClose($this->cureentConn); |
185
|
|
|
} catch (WebSocketException $e) { |
186
|
|
|
$e->printStack(); |
187
|
|
|
} |
188
|
|
|
// to avoid event leaks |
189
|
|
|
unset($this->clients[array_search($sock, $this->clients)], $readSocks[$kSock]); |
190
|
|
|
continue; |
191
|
|
|
} |
192
|
|
|
|
193
|
|
|
if (method_exists($this->handler, self::MAP_EVENT_TYPE_TO_METHODS[$dataType])) { |
194
|
|
|
try { |
195
|
|
|
// dynamic call: onMessage, onPing, onPong |
196
|
|
|
$this->handler->{self::MAP_EVENT_TYPE_TO_METHODS[$dataType]}($this->cureentConn, $dataPayload); |
197
|
|
|
} catch (WebSocketException $e) { |
198
|
|
|
$e->printStack(); |
199
|
|
|
} |
200
|
|
|
} |
201
|
|
|
} |
202
|
|
|
} |
203
|
|
|
|
204
|
|
|
/** |
205
|
|
|
* Message frames decoder |
206
|
|
|
* |
207
|
|
|
* @param string $data |
208
|
|
|
* @return mixed null on empty data|false on improper data|array - on success |
209
|
|
|
*/ |
210
|
|
|
private function decode(string $data) |
211
|
|
|
{ |
212
|
|
|
if (empty($data)) { |
213
|
|
|
return NULL; // close has been sent |
214
|
|
|
} |
215
|
|
|
|
216
|
|
|
$unmaskedPayload = ''; |
217
|
|
|
$decodedData = []; |
218
|
|
|
|
219
|
|
|
// estimate frame type: |
220
|
|
|
$firstByteBinary = sprintf('%08b', ord($data[0])); |
221
|
|
|
$secondByteBinary = sprintf('%08b', ord($data[1])); |
222
|
|
|
$opcode = bindec(substr($firstByteBinary, 4, 4)); |
223
|
|
|
$isMasked = $secondByteBinary[0] === '1'; |
224
|
|
|
$payloadLength = ord($data[1]) & self::MASK_127; |
225
|
|
|
|
226
|
|
|
// unmasked frame is received: |
227
|
|
|
if (!$isMasked) { |
228
|
|
|
return ['type' => '', 'payload' => '', 'error' => self::ERR_PROTOCOL]; |
229
|
|
|
} |
230
|
|
|
|
231
|
|
|
switch ($opcode) { |
232
|
|
|
// text frame: |
233
|
|
|
case self::DECODE_TEXT: |
234
|
|
|
$decodedData['type'] = self::EVENT_TYPE_TEXT; |
235
|
|
|
break; |
236
|
|
|
case self::DECODE_BINARY: |
237
|
|
|
$decodedData['type'] = self::EVENT_TYPE_BINARY; |
238
|
|
|
break; |
239
|
|
|
// connection close frame: |
240
|
|
|
case self::DECODE_CLOSE: |
241
|
|
|
$decodedData['type'] = self::EVENT_TYPE_CLOSE; |
242
|
|
|
break; |
243
|
|
|
// ping frame: |
244
|
|
|
case self::DECODE_PING: |
245
|
|
|
$decodedData['type'] = self::EVENT_TYPE_PING; |
246
|
|
|
break; |
247
|
|
|
// pong frame: |
248
|
|
|
case self::DECODE_PONG: |
249
|
|
|
$decodedData['type'] = self::EVENT_TYPE_PONG; |
250
|
|
|
break; |
251
|
|
|
default: |
252
|
|
|
return ['type' => '', 'payload' => '', 'error' => self::ERR_UNKNOWN_OPCODE]; |
253
|
|
|
} |
254
|
|
|
|
255
|
|
|
if ($payloadLength === self::MASK_126) { |
256
|
|
|
$mask = substr($data, 4, 4); |
257
|
|
|
$payloadOffset = self::PAYLOAD_OFFSET_8; |
258
|
|
|
$dataLength = bindec(sprintf('%08b', ord($data[2])) . sprintf('%08b', ord($data[3]))) + $payloadOffset; |
259
|
|
|
} elseif ($payloadLength === self::MASK_127) { |
260
|
|
|
$mask = substr($data, 10, 4); |
261
|
|
|
$payloadOffset = self::PAYLOAD_OFFSET_14; |
262
|
|
|
$tmp = ''; |
263
|
|
|
for ($i = 0; $i < 8; $i++) { |
264
|
|
|
$tmp .= sprintf('%08b', ord($data[$i + 2])); |
265
|
|
|
} |
266
|
|
|
$dataLength = bindec($tmp) + $payloadOffset; |
267
|
|
|
unset($tmp); |
268
|
|
|
} else { |
269
|
|
|
$mask = substr($data, 2, 4); |
270
|
|
|
$payloadOffset = self::PAYLOAD_OFFSET_6; |
271
|
|
|
$dataLength = $payloadLength + $payloadOffset; |
272
|
|
|
} |
273
|
|
|
|
274
|
|
|
/** |
275
|
|
|
* We have to check for large frames here. socket_recv cuts at 1024 bytes |
276
|
|
|
* so if websocket-frame is > 1024 bytes we have to wait until whole |
277
|
|
|
* data is transferd. |
278
|
|
|
*/ |
279
|
|
|
if (strlen($data) < $dataLength) { |
280
|
|
|
return false; |
281
|
|
|
} |
282
|
|
|
|
283
|
|
|
if ($isMasked) { |
284
|
|
|
for ($i = $payloadOffset; $i < $dataLength; $i++) { |
285
|
|
|
$j = $i - $payloadOffset; |
286
|
|
|
if (isset($data[$i])) { |
287
|
|
|
$unmaskedPayload .= $data[$i] ^ $mask[$j % 4]; |
288
|
|
|
} |
289
|
|
|
} |
290
|
|
|
$decodedData['payload'] = $unmaskedPayload; |
291
|
|
|
} else { |
292
|
|
|
$payloadOffset -= 4; |
293
|
|
|
$decodedData['payload'] = substr($data, $payloadOffset); |
294
|
|
|
} |
295
|
|
|
|
296
|
|
|
return $decodedData; |
297
|
|
|
} |
298
|
|
|
|
299
|
|
|
/** |
300
|
|
|
* Handshakes/upgrade and key parse |
301
|
|
|
* |
302
|
|
|
* @param resource $client Source client socket to write |
303
|
|
|
* @param string $headers Headers that client has been sent |
304
|
|
|
* @return string socket handshake key (Sec-WebSocket-Key)| false on parse error |
305
|
|
|
*/ |
306
|
|
|
private function handshake($client, string $headers) : string |
307
|
|
|
{ |
308
|
|
|
$match = []; |
309
|
|
|
preg_match(self::SEC_WEBSOCKET_KEY_PTRN, $headers, $match); |
310
|
|
|
if (empty($match[1])) { |
311
|
|
|
return false; |
312
|
|
|
} |
313
|
|
|
|
314
|
|
|
$key = $match[1]; |
315
|
|
|
$this->handshakes[(int)$client] = $key; |
316
|
|
|
// sending header according to WebSocket Protocol |
317
|
|
|
$secWebSocketAccept = base64_encode(sha1(trim($key) . self::HEADER_WEBSOCKET_ACCEPT_HASH, true)); |
318
|
|
|
$this->setHeadersUpgrade($secWebSocketAccept); |
319
|
|
|
$upgradeHeaders = $this->getHeadersUpgrade(); |
320
|
|
|
|
321
|
|
|
fwrite($client, $upgradeHeaders); |
322
|
|
|
return $key; |
323
|
|
|
} |
324
|
|
|
|
325
|
|
|
/** |
326
|
|
|
* Sets an array of headers needed to upgrade server/client connection |
327
|
|
|
* |
328
|
|
|
* @param string $secWebSocketAccept base64 encoded Sec-WebSocket-Accept header |
329
|
|
|
*/ |
330
|
|
|
private function setHeadersUpgrade($secWebSocketAccept) : void |
331
|
|
|
{ |
332
|
|
|
$this->headersUpgrade = [ |
333
|
|
|
self::HEADERS_UPGRADE_KEY => self::HEADERS_UPGRADE_VALUE, |
334
|
|
|
self::HEADERS_CONNECTION_KEY => self::HEADERS_CONNECTION_VALUE, |
335
|
|
|
self::HEADERS_SEC_WEBSOCKET_ACCEPT_KEY => ' ' . $secWebSocketAccept // the space before key is really important |
336
|
|
|
]; |
337
|
|
|
} |
338
|
|
|
|
339
|
|
|
/** |
340
|
|
|
* Retreives headers from an array of headers to upgrade server/client connection |
341
|
|
|
* |
342
|
|
|
* @return string Headers to Upgrade communication connection |
343
|
|
|
*/ |
344
|
|
|
private function getHeadersUpgrade() : string |
345
|
|
|
{ |
346
|
|
|
$handShakeHeaders = self::HEADER_HTTP1_1 . self::HEADERS_EOL; |
347
|
|
|
if (empty($this->headersUpgrade)) { |
348
|
|
|
die('Headers array is not set' . PHP_EOL); |
349
|
|
|
} |
350
|
|
|
foreach ($this->headersUpgrade as $key => $header) { |
351
|
|
|
$handShakeHeaders .= $key . ':' . $header . self::HEADERS_EOL; |
352
|
|
|
if ($key === self::HEADERS_SEC_WEBSOCKET_ACCEPT_KEY) { // add additional EOL fo Sec-WebSocket-Accept |
353
|
|
|
$handShakeHeaders .= self::HEADERS_EOL; |
354
|
|
|
} |
355
|
|
|
} |
356
|
|
|
return $handShakeHeaders; |
357
|
|
|
} |
358
|
|
|
|
359
|
|
|
/** |
360
|
|
|
* Parses parameters from GET on web-socket client connection before handshake |
361
|
|
|
* |
362
|
|
|
* @param string $headers |
363
|
|
|
*/ |
364
|
|
|
private function setPathParams(string $headers) : void |
365
|
|
|
{ |
366
|
|
|
/** @var WebSocketMessageContract $handler */ |
367
|
|
|
if (empty($this->handler->pathParams) === false) { |
|
|
|
|
368
|
|
|
$matches = []; |
369
|
|
|
preg_match('/GET\s(.*?)\s/', $headers, $matches); |
370
|
|
|
$left = $matches[1]; |
371
|
|
|
foreach ($this->handler->pathParams as $k => $param) { |
|
|
|
|
372
|
|
|
if (empty($this->handler->pathParams[$k + 1]) && strpos($left, '/', 1) === false) { |
|
|
|
|
373
|
|
|
// do not eat last char if there is no / at the end |
374
|
|
|
$this->handler->pathParams[$param] = substr($left, strpos($left, '/') + 1); |
|
|
|
|
375
|
|
|
} else { |
376
|
|
|
// eat both slashes |
377
|
|
|
$this->handler->pathParams[$param] = substr($left, strpos($left, '/') + 1, strpos($left, '/', 1) - 1); |
|
|
|
|
378
|
|
|
} |
379
|
|
|
// clear the declaration of parsed param |
380
|
|
|
unset($this->handler->pathParams[array_search($param, $this->handler->pathParams, false)]); |
381
|
|
|
$left = substr($left, strpos($left, '/', 1)); |
382
|
|
|
} |
383
|
|
|
} |
384
|
|
|
} |
385
|
|
|
|
386
|
|
|
} |
387
|
|
|
|
If you suppress an error, we recommend checking for the error condition explicitly: