<?php
/**
 * This file is part of workerman.
 *
 * Licensed under The MIT License
 * For full copyright and license information, please see the MIT-LICENSE.txt
 * Redistributions of files must retain the above copyright notice.
 *
 * @author walkor<walkor@workerman.net>
 * @copyright walkor<walkor@workerman.net>
 * @link http://www.workerman.net/
 * @license http://www.opensource.org/licenses/mit-license.php MIT License
 */
namespace Workerman\Connection;

use Workerman\Events\EventInterface;
use Workerman\Worker;
use \Exception;

/**
 * TcpConnection.
 */
class TcpConnection extends ConnectionInterface
{
 /**
 * Read buffer size.
 *
 * @var int
 */
 const READ_BUFFER_SIZE = 65535;

 /**
 * Status initial.
 *
 * @var int
 */
 const STATUS_INITIAL = 0;

 /**
 * Status connecting.
 *
 * @var int
 */
 const STATUS_CONNECTING = 1;

 /**
 * Status connection established.
 *
 * @var int
 */
 const STATUS_ESTABLISHED = 2;

 /**
 * Status closing.
 *
 * @var int
 */
 const STATUS_CLOSING = 4;

 /**
 * Status closed.
 *
 * @var int
 */
 const STATUS_CLOSED = 8;

 /**
 * Emitted when data is received.
 *
 * @var callable
 */
 public $onMessage = null;

 /**
 * Emitted when the other end of the socket sends a FIN packet.
 *
 * @var callable
 */
 public $onClose = null;

 /**
 * Emitted when an error occurs with connection.
 *
 * @var callable
 */
 public $onError = null;

 /**
 * Emitted when the send buffer becomes full.
 *
 * @var callable
 */
 public $onBufferFull = null;

 /**
 * Emitted when the send buffer becomes empty.
 *
 * @var callable
 */
 public $onBufferDrain = null;

 /**
 * Application layer protocol.
 * The format is like this Workerman\\Protocols\\Http.
 *
 * @var \Workerman\Protocols\ProtocolInterface
 */
 public $protocol = null;

 /**
 * Transport (tcp/udp/unix/ssl).
 *
 * @var string
 */
 public $transport = 'tcp';

 /**
 * Which worker belong to.
 *
 * @var Worker
 */
 public $worker = null;

 /**
 * Bytes read.
 *
 * @var int
 */
 public $bytesRead = 0;

 /**
 * Bytes written.
 *
 * @var int
 */
 public $bytesWritten = 0;

 /**
 * Connection->id.
 *
 * @var int
 */
 public $id = 0;

 /**
 * A copy of $worker->id which used to clean up the connection in worker->connections
 *
 * @var int
 */
 protected $_id = 0;

 /**
 * Sets the maximum send buffer size for the current connection.
 * OnBufferFull callback will be emited When the send buffer is full.
 *
 * @var int
 */
 public $maxSendBufferSize = 1048576;

 /**
 * Default send buffer size.
 *
 * @var int
 */
 public static $defaultMaxSendBufferSize = 1048576;

 /**
 * Sets the maximum acceptable packet size for the current connection.
 *
 * @var int
 */
 public $maxPackageSize = 1048576;
 
 /**
 * Default maximum acceptable packet size.
 *
 * @var int
 */
 public static $defaultMaxPackageSize = 10485760;

 /**
 * Id recorder.
 *
 * @var int
 */
 protected static $_idRecorder = 1;

 /**
 * Socket
 *
 * @var resource
 */
 protected $_socket = null;

 /**
 * Send buffer.
 *
 * @var string
 */
 protected $_sendBuffer = '';

 /**
 * Receive buffer.
 *
 * @var string
 */
 protected $_recvBuffer = '';

 /**
 * Current package length.
 *
 * @var int
 */
 protected $_currentPackageLength = 0;

 /**
 * Connection status.
 *
 * @var int
 */
 protected $_status = self::STATUS_ESTABLISHED;

 /**
 * Remote address.
 *
 * @var string
 */
 protected $_remoteAddress = '';

