TcpConnection.php 29 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113
  1. <?php
  2. /**
  3. * This file is part of workerman.
  4. *
  5. * Licensed under The MIT License
  6. * For full copyright and license information, please see the MIT-LICENSE.txt
  7. * Redistributions of files must retain the above copyright notice.
  8. *
  9. * @author walkor<walkor@workerman.net>
  10. * @copyright walkor<walkor@workerman.net>
  11. * @link http://www.workerman.net/
  12. * @license http://www.opensource.org/licenses/mit-license.php MIT License
  13. */
  14. declare(strict_types=1);
  15. namespace Workerman\Connection;
  16. use JsonSerializable;
  17. use RuntimeException;
  18. use stdClass;
  19. use Throwable;
  20. use Workerman\Events\EventInterface;
  21. use Workerman\Protocols\Http\Request;
  22. use Workerman\Protocols\ProtocolInterface;
  23. use Workerman\Worker;
  24. use function ceil;
  25. use function count;
  26. use function fclose;
  27. use function feof;
  28. use function fread;
  29. use function function_exists;
  30. use function fwrite;
  31. use function is_object;
  32. use function is_resource;
  33. use function key;
  34. use function method_exists;
  35. use function posix_getpid;
  36. use function restore_error_handler;
  37. use function set_error_handler;
  38. use function stream_set_blocking;
  39. use function stream_set_read_buffer;
  40. use function stream_socket_enable_crypto;
  41. use function stream_socket_get_name;
  42. use function strlen;
  43. use function strrchr;
  44. use function strrpos;
  45. use function substr;
  46. use function var_export;
  47. use const PHP_INT_MAX;
  48. use const STREAM_CRYPTO_METHOD_SSLv23_CLIENT;
  49. use const STREAM_CRYPTO_METHOD_SSLv23_SERVER;
  50. use const STREAM_CRYPTO_METHOD_SSLv2_CLIENT;
  51. use const STREAM_CRYPTO_METHOD_SSLv2_SERVER;
  52. /**
  53. * TcpConnection.
  54. * @property string $websocketType
  55. */
  56. class TcpConnection extends ConnectionInterface implements JsonSerializable
  57. {
  58. /**
  59. * Read buffer size.
  60. *
  61. * @var int
  62. */
  63. public const READ_BUFFER_SIZE = 87380;
  64. /**
  65. * Status initial.
  66. *
  67. * @var int
  68. */
  69. public const STATUS_INITIAL = 0;
  70. /**
  71. * Status connecting.
  72. *
  73. * @var int
  74. */
  75. public const STATUS_CONNECTING = 1;
  76. /**
  77. * Status connection established.
  78. *
  79. * @var int
  80. */
  81. public const STATUS_ESTABLISHED = 2;
  82. /**
  83. * Status closing.
  84. *
  85. * @var int
  86. */
  87. public const STATUS_CLOSING = 4;
  88. /**
  89. * Status closed.
  90. *
  91. * @var int
  92. */
  93. public const STATUS_CLOSED = 8;
  94. /**
  95. * Maximum string length for cache
  96. *
  97. * @var int
  98. */
  99. public const MAX_CACHE_STRING_LENGTH = 2048;
  100. /**
  101. * Maximum cache size.
  102. *
  103. * @var int
  104. */
  105. public const MAX_CACHE_SIZE = 512;
  106. /**
  107. * Emitted when socket connection is successfully established.
  108. *
  109. * @var ?callable
  110. */
  111. public $onConnect = null;
  112. /**
  113. * Emitted before websocket handshake (Only works when protocol is ws).
  114. *
  115. * @var ?callable
  116. */
  117. public $onWebSocketConnect = null;
  118. /**
  119. * Emitted after websocket handshake (Only works when protocol is ws).
  120. *
  121. * @var ?callable
  122. */
  123. public $onWebSocketConnected = null;
  124. /**
  125. * Emitted when data is received.
  126. *
  127. * @var ?callable
  128. */
  129. public $onMessage = null;
  130. /**
  131. * Emitted when the other end of the socket sends a FIN packet.
  132. *
  133. * @var ?callable
  134. */
  135. public $onClose = null;
  136. /**
  137. * Emitted when an error occurs with connection.
  138. *
  139. * @var ?callable
  140. */
  141. public $onError = null;
  142. /**
  143. * Emitted when the send buffer becomes full.
  144. *
  145. * @var ?callable
  146. */
  147. public $onBufferFull = null;
  148. /**
  149. * Emitted when send buffer becomes empty.
  150. *
  151. * @var ?callable
  152. */
  153. public $onBufferDrain = null;
  154. /**
  155. * Transport (tcp/udp/unix/ssl).
  156. *
  157. * @var string
  158. */
  159. public string $transport = 'tcp';
  160. /**
  161. * Which worker belong to.
  162. *
  163. * @var ?Worker
  164. */
  165. public ?Worker $worker = null;
  166. /**
  167. * Bytes read.
  168. *
  169. * @var int
  170. */
  171. public int $bytesRead = 0;
  172. /**
  173. * Bytes written.
  174. *
  175. * @var int
  176. */
  177. public int $bytesWritten = 0;
  178. /**
  179. * Connection->id.
  180. *
  181. * @var int
  182. */
  183. public int $id = 0;
  184. /**
  185. * A copy of $worker->id which used to clean up the connection in worker->connections
  186. *
  187. * @var int
  188. */
  189. protected int $realId = 0;
  190. /**
  191. * Sets the maximum send buffer size for the current connection.
  192. * OnBufferFull callback will be emitted When send buffer is full.
  193. *
  194. * @var int
  195. */
  196. public int $maxSendBufferSize = 1048576;
  197. /**
  198. * Context.
  199. *
  200. * @var ?stdClass
  201. */
  202. public ?stdClass $context = null;
  203. /**
  204. * @var array
  205. */
  206. public array $headers = [];
  207. /**
  208. * @var ?Request
  209. */
  210. public ?Request $request = null;
  211. /**
  212. * Default send buffer size.
  213. *
  214. * @var int
  215. */
  216. public static int $defaultMaxSendBufferSize = 1048576;
  217. /**
  218. * Sets the maximum acceptable packet size for the current connection.
  219. *
  220. * @var int
  221. */
  222. public int $maxPackageSize = 1048576;
  223. /**
  224. * Default maximum acceptable packet size.
  225. *
  226. * @var int
  227. */
  228. public static int $defaultMaxPackageSize = 10485760;
  229. /**
  230. * Id recorder.
  231. *
  232. * @var int
  233. */
  234. protected static int $idRecorder = 1;
  235. /**
  236. * Cache.
  237. *
  238. * @var bool.
  239. */
  240. protected static bool $enableCache = true;
  241. /**
  242. * Socket
  243. *
  244. * @var resource
  245. */
  246. protected $socket = null;
  247. /**
  248. * Send buffer.
  249. *
  250. * @var string
  251. */
  252. protected string $sendBuffer = '';
  253. /**
  254. * Receive buffer.
  255. *
  256. * @var string
  257. */
  258. protected string $recvBuffer = '';
  259. /**
  260. * Current package length.
  261. *
  262. * @var int
  263. */
  264. protected int $currentPackageLength = 0;
  265. /**
  266. * Connection status.
  267. *
  268. * @var int
  269. */
  270. protected int $status = self::STATUS_ESTABLISHED;
  271. /**
  272. * Remote address.
  273. *
  274. * @var string
  275. */
  276. protected string $remoteAddress = '';
  277. /**
  278. * Is paused.
  279. *
  280. * @var bool
  281. */
  282. protected bool $isPaused = false;
  283. /**
  284. * SSL handshake completed or not.
  285. *
  286. * @var bool
  287. */
  288. protected bool|int $sslHandshakeCompleted = false;
  289. /**
  290. * All connection instances.
  291. *
  292. * @var array
  293. */
  294. public static array $connections = [];
  295. /**
  296. * Status to string.
  297. *
  298. * @var array
  299. */
  300. public const STATUS_TO_STRING = [
  301. self::STATUS_INITIAL => 'INITIAL',
  302. self::STATUS_CONNECTING => 'CONNECTING',
  303. self::STATUS_ESTABLISHED => 'ESTABLISHED',
  304. self::STATUS_CLOSING => 'CLOSING',
  305. self::STATUS_CLOSED => 'CLOSED',
  306. ];
  307. /**
  308. * Construct.
  309. *
  310. * @param EventInterface $eventLoop
  311. * @param resource $socket
  312. * @param string $remoteAddress
  313. */
  314. public function __construct(EventInterface $eventLoop, $socket, string $remoteAddress = '')
  315. {
  316. ++self::$statistics['connection_count'];
  317. $this->id = $this->realId = self::$idRecorder++;
  318. if (self::$idRecorder === PHP_INT_MAX) {
  319. self::$idRecorder = 0;
  320. }
  321. $this->socket = $socket;
  322. stream_set_blocking($this->socket, false);
  323. // Compatible with hhvm
  324. if (function_exists('stream_set_read_buffer')) {
  325. stream_set_read_buffer($this->socket, 0);
  326. }
  327. $this->eventLoop = $eventLoop;
  328. $this->eventLoop->onReadable($this->socket, $this->baseRead(...));
  329. $this->maxSendBufferSize = self::$defaultMaxSendBufferSize;
  330. $this->maxPackageSize = self::$defaultMaxPackageSize;
  331. $this->remoteAddress = $remoteAddress;
  332. static::$connections[$this->id] = $this;
  333. $this->context = new stdClass();
  334. }
  335. /**
  336. * Get status.
  337. *
  338. * @param bool $rawOutput
  339. *
  340. * @return int|string
  341. */
  342. public function getStatus(bool $rawOutput = true): int|string
  343. {
  344. if ($rawOutput) {
  345. return $this->status;
  346. }
  347. return self::STATUS_TO_STRING[$this->status];
  348. }
  349. /**
  350. * Sends data on the connection.
  351. *
  352. * @param mixed $sendBuffer
  353. * @param bool $raw
  354. * @return bool|null
  355. */
  356. public function send(mixed $sendBuffer, bool $raw = false): bool|null
  357. {
  358. if ($this->status === self::STATUS_CLOSING || $this->status === self::STATUS_CLOSED) {
  359. return false;
  360. }
  361. // Try to call protocol::encode($sendBuffer) before sending.
  362. if (false === $raw && $this->protocol !== null) {
  363. /** @var ProtocolInterface $parser */
  364. $parser = $this->protocol;
  365. try {
  366. $sendBuffer = $parser::encode($sendBuffer, $this);
  367. } catch(Throwable $e) {
  368. $this->error($e);
  369. }
  370. if ($sendBuffer === '') {
  371. return null;
  372. }
  373. }
  374. if ($this->status !== self::STATUS_ESTABLISHED ||
  375. ($this->transport === 'ssl' && $this->sslHandshakeCompleted !== true)
  376. ) {
  377. if ($this->sendBuffer && $this->bufferIsFull()) {
  378. ++self::$statistics['send_fail'];
  379. return false;
  380. }
  381. $this->sendBuffer .= $sendBuffer;
  382. $this->checkBufferWillFull();
  383. return null;
  384. }
  385. // Attempt to send data directly.
  386. if ($this->sendBuffer === '') {
  387. if ($this->transport === 'ssl') {
  388. $this->eventLoop->onWritable($this->socket, $this->baseWrite(...));
  389. $this->sendBuffer = $sendBuffer;
  390. $this->checkBufferWillFull();
  391. return null;
  392. }
  393. $len = 0;
  394. try {
  395. $len = @fwrite($this->socket, $sendBuffer);
  396. } catch (Throwable $e) {
  397. Worker::log($e);
  398. }
  399. // send successful.
  400. if ($len === strlen($sendBuffer)) {
  401. $this->bytesWritten += $len;
  402. return true;
  403. }
  404. // Send only part of the data.
  405. if ($len > 0) {
  406. $this->sendBuffer = substr($sendBuffer, $len);
  407. $this->bytesWritten += $len;
  408. } else {
  409. // Connection closed?
  410. if (!is_resource($this->socket) || feof($this->socket)) {
  411. ++self::$statistics['send_fail'];
  412. if ($this->onError) {
  413. try {
  414. ($this->onError)($this, static::SEND_FAIL, 'client closed');
  415. } catch (Throwable $e) {
  416. $this->error($e);
  417. }
  418. }
  419. $this->destroy();
  420. return false;
  421. }
  422. $this->sendBuffer = $sendBuffer;
  423. }
  424. $this->eventLoop->onWritable($this->socket, $this->baseWrite(...));
  425. // Check if send buffer will be full.
  426. $this->checkBufferWillFull();
  427. return null;
  428. }
  429. if ($this->bufferIsFull()) {
  430. ++self::$statistics['send_fail'];
  431. return false;
  432. }
  433. $this->sendBuffer .= $sendBuffer;
  434. // Check if send buffer is full.
  435. $this->checkBufferWillFull();
  436. return null;
  437. }
  438. /**
  439. * Get remote IP.
  440. *
  441. * @return string
  442. */
  443. public function getRemoteIp(): string
  444. {
  445. $pos = strrpos($this->remoteAddress, ':');
  446. if ($pos) {
  447. return substr($this->remoteAddress, 0, $pos);
  448. }
  449. return '';
  450. }
  451. /**
  452. * Get remote port.
  453. *
  454. * @return int
  455. */
  456. public function getRemotePort(): int
  457. {
  458. if ($this->remoteAddress) {
  459. return (int)substr(strrchr($this->remoteAddress, ':'), 1);
  460. }
  461. return 0;
  462. }
  463. /**
  464. * Get remote address.
  465. *
  466. * @return string
  467. */
  468. public function getRemoteAddress(): string
  469. {
  470. return $this->remoteAddress;
  471. }
  472. /**
  473. * Get local IP.
  474. *
  475. * @return string
  476. */
  477. public function getLocalIp(): string
  478. {
  479. $address = $this->getLocalAddress();
  480. $pos = strrpos($address, ':');
  481. if (!$pos) {
  482. return '';
  483. }
  484. return substr($address, 0, $pos);
  485. }
  486. /**
  487. * Get local port.
  488. *
  489. * @return int
  490. */
  491. public function getLocalPort(): int
  492. {
  493. $address = $this->getLocalAddress();
  494. $pos = strrpos($address, ':');
  495. if (!$pos) {
  496. return 0;
  497. }
  498. return (int)substr(strrchr($address, ':'), 1);
  499. }
  500. /**
  501. * Get local address.
  502. *
  503. * @return string
  504. */
  505. public function getLocalAddress(): string
  506. {
  507. if (!is_resource($this->socket)) {
  508. return '';
  509. }
  510. return (string)@stream_socket_get_name($this->socket, false);
  511. }
  512. /**
  513. * Get send buffer queue size.
  514. *
  515. * @return integer
  516. */
  517. public function getSendBufferQueueSize(): int
  518. {
  519. return strlen($this->sendBuffer);
  520. }
  521. /**
  522. * Get receive buffer queue size.
  523. *
  524. * @return integer
  525. */
  526. public function getRecvBufferQueueSize(): int
  527. {
  528. return strlen($this->recvBuffer);
  529. }
  530. /**
  531. * Pauses the reading of data. That is onMessage will not be emitted. Useful to throttle back an upload.
  532. *
  533. * @return void
  534. */
  535. public function pauseRecv(): void
  536. {
  537. $this->eventLoop->offReadable($this->socket);
  538. $this->isPaused = true;
  539. }
  540. /**
  541. * Resumes reading after a call to pauseRecv.
  542. *
  543. * @return void
  544. */
  545. public function resumeRecv(): void
  546. {
  547. if ($this->isPaused === true) {
  548. $this->eventLoop->onReadable($this->socket, $this->baseRead(...));
  549. $this->isPaused = false;
  550. $this->baseRead($this->socket, false);
  551. }
  552. }
  553. /**
  554. * Base read handler.
  555. *
  556. * @param resource $socket
  557. * @param bool $checkEof
  558. * @return void
  559. */
  560. public function baseRead($socket, bool $checkEof = true): void
  561. {
  562. static $requests = [];
  563. // SSL handshake.
  564. if ($this->transport === 'ssl' && $this->sslHandshakeCompleted !== true) {
  565. if ($this->doSslHandshake($socket)) {
  566. $this->sslHandshakeCompleted = true;
  567. if ($this->sendBuffer) {
  568. $this->eventLoop->onWritable($socket, $this->baseWrite(...));
  569. }
  570. } else {
  571. return;
  572. }
  573. }
  574. $buffer = '';
  575. try {
  576. $buffer = @fread($socket, self::READ_BUFFER_SIZE);
  577. } catch (Throwable) {
  578. // do nothing
  579. }
  580. // Check connection closed.
  581. if ($buffer === '' || $buffer === false) {
  582. if ($checkEof && (feof($socket) || !is_resource($socket) || $buffer === false)) {
  583. $this->destroy();
  584. return;
  585. }
  586. } else {
  587. $this->bytesRead += strlen($buffer);
  588. if ($this->recvBuffer === '') {
  589. if (static::$enableCache && isset($requests[$buffer])) {
  590. ++self::$statistics['total_request'];
  591. $request = $requests[$buffer];
  592. if ($request instanceof Request) {
  593. $request = clone $request;
  594. $requests[$buffer] = $request;
  595. $request->connection = $this;
  596. $this->request = $request;
  597. $request->properties = [];
  598. }
  599. try {
  600. ($this->onMessage)($this, $request);
  601. } catch (Throwable $e) {
  602. $this->error($e);
  603. }
  604. return;
  605. }
  606. $this->recvBuffer = $buffer;
  607. } else {
  608. $this->recvBuffer .= $buffer;
  609. }
  610. }
  611. // If the application layer protocol has been set up.
  612. if ($this->protocol !== null) {
  613. while ($this->recvBuffer !== '' && !$this->isPaused) {
  614. // The current packet length is known.
  615. if ($this->currentPackageLength) {
  616. // Data is not enough for a package.
  617. if ($this->currentPackageLength > strlen($this->recvBuffer)) {
  618. break;
  619. }
  620. } else {
  621. // Get current package length.
  622. try {
  623. /** @var ProtocolInterface $parser */
  624. $parser = $this->protocol;
  625. $this->currentPackageLength = $parser::input($this->recvBuffer, $this);
  626. } catch (Throwable) {
  627. }
  628. // The packet length is unknown.
  629. if ($this->currentPackageLength === 0) {
  630. break;
  631. } elseif ($this->currentPackageLength > 0 && $this->currentPackageLength <= $this->maxPackageSize) {
  632. // Data is not enough for a package.
  633. if ($this->currentPackageLength > strlen($this->recvBuffer)) {
  634. break;
  635. }
  636. } // Wrong package.
  637. else {
  638. Worker::safeEcho((string)(new RuntimeException("Protocol $this->protocol Error package. package_length=" . var_export($this->currentPackageLength, true))));
  639. $this->destroy();
  640. return;
  641. }
  642. }
  643. // The data is enough for a packet.
  644. ++self::$statistics['total_request'];
  645. // The current packet length is equal to the length of the buffer.
  646. if ($one = (strlen($this->recvBuffer) === $this->currentPackageLength)) {
  647. $oneRequestBuffer = $this->recvBuffer;
  648. $this->recvBuffer = '';
  649. } else {
  650. // Get a full package from the buffer.
  651. $oneRequestBuffer = substr($this->recvBuffer, 0, $this->currentPackageLength);
  652. // Remove the current package from receive buffer.
  653. $this->recvBuffer = substr($this->recvBuffer, $this->currentPackageLength);
  654. }
  655. // Reset the current packet length to 0.
  656. $this->currentPackageLength = 0;
  657. try {
  658. // Decode request buffer before Emitting onMessage callback.
  659. /** @var ProtocolInterface $parser */
  660. $parser = $this->protocol;
  661. $request = $parser::decode($oneRequestBuffer, $this);
  662. if (static::$enableCache && (!is_object($request) || $request instanceof Request) && $one && !isset($oneRequestBuffer[static::MAX_CACHE_STRING_LENGTH])) {
  663. $requests[$oneRequestBuffer] = $request;
  664. if (count($requests) > static::MAX_CACHE_SIZE) {
  665. unset($requests[key($requests)]);
  666. }
  667. }
  668. ($this->onMessage)($this, $request);
  669. } catch (Throwable $e) {
  670. $this->error($e);
  671. }
  672. }
  673. return;
  674. }
  675. if ($this->recvBuffer === '' || $this->isPaused) {
  676. return;
  677. }
  678. // Applications protocol is not set.
  679. ++self::$statistics['total_request'];
  680. try {
  681. ($this->onMessage)($this, $this->recvBuffer);
  682. } catch (Throwable $e) {
  683. $this->error($e);
  684. }
  685. // Clean receive buffer.
  686. $this->recvBuffer = '';
  687. }
  688. /**
  689. * Base write handler.
  690. *
  691. * @return void
  692. */
  693. public function baseWrite(): void
  694. {
  695. $len = 0;
  696. try {
  697. if ($this->transport === 'ssl') {
  698. $len = @fwrite($this->socket, $this->sendBuffer, 8192);
  699. } else {
  700. $len = @fwrite($this->socket, $this->sendBuffer);
  701. }
  702. } catch (Throwable) {
  703. }
  704. if ($len === strlen($this->sendBuffer)) {
  705. $this->bytesWritten += $len;
  706. $this->eventLoop->offWritable($this->socket);
  707. $this->sendBuffer = '';
  708. // Try to emit onBufferDrain callback when send buffer becomes empty.
  709. if ($this->onBufferDrain) {
  710. try {
  711. ($this->onBufferDrain)($this);
  712. } catch (Throwable $e) {
  713. $this->error($e);
  714. }
  715. }
  716. if ($this->status === self::STATUS_CLOSING) {
  717. if (!empty($this->context->streamSending)) {
  718. return;
  719. }
  720. $this->destroy();
  721. }
  722. return;
  723. }
  724. if ($len > 0) {
  725. $this->bytesWritten += $len;
  726. $this->sendBuffer = substr($this->sendBuffer, $len);
  727. } else {
  728. ++self::$statistics['send_fail'];
  729. $this->destroy();
  730. }
  731. }
  732. /**
  733. * SSL handshake.
  734. *
  735. * @param resource $socket
  736. * @return bool|int
  737. */
  738. public function doSslHandshake($socket): bool|int
  739. {
  740. if (feof($socket)) {
  741. $this->destroy();
  742. return false;
  743. }
  744. $async = $this instanceof AsyncTcpConnection;
  745. /**
  746. * We disabled ssl3 because https://blog.qualys.com/ssllabs/2014/10/15/ssl-3-is-dead-killed-by-the-poodle-attack.
  747. * You can enable ssl3 by the codes below.
  748. */
  749. /*if($async){
  750. $type = STREAM_CRYPTO_METHOD_SSLv2_CLIENT | STREAM_CRYPTO_METHOD_SSLv23_CLIENT | STREAM_CRYPTO_METHOD_SSLv3_CLIENT;
  751. }else{
  752. $type = STREAM_CRYPTO_METHOD_SSLv2_SERVER | STREAM_CRYPTO_METHOD_SSLv23_SERVER | STREAM_CRYPTO_METHOD_SSLv3_SERVER;
  753. }*/
  754. if ($async) {
  755. $type = STREAM_CRYPTO_METHOD_SSLv2_CLIENT | STREAM_CRYPTO_METHOD_SSLv23_CLIENT;
  756. } else {
  757. $type = STREAM_CRYPTO_METHOD_SSLv2_SERVER | STREAM_CRYPTO_METHOD_SSLv23_SERVER;
  758. }
  759. // Hidden error.
  760. set_error_handler(static function (int $code, string $msg): bool {
  761. if (!Worker::$daemonize) {
  762. Worker::safeEcho(sprintf("SSL handshake error: %s\n", $msg));
  763. }
  764. return true;
  765. });
  766. $ret = stream_socket_enable_crypto($socket, true, $type);
  767. restore_error_handler();
  768. // Negotiation has failed.
  769. if (false === $ret) {
  770. $this->destroy();
  771. return false;
  772. }
  773. if (0 === $ret) {
  774. // There isn't enough data and should try again.
  775. return 0;
  776. }
  777. return true;
  778. }
  779. /**
  780. * This method pulls all the data out of a readable stream, and writes it to the supplied destination.
  781. *
  782. * @param self $dest
  783. * @return void
  784. */
  785. public function pipe(self $dest): void
  786. {
  787. $source = $this;
  788. $this->onMessage = function ($source, $data) use ($dest) {
  789. $dest->send($data);
  790. };
  791. $this->onClose = function () use ($dest) {
  792. $dest->close();
  793. };
  794. $dest->onBufferFull = function () use ($source) {
  795. $source->pauseRecv();
  796. };
  797. $dest->onBufferDrain = function () use ($source) {
  798. $source->resumeRecv();
  799. };
  800. }
  801. /**
  802. * Remove $length of data from receive buffer.
  803. *
  804. * @param int $length
  805. * @return void
  806. */
  807. public function consumeRecvBuffer(int $length): void
  808. {
  809. $this->recvBuffer = substr($this->recvBuffer, $length);
  810. }
  811. /**
  812. * Close connection.
  813. *
  814. * @param mixed $data
  815. * @param bool $raw
  816. * @return void
  817. */
  818. public function close(mixed $data = null, bool $raw = false): void
  819. {
  820. if ($this->status === self::STATUS_CONNECTING) {
  821. $this->destroy();
  822. return;
  823. }
  824. if ($this->status === self::STATUS_CLOSING || $this->status === self::STATUS_CLOSED) {
  825. return;
  826. }
  827. if ($data !== null) {
  828. $this->send($data, $raw);
  829. }
  830. $this->status = self::STATUS_CLOSING;
  831. if ($this->sendBuffer === '') {
  832. $this->destroy();
  833. } else {
  834. $this->pauseRecv();
  835. }
  836. }
  837. /**
  838. * Is ipv4.
  839. *
  840. * return bool.
  841. */
  842. public function isIpV4(): bool
  843. {
  844. if ($this->transport === 'unix') {
  845. return false;
  846. }
  847. return !str_contains($this->getRemoteIp(), ':');
  848. }
  849. /**
  850. * Is ipv6.
  851. *
  852. * return bool.
  853. */
  854. public function isIpV6(): bool
  855. {
  856. if ($this->transport === 'unix') {
  857. return false;
  858. }
  859. return str_contains($this->getRemoteIp(), ':');
  860. }
  861. /**
  862. * Get the real socket.
  863. *
  864. * @return resource
  865. */
  866. public function getSocket()
  867. {
  868. return $this->socket;
  869. }
  870. /**
  871. * Check whether send buffer will be full.
  872. *
  873. * @return void
  874. */
  875. protected function checkBufferWillFull(): void
  876. {
  877. if ($this->onBufferFull && $this->maxSendBufferSize <= strlen($this->sendBuffer)) {
  878. try {
  879. ($this->onBufferFull)($this);
  880. } catch (Throwable $e) {
  881. $this->error($e);
  882. }
  883. }
  884. }
  885. /**
  886. * Whether send buffer is full.
  887. *
  888. * @return bool
  889. */
  890. protected function bufferIsFull(): bool
  891. {
  892. // Buffer has been marked as full but still has data to send then the packet is discarded.
  893. if ($this->maxSendBufferSize <= strlen($this->sendBuffer)) {
  894. if ($this->onError) {
  895. try {
  896. ($this->onError)($this, static::SEND_FAIL, 'send buffer full and drop package');
  897. } catch (Throwable $e) {
  898. $this->error($e);
  899. }
  900. }
  901. return true;
  902. }
  903. return false;
  904. }
  905. /**
  906. * Whether send buffer is Empty.
  907. *
  908. * @return bool
  909. */
  910. public function bufferIsEmpty(): bool
  911. {
  912. return empty($this->sendBuffer);
  913. }
  914. /**
  915. * Destroy connection.
  916. *
  917. * @return void
  918. */
  919. public function destroy(): void
  920. {
  921. // Avoid repeated calls.
  922. if ($this->status === self::STATUS_CLOSED) {
  923. return;
  924. }
  925. // Remove event listener.
  926. $this->eventLoop->offReadable($this->socket);
  927. $this->eventLoop->offWritable($this->socket);
  928. if (DIRECTORY_SEPARATOR === '\\' && method_exists($this->eventLoop, 'offExcept')) {
  929. $this->eventLoop->offExcept($this->socket);
  930. }
  931. // Close socket.
  932. try {
  933. @fclose($this->socket);
  934. } catch (Throwable) {
  935. }
  936. $this->status = self::STATUS_CLOSED;
  937. // Try to emit onClose callback.
  938. if ($this->onClose) {
  939. try {
  940. ($this->onClose)($this);
  941. } catch (Throwable $e) {
  942. $this->error($e);
  943. }
  944. }
  945. // Try to emit protocol::onClose
  946. if ($this->protocol && method_exists($this->protocol, 'onClose')) {
  947. try {
  948. $this->protocol::onClose($this);
  949. } catch (Throwable $e) {
  950. $this->error($e);
  951. }
  952. }
  953. $this->sendBuffer = $this->recvBuffer = '';
  954. $this->currentPackageLength = 0;
  955. $this->isPaused = $this->sslHandshakeCompleted = false;
  956. if ($this->status === self::STATUS_CLOSED) {
  957. // Cleaning up the callback to avoid memory leaks.
  958. $this->onMessage = $this->onClose = $this->onError = $this->onBufferFull = $this->onBufferDrain = $this->eventLoop = $this->errorHandler = null;
  959. // Remove from worker->connections.
  960. if ($this->worker) {
  961. unset($this->worker->connections[$this->realId]);
  962. }
  963. unset(static::$connections[$this->realId]);
  964. }
  965. }
  966. /**
  967. * Enable or disable Cache.
  968. *
  969. * @param bool $value
  970. */
  971. public static function enableCache(bool $value = true): void
  972. {
  973. static::$enableCache = $value;
  974. }
  975. /**
  976. * Get the json_encode information.
  977. *
  978. * @return array
  979. */
  980. public function jsonSerialize(): array
  981. {
  982. return [
  983. 'id' => $this->id,
  984. 'status' => $this->getStatus(),
  985. 'transport' => $this->transport,
  986. 'getRemoteIp' => $this->getRemoteIp(),
  987. 'remotePort' => $this->getRemotePort(),
  988. 'getRemoteAddress' => $this->getRemoteAddress(),
  989. 'getLocalIp' => $this->getLocalIp(),
  990. 'getLocalPort' => $this->getLocalPort(),
  991. 'getLocalAddress' => $this->getLocalAddress(),
  992. 'isIpV4' => $this->isIpV4(),
  993. 'isIpV6' => $this->isIpV6(),
  994. ];
  995. }
  996. /**
  997. * Destruct.
  998. *
  999. * @return void
  1000. */
  1001. public function __destruct()
  1002. {
  1003. static $mod;
  1004. self::$statistics['connection_count']--;
  1005. if (Worker::getGracefulStop()) {
  1006. $mod ??= ceil((self::$statistics['connection_count'] + 1) / 3);
  1007. if (0 === self::$statistics['connection_count'] % $mod) {
  1008. $pid = function_exists('posix_getpid') ? posix_getpid() : 0;
  1009. Worker::log('worker[' . $pid . '] remains ' . self::$statistics['connection_count'] . ' connection(s)');
  1010. }
  1011. if (0 === self::$statistics['connection_count']) {
  1012. Worker::stopAll();
  1013. }
  1014. }
  1015. }
  1016. }