TcpConnection.php 29 KB

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