 /**
 * Is paused.
 *
 * @var bool
 */
 protected $_isPaused = false;

 /**
 * SSL handshake completed or not.
 *
 * @var bool
 */
 protected $_sslHandshakeCompleted = false;

 /**
 * All connection instances.
 *
 * @var array
 */
 public static $connections = array();

 /**
 * Status to string.
 *
 * @var array
 */
 public static $_statusToString = array(
 self::STATUS_INITIAL => 'INITIAL',
 self::STATUS_CONNECTING => 'CONNECTING',
 self::STATUS_ESTABLISHED => 'ESTABLISHED',
 self::STATUS_CLOSING => 'CLOSING',
 self::STATUS_CLOSED => 'CLOSED',
 );

 /**
 * Construct.
 *
 * @param resource $socket
 * @param string $remote_address
 */
 public function __construct($socket, $remote_address = '')
 {
 ++self::$statistics['connection_count'];
 $this->id = $this->_id = self::$_idRecorder++;
 if(self::$_idRecorder === \PHP_INT_MAX){
 self::$_idRecorder = 0;
 }
 $this->_socket = $socket;
 \stream_set_blocking($this->_socket, 0);
 // Compatible with hhvm
 if (\function_exists('stream_set_read_buffer')) {
 \stream_set_read_buffer($this->_socket, 0);
 }
 Worker::$globalEvent->add($this->_socket, EventInterface::EV_READ, array($this, 'baseRead'));
 $this->maxSendBufferSize = self::$defaultMaxSendBufferSize;
 $this->maxPackageSize = self::$defaultMaxPackageSize;
 $this->_remoteAddress = $remote_address;
 static::$connections[$this->id] = $this;
 }

 /**
 * Get status.
 *
 * @param bool $raw_output
 *
 * @return int|string
 */
 public function getStatus($raw_output = true)
 {
 if ($raw_output) {
 return $this->_status;
 }
 return self::$_statusToString[$this->_status];
 }

 /**
 * Sends data on the connection.
 *
 * @param mixed $send_buffer
 * @param bool $raw
 * @return bool|null
 */
 public function send($send_buffer, $raw = false)
 {
 if ($this->_status === self::STATUS_CLOSING || $this->_status === self::STATUS_CLOSED) {
 return false;
 }

 // Try to call protocol::encode($send_buffer) before sending.
 if (false === $raw && $this->protocol !== null) {
 $parser = $this->protocol;
 $send_buffer = $parser::encode($send_buffer, $this);
 if ($send_buffer === '') {
 return;
 }
 }

 if ($this->_status !== self::STATUS_ESTABLISHED ||
 ($this->transport === 'ssl' && $this->_sslHandshakeCompleted !== true)
 ) {
 if ($this->_sendBuffer && $this->bufferIsFull()) {
 ++self::$statistics['send_fail'];
 return false;
 }
 $this->_sendBuffer .= $send_buffer;
 $this->checkBufferWillFull();
 return;
 }

 // Attempt to send data directly.
 if ($this->_sendBuffer === '') {
 if ($this->transport === 'ssl') {
 Worker::$globalEvent->add($this->_socket, EventInterface::EV_WRITE, array($this, 'baseWrite'));
 $this->_sendBuffer = $send_buffer;
 $this->checkBufferWillFull();
 return;
 }
 $len = 0;
 try {
 $len = @\fwrite($this->_socket, $send_buffer);
 } catch (\Exception $e) {
 Worker::log($e);
 } catch (\Error $e) {
 Worker::log($e);
 }
 // send successful.
 if ($len === \strlen($send_buffer)) {
 $this->bytesWritten += $len;
 return true;
 }
 // Send only part of the data.
 if ($len > 0) {
 $this->_sendBuffer = \substr($send_buffer, $len);
 $this->bytesWritten += $len;
 } else {
 // Connection closed?
 if (!\is_resource($this->_socket) || \feof($this->_socket)) {
 ++self::$statistics['send_fail'];
 if ($this->onError) {
 try {
 \call_user_func($this->onError, $this, \WORKERMAN_SEND_FAIL, 'client closed');
 } catch (\Exception $e) {
 Worker::stopAll(250, $e);
 } catch (\Error $e) {
 Worker::stopAll(250, $e);
 }
 }
 $this->destroy();
 return false;
 }
 $this->_sendBuffer = $send_buffer;
 }
 Worker::$globalEvent->add($this->_socket, EventInterface::EV_WRITE, array($this, 'baseWrite'));
 // Check if the send buffer will be full.
 $this->checkBufferWillFull();
 return;
 }

 if ($this->bufferIsFull()) {
 ++self::$statistics['send_fail'];
 return false;
 }

 $this->_sendBuffer .= $send_buffer;
 // Check if the send buffer is full.
 $this->checkBufferWillFull();
 }

