TcpConnection.php 29 KB

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