TcpConnection.php 29 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091
  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. $sendBuffer = $parser::encode($sendBuffer, $this);
  348. if ($sendBuffer === '') {
  349. return;
  350. }
  351. }
  352. if ($this->status !== self::STATUS_ESTABLISHED ||
  353. ($this->transport === 'ssl' && $this->sslHandshakeCompleted !== true)
  354. ) {
  355. if ($this->sendBuffer && $this->bufferIsFull()) {
  356. ++self::$statistics['send_fail'];
  357. return false;
  358. }
  359. $this->sendBuffer .= $sendBuffer;
  360. $this->checkBufferWillFull();
  361. return;
  362. }
  363. // Attempt to send data directly.
  364. if ($this->sendBuffer === '') {
  365. if ($this->transport === 'ssl') {
  366. $this->eventLoop->onWritable($this->socket, [$this, 'baseWrite']);
  367. $this->sendBuffer = $sendBuffer;
  368. $this->checkBufferWillFull();
  369. return;
  370. }
  371. $len = 0;
  372. try {
  373. $len = @fwrite($this->socket, $sendBuffer);
  374. } catch (Throwable $e) {
  375. Worker::log($e);
  376. }
  377. // send successful.
  378. if ($len === strlen($sendBuffer)) {
  379. $this->bytesWritten += $len;
  380. return true;
  381. }
  382. // Send only part of the data.
  383. if ($len > 0) {
  384. $this->sendBuffer = substr($sendBuffer, $len);
  385. $this->bytesWritten += $len;
  386. } else {
  387. // Connection closed?
  388. if (!is_resource($this->socket) || feof($this->socket)) {
  389. ++self::$statistics['send_fail'];
  390. if ($this->onError) {
  391. try {
  392. ($this->onError)($this, static::SEND_FAIL, 'client closed');
  393. } catch (Throwable $e) {
  394. $this->error($e);
  395. }
  396. }
  397. $this->destroy();
  398. return false;
  399. }
  400. $this->sendBuffer = $sendBuffer;
  401. }
  402. $this->eventLoop->onWritable($this->socket, [$this, 'baseWrite']);
  403. // Check if send buffer will be full.
  404. $this->checkBufferWillFull();
  405. return;
  406. }
  407. if ($this->bufferIsFull()) {
  408. ++self::$statistics['send_fail'];
  409. return false;
  410. }
  411. $this->sendBuffer .= $sendBuffer;
  412. // Check if send buffer is full.
  413. $this->checkBufferWillFull();
  414. }
  415. /**
  416. * Get remote IP.
  417. *
  418. * @return string
  419. */
  420. public function getRemoteIp(): string
  421. {
  422. $pos = strrpos($this->remoteAddress, ':');
  423. if ($pos) {
  424. return substr($this->remoteAddress, 0, $pos);
  425. }
  426. return '';
  427. }
  428. /**
  429. * Get remote port.
  430. *
  431. * @return int
  432. */
  433. public function getRemotePort(): int
  434. {
  435. if ($this->remoteAddress) {
  436. return (int)substr(strrchr($this->remoteAddress, ':'), 1);
  437. }
  438. return 0;
  439. }
  440. /**
  441. * Get remote address.
  442. *
  443. * @return string
  444. */
  445. public function getRemoteAddress(): string
  446. {
  447. return $this->remoteAddress;
  448. }
  449. /**
  450. * Get local IP.
  451. *
  452. * @return string
  453. */
  454. public function getLocalIp(): string
  455. {
  456. $address = $this->getLocalAddress();
  457. $pos = strrpos($address, ':');
  458. if (!$pos) {
  459. return '';
  460. }
  461. return substr($address, 0, $pos);
  462. }
  463. /**
  464. * Get local port.
  465. *
  466. * @return int
  467. */
  468. public function getLocalPort(): int
  469. {
  470. $address = $this->getLocalAddress();
  471. $pos = strrpos($address, ':');
  472. if (!$pos) {
  473. return 0;
  474. }
  475. return (int)substr(strrchr($address, ':'), 1);
  476. }
  477. /**
  478. * Get local address.
  479. *
  480. * @return string
  481. */
  482. public function getLocalAddress(): string
  483. {
  484. if (!is_resource($this->socket)) {
  485. return '';
  486. }
  487. return (string)@stream_socket_get_name($this->socket, false);
  488. }
  489. /**
  490. * Get send buffer queue size.
  491. *
  492. * @return integer
  493. */
  494. public function getSendBufferQueueSize(): int
  495. {
  496. return strlen($this->sendBuffer);
  497. }
  498. /**
  499. * Get receive buffer queue size.
  500. *
  501. * @return integer
  502. */
  503. public function getRecvBufferQueueSize(): int
  504. {
  505. return strlen($this->recvBuffer);
  506. }
  507. /**
  508. * Pauses the reading of data. That is onMessage will not be emitted. Useful to throttle back an upload.
  509. *
  510. * @return void
  511. */
  512. public function pauseRecv(): void
  513. {
  514. $this->eventLoop->offReadable($this->socket);
  515. $this->isPaused = true;
  516. }
  517. /**
  518. * Resumes reading after a call to pauseRecv.
  519. *
  520. * @return void
  521. * @throws Throwable
  522. */
  523. public function resumeRecv(): void
  524. {
  525. if ($this->isPaused === true) {
  526. $this->eventLoop->onReadable($this->socket, [$this, 'baseRead']);
  527. $this->isPaused = false;
  528. $this->baseRead($this->socket, false);
  529. }
  530. }
  531. /**
  532. * Base read handler.
  533. *
  534. * @param resource $socket
  535. * @param bool $checkEof
  536. * @return void
  537. * @throws Throwable
  538. */
  539. public function baseRead($socket, bool $checkEof = true): void
  540. {
  541. static $requests = [];
  542. // SSL handshake.
  543. if ($this->transport === 'ssl' && $this->sslHandshakeCompleted !== true) {
  544. if ($this->doSslHandshake($socket)) {
  545. $this->sslHandshakeCompleted = true;
  546. if ($this->sendBuffer) {
  547. $this->eventLoop->onWritable($socket, [$this, 'baseWrite']);
  548. }
  549. } else {
  550. return;
  551. }
  552. }
  553. $buffer = '';
  554. try {
  555. $buffer = @fread($socket, self::READ_BUFFER_SIZE);
  556. } catch (Throwable) {
  557. }
  558. // Check connection closed.
  559. if ($buffer === '' || $buffer === false) {
  560. if ($checkEof && (feof($socket) || !is_resource($socket) || $buffer === false)) {
  561. $this->destroy();
  562. return;
  563. }
  564. } else {
  565. $this->bytesRead += strlen($buffer);
  566. if ($this->recvBuffer === '') {
  567. if (static::$enableCache && !isset($buffer[512]) && isset($requests[$buffer])) {
  568. ++self::$statistics['total_request'];
  569. $request = $requests[$buffer];
  570. if ($request instanceof Request) {
  571. $request = clone $request;
  572. $requests[$buffer] = $request;
  573. $request->connection = $this;
  574. $this->request = $request;
  575. $request->properties = [];
  576. }
  577. try {
  578. ($this->onMessage)($this, $request);
  579. } catch (Throwable $e) {
  580. $this->error($e);
  581. }
  582. return;
  583. }
  584. $this->recvBuffer = $buffer;
  585. } else {
  586. $this->recvBuffer .= $buffer;
  587. }
  588. }
  589. // If the application layer protocol has been set up.
  590. if ($this->protocol !== null) {
  591. while ($this->recvBuffer !== '' && !$this->isPaused) {
  592. // The current packet length is known.
  593. if ($this->currentPackageLength) {
  594. // Data is not enough for a package.
  595. if ($this->currentPackageLength > strlen($this->recvBuffer)) {
  596. break;
  597. }
  598. } else {
  599. // Get current package length.
  600. try {
  601. /** @var ProtocolInterface $parser */
  602. $parser = $this->protocol;
  603. $this->currentPackageLength = $parser::input($this->recvBuffer, $this);
  604. } catch (Throwable) {
  605. }
  606. // The packet length is unknown.
  607. if ($this->currentPackageLength === 0) {
  608. break;
  609. } elseif ($this->currentPackageLength > 0 && $this->currentPackageLength <= $this->maxPackageSize) {
  610. // Data is not enough for a package.
  611. if ($this->currentPackageLength > strlen($this->recvBuffer)) {
  612. break;
  613. }
  614. } // Wrong package.
  615. else {
  616. Worker::safeEcho('Error package. package_length=' . var_export($this->currentPackageLength, true));
  617. $this->destroy();
  618. return;
  619. }
  620. }
  621. // The data is enough for a packet.
  622. ++self::$statistics['total_request'];
  623. // The current packet length is equal to the length of the buffer.
  624. if ($one = (strlen($this->recvBuffer) === $this->currentPackageLength)) {
  625. $oneRequestBuffer = $this->recvBuffer;
  626. $this->recvBuffer = '';
  627. } else {
  628. // Get a full package from the buffer.
  629. $oneRequestBuffer = substr($this->recvBuffer, 0, $this->currentPackageLength);
  630. // Remove the current package from receive buffer.
  631. $this->recvBuffer = substr($this->recvBuffer, $this->currentPackageLength);
  632. }
  633. // Reset the current packet length to 0.
  634. $this->currentPackageLength = 0;
  635. try {
  636. // Decode request buffer before Emitting onMessage callback.
  637. /** @var ProtocolInterface $parser */
  638. $parser = $this->protocol;
  639. $request = $parser::decode($oneRequestBuffer, $this);
  640. if (static::$enableCache && (!is_object($request) || $request instanceof Request) && $one && !isset($oneRequestBuffer[512])) {
  641. $requests[$oneRequestBuffer] = $request;
  642. if (count($requests) > 512) {
  643. unset($requests[key($requests)]);
  644. }
  645. }
  646. ($this->onMessage)($this, $request);
  647. } catch (Throwable $e) {
  648. $this->error($e);
  649. }
  650. }
  651. return;
  652. }
  653. if ($this->recvBuffer === '' || $this->isPaused) {
  654. return;
  655. }
  656. // Applications protocol is not set.
  657. ++self::$statistics['total_request'];
  658. try {
  659. ($this->onMessage)($this, $this->recvBuffer);
  660. } catch (Throwable $e) {
  661. $this->error($e);
  662. }
  663. // Clean receive buffer.
  664. $this->recvBuffer = '';
  665. }
  666. /**
  667. * Base write handler.
  668. *
  669. * @return void
  670. * @throws Throwable
  671. */
  672. public function baseWrite(): void
  673. {
  674. $len = 0;
  675. try {
  676. if ($this->transport === 'ssl') {
  677. $len = @fwrite($this->socket, $this->sendBuffer, 8192);
  678. } else {
  679. $len = @fwrite($this->socket, $this->sendBuffer);
  680. }
  681. } catch (Throwable) {
  682. }
  683. if ($len === strlen($this->sendBuffer)) {
  684. $this->bytesWritten += $len;
  685. $this->eventLoop->offWritable($this->socket);
  686. $this->sendBuffer = '';
  687. // Try to emit onBufferDrain callback when send buffer becomes empty.
  688. if ($this->onBufferDrain) {
  689. try {
  690. ($this->onBufferDrain)($this);
  691. } catch (Throwable $e) {
  692. $this->error($e);
  693. }
  694. }
  695. if ($this->status === self::STATUS_CLOSING) {
  696. if (!empty($this->context->streamSending)) {
  697. return;
  698. }
  699. $this->destroy();
  700. }
  701. return;
  702. }
  703. if ($len > 0) {
  704. $this->bytesWritten += $len;
  705. $this->sendBuffer = substr($this->sendBuffer, $len);
  706. } else {
  707. ++self::$statistics['send_fail'];
  708. $this->destroy();
  709. }
  710. }
  711. /**
  712. * SSL handshake.
  713. *
  714. * @param resource $socket
  715. * @return bool|int
  716. * @throws Throwable
  717. */
  718. public function doSslHandshake($socket): bool|int
  719. {
  720. if (feof($socket)) {
  721. $this->destroy();
  722. return false;
  723. }
  724. $async = $this instanceof AsyncTcpConnection;
  725. /**
  726. * We disabled ssl3 because https://blog.qualys.com/ssllabs/2014/10/15/ssl-3-is-dead-killed-by-the-poodle-attack.
  727. * You can enable ssl3 by the codes below.
  728. */
  729. /*if($async){
  730. $type = STREAM_CRYPTO_METHOD_SSLv2_CLIENT | STREAM_CRYPTO_METHOD_SSLv23_CLIENT | STREAM_CRYPTO_METHOD_SSLv3_CLIENT;
  731. }else{
  732. $type = STREAM_CRYPTO_METHOD_SSLv2_SERVER | STREAM_CRYPTO_METHOD_SSLv23_SERVER | STREAM_CRYPTO_METHOD_SSLv3_SERVER;
  733. }*/
  734. if ($async) {
  735. $type = STREAM_CRYPTO_METHOD_SSLv2_CLIENT | STREAM_CRYPTO_METHOD_SSLv23_CLIENT;
  736. } else {
  737. $type = STREAM_CRYPTO_METHOD_SSLv2_SERVER | STREAM_CRYPTO_METHOD_SSLv23_SERVER;
  738. }
  739. // Hidden error.
  740. set_error_handler(function ($errno, $err_str) {
  741. if (!Worker::$daemonize) {
  742. Worker::safeEcho("SSL handshake error: $err_str \n");
  743. }
  744. });
  745. $ret = stream_socket_enable_crypto($socket, true, $type);
  746. restore_error_handler();
  747. // Negotiation has failed.
  748. if (false === $ret) {
  749. $this->destroy();
  750. return false;
  751. }
  752. if (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. public function isIpV4(): bool
  823. {
  824. if ($this->transport === 'unix') {
  825. return false;
  826. }
  827. return !str_contains($this->getRemoteIp(), ':');
  828. }
  829. /**
  830. * Is ipv6.
  831. *
  832. * return bool.
  833. */
  834. public function isIpV6(): bool
  835. {
  836. if ($this->transport === 'unix') {
  837. return false;
  838. }
  839. return str_contains($this->getRemoteIp(), ':');
  840. }
  841. /**
  842. * Get the real socket.
  843. *
  844. * @return resource
  845. */
  846. public function getSocket()
  847. {
  848. return $this->socket;
  849. }
  850. /**
  851. * Check whether send buffer will be full.
  852. *
  853. * @return void
  854. * @throws Throwable
  855. */
  856. protected function checkBufferWillFull(): void
  857. {
  858. if ($this->onBufferFull && $this->maxSendBufferSize <= strlen($this->sendBuffer)) {
  859. try {
  860. ($this->onBufferFull)($this);
  861. } catch (Throwable $e) {
  862. $this->error($e);
  863. }
  864. }
  865. }
  866. /**
  867. * Whether send buffer is full.
  868. *
  869. * @return bool
  870. * @throws Throwable
  871. */
  872. protected function bufferIsFull(): bool
  873. {
  874. // Buffer has been marked as full but still has data to send then the packet is discarded.
  875. if ($this->maxSendBufferSize <= strlen($this->sendBuffer)) {
  876. if ($this->onError) {
  877. try {
  878. ($this->onError)($this, static::SEND_FAIL, 'send buffer full and drop package');
  879. } catch (Throwable $e) {
  880. $this->error($e);
  881. }
  882. }
  883. return true;
  884. }
  885. return false;
  886. }
  887. /**
  888. * Whether send buffer is Empty.
  889. *
  890. * @return bool
  891. */
  892. public function bufferIsEmpty(): bool
  893. {
  894. return empty($this->sendBuffer);
  895. }
  896. /**
  897. * Destroy connection.
  898. *
  899. * @return void
  900. * @throws Throwable
  901. */
  902. public function destroy(): void
  903. {
  904. // Avoid repeated calls.
  905. if ($this->status === self::STATUS_CLOSED) {
  906. return;
  907. }
  908. // Remove event listener.
  909. $this->eventLoop->offReadable($this->socket);
  910. $this->eventLoop->offWritable($this->socket);
  911. // Close socket.
  912. try {
  913. @fclose($this->socket);
  914. } catch (Throwable) {
  915. }
  916. $this->status = self::STATUS_CLOSED;
  917. // Try to emit onClose callback.
  918. if ($this->onClose) {
  919. try {
  920. ($this->onClose)($this);
  921. } catch (Throwable $e) {
  922. $this->error($e);
  923. }
  924. }
  925. // Try to emit protocol::onClose
  926. if ($this->protocol && method_exists($this->protocol, 'onClose')) {
  927. try {
  928. ([$this->protocol, 'onClose'])($this);
  929. } catch (Throwable $e) {
  930. $this->error($e);
  931. }
  932. }
  933. $this->sendBuffer = $this->recvBuffer = '';
  934. $this->currentPackageLength = 0;
  935. $this->isPaused = $this->sslHandshakeCompleted = false;
  936. if ($this->status === self::STATUS_CLOSED) {
  937. // Cleaning up the callback to avoid memory leaks.
  938. $this->onMessage = $this->onClose = $this->onError = $this->onBufferFull = $this->onBufferDrain = $this->eventLoop = $this->errorHandler = null;
  939. // Remove from worker->connections.
  940. if ($this->worker) {
  941. unset($this->worker->connections[$this->realId]);
  942. }
  943. unset(static::$connections[$this->realId]);
  944. }
  945. }
  946. /**
  947. * Enable or disable Cache.
  948. *
  949. * @param bool $value
  950. */
  951. public static function enableCache(bool $value = true): void
  952. {
  953. static::$enableCache = $value;
  954. }
  955. /**
  956. * Get the json_encode information.
  957. *
  958. * @return array
  959. */
  960. public function jsonSerialize(): array
  961. {
  962. return [
  963. 'id' => $this->id,
  964. 'status' => $this->getStatus(),
  965. 'transport' => $this->transport,
  966. 'getRemoteIp' => $this->getRemoteIp(),
  967. 'remotePort' => $this->getRemotePort(),
  968. 'getRemoteAddress' => $this->getRemoteAddress(),
  969. 'getLocalIp' => $this->getLocalIp(),
  970. 'getLocalPort' => $this->getLocalPort(),
  971. 'getLocalAddress' => $this->getLocalAddress(),
  972. 'isIpV4' => $this->isIpV4(),
  973. 'isIpV6' => $this->isIpV6(),
  974. ];
  975. }
  976. /**
  977. * Destruct.
  978. *
  979. * @return void
  980. * @throws Throwable
  981. */
  982. public function __destruct()
  983. {
  984. static $mod;
  985. self::$statistics['connection_count']--;
  986. if (Worker::getGracefulStop()) {
  987. if (!isset($mod)) {
  988. $mod = ceil((self::$statistics['connection_count'] + 1) / 3);
  989. }
  990. if (0 === self::$statistics['connection_count'] % $mod) {
  991. $pid = function_exists('posix_getpid') ? posix_getpid() : 0;
  992. Worker::log('worker[' . $pid . '] remains ' . self::$statistics['connection_count'] . ' connection(s)');
  993. }
  994. if (0 === self::$statistics['connection_count']) {
  995. Worker::stopAll();
  996. }
  997. }
  998. }
  999. }