 /**
 * Get remote IP.
 *
 * @return string
 */
 public function getRemoteIp()
 {
 $pos = \strrpos($this->_remoteAddress, ':');
 if ($pos) {
 return (string) \substr($this->_remoteAddress, 0, $pos);
 }
 return '';
 }

 /**
 * Get remote port.
 *
 * @return int
 */
 public function getRemotePort()
 {
 if ($this->_remoteAddress) {
 return (int) \substr(\strrchr($this->_remoteAddress, ':'), 1);
 }
 return 0;
 }

 /**
 * Get remote address.
 *
 * @return string
 */
 public function getRemoteAddress()
 {
 return $this->_remoteAddress;
 }

 /**
 * Get local IP.
 *
 * @return string
 */
 public function getLocalIp()
 {
 $address = $this->getLocalAddress();
 $pos = \strrpos($address, ':');
 if (!$pos) {
 return '';
 }
 return \substr($address, 0, $pos);
 }

 /**
 * Get local port.
 *
 * @return int
 */
 public function getLocalPort()
 {
 $address = $this->getLocalAddress();
 $pos = \strrpos($address, ':');
 if (!$pos) {
 return 0;
 }
 return (int)\substr(\strrchr($address, ':'), 1);
 }

 /**
 * Get local address.
 *
 * @return string
 */
 public function getLocalAddress()
 {
 if (!\is_resource($this->_socket)) {
 return '';
 }
 return (string)@\stream_socket_get_name($this->_socket, false);
 }

 /**
 * Get send buffer queue size.
 *
 * @return integer
 */
 public function getSendBufferQueueSize()
 {
 return \strlen($this->_sendBuffer);
 }

 /**
 * Get recv buffer queue size.
 *
 * @return integer
 */
 public function getRecvBufferQueueSize()
 {
 return \strlen($this->_recvBuffer);
 }

 /**
 * Is ipv4.
 *
 * return bool.
 */
 public function isIpV4()
 {
 if ($this->transport === 'unix') {
 return false;
 }
 return \strpos($this->getRemoteIp(), ':') === false;
 }

 /**
 * Is ipv6.
 *
 * return bool.
 */
 public function isIpV6()
 {
 if ($this->transport === 'unix') {
 return false;
 }
 return \strpos($this->getRemoteIp(), ':') !== false;
 }

 /**
 * Pauses the reading of data. That is onMessage will not be emitted. Useful to throttle back an upload.
 *
 * @return void
 */
 public function pauseRecv()
 {
 Worker::$globalEvent->del($this->_socket, EventInterface::EV_READ);
 $this->_isPaused = true;
 }

 /**
 * Resumes reading after a call to pauseRecv.
 *
 * @return void
 */
 public function resumeRecv()
 {
 if ($this->_isPaused === true) {
 Worker::$globalEvent->add($this->_socket, EventInterface::EV_READ, array($this, 'baseRead'));
 $this->_isPaused = false;
 $this->baseRead($this->_socket, false);
 }
 }



 /**
 * Base read handler.
 *
 * @param resource $socket
 * @param bool $check_eof
 * @return void
 */
 public function baseRead($socket, $check_eof = true)
 {
 // SSL handshake.
 if ($this->transport === 'ssl' && $this->_sslHandshakeCompleted !== true) {
 if ($this->doSslHandshake($socket)) {
 $this->_sslHandshakeCompleted = true;
 if ($this->_sendBuffer) {
 Worker::$globalEvent->add($socket, EventInterface::EV_WRITE, array($this, 'baseWrite'));
 }
 } else {
 return;
 }
 }

 $buffer = '';
 try {
 $buffer = @\fread($socket, self::READ_BUFFER_SIZE);
 } catch (\Exception $e) {} catch (\Error $e) {}

 // Check connection closed.
 if ($buffer === '' || $buffer === false) {
 if ($check_eof && (\feof($socket) || !\is_resource($socket) || $buffer === false)) {
 $this->destroy();
 return;
 }
 } else {
 $this->bytesRead += \strlen($buffer);
 $this->_recvBuffer .= $buffer;
 }

 // If the application layer protocol has been set up.
 if ($this->protocol !== null) {
 $parser = $this->protocol;
 while ($this->_recvBuffer !== '' && !$this->_isPaused) {
 // The current packet length is known.
 if ($this->_currentPackageLength) {
 // Data is not enough for a package.
 if ($this->_currentPackageLength > \strlen($this->_recvBuffer)) {
 break;
 }
 } else {
 // Get current package length.
 try {
 $this->_currentPackageLength = $parser::input($this->_recvBuffer, $this);
 } catch (\Exception $e) {} catch (\Error $e) {}
 // The packet length is unknown.
 if ($this->_currentPackageLength === 0) {
 break;
 } elseif ($this->_currentPackageLength > 0 && $this->_currentPackageLength <= $this->maxPackageSize) {
 // Data is not enough for a package.
 if ($this->_currentPackageLength > \strlen($this->_recvBuffer)) {
 break;
 }
 } // Wrong package.
 else {
 Worker::safeEcho('Error package. package_length=' . \var_export($this->_currentPackageLength, true));
 $this->destroy();
 return;
 }
 }

 // The data is enough for a packet.
 ++self::$statistics['total_request'];
 // The current packet length is equal to the length of the buffer.
 if (\strlen($this->_recvBuffer) === $this->_currentPackageLength) {
 $one_request_buffer = $this->_recvBuffer;
 $this->_recvBuffer = '';
 } else {
 // Get a full package from the buffer.
 $one_request_buffer = \substr($this->_recvBuffer, 0, $this->_currentPackageLength);
 // Remove the current package from the receive buffer.
 $this->_recvBuffer = \substr($this->_recvBuffer, $this->_currentPackageLength);
 }
 // Reset the current packet length to 0.
 $this->_currentPackageLength = 0;
 if (!$this->onMessage) {
 continue;
 }
 try {
 // Decode request buffer before Emitting onMessage callback.
 \call_user_func($this->onMessage, $this, $parser::decode($one_request_buffer, $this));
 } catch (\Exception $e) {
 Worker::stopAll(250, $e);
 } catch (\Error $e) {
 Worker::stopAll(250, $e);
 }
 }
 return;
 }

 if ($this->_recvBuffer === '' || $this->_isPaused) {
 return;
 }

 // Applications protocol is not set.
 ++self::$statistics['total_request'];
 if (!$this->onMessage) {
 $this->_recvBuffer = '';
 return;
 }
 try {
 \call_user_func($this->onMessage, $this, $this->_recvBuffer);
 } catch (\Exception $e) {
 Worker::stopAll(250, $e);
 } catch (\Error $e) {
 Worker::stopAll(250, $e);
 }
 // Clean receive buffer.
 $this->_recvBuffer = '';
 }

 /**
 * Base write handler.
 *
 * @return void|bool
 */
 public function baseWrite()
 {
 \set_error_handler(function(){});
 if ($this->transport === 'ssl') {
 $len = @\fwrite($this->_socket, $this->_sendBuffer, 8192);
 } else {
 $len = @\fwrite($this->_socket, $this->_sendBuffer);
 }
 \restore_error_handler();
 if ($len === \strlen($this->_sendBuffer)) {
 $this->bytesWritten += $len;
 Worker::$globalEvent->del($this->_socket, EventInterface::EV_WRITE);
 $this->_sendBuffer = '';
 // Try to emit onBufferDrain callback when the send buffer becomes empty.
 if ($this->onBufferDrain) {
 try {
 \call_user_func($this->onBufferDrain, $this);
 } catch (\Exception $e) {
 Worker::stopAll(250, $e);
 } catch (\Error $e) {
 Worker::stopAll(250, $e);
 }
 }
 if ($this->_status === self::STATUS_CLOSING) {
 $this->destroy();
 }
 return true;
 }
 if ($len > 0) {
 $this->bytesWritten += $len;
 $this->_sendBuffer = \substr($this->_sendBuffer, $len);
 } else {
 ++self::$statistics['send_fail'];
 $this->destroy();
 }
 }

 /**
 * SSL handshake.
 *
 * @param resource $socket
 * @return bool
 */
 public function doSslHandshake($socket){
 if (\feof($socket)) {
 $this->destroy();
 return false;
 }
 $async = $this instanceof AsyncTcpConnection;
 
 /**
 * We disabled ssl3 because https://blog.qualys.com/ssllabs/2014/10/15/ssl-3-is-dead-killed-by-the-poodle-attack.
 * You can enable ssl3 by the codes below.
 */
 /*if($async){
 $type = STREAM_CRYPTO_METHOD_SSLv2_CLIENT | STREAM_CRYPTO_METHOD_SSLv23_CLIENT | STREAM_CRYPTO_METHOD_SSLv3_CLIENT;
 }else{
 $type = STREAM_CRYPTO_METHOD_SSLv2_SERVER | STREAM_CRYPTO_METHOD_SSLv23_SERVER | STREAM_CRYPTO_METHOD_SSLv3_SERVER;
 }*/
 
 if($async){
 $type = \STREAM_CRYPTO_METHOD_SSLv2_CLIENT | \STREAM_CRYPTO_METHOD_SSLv23_CLIENT;
 }else{
 $type = \STREAM_CRYPTO_METHOD_SSLv2_SERVER | \STREAM_CRYPTO_METHOD_SSLv23_SERVER;
 }
 
 // Hidden error.
 \set_error_handler(function($errno, $errstr, $file){
 if (!Worker::$daemonize) {
 Worker::safeEcho("SSL handshake error: $errstr\n");
 }
 });
 $ret = \stream_socket_enable_crypto($socket, true, $type);
 \restore_error_handler();
 // Negotiation has failed.
 if (false === $ret) {
 $this->destroy();
 return false;
 } elseif (0 === $ret) {
 // There isn't enough data and should try again.
 return 0;
 }
 if (isset($this->onSslHandshake)) {
 try {
 \call_user_func($this->onSslHandshake, $this);
 } catch (\Exception $e) {
 Worker::stopAll(250, $e);
 } catch (\Error $e) {
 Worker::stopAll(250, $e);
 }
 }
 return true;
 }

 /**
 * This method pulls all the data out of a readable stream, and writes it to the supplied destination.
 *
 * @param self $dest
 * @return void
 */
 public function pipe(self $dest)
 {
 $source = $this;
 $this->onMessage = function ($source, $data) use ($dest) {
 $dest->send($data);
 };
 $this->onClose = function ($source) use ($dest) {
 $dest->close();
 };
 $dest->onBufferFull = function ($dest) use ($source) {
 $source->pauseRecv();
 };
 $dest->onBufferDrain = function ($dest) use ($source) {
 $source->resumeRecv();
 };
 }

 /**
 * Remove $length of data from receive buffer.
 *
 * @param int $length
 * @return void
 */
 public function consumeRecvBuffer($length)
 {
 $this->_recvBuffer = \substr($this->_recvBuffer, $length);
 }

 /**
 * Close connection.
 *
 * @param mixed $data
 * @param bool $raw
 * @return void
 */
 public function close($data = null, $raw = false)
 {
 if($this->_status === self::STATUS_CONNECTING){
 $this->destroy();
 return;
 }

 if ($this->_status === self::STATUS_CLOSING || $this->_status === self::STATUS_CLOSED) {
 return;
 }

 if ($data !== null) {
 $this->send($data, $raw);
 }

 $this->_status = self::STATUS_CLOSING;
 
 if ($this->_sendBuffer === '') {
 $this->destroy();
 } else {
 $this->pauseRecv();
 }
 }

 /**
 * Get the real socket.
 *
 * @return resource
 */
 public function getSocket()
 {
 return $this->_socket;
 }

 /**
 * Check whether the send buffer will be full.
 *
 * @return void
 */
 protected function checkBufferWillFull()
 {
 if ($this->maxSendBufferSize <= \strlen($this->_sendBuffer)) {
 if ($this->onBufferFull) {
 try {
 \call_user_func($this->onBufferFull, $this);
 } catch (\Exception $e) {
 Worker::stopAll(250, $e);
 } catch (\Error $e) {
 Worker::stopAll(250, $e);
 }
 }
 }
 }

 /**
 * Whether send buffer is full.
 *
 * @return bool
 */
 protected function bufferIsFull()
 {
 // Buffer has been marked as full but still has data to send then the packet is discarded.
 if ($this->maxSendBufferSize <= \strlen($this->_sendBuffer)) {
 if ($this->onError) {
 try {
 \call_user_func($this->onError, $this, \WORKERMAN_SEND_FAIL, 'send buffer full and drop package');
 } catch (\Exception $e) {
 Worker::stopAll(250, $e);
 } catch (\Error $e) {
 Worker::stopAll(250, $e);
 }
 }
 return true;
 }
 return false;
 }
 
 /**
 * Whether send buffer is Empty.
 *
 * @return bool
 */
 public function bufferIsEmpty()
 {
 	return empty($this->_sendBuffer);
 }

 /**
 * Destroy connection.
 *
 * @return void
 */
 public function destroy()
 {
 // Avoid repeated calls.
 if ($this->_status === self::STATUS_CLOSED) {
 return;
 }
 // Remove event listener.
 Worker::$globalEvent->del($this->_socket, EventInterface::EV_READ);
 Worker::$globalEvent->del($this->_socket, EventInterface::EV_WRITE);

 // Close socket.
 try {
 @\fclose($this->_socket);
 } catch (\Exception $e) {} catch (\Error $e) {}

 $this->_status = self::STATUS_CLOSED;
 // Try to emit onClose callback.
 if ($this->onClose) {
 try {
 \call_user_func($this->onClose, $this);
 } catch (\Exception $e) {
 Worker::stopAll(250, $e);
 } catch (\Error $e) {
 Worker::stopAll(250, $e);
 }
 }
 // Try to emit protocol::onClose
 if ($this->protocol && \method_exists($this->protocol, 'onClose')) {
 try {
 \call_user_func(array($this->protocol, 'onClose'), $this);
 } catch (\Exception $e) {
 Worker::stopAll(250, $e);
 } catch (\Error $e) {
 Worker::stopAll(250, $e);
 }
 }
 $this->_sendBuffer = $this->_recvBuffer = '';
 $this->_currentPackageLength = 0;
 $this->_isPaused = $this->_sslHandshakeCompleted = false;
 if ($this->_status === self::STATUS_CLOSED) {
 // Cleaning up the callback to avoid memory leaks.
 $this->onMessage = $this->onClose = $this->onError = $this->onBufferFull = $this->onBufferDrain = null;
 // Remove from worker->connections.
 if ($this->worker) {
 unset($this->worker->connections[$this->_id]);
 }
 unset(static::$connections[$this->_id]);
 }
 }

 /**
 * Destruct.
 *
 * @return void
 */
 public function __destruct()
 {
 static $mod;
 self::$statistics['connection_count']--;
 if (Worker::getGracefulStop()) {
 if (!isset($mod)) {
 $mod = \ceil((self::$statistics['connection_count'] + 1) / 3);
 }

 if (0 === self::$statistics['connection_count'] % $mod) {
 Worker::log('worker[' . \posix_getpid() . '] remains ' . self::$statistics['connection_count'] . ' connection(s)');
 }

 if(0 === self::$statistics['connection_count']) {
 Worker::stopAll();
 }
 }
 }
}