Merge pull request #3 from walkor/master

sync

.gitignore

@@ -2,4 +2,5 @@ logs
 .buildpath
 .project
 .settings
-.idea
+.idea
+.DS_Store

Connection/AsyncTcpConnection.php

@@ -14,6 +14,7 @@
 namespace Workerman\Connection;
 
 use Workerman\Events\EventInterface;
+use Workerman\Lib\Timer;
 use Workerman\Worker;
 use Exception;
 
@@ -30,11 +31,18 @@ class AsyncTcpConnection extends TcpConnection
     public $onConnect = null;
 
     /**
+     * Transport layer protocol.
+     *
+     * @var string
+     */
+    public $transport = 'tcp';
+
+    /**
      * Status.
      *
      * @var int
      */
-    protected $_status = self::STATUS_CONNECTING;
+    protected $_status = self::STATUS_INITIAL;
 
     /**
      * Remote host.
@@ -44,46 +52,177 @@ class AsyncTcpConnection extends TcpConnection
     protected $_remoteHost = '';
 
     /**
+     * Remote port.
+     *
+     * @var int
+     */
+    protected $_remotePort = 80;
+
+    /**
+     * Connect start time.
+     *
+     * @var string
+     */
+    protected $_connectStartTime = 0;
+
+    /**
+     * Remote URI.
+     *
+     * @var string
+     */
+    protected $_remoteURI = '';
+
+    /**
+     * Context option.
+     *
+     * @var resource
+     */
+    protected $_contextOption = null;
+
+    /**
+     * Reconnect timer.
+     *
+     * @var int
+     */
+    protected $_reconnectTimer = null;
+
+
+    /**
+     * PHP built-in protocols.
+     *
+     * @var array
+     */
+    protected static $_builtinTransports = array(
+        'tcp'   => 'tcp',
+        'udp'   => 'udp',
+        'unix'  => 'unix',
+        'ssl'   => 'ssl',
+        'sslv2' => 'sslv2',
+        'sslv3' => 'sslv3',
+        'tls'   => 'tls'
+    );
+
+    /**
      * Construct.
      *
      * @param string $remote_address
+     * @param array $context_option
      * @throws Exception
      */
-    public function __construct($remote_address)
+    public function __construct($remote_address, $context_option = null)
     {
-        list($scheme, $address) = explode(':', $remote_address, 2);
-        if ($scheme != 'tcp') {
-            // Get application layer protocol.
+        $address_info = parse_url($remote_address);
+        if (!$address_info) {
+            list($scheme, $this->_remoteAddress) = explode(':', $remote_address, 2);
+            if (!$this->_remoteAddress) {
+                echo new \Exception('bad remote_address');
+            }
+        } else {
+            if (!isset($address_info['port'])) {
+                $address_info['port'] = 80;
+            }
+            if (!isset($address_info['path'])) {
+                $address_info['path'] = '/';
+            }
+            if (!isset($address_info['query'])) {
+                $address_info['query'] = '';
+            } else {
+                $address_info['query'] = '?' . $address_info['query'];
+            }
+            $this->_remoteAddress = "{$address_info['host']}:{$address_info['port']}";
+            $this->_remoteHost    = $address_info['host'];
+            $this->_remotePort    = $address_info['port'];
+            $this->_remoteURI     = "{$address_info['path']}{$address_info['query']}";
+            $scheme               = isset($address_info['scheme']) ? $address_info['scheme'] : 'tcp';
+        }
+
+        $this->id = $this->_id = self::$_idRecorder++;
+        if(PHP_INT_MAX === self::$_idRecorder){
+            self::$_idRecorder = 0;
+        }
+        // Check application layer protocol class.
+        if (!isset(self::$_builtinTransports[$scheme])) {
             $scheme         = ucfirst($scheme);
             $this->protocol = '\\Protocols\\' . $scheme;
             if (!class_exists($this->protocol)) {
-                $this->protocol = '\\Workerman\\Protocols\\' . $scheme;
+                $this->protocol = "\\Workerman\\Protocols\\$scheme";
                 if (!class_exists($this->protocol)) {
                     throw new Exception("class \\Protocols\\$scheme not exist");
                 }
             }
+        } else {
+            $this->transport = self::$_builtinTransports[$scheme];
         }
-        $this->_remoteAddress = substr($address, 2);
-        $this->_remoteHost    = substr($this->_remoteAddress, 0, strrpos($this->_remoteAddress, ':'));
-        $this->id             = self::$_idRecorder++;
+
         // For statistics.
         self::$statistics['connection_count']++;
-        $this->maxSendBufferSize = self::$defaultMaxSendBufferSize;
+        $this->maxSendBufferSize        = self::$defaultMaxSendBufferSize;
+        $this->_contextOption           = $context_option;
+        static::$connections[$this->id] = $this;
     }
 
+    /**
+     * Do connect.
+     *
+     * @return void 
+     */
     public function connect()
     {
-        // Open socket connection asynchronously.
-        $this->_socket = stream_socket_client("tcp://{$this->_remoteAddress}", $errno, $errstr, 0,
-            STREAM_CLIENT_ASYNC_CONNECT);
+        if ($this->_status !== self::STATUS_INITIAL && $this->_status !== self::STATUS_CLOSING &&
+             $this->_status !== self::STATUS_CLOSED) {
+            return;
+        }
+        $this->_status           = self::STATUS_CONNECTING;
+        $this->_connectStartTime = microtime(true);
+        if ($this->transport !== 'unix') {
+            // Open socket connection asynchronously.
+            if ($this->_contextOption) {
+                $context = stream_context_create($this->_contextOption);
+                $this->_socket = stream_socket_client("{$this->transport}://{$this->_remoteHost}:{$this->_remotePort}",
+                    $errno, $errstr, 0, STREAM_CLIENT_ASYNC_CONNECT, $context);
+            } else {
+                $this->_socket = stream_socket_client("{$this->transport}://{$this->_remoteHost}:{$this->_remotePort}",
+                    $errno, $errstr, 0, STREAM_CLIENT_ASYNC_CONNECT);
+            }
+        } else {
+            $this->_socket = stream_socket_client("{$this->transport}://{$this->_remoteAddress}", $errno, $errstr, 0,
+                STREAM_CLIENT_ASYNC_CONNECT);
+        }
         // If failed attempt to emit onError callback.
         if (!$this->_socket) {
-            $this->_status = self::STATUS_CLOSED;
             $this->emitError(WORKERMAN_CONNECT_FAIL, $errstr);
+            if ($this->_status === self::STATUS_CLOSING) {
+                $this->destroy();
+            }
+            if ($this->_status === self::STATUS_CLOSED) {
+                $this->onConnect = null;
+            }
             return;
         }
         // Add socket to global event loop waiting connection is successfully established or faild. 
         Worker::$globalEvent->add($this->_socket, EventInterface::EV_WRITE, array($this, 'checkConnection'));
+        // For windows.
+        if(DIRECTORY_SEPARATOR === '\\') {
+            Worker::$globalEvent->add($this->_socket, EventInterface::EV_EXCEPT, array($this, 'checkConnection'));
+        }
+    }
+
+    /**
+     * Reconnect.
+     *
+     * @param int $after
+     * @return void
+     */
+    public function reConnect($after = 0) {
+        $this->_status = self::STATUS_INITIAL;
+        if ($this->_reconnectTimer) {
+            Timer::del($this->_reconnectTimer);
+        }
+        if ($after > 0) {
+            $this->_reconnectTimer = Timer::add($after, array($this, 'connect'), null, false);
+            return;
+        }
+        $this->connect();
     }
 
     /**
@@ -97,6 +236,16 @@ class AsyncTcpConnection extends TcpConnection
     }
 
     /**
+     * Get remote URI.
+     *
+     * @return string
+     */
+    public function getRemoteURI()
+    {
+        return $this->_remoteURI;
+    }
+
+    /**
      * Try to emit onError callback.
      *
      * @param int    $code
@@ -105,11 +254,15 @@ class AsyncTcpConnection extends TcpConnection
      */
     protected function emitError($code, $msg)
     {
+        $this->_status = self::STATUS_CLOSING;
         if ($this->onError) {
             try {
                 call_user_func($this->onError, $this, $code, $msg);
             } catch (\Exception $e) {
-                echo $e;
+                Worker::log($e);
+                exit(250);
+            } catch (\Error $e) {
+                Worker::log($e);
                 exit(250);
             }
         }
@@ -123,14 +276,22 @@ class AsyncTcpConnection extends TcpConnection
      */
     public function checkConnection($socket)
     {
+        // Remove EV_EXPECT for windows.
+        if(DIRECTORY_SEPARATOR === '\\') {
+            Worker::$globalEvent->del($socket, EventInterface::EV_EXCEPT);
+        }
         // Check socket state.
-        if (stream_socket_get_name($socket, true)) {
+        if ($address = stream_socket_get_name($socket, true)) {
             // Remove write listener.
             Worker::$globalEvent->del($socket, EventInterface::EV_WRITE);
             // Nonblocking.
             stream_set_blocking($socket, 0);
+            // Compatible with hhvm
+            if (function_exists('stream_set_read_buffer')) {
+                stream_set_read_buffer($socket, 0);
+            }
             // Try to open keepalive for tcp and disable Nagle algorithm.
-            if (function_exists('socket_import_stream')) {
+            if (function_exists('socket_import_stream') && $this->transport === 'tcp') {
                 $raw_socket = socket_import_stream($socket);
                 socket_set_option($raw_socket, SOL_SOCKET, SO_KEEPALIVE, 1);
                 socket_set_option($raw_socket, SOL_TCP, TCP_NODELAY, 1);
@@ -141,22 +302,43 @@ class AsyncTcpConnection extends TcpConnection
             if ($this->_sendBuffer) {
                 Worker::$globalEvent->add($socket, EventInterface::EV_WRITE, array($this, 'baseWrite'));
             }
-            $this->_status        = self::STATUS_ESTABLISH;
-            $this->_remoteAddress = stream_socket_get_name($socket, true);
+            $this->_status                = self::STATUS_ESTABLISHED;
+            $this->_remoteAddress         = $address;
+            $this->_sslHandshakeCompleted = true;
+
             // Try to emit onConnect callback.
             if ($this->onConnect) {
                 try {
                     call_user_func($this->onConnect, $this);
                 } catch (\Exception $e) {
-                    echo $e;
+                    Worker::log($e);
+                    exit(250);
+                } catch (\Error $e) {
+                    Worker::log($e);
+                    exit(250);
+                }
+            }
+            // Try to emit protocol::onConnect
+            if (method_exists($this->protocol, 'onConnect')) {
+                try {
+                    call_user_func(array($this->protocol, 'onConnect'), $this);
+                } catch (\Exception $e) {
+                    Worker::log($e);
+                    exit(250);
+                } catch (\Error $e) {
+                    Worker::log($e);
                     exit(250);
                 }
             }
         } else {
             // Connection failed.
-            $this->emitError(WORKERMAN_CONNECT_FAIL, 'connect fail');
-            $this->destroy();
-            $this->onConnect = null;
+            $this->emitError(WORKERMAN_CONNECT_FAIL, 'connect ' . $this->_remoteAddress . ' fail after ' . round(microtime(true) - $this->_connectStartTime, 4) . ' seconds');
+            if ($this->_status === self::STATUS_CLOSING) {
+                $this->destroy();
+            }
+            if ($this->_status === self::STATUS_CLOSED) {
+                $this->onConnect = null;
+            }
         }
     }
 }

Connection/ConnectionInterface.php

@@ -74,6 +74,48 @@ abstract class  ConnectionInterface
     abstract public function getRemotePort();
 
     /**
+     * Get remote address.
+     *
+     * @return string
+     */
+    abstract public function getRemoteAddress();
+
+    /**
+     * Get local IP.
+     *
+     * @return string
+     */
+    abstract public function getLocalIp();
+
+    /**
+     * Get local port.
+     *
+     * @return int
+     */
+    abstract public function getLocalPort();
+
+    /**
+     * Get local address.
+     *
+     * @return string
+     */
+    abstract public function getLocalAddress();
+
+    /**
+     * Is ipv4.
+     *
+     * @return bool
+     */
+    abstract public function isIPv4();
+
+    /**
+     * Is ipv6.
+     *
+     * @return bool
+     */
+    abstract public function isIPv6();
+
+    /**
      * Close connection.
      *
      * @param $data

Connection/TcpConnection.php

@@ -30,6 +30,13 @@ class TcpConnection extends ConnectionInterface
     const READ_BUFFER_SIZE = 65535;
 
     /**
+     * Status initial.
+     *
+     * @var int
+     */
+    const STATUS_INITIAL = 0;
+
+    /**
      * Status connecting.
      *
      * @var int
@@ -41,7 +48,7 @@ class TcpConnection extends ConnectionInterface
      *
      * @var int
      */
-    const STATUS_ESTABLISH = 2;
+    const STATUS_ESTABLISHED = 2;
 
     /**
      * Status closing.
@@ -101,6 +108,13 @@ class TcpConnection extends ConnectionInterface
     public $protocol = null;
 
     /**
+     * Transport (tcp/udp/unix/ssl).
+     *
+     * @var string
+     */
+    public $transport = 'tcp';
+
+    /**
      * Which worker belong to.
      *
      * @var Worker
@@ -108,6 +122,20 @@ class TcpConnection extends ConnectionInterface
     public $worker = null;
 
     /**
+     * Bytes read.
+     *
+     * @var int
+     */
+    public $bytesRead = 0;
+
+    /**
+     * Bytes written.
+     *
+     * @var int
+     */
+    public $bytesWritten = 0;
+
+    /**
      * Connection->id.
      *
      * @var int
@@ -183,7 +211,7 @@ class TcpConnection extends ConnectionInterface
      *
      * @var int
      */
-    protected $_status = self::STATUS_ESTABLISH;
+    protected $_status = self::STATUS_ESTABLISHED;
 
     /**
      * Remote address.
@@ -200,6 +228,58 @@ class TcpConnection extends ConnectionInterface
     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',
+    );
+
+
+    /**
+     * Adding support of custom functions within protocols
+     *
+     * @param string $name
+     * @param array  $arguments
+     */
+    public function __call($name, $arguments) {
+        // Try to emit custom function within protocol
+        if (method_exists($this->protocol, $name)) {
+            try {
+                return call_user_func(array($this->protocol, $name), $this, $arguments);
+            } catch (\Exception $e) {
+                Worker::log($e);
+                exit(250);
+            } catch (\Error $e) {
+                Worker::log($e);
+                exit(250);
+            }
+	} else {
+	    trigger_error('Call to undefined method '.__CLASS__.'::'.$name.'()', E_USER_ERROR);
+	}
+
+    }
+
+    /**
      * Construct.
      *
      * @param resource $socket
@@ -208,25 +288,52 @@ class TcpConnection extends ConnectionInterface
     public function __construct($socket, $remote_address = '')
     {
         self::$statistics['connection_count']++;
-        $this->id      = $this->_id = self::$_idRecorder++;
+        $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->_remoteAddress    = $remote_address;
+        static::$connections[$this->id] = $this;
+    }
+
+    /**
+     * Get status.
+     *
+     * @param bool $raw_output
+     *
+     * @return int
+     */
+    public function getStatus($raw_output = true)
+    {
+        if ($raw_output) {
+            return $this->_status;
+        }
+        return self::$_statusToString[$this->_status];
     }
 
     /**
      * Sends data on the connection.
      *
      * @param string $send_buffer
-     * @param bool   $raw
+     * @param bool  $raw
      * @return void|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) {
+        if (false === $raw && $this->protocol !== null) {
             $parser      = $this->protocol;
             $send_buffer = $parser::encode($send_buffer, $this);
             if ($send_buffer === '') {
@@ -234,23 +341,33 @@ class TcpConnection extends ConnectionInterface
             }
         }
 
-        if ($this->_status === self::STATUS_CONNECTING) {
+        if ($this->_status !== self::STATUS_ESTABLISHED ||
+            ($this->transport === 'ssl' && $this->_sslHandshakeCompleted !== true)
+        ) {
+            if ($this->_sendBuffer) {
+                if ($this->bufferIsFull()) {
+                    self::$statistics['send_fail']++;
+                    return false;
+                }
+            }
             $this->_sendBuffer .= $send_buffer;
+            $this->checkBufferWillFull();
             return null;
-        } elseif ($this->_status === self::STATUS_CLOSING || $this->_status === self::STATUS_CLOSED) {
-            return false;
         }
 
+
         // Attempt to send data directly.
         if ($this->_sendBuffer === '') {
-            $len = @fwrite($this->_socket, $send_buffer);
+            $len = @fwrite($this->_socket, $send_buffer, 8192);
             // 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)) {
@@ -259,7 +376,10 @@ class TcpConnection extends ConnectionInterface
                         try {
                             call_user_func($this->onError, $this, WORKERMAN_SEND_FAIL, 'client closed');
                         } catch (\Exception $e) {
-                            echo $e;
+                            Worker::log($e);
+                            exit(250);
+                        } catch (\Error $e) {
+                            Worker::log($e);
                             exit(250);
                         }
                     }
@@ -269,26 +389,18 @@ class TcpConnection extends ConnectionInterface
                 $this->_sendBuffer = $send_buffer;
             }
             Worker::$globalEvent->add($this->_socket, EventInterface::EV_WRITE, array($this, 'baseWrite'));
-            // Check if the send buffer is full.
-            $this->checkBufferIsFull();
+            // Check if the send buffer will be full.
+            $this->checkBufferWillFull();
             return null;
         } else {
-            // Buffer has been marked as full but still has data to send the packet is discarded.
-            if ($this->maxSendBufferSize <= strlen($this->_sendBuffer)) {
+            if ($this->bufferIsFull()) {
                 self::$statistics['send_fail']++;
-                if ($this->onError) {
-                    try {
-                        call_user_func($this->onError, $this, WORKERMAN_SEND_FAIL, 'send buffer full and drop package');
-                    } catch (\Exception $e) {
-                        echo $e;
-                        exit(250);
-                    }
-                }
                 return false;
             }
+
             $this->_sendBuffer .= $send_buffer;
             // Check if the send buffer is full.
-            $this->checkBufferIsFull();
+            $this->checkBufferWillFull();
         }
     }
 
@@ -320,6 +432,102 @@ class TcpConnection extends ConnectionInterface
     }
 
     /**
+     * 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()
+    {
+        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
@@ -348,28 +556,58 @@ class TcpConnection extends ConnectionInterface
      * Base read handler.
      *
      * @param resource $socket
+     * @param bool $check_eof
      * @return void
      */
     public function baseRead($socket, $check_eof = true)
     {
-        $read_data = false;
-        while (1) {
-            $buffer = fread($socket, self::READ_BUFFER_SIZE);
-            if ($buffer === '' || $buffer === false) {
-                break;
+        // SSL handshake.
+        if ($this->transport === 'ssl' && $this->_sslHandshakeCompleted !== true) {
+            $ret = stream_socket_enable_crypto($socket, true, STREAM_CRYPTO_METHOD_SSLv2_SERVER | 
+					       STREAM_CRYPTO_METHOD_SSLv23_SERVER);
+            // Negotiation has failed.
+            if(false === $ret) {
+                if (!feof($socket)) {
+                    echo "\nSSL Handshake fail. \nBuffer:".bin2hex(fread($socket, 8182))."\n";
+                }
+                return $this->destroy();
+            } elseif(0 === $ret) {
+                // There isn't enough data and should try again.
+                return;
             }
-            $read_data = true;
-            $this->_recvBuffer .= $buffer;
+            if (isset($this->onSslHandshake)) {
+                try {
+                    call_user_func($this->onSslHandshake, $this);
+                } catch (\Exception $e) {
+                    Worker::log($e);
+                    exit(250);
+                } catch (\Error $e) {
+                    Worker::log($e);
+                    exit(250);
+                }
+            }
+            $this->_sslHandshakeCompleted = true;
+            if ($this->_sendBuffer) {
+                Worker::$globalEvent->add($socket, EventInterface::EV_WRITE, array($this, 'baseWrite'));
+            }
+            return;
         }
 
+        $buffer = @fread($socket, self::READ_BUFFER_SIZE);
+
         // Check connection closed.
-        if (!$read_data && $check_eof) {
-            $this->destroy();
-            return;
+        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) {
+        if ($this->protocol !== null) {
             $parser = $this->protocol;
             while ($this->_recvBuffer !== '' && !$this->_isPaused) {
                 // The current packet length is known.
@@ -415,10 +653,13 @@ class TcpConnection extends ConnectionInterface
                     continue;
                 }
                 try {
-                    // Decode request buffer before Emiting onMessage callback.
+                    // Decode request buffer before Emitting onMessage callback.
                     call_user_func($this->onMessage, $this, $parser::decode($one_request_buffer, $this));
                 } catch (\Exception $e) {
-                    echo $e;
+                    Worker::log($e);
+                    exit(250);
+                } catch (\Error $e) {
+                    Worker::log($e);
                     exit(250);
                 }
             }
@@ -438,7 +679,10 @@ class TcpConnection extends ConnectionInterface
         try {
             call_user_func($this->onMessage, $this, $this->_recvBuffer);
         } catch (\Exception $e) {
-            echo $e;
+            Worker::log($e);
+            exit(250);
+        } catch (\Error $e) {
+            Worker::log($e);
             exit(250);
         }
         // Clean receive buffer.
@@ -452,8 +696,9 @@ class TcpConnection extends ConnectionInterface
      */
     public function baseWrite()
     {
-        $len = @fwrite($this->_socket, $this->_sendBuffer);
+        $len = @fwrite($this->_socket, $this->_sendBuffer, 8192);
         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. 
@@ -461,7 +706,10 @@ class TcpConnection extends ConnectionInterface
                 try {
                     call_user_func($this->onBufferDrain, $this);
                 } catch (\Exception $e) {
-                    echo $e;
+                    Worker::log($e);
+                    exit(250);
+                } catch (\Error $e) {
+                    Worker::log($e);
                     exit(250);
                 }
             }
@@ -471,6 +719,7 @@ class TcpConnection extends ConnectionInterface
             return true;
         }
         if ($len > 0) {
+            $this->bytesWritten += $len;
             $this->_sendBuffer = substr($this->_sendBuffer, $len);
         } else {
             self::$statistics['send_fail']++;
@@ -516,15 +765,16 @@ class TcpConnection extends ConnectionInterface
      * Close connection.
      *
      * @param mixed $data
+     * @param bool $raw
      * @return void
      */
-    public function close($data = null)
+    public function close($data = null, $raw = false)
     {
         if ($this->_status === self::STATUS_CLOSING || $this->_status === self::STATUS_CLOSED) {
             return;
         } else {
             if ($data !== null) {
-                $this->send($data);
+                $this->send($data, $raw);
             }
             $this->_status = self::STATUS_CLOSING;
         }
@@ -544,18 +794,21 @@ class TcpConnection extends ConnectionInterface
     }
 
     /**
-     * Check whether the send buffer is full.
+     * Check whether the send buffer will be full.
      *
      * @return void
      */
-    protected function checkBufferIsFull()
+    protected function checkBufferWillFull()
     {
         if ($this->maxSendBufferSize <= strlen($this->_sendBuffer)) {
             if ($this->onBufferFull) {
                 try {
                     call_user_func($this->onBufferFull, $this);
                 } catch (\Exception $e) {
-                    echo $e;
+                    Worker::log($e);
+                    exit(250);
+                } catch (\Error $e) {
+                    Worker::log($e);
                     exit(250);
                 }
             }
@@ -563,6 +816,31 @@ class TcpConnection extends ConnectionInterface
     }
 
     /**
+     * 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::log($e);
+                    exit(250);
+                } catch (\Error $e) {
+                    Worker::log($e);
+                    exit(250);
+                }
+            }
+            return true;
+        }
+        return false;
+    }
+
+    /**
      * Destroy connection.
      *
      * @return void
@@ -582,18 +860,36 @@ class TcpConnection extends ConnectionInterface
         if ($this->worker) {
             unset($this->worker->connections[$this->_id]);
         }
+        unset(static::$connections[$this->_id]);
         $this->_status = self::STATUS_CLOSED;
         // Try to emit onClose callback.
         if ($this->onClose) {
             try {
                 call_user_func($this->onClose, $this);
             } catch (\Exception $e) {
-                echo $e;
+                Worker::log($e);
+                exit(250);
+            } catch (\Error $e) {
+                Worker::log($e);
+                exit(250);
+            }
+        }
+        // Try to emit protocol::onClose
+        if (method_exists($this->protocol, 'onClose')) {
+            try {
+                call_user_func(array($this->protocol, 'onClose'), $this);
+            } catch (\Exception $e) {
+                Worker::log($e);
+                exit(250);
+            } catch (\Error $e) {
+                Worker::log($e);
                 exit(250);
             }
         }
-        // Cleaning up the callback to avoid memory leaks.
-        $this->onMessage = $this->onClose = $this->onError = $this->onBufferFull = $this->onBufferDrain = null;
+        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;
+        }
     }
 
     /**
@@ -603,6 +899,21 @@ class TcpConnection extends ConnectionInterface
      */
     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::$globalEvent->destroy();
+                exit(0);
+            }
+        }
     }
 }

Connection/UdpConnection.php

@@ -34,20 +34,6 @@ class UdpConnection extends ConnectionInterface
     protected $_socket = null;
 
     /**
-     * Remote ip.
-     *
-     * @var string
-     */
-    protected $_remoteIp = '';
-
-    /**
-     * Remote port.
-     *
-     * @var int
-     */
-    protected $_remotePort = 0;
-
-    /**
      * Remote address.
      *
      * @var string
@@ -92,10 +78,11 @@ class UdpConnection extends ConnectionInterface
      */
     public function getRemoteIp()
     {
-        if (!$this->_remoteIp) {
-            list($this->_remoteIp, $this->_remotePort) = explode(':', $this->_remoteAddress, 2);
+        $pos = strrpos($this->_remoteAddress, ':');
+        if ($pos) {
+            return trim(substr($this->_remoteAddress, 0, $pos), '[]');
         }
-        return $this->_remoteIp;
+        return '';
     }
 
     /**
@@ -105,22 +92,99 @@ class UdpConnection extends ConnectionInterface
      */
     public function getRemotePort()
     {
-        if (!$this->_remotePort) {
-            list($this->_remoteIp, $this->_remotePort) = explode(':', $this->_remoteAddress, 2);
+        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()
+    {
+        return (string)@stream_socket_get_name($this->_socket, false);
+    }
+
+    /**
+     * 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 $this->_remotePort;
+        return strpos($this->getRemoteIp(), ':') !== false;
     }
 
     /**
      * Close connection.
      *
      * @param mixed $data
+     * @param bool  $raw
      * @return bool
      */
-    public function close($data = null)
+    public function close($data = null, $raw = false)
     {
         if ($data !== null) {
-            $this->send($data);
+            $this->send($data, $raw);
         }
         return true;
     }

Events/Ev.php

@@ -12,6 +12,8 @@
  */
 namespace Workerman\Events;
 
+use Workerman\Worker;
+
 /**
  * ev eventloop
  */
@@ -56,11 +58,13 @@ class Ev implements EventInterface
             try {
                 call_user_func($func, $fd);
             } catch (\Exception $e) {
-                echo $e;
+                Worker::log($e);
+                exit(250);
+            } catch (\Error $e) {
+                Worker::log($e);
                 exit(250);
             }
         };
-
         switch ($flag) {
             case self::EV_SIGNAL:
                 $event                   = new \EvSignal($fd, $callback);
@@ -104,7 +108,7 @@ class Ev implements EventInterface
             case  self::EV_SIGNAL:
                 $fd_key = (int)$fd;
                 if (isset($this->_eventSignal[$fd_key])) {
-                    $this->_allEvents[$fd_key][$flag]->stop();
+                    $this->_eventSignal[$fd_key]->stop();
                     unset($this->_eventSignal[$fd_key]);
                 }
                 break;
@@ -135,7 +139,10 @@ class Ev implements EventInterface
         try {
             call_user_func_array($param[0], $param[1]);
         } catch (\Exception $e) {
-            echo $e;
+            Worker::log($e);
+            exit(250);
+        } catch (\Error $e) {
+            Worker::log($e);
             exit(250);
         }
     }
@@ -162,4 +169,26 @@ class Ev implements EventInterface
     {
         \Ev::run();
     }
+
+    /**
+     * Destroy loop.
+     *
+     * @return void
+     */
+    public function destroy()
+    {
+        foreach ($this->_allEvents as $event) {
+            $event->stop();
+        }
+    }
+
+    /**
+     * Get timer count.
+     *
+     * @return integer
+     */
+    public function getTimerCount()
+    {
+        return count($this->_eventTimer);
+    }
 }

Events/Event.php

@@ -13,6 +13,8 @@
  */
 namespace Workerman\Events;
 
+use Workerman\Worker;
+
 /**
  * libevent eventloop
  */
@@ -118,10 +120,9 @@ class Event implements EventInterface
                 break;
 
             case  self::EV_SIGNAL:
-
                 $fd_key = (int)$fd;
                 if (isset($this->_eventSignal[$fd_key])) {
-                    $this->_allEvents[$fd_key][$flag]->del();
+                    $this->_eventSignal[$fd_key]->del();
                     unset($this->_eventSignal[$fd_key]);
                 }
                 break;
@@ -155,7 +156,10 @@ class Event implements EventInterface
         try {
             call_user_func_array($param[0], $param[1]);
         } catch (\Exception $e) {
-            echo $e;
+            Worker::log($e);
+            exit(250);
+        } catch (\Error $e) {
+            Worker::log($e);
             exit(250);
         }
     }
@@ -180,4 +184,26 @@ class Event implements EventInterface
     {
         $this->_eventBase->loop();
     }
+
+    /**
+     * Destroy loop.
+     *
+     * @return void
+     */
+    public function destroy()
+    {
+        foreach ($this->_eventSignal as $event) {
+            $event->del();
+        }
+    }
+
+    /**
+     * Get timer count.
+     *
+     * @return integer
+     */
+    public function getTimerCount()
+    {
+        return count($this->_eventTimer);
+    }
 }

Events/EventInterface.php

@@ -30,6 +30,13 @@ interface EventInterface
     const EV_WRITE = 2;
 
     /**
+     * Except event
+     *
+     * @var int
+     */
+    const EV_EXCEPT = 3;
+
+    /**
      * Signal event.
      *
      * @var int
@@ -83,4 +90,18 @@ interface EventInterface
      * @return void
      */
     public function loop();
+
+    /**
+     * Destroy loop.
+     *
+     * @return mixed
+     */
+    public function destroy();
+
+    /**
+     * Get Timer count.
+     *
+     * @return mixed
+     */
+    public function getTimerCount();
 }

Events/Libevent.php

@@ -13,6 +13,8 @@
  */
 namespace Workerman\Events;
 
+use Workerman\Worker;
+
 /**
  * libevent eventloop
  */
@@ -170,7 +172,10 @@ class Libevent implements EventInterface
         try {
             call_user_func_array($this->_eventTimer[$timer_id][0], $this->_eventTimer[$timer_id][1]);
         } catch (\Exception $e) {
-            echo $e;
+            Worker::log($e);
+            exit(250);
+        } catch (\Error $e) {
+            Worker::log($e);
             exit(250);
         }
         if (isset($this->_eventTimer[$timer_id]) && $this->_eventTimer[$timer_id][3] === self::EV_TIMER_ONCE) {
@@ -196,5 +201,27 @@ class Libevent implements EventInterface
     {
         event_base_loop($this->_eventBase);
     }
+
+    /**
+     * Destroy loop.
+     *
+     * @return void
+     */
+    public function destroy()
+    {
+        foreach ($this->_eventSignal as $event) {
+            event_del($event);
+        }
+    }
+
+    /**
+     * Get timer count.
+     *
+     * @return integer
+     */
+    public function getTimerCount()
+    {
+        return count($this->_eventTimer);
+    }
 }
 

Events/React/ExtEventLoop.php

@@ -0,0 +1,186 @@
+<?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\Events\React;
+use Workerman\Events\EventInterface;
+
+/**
+ * Class ExtEventLoop
+ * @package Workerman\Events\React
+ */
+class ExtEventLoop extends \React\EventLoop\ExtEventLoop
+{
+    /**
+     * Event base.
+     *
+     * @var EventBase
+     */
+    protected $_eventBase = null;
+
+    /**
+     * All signal Event instances.
+     *
+     * @var array
+     */
+    protected $_signalEvents = array();
+
+    /**
+     * @var array
+     */
+    protected $_timerIdMap = array();
+
+    /**
+     * @var int
+     */
+    protected $_timerIdIndex = 0;
+
+    /**
+     * Add event listener to event loop.
+     *
+     * @param $fd
+     * @param $flag
+     * @param $func
+     * @param array $args
+     * @return bool
+     */
+    public function add($fd, $flag, $func, $args = array())
+    {
+        $args = (array)$args;
+        switch ($flag) {
+            case EventInterface::EV_READ:
+                return $this->addReadStream($fd, $func);
+            case EventInterface::EV_WRITE:
+                return $this->addWriteStream($fd, $func);
+            case EventInterface::EV_SIGNAL:
+                return $this->addSignal($fd, $func);
+            case EventInterface::EV_TIMER:
+                $timer_id = ++$this->_timerIdIndex;
+                $timer_obj = $this->addPeriodicTimer($fd, function() use ($func, $args) {
+                    call_user_func_array($func, $args);
+                });
+                $this->_timerIdMap[$timer_id] = $timer_obj;
+                return $timer_id;
+            case EventInterface::EV_TIMER_ONCE:
+                $timer_id = ++$this->_timerIdIndex;
+                $timer_obj = $this->addTimer($fd, function() use ($func, $args, $timer_id) {
+                    unset($this->_timerIdMap[$timer_id]);
+                    call_user_func_array($func, $args);
+                });
+                $this->_timerIdMap[$timer_id] = $timer_obj;
+                return $timer_id;
+        }
+        return false;
+    }
+
+    /**
+     * Remove event listener from event loop.
+     *
+     * @param mixed $fd
+     * @param int   $flag
+     * @return bool
+     */
+    public function del($fd, $flag)
+    {
+        switch ($flag) {
+            case EventInterface::EV_READ:
+                return $this->removeReadStream($fd);
+            case EventInterface::EV_WRITE:
+                return $this->removeWriteStream($fd);
+            case EventInterface::EV_SIGNAL:
+                return $this->removeSignal($fd);
+            case EventInterface::EV_TIMER:
+            case EventInterface::EV_TIMER_ONCE;
+                if (isset($this->_timerIdMap[$fd])){
+                    $timer_obj = $this->_timerIdMap[$fd];
+                    unset($this->_timerIdMap[$fd]);
+                    $this->cancelTimer($timer_obj);
+                    return true;
+                }
+        }
+        return false;
+    }
+
+
+    /**
+     * Main loop.
+     *
+     * @return void
+     */
+    public function loop()
+    {
+        $this->run();
+    }
+
+    /**
+     * Construct
+     */
+    public function __construct()
+    {
+        parent::__construct();
+        $class = new \ReflectionClass('\React\EventLoop\ExtEventLoop');
+        $property = $class->getProperty('eventBase');
+        $property->setAccessible(true);
+        $this->_eventBase = $property->getValue($this);
+    }
+
+    /**
+     * Add signal handler.
+     *
+     * @param $signal
+     * @param $callback
+     * @return bool
+     */
+    public function addSignal($signal, $callback)
+    {
+        $event = \Event::signal($this->_eventBase, $signal, $callback);
+        if (!$event||!$event->add()) {
+            return false;
+        }
+        $this->_signalEvents[$signal] = $event;
+    }
+
+    /**
+     * Remove signal handler.
+     *
+     * @param $signal
+     */
+    public function removeSignal($signal)
+    {
+        if (isset($this->_signalEvents[$signal])) {
+            $this->_signalEvents[$signal]->del();
+            unset($this->_signalEvents[$signal]);
+        }
+    }
+
+    /**
+     * Destroy loop.
+     *
+     * @return void
+     */
+    public function destroy()
+    {
+        foreach ($this->_signalEvents as $event) {
+            $event->del();
+        }
+    }
+
+    /**
+     * Get timer count.
+     *
+     * @return integer
+     */
+    public function getTimerCount()
+    {
+        return count($this->_timerIdMap);
+    }
+}

Events/React/LibEventLoop.php

@@ -0,0 +1,187 @@
+<?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\Events\React;
+use Workerman\Events\EventInterface;
+
+/**
+ * Class LibEventLoop
+ * @package Workerman\Events\React
+ */
+class LibEventLoop extends \React\EventLoop\LibEventLoop
+{
+    /**
+     * Event base.
+     *
+     * @var event_base resource
+     */
+    protected $_eventBase = null;
+
+    /**
+     * All signal Event instances.
+     *
+     * @var array
+     */
+    protected $_signalEvents = array();
+
+    /**
+     * @var array
+     */
+    protected $_timerIdMap = array();
+
+    /**
+     * @var int
+     */
+    protected $_timerIdIndex = 0;
+
+    /**
+     * Add event listener to event loop.
+     *
+     * @param $fd
+     * @param $flag
+     * @param $func
+     * @param array $args
+     * @return bool
+     */
+    public function add($fd, $flag, $func, $args = array())
+    {
+        $args = (array)$args;
+        switch ($flag) {
+            case EventInterface::EV_READ:
+                return $this->addReadStream($fd, $func);
+            case EventInterface::EV_WRITE:
+                return $this->addWriteStream($fd, $func);
+            case EventInterface::EV_SIGNAL:
+                return $this->addSignal($fd, $func);
+            case EventInterface::EV_TIMER:
+                $timer_id = ++$this->_timerIdIndex;
+                $timer_obj = $this->addPeriodicTimer($fd, function() use ($func, $args) {
+                    call_user_func_array($func, $args);
+                });
+                $this->_timerIdMap[$timer_id] = $timer_obj;
+                return $timer_id;
+            case EventInterface::EV_TIMER_ONCE:
+                $timer_id = ++$this->_timerIdIndex;
+                $timer_obj = $this->addTimer($fd, function() use ($func, $args, $timer_id) {
+                    unset($this->_timerIdMap[$timer_id]);
+                    call_user_func_array($func, $args);
+                });
+                $this->_timerIdMap[$timer_id] = $timer_obj;
+                return $timer_id;
+        }
+        return false;
+    }
+
+    /**
+     * Remove event listener from event loop.
+     *
+     * @param mixed $fd
+     * @param int   $flag
+     * @return bool
+     */
+    public function del($fd, $flag)
+    {
+        switch ($flag) {
+            case EventInterface::EV_READ:
+                return $this->removeReadStream($fd);
+            case EventInterface::EV_WRITE:
+                return $this->removeWriteStream($fd);
+            case EventInterface::EV_SIGNAL:
+                return $this->removeSignal($fd);
+            case EventInterface::EV_TIMER:
+            case EventInterface::EV_TIMER_ONCE;
+                if (isset($this->_timerIdMap[$fd])){
+                    $timer_obj = $this->_timerIdMap[$fd];
+                    unset($this->_timerIdMap[$fd]);
+                    $this->cancelTimer($timer_obj);
+                    return true;
+                }
+        }
+        return false;
+    }
+
+
+    /**
+     * Main loop.
+     *
+     * @return void
+     */
+    public function loop()
+    {
+        $this->run();
+    }
+
+    /**
+     * Construct.
+     */
+    public function __construct()
+    {
+        parent::__construct();
+        $class = new \ReflectionClass('\React\EventLoop\LibEventLoop');
+        $property = $class->getProperty('eventBase');
+        $property->setAccessible(true);
+        $this->_eventBase = $property->getValue($this);
+    }
+
+    /**
+     * Add signal handler.
+     *
+     * @param $signal
+     * @param $callback
+     * @return bool
+     */
+    public function addSignal($signal, $callback)
+    {
+        $event = event_new();
+        $this->_signalEvents[$signal] = $event;
+        event_set($event, $signal, EV_SIGNAL | EV_PERSIST, $callback);
+        event_base_set($event, $this->_eventBase);
+        event_add($event);
+    }
+
+    /**
+     * Remove signal handler.
+     *
+     * @param $signal
+     */
+    public function removeSignal($signal)
+    {
+        if (isset($this->_signalEvents[$signal])) {
+            $event = $this->_signalEvents[$signal];
+            event_del($event);
+            unset($this->_signalEvents[$signal]);
+        }
+    }
+
+    /**
+     * Destroy loop.
+     *
+     * @return void
+     */
+    public function destroy()
+    {
+        foreach ($this->_signalEvents as $event) {
+            event_del($event);
+        }
+    }
+
+    /**
+     * Get timer count.
+     *
+     * @return integer
+     */
+    public function getTimerCount()
+    {
+        return count($this->_timerIdMap);
+    }
+}

Events/React/StreamSelectLoop.php

@@ -0,0 +1,186 @@
+<?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\Events\React;
+use Workerman\Events\EventInterface;
+
+/**
+ * Class StreamSelectLoop
+ * @package Workerman\Events\React
+ */
+class StreamSelectLoop extends \React\EventLoop\StreamSelectLoop
+{
+    /**
+     * @var array
+     */
+    protected $_timerIdMap = array();
+
+    /**
+     * @var int
+     */
+    protected $_timerIdIndex = 0;
+
+    /**
+     * Add event listener to event loop.
+     *
+     * @param $fd
+     * @param $flag
+     * @param $func
+     * @param array $args
+     * @return bool
+     */
+    public function add($fd, $flag, $func, $args = array())
+    {
+        $args = (array)$args;
+        switch ($flag) {
+            case EventInterface::EV_READ:
+                return $this->addReadStream($fd, $func);
+            case EventInterface::EV_WRITE:
+                return $this->addWriteStream($fd, $func);
+            case EventInterface::EV_SIGNAL:
+                return $this->addSignal($fd, $func);
+            case EventInterface::EV_TIMER:
+                $timer_obj = $this->addPeriodicTimer($fd, function() use ($func, $args) {
+                    call_user_func_array($func, $args);
+                });
+                $this->_timerIdMap[++$this->_timerIdIndex] = $timer_obj;
+                return $this->_timerIdIndex;
+            case EventInterface::EV_TIMER_ONCE:
+                $index = ++$this->_timerIdIndex;
+                $timer_obj = $this->addTimer($fd, function() use ($func, $args, $index) {
+                    $this->del($index,EventInterface::EV_TIMER_ONCE);
+                    call_user_func_array($func, $args);
+                });
+                $this->_timerIdMap[$index] = $timer_obj;
+                return $this->_timerIdIndex;
+        }
+        return false;
+    }
+
+    /**
+     * Remove event listener from event loop.
+     *
+     * @param mixed $fd
+     * @param int   $flag
+     * @return bool
+     */
+    public function del($fd, $flag)
+    {
+        switch ($flag) {
+            case EventInterface::EV_READ:
+                return $this->removeReadStream($fd);
+            case EventInterface::EV_WRITE:
+                return $this->removeWriteStream($fd);
+            case EventInterface::EV_SIGNAL:
+                return $this->removeSignal($fd);
+            case EventInterface::EV_TIMER:
+            case EventInterface::EV_TIMER_ONCE;
+                if (isset($this->_timerIdMap[$fd])){
+                    $timer_obj = $this->_timerIdMap[$fd];
+                    unset($this->_timerIdMap[$fd]);
+                    $this->cancelTimer($timer_obj);
+                    return true;
+                }
+        }
+        return false;
+    }
+
+
+    /**
+     * Main loop.
+     *
+     * @return void
+     */
+    public function loop()
+    {
+        $this->run();
+    }
+
+    /**
+     * Add signal handler.
+     *
+     * @param $signal
+     * @param $callback
+     * @return bool
+     */
+    public function addSignal($signal, $callback)
+    {
+        if(DIRECTORY_SEPARATOR === '/') {
+            pcntl_signal($signal, $callback);
+        }
+    }
+
+    /**
+     * Remove signal handler.
+     *
+     * @param $signal
+     */
+    public function removeSignal($signal)
+    {
+        if(DIRECTORY_SEPARATOR === '/') {
+            pcntl_signal($signal, SIG_IGN);
+        }
+    }
+
+    /**
+     * Emulate a stream_select() implementation that does not break when passed
+     * empty stream arrays.
+     *
+     * @param array        &$read   An array of read streams to select upon.
+     * @param array        &$write  An array of write streams to select upon.
+     * @param integer|null $timeout Activity timeout in microseconds, or null to wait forever.
+     *
+     * @return integer|false The total number of streams that are ready for read/write.
+     * Can return false if stream_select() is interrupted by a signal.
+     */
+    protected function streamSelect(array &$read, array &$write, $timeout)
+    {
+        if ($read || $write) {
+            $except = null;
+            // Calls signal handlers for pending signals
+            if(DIRECTORY_SEPARATOR === '/') {
+                pcntl_signal_dispatch();
+            }
+            // suppress warnings that occur, when stream_select is interrupted by a signal
+            return @stream_select($read, $write, $except, $timeout === null ? null : 0, $timeout);
+        }
+
+        // Calls signal handlers for pending signals
+        if(DIRECTORY_SEPARATOR === '/') {
+            pcntl_signal_dispatch();
+        }
+        $timeout && usleep($timeout);
+
+        return 0;
+    }
+
+    /**
+     * Destroy loop.
+     *
+     * @return void
+     */
+    public function destroy()
+    {
+
+    }
+
+    /**
+     * Get timer count.
+     *
+     * @return integer
+     */
+    public function getTimerCount()
+    {
+        return count($this->_timerIdMap);
+    }
+}

Events/Select.php

@@ -47,6 +47,13 @@ class Select implements EventInterface
     protected $_writeFds = array();
 
     /**
+     * Fds waiting for except event.
+     *
+     * @var array
+     */
+    protected $_exceptFds = array();
+
+    /**
      * Timer scheduler.
      * {['data':timer_id, 'priority':run_timestamp], ..}
      *
@@ -60,7 +67,7 @@ class Select implements EventInterface
      *
      * @var array
      */
-    protected $_task = array();
+    protected $_eventTimer = array();
 
     /**
      * Timer id.
@@ -89,8 +96,9 @@ class Select implements EventInterface
     public function __construct()
     {
         // Create a pipeline and put into the collection of the read to read the descriptor to avoid empty polling.
-        $this->channel = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
-        if ($this->channel) {
+        $this->channel = stream_socket_pair(DIRECTORY_SEPARATOR === '/' ? STREAM_PF_UNIX : STREAM_PF_INET,
+            STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
+        if($this->channel) {
             stream_set_blocking($this->channel[0], 0);
             $this->_readFds[0] = $this->channel[0];
         }
@@ -115,18 +123,31 @@ class Select implements EventInterface
                 $this->_allEvents[$fd_key][$flag] = array($func, $fd);
                 $this->_writeFds[$fd_key]         = $fd;
                 break;
+            case self::EV_EXCEPT:
+                $fd_key = (int)$fd;
+                $this->_allEvents[$fd_key][$flag] = array($func, $fd);
+                $this->_exceptFds[$fd_key] = $fd;
+                break;
             case self::EV_SIGNAL:
+                // Windows not support signal.
+                if(DIRECTORY_SEPARATOR !== '/') {
+                    return false;
+                }
                 $fd_key                              = (int)$fd;
                 $this->_signalEvents[$fd_key][$flag] = array($func, $fd);
                 pcntl_signal($fd, array($this, 'signalHandler'));
                 break;
             case self::EV_TIMER:
             case self::EV_TIMER_ONCE:
+                $timer_id = $this->_timerId++;
                 $run_time = microtime(true) + $fd;
-                $this->_scheduler->insert($this->_timerId, -$run_time);
-                $this->_task[$this->_timerId] = array($func, (array)$args, $flag, $fd);
-                $this->tick();
-                return $this->_timerId++;
+                $this->_scheduler->insert($timer_id, -$run_time);
+                $this->_eventTimer[$timer_id] = array($func, (array)$args, $flag, $fd);
+                $select_timeout = ($run_time - microtime(true)) * 1000000;
+                if( $this->_selectTimeout > $select_timeout ){ 
+                    $this->_selectTimeout = $select_timeout;   
+                }  
+                return $timer_id;
         }
 
         return true;
@@ -161,13 +182,23 @@ class Select implements EventInterface
                     unset($this->_allEvents[$fd_key]);
                 }
                 return true;
+            case self::EV_EXCEPT:
+                unset($this->_allEvents[$fd_key][$flag], $this->_exceptFds[$fd_key]);
+                if(empty($this->_allEvents[$fd_key]))
+                {
+                    unset($this->_allEvents[$fd_key]);
+                }
+                return true;
             case self::EV_SIGNAL:
+                if(DIRECTORY_SEPARATOR !== '/') {
+                    return false;
+                }
                 unset($this->_signalEvents[$fd_key]);
                 pcntl_signal($fd, SIG_IGN);
                 break;
             case self::EV_TIMER:
             case self::EV_TIMER_ONCE;
-                unset($this->_task[$fd_key]);
+                unset($this->_eventTimer[$fd_key]);
                 return true;
         }
         return false;
@@ -189,18 +220,18 @@ class Select implements EventInterface
             if ($this->_selectTimeout <= 0) {
                 $this->_scheduler->extract();
 
-                if (!isset($this->_task[$timer_id])) {
+                if (!isset($this->_eventTimer[$timer_id])) {
                     continue;
                 }
 
                 // [func, args, flag, timer_interval]
-                $task_data = $this->_task[$timer_id];
+                $task_data = $this->_eventTimer[$timer_id];
                 if ($task_data[2] === self::EV_TIMER) {
                     $next_run_time = $time_now + $task_data[3];
                     $this->_scheduler->insert($timer_id, -$next_run_time);
                 }
                 call_user_func_array($task_data[0], $task_data[1]);
-                if (isset($this->_task[$timer_id]) && $task_data[2] === self::EV_TIMER_ONCE) {
+                if (isset($this->_eventTimer[$timer_id]) && $task_data[2] === self::EV_TIMER_ONCE) {
                     $this->del($timer_id, self::EV_TIMER_ONCE);
                 }
                 continue;
@@ -217,7 +248,7 @@ class Select implements EventInterface
     {
         $this->_scheduler = new \SplPriorityQueue();
         $this->_scheduler->setExtractFlags(\SplPriorityQueue::EXTR_BOTH);
-        $this->_task = array();
+        $this->_eventTimer = array();
     }
 
     /**
@@ -227,13 +258,17 @@ class Select implements EventInterface
     {
         $e = null;
         while (1) {
-            // Calls signal handlers for pending signals
-            pcntl_signal_dispatch();
+            if(DIRECTORY_SEPARATOR === '/') {
+                // Calls signal handlers for pending signals
+                pcntl_signal_dispatch();
+            }
 
             $read  = $this->_readFds;
             $write = $this->_writeFds;
+            $except = $this->_writeFds;
+
             // Waiting read/write/signal/timeout events.
-            $ret = @stream_select($read, $write, $e, 0, $this->_selectTimeout);
+            $ret = @stream_select($read, $write, $except, 0, $this->_selectTimeout);
 
             if (!$this->_scheduler->isEmpty()) {
                 $this->tick();
@@ -243,21 +278,55 @@ class Select implements EventInterface
                 continue;
             }
 
-            foreach ($read as $fd) {
-                $fd_key = (int)$fd;
-                if (isset($this->_allEvents[$fd_key][self::EV_READ])) {
-                    call_user_func_array($this->_allEvents[$fd_key][self::EV_READ][0],
-                        array($this->_allEvents[$fd_key][self::EV_READ][1]));
+            if ($read) {
+                foreach ($read as $fd) {
+                    $fd_key = (int)$fd;
+                    if (isset($this->_allEvents[$fd_key][self::EV_READ])) {
+                        call_user_func_array($this->_allEvents[$fd_key][self::EV_READ][0],
+                            array($this->_allEvents[$fd_key][self::EV_READ][1]));
+                    }
                 }
             }
 
-            foreach ($write as $fd) {
-                $fd_key = (int)$fd;
-                if (isset($this->_allEvents[$fd_key][self::EV_WRITE])) {
-                    call_user_func_array($this->_allEvents[$fd_key][self::EV_WRITE][0],
-                        array($this->_allEvents[$fd_key][self::EV_WRITE][1]));
+            if ($write) {
+                foreach ($write as $fd) {
+                    $fd_key = (int)$fd;
+                    if (isset($this->_allEvents[$fd_key][self::EV_WRITE])) {
+                        call_user_func_array($this->_allEvents[$fd_key][self::EV_WRITE][0],
+                            array($this->_allEvents[$fd_key][self::EV_WRITE][1]));
+                    }
+                }
+            }
+
+            if($except) {
+                foreach($except as $fd) {
+                    $fd_key = (int) $fd;
+                    if(isset($this->_allEvents[$fd_key][self::EV_EXCEPT])) {
+                        call_user_func_array($this->_allEvents[$fd_key][self::EV_EXCEPT][0],
+                            array($this->_allEvents[$fd_key][self::EV_EXCEPT][1]));
+                    }
                 }
             }
         }
     }
+
+    /**
+     * Destroy loop.
+     *
+     * @return void
+     */
+    public function destroy()
+    {
+
+    }
+
+    /**
+     * Get timer count.
+     *
+     * @return integer
+     */
+    public function getTimerCount()
+    {
+        return count($this->_eventTimer);
+    }
 }

Lib/Constants.php

@@ -21,7 +21,20 @@ ini_set('display_errors', 'on');
 // Reporting all.
 error_reporting(E_ALL);
 
+// Reset opcache.
+if (function_exists('opcache_reset')) {
+    opcache_reset();
+}
+
 // For onError callback.
 define('WORKERMAN_CONNECT_FAIL', 1);
 // For onError callback.
-define('WORKERMAN_SEND_FAIL', 2);
+define('WORKERMAN_SEND_FAIL', 2);
+
+// Compatible with php7
+if(!class_exists('Error'))
+{
+    class Error extends Exception
+    {
+    }
+}

Lib/Timer.php

@@ -54,7 +54,9 @@ class Timer
         if ($event) {
             self::$_event = $event;
         } else {
-            pcntl_signal(SIGALRM, array('\Workerman\Lib\Timer', 'signalHandle'), false);
+            if (function_exists('pcntl_signal')) {
+                pcntl_signal(SIGALRM, array('\Workerman\Lib\Timer', 'signalHandle'), false);
+            }
         }
     }
 
@@ -74,11 +76,11 @@ class Timer
     /**
      * Add a timer.
      *
-     * @param int      $time_interval
-     * @param callback $func
+     * @param float    $time_interval
+     * @param callable $func
      * @param mixed    $args
      * @param bool     $persistent
-     * @return bool
+     * @return int/false
      */
     public static function add($time_interval, $func, $args = array(), $persistent = true)
     {
@@ -107,7 +109,7 @@ class Timer
             self::$_tasks[$run_time] = array();
         }
         self::$_tasks[$run_time][] = array($func, (array)$args, $persistent, $time_interval);
-        return true;
+        return 1;
     }
 
 

Protocols/Frame.php

@@ -37,7 +37,7 @@ class Frame
     }
 
     /**
-     * Encode.
+     * Decode.
      *
      * @param string $buffer
      * @return string
@@ -48,7 +48,7 @@ class Frame
     }
 
     /**
-     * Decode.
+     * Encode.
      *
      * @param string $buffer
      * @return string

Protocols/Http.php

@@ -14,6 +14,7 @@
 namespace Workerman\Protocols;
 
 use Workerman\Connection\TcpConnection;
+use Workerman\Worker;
 
 /**
  * http protocol
@@ -21,6 +22,12 @@ use Workerman\Connection\TcpConnection;
 class Http
 {
     /**
+      * The supported HTTP methods
+      * @var array
+      */
+    public static $methods = array('GET', 'POST', 'PUT', 'DELETE', 'HEAD', 'OPTIONS');
+
+    /**
      * Check the integrity of the package.
      *
      * @param string        $recv_buffer
@@ -39,18 +46,34 @@ class Http
         }
 
         list($header,) = explode("\r\n\r\n", $recv_buffer, 2);
-        if (0 === strpos($recv_buffer, "POST")) {
-            // find Content-Length
-            $match = array();
-            if (preg_match("/\r\nContent-Length: ?(\d+)/i", $header, $match)) {
-                $content_length = $match[1];
-                return $content_length + strlen($header) + 4;
-            } else {
-                return 0;
-            }
-        } else {
+        $method = substr($header, 0, strpos($header, ' '));
+
+        if(in_array($method, static::$methods)) {
+            return static::getRequestSize($header, $method);
+        }else{
+            $connection->send("HTTP/1.1 400 Bad Request\r\n\r\n", true);
+            return 0;
+        }
+    }
+
+    /**
+      * Get whole size of the request
+      * includes the request headers and request body.
+      * @param string $header The request headers
+      * @param string $method The request method
+      * @return integer
+      */
+    protected static function getRequestSize($header, $method)
+    {
+        if($method === 'GET' || $method === 'OPTIONS' || $method === 'HEAD') {
             return strlen($header) + 4;
         }
+        $match = array();
+        if (preg_match("/\r\nContent-Length: ?(\d+)/i", $header, $match)) {
+            $content_length = isset($match[1]) ? $match[1] : 0;
+            return $content_length + strlen($header) + 4;
+        }
+        return 0;
     }
 
     /**
@@ -74,7 +97,7 @@ class Http
             'REQUEST_METHOD'       => '',
             'REQUEST_URI'          => '',
             'SERVER_PROTOCOL'      => '',
-            'SERVER_SOFTWARE'      => 'workerman/3.0',
+            'SERVER_SOFTWARE'      => 'workerman/'.Worker::VERSION,
             'SERVER_NAME'          => '',
             'HTTP_HOST'            => '',
             'HTTP_USER_AGENT'      => '',
@@ -85,6 +108,7 @@ class Http
             'HTTP_CONNECTION'      => '',
             'REMOTE_ADDR'          => '',
             'REMOTE_PORT'          => '0',
+            'REQUEST_TIME'         => time()
         );
 
         // Parse headers.
@@ -101,13 +125,13 @@ class Http
             if (empty($content)) {
                 continue;
             }
-            list($key, $value) = explode(':', $content, 2);
-            $key   = strtolower($key);
-            $value = trim($value);
+            list($key, $value)       = explode(':', $content, 2);
+            $key                     = str_replace('-', '_', strtoupper($key));
+            $value                   = trim($value);
+            $_SERVER['HTTP_' . $key] = $value;
             switch ($key) {
                 // HTTP_HOST
-                case 'host':
-                    $_SERVER['HTTP_HOST']   = $value;
+                case 'HOST':
                     $tmp                    = explode(':', $value);
                     $_SERVER['SERVER_NAME'] = $tmp[0];
                     if (isset($tmp[1])) {
@@ -115,61 +139,48 @@ class Http
                     }
                     break;
                 // cookie
-                case 'cookie':
-                    $_SERVER['HTTP_COOKIE'] = $value;
+                case 'COOKIE':
                     parse_str(str_replace('; ', '&', $_SERVER['HTTP_COOKIE']), $_COOKIE);
                     break;
-                // user-agent
-                case 'user-agent':
-                    $_SERVER['HTTP_USER_AGENT'] = $value;
-                    break;
-                // accept
-                case 'accept':
-                    $_SERVER['HTTP_ACCEPT'] = $value;
-                    break;
-                // accept-language
-                case 'accept-language':
-                    $_SERVER['HTTP_ACCEPT_LANGUAGE'] = $value;
-                    break;
-                // accept-encoding
-                case 'accept-encoding':
-                    $_SERVER['HTTP_ACCEPT_ENCODING'] = $value;
-                    break;
-                // connection
-                case 'connection':
-                    $_SERVER['HTTP_CONNECTION'] = $value;
-                    break;
-                case 'referer':
-                    $_SERVER['HTTP_REFERER'] = $value;
-                    break;
-                case 'if-modified-since':
-                    $_SERVER['HTTP_IF_MODIFIED_SINCE'] = $value;
-                    break;
-                case 'if-none-match':
-                    $_SERVER['HTTP_IF_NONE_MATCH'] = $value;
-                    break;
-                case 'content-type':
+                // content-type
+                case 'CONTENT_TYPE':
                     if (!preg_match('/boundary="?(\S+)"?/', $value, $match)) {
-                        $_SERVER['CONTENT_TYPE'] = $value;
+                        if ($pos = strpos($value, ';')) {
+                            $_SERVER['CONTENT_TYPE'] = substr($value, 0, $pos);
+                        } else {
+                            $_SERVER['CONTENT_TYPE'] = $value;
+                        }
                     } else {
                         $_SERVER['CONTENT_TYPE'] = 'multipart/form-data';
                         $http_post_boundary      = '--' . $match[1];
                     }
                     break;
+                case 'CONTENT_LENGTH':
+                    $_SERVER['CONTENT_LENGTH'] = $value;
+                    break;
             }
         }
 
         // Parse $_POST.
         if ($_SERVER['REQUEST_METHOD'] === 'POST') {
-            if (isset($_SERVER['CONTENT_TYPE']) && $_SERVER['CONTENT_TYPE'] === 'multipart/form-data') {
-                self::parseUploadFiles($http_body, $http_post_boundary);
-            } else {
-                parse_str($http_body, $_POST);
-                // $GLOBALS['HTTP_RAW_POST_DATA']
-                $GLOBALS['HTTP_RAW_POST_DATA'] = $http_body;
+            if (isset($_SERVER['CONTENT_TYPE'])) {
+                switch ($_SERVER['CONTENT_TYPE']) {
+                    case 'multipart/form-data':
+                        self::parseUploadFiles($http_body, $http_post_boundary);
+                        break;
+                    case 'application/json':
+                        $_POST = json_decode($http_body, true);
+                        break;
+                    case 'application/x-www-form-urlencoded':
+                        parse_str($http_body, $_POST);
+                        break;
+                }
             }
         }
 
+        // HTTP_RAW_REQUEST_DATA HTTP_RAW_POST_DATA
+        $GLOBALS['HTTP_RAW_REQUEST_DATA'] = $GLOBALS['HTTP_RAW_POST_DATA'] = $http_body;
+
         // QUERY_STRING
         $_SERVER['QUERY_STRING'] = parse_url($_SERVER['REQUEST_URI'], PHP_URL_QUERY);
         if ($_SERVER['QUERY_STRING']) {
@@ -223,7 +234,7 @@ class Http
         }
 
         // header
-        $header .= "Server: WorkerMan/3.0\r\nContent-Length: " . strlen($content) . "\r\n\r\n";
+        $header .= "Server: workerman/" . Worker::VERSION . "\r\nContent-Length: " . strlen($content) . "\r\n\r\n";
 
         // save session
         self::sessionWriteClose();
@@ -329,8 +340,11 @@ class Http
         if (PHP_SAPI != 'cli') {
             return session_start();
         }
+
+        self::tryGcSessions();
+
         if (HttpCache::$instance->sessionStarted) {
-            echo "already sessionStarted\nn";
+            echo "already sessionStarted\n";
             return true;
         }
         HttpCache::$instance->sessionStarted = true;
@@ -359,9 +373,10 @@ class Http
         if (HttpCache::$instance->sessionFile) {
             $raw = file_get_contents(HttpCache::$instance->sessionFile);
             if ($raw) {
-                session_decode($raw);
+                $_SESSION = unserialize($raw);
             }
         }
+        return true;
     }
 
     /**
@@ -375,7 +390,7 @@ class Http
             return session_write_close();
         }
         if (!empty(HttpCache::$instance->sessionStarted) && !empty($_SESSION)) {
-            $session_str = session_encode();
+            $session_str = serialize($_SESSION);
             if ($session_str && HttpCache::$instance->sessionFile) {
                 return file_put_contents(HttpCache::$instance->sessionFile, $session_str);
             }
@@ -424,20 +439,23 @@ class Http
         if ($boundary_data_array[0] === '') {
             unset($boundary_data_array[0]);
         }
+        $key = -1;
         foreach ($boundary_data_array as $boundary_data_buffer) {
             list($boundary_header_buffer, $boundary_value) = explode("\r\n\r\n", $boundary_data_buffer, 2);
             // Remove \r\n from the end of buffer.
             $boundary_value = substr($boundary_value, 0, -2);
+            $key ++;
             foreach (explode("\r\n", $boundary_header_buffer) as $item) {
                 list($header_key, $header_value) = explode(": ", $item);
                 $header_key = strtolower($header_key);
                 switch ($header_key) {
                     case "content-disposition":
                         // Is file data.
-                        if (preg_match('/name=".*?"; filename="(.*?)"$/', $header_value, $match)) {
+                        if (preg_match('/name="(.*?)"; filename="(.*?)"$/', $header_value, $match)) {
                             // Parse $_FILES.
-                            $_FILES[] = array(
-                                'file_name' => $match[1],
+                            $_FILES[$key] = array(
+                                'name' => $match[1],
+                                'file_name' => $match[2],
                                 'file_data' => $boundary_value,
                                 'file_size' => strlen($boundary_value),
                             );
@@ -450,10 +468,35 @@ class Http
                             }
                         }
                         break;
+                    case "content-type":
+                        // add file_type
+                        $_FILES[$key]['file_type'] = trim($header_value);
+                        break;
                 }
             }
         }
     }
+
+    /**
+     * Try GC sessions.
+     *
+     * @return void
+     */
+    public static function tryGcSessions()
+    {
+        if (HttpCache::$sessionGcProbability <= 0 ||
+            HttpCache::$sessionGcDivisor     <= 0 ||
+            rand(1, HttpCache::$sessionGcDivisor) > HttpCache::$sessionGcProbability) {
+            return;
+        }
+
+        $time_now = time();
+        foreach(glob(HttpCache::$sessionPath.'/ses*') as $file) {
+            if(is_file($file) && $time_now - filemtime($file) > HttpCache::$sessionGcMaxLifeTime) {
+                unlink($file);
+            }
+        }
+    }
 }
 
 /**
@@ -510,21 +553,36 @@ class HttpCache
     /**
      * @var HttpCache
      */
-    public static $instance = null;
-
-    public static $header = array();
-    public static $sessionPath = '';
-    public static $sessionName = '';
+    public static $instance             = null;
+    public static $header               = array();
+    public static $sessionPath          = '';
+    public static $sessionName          = '';
+    public static $sessionGcProbability = 1;
+    public static $sessionGcDivisor     = 1000;
+    public static $sessionGcMaxLifeTime = 1440;
     public $sessionStarted = false;
     public $sessionFile = '';
 
     public static function init()
     {
         self::$sessionName = ini_get('session.name');
-        self::$sessionPath = session_save_path();
-        if (!self::$sessionPath) {
+        self::$sessionPath = @session_save_path();
+        if (!self::$sessionPath || strpos(self::$sessionPath, 'tcp://') === 0) {
             self::$sessionPath = sys_get_temp_dir();
         }
-        @\session_start();
+
+        if ($gc_probability = ini_get('session.gc_probability')) {
+            self::$sessionGcProbability = $gc_probability;
+        }
+
+        if ($gc_divisor = ini_get('session.gc_divisor')) {
+            self::$sessionGcDivisor = $gc_divisor;
+        }
+
+        if ($gc_max_life_time = ini_get('session.gc_maxlifetime')) {
+            self::$sessionGcMaxLifeTime = $gc_max_life_time;
+        }
     }
 }
+
+HttpCache::init();

Protocols/Websocket.php

@@ -14,6 +14,8 @@
 namespace Workerman\Protocols;
 
 use Workerman\Connection\ConnectionInterface;
+use Workerman\Connection\TcpConnection;
+use Workerman\Worker;
 
 /**
  * WebSocket protocol.
@@ -21,13 +23,6 @@ use Workerman\Connection\ConnectionInterface;
 class Websocket implements \Workerman\Protocols\ProtocolInterface
 {
     /**
-     * Minimum head length of websocket protocol.
-     *
-     * @var int
-     */
-    const MIN_HEAD_LEN = 2;
-
-    /**
      * Websocket blob type.
      *
      * @var string
@@ -53,13 +48,13 @@ class Websocket implements \Workerman\Protocols\ProtocolInterface
         // Receive length.
         $recv_len = strlen($buffer);
         // We need more data.
-        if ($recv_len < self::MIN_HEAD_LEN) {
+        if ($recv_len < 2) {
             return 0;
         }
 
         // Has not yet completed the handshake.
         if (empty($connection->websocketHandshake)) {
-            return self::dealHandshake($buffer, $connection);
+            return static::dealHandshake($buffer, $connection);
         }
 
         // Buffer websocket frame data.
@@ -70,9 +65,11 @@ class Websocket implements \Workerman\Protocols\ProtocolInterface
                 return 0;
             }
         } else {
-            $data_len     = ord($buffer[1]) & 127;
             $firstbyte    = ord($buffer[0]);
+            $secondbyte   = ord($buffer[1]);
+            $data_len     = $secondbyte & 127;
             $is_fin_frame = $firstbyte >> 7;
+            $masked       = $secondbyte >> 7;
             $opcode       = $firstbyte & 0xf;
             switch ($opcode) {
                 case 0x0:
@@ -86,11 +83,14 @@ class Websocket implements \Workerman\Protocols\ProtocolInterface
                 // Close package.
                 case 0x8:
                     // Try to emit onWebSocketClose callback.
-                    if (isset($connection->onWebSocketClose)) {
+                    if (isset($connection->onWebSocketClose) || isset($connection->worker->onWebSocketClose)) {
                         try {
-                            call_user_func($connection->onWebSocketClose, $connection);
+                            call_user_func(isset($connection->onWebSocketClose)?$connection->onWebSocketClose:$connection->worker->onWebSocketClose, $connection);
                         } catch (\Exception $e) {
-                            echo $e;
+                            Worker::log($e);
+                            exit(250);
+                        } catch (\Error $e) {
+                            Worker::log($e);
                             exit(250);
                         }
                     } // Close connection.
@@ -101,11 +101,14 @@ class Websocket implements \Workerman\Protocols\ProtocolInterface
                 // Ping package.
                 case 0x9:
                     // Try to emit onWebSocketPing callback.
-                    if (isset($connection->onWebSocketPing)) {
+                    if (isset($connection->onWebSocketPing) || isset($connection->worker->onWebSocketPing)) {
                         try {
-                            call_user_func($connection->onWebSocketPing, $connection);
+                            call_user_func(isset($connection->onWebSocketPing)?$connection->onWebSocketPing:$connection->worker->onWebSocketPing, $connection);
                         } catch (\Exception $e) {
-                            echo $e;
+                            Worker::log($e);
+                            exit(250);
+                        } catch (\Error $e) {
+                            Worker::log($e);
                             exit(250);
                         }
                     } // Send pong package to client.
@@ -115,9 +118,10 @@ class Websocket implements \Workerman\Protocols\ProtocolInterface
 
                     // Consume data from receive buffer.
                     if (!$data_len) {
-                        $connection->consumeRecvBuffer(self::MIN_HEAD_LEN);
-                        if ($recv_len > self::MIN_HEAD_LEN) {
-                            return self::input(substr($buffer, self::MIN_HEAD_LEN), $connection);
+                        $head_len = $masked ? 6 : 2;
+                        $connection->consumeRecvBuffer($head_len);
+                        if ($recv_len > $head_len) {
+                            return static::input(substr($buffer, $head_len), $connection);
                         }
                         return 0;
                     }
@@ -125,26 +129,30 @@ class Websocket implements \Workerman\Protocols\ProtocolInterface
                 // Pong package.
                 case 0xa:
                     // Try to emit onWebSocketPong callback.
-                    if (isset($connection->onWebSocketPong)) {
+                    if (isset($connection->onWebSocketPong) || isset($connection->worker->onWebSocketPong)) {
                         try {
-                            call_user_func($connection->onWebSocketPong, $connection);
+                            call_user_func(isset($connection->onWebSocketPong)?$connection->onWebSocketPong:$connection->worker->onWebSocketPong, $connection);
                         } catch (\Exception $e) {
-                            echo $e;
+                            Worker::log($e);
+                            exit(250);
+                        } catch (\Error $e) {
+                            Worker::log($e);
                             exit(250);
                         }
                     }
                     //  Consume data from receive buffer.
                     if (!$data_len) {
-                        $connection->consumeRecvBuffer(self::MIN_HEAD_LEN);
-                        if ($recv_len > self::MIN_HEAD_LEN) {
-                            return self::input(substr($buffer, self::MIN_HEAD_LEN), $connection);
+                        $head_len = $masked ? 6 : 2;
+                        $connection->consumeRecvBuffer($head_len);
+                        if ($recv_len > $head_len) {
+                            return static::input(substr($buffer, $head_len), $connection);
                         }
                         return 0;
                     }
                     break;
                 // Wrong opcode. 
                 default :
-                    echo "error opcode $opcode and close websocket connection\n";
+                    echo "error opcode $opcode and close websocket connection. Buffer:" . bin2hex($buffer) . "\n";
                     $connection->close();
                     return 0;
             }
@@ -169,6 +177,14 @@ class Websocket implements \Workerman\Protocols\ProtocolInterface
                 }
             }
             $current_frame_length = $head_len + $data_len;
+
+            $total_package_size = strlen($connection->websocketDataBuffer) + $current_frame_length;
+            if ($total_package_size > TcpConnection::$maxPackageSize) {
+                echo "error package. package_length=$total_package_size\n";
+                $connection->close();
+                return 0;
+            }
+
             if ($is_fin_frame) {
                 return $current_frame_length;
             } else {
@@ -178,18 +194,18 @@ class Websocket implements \Workerman\Protocols\ProtocolInterface
 
         // Received just a frame length data.
         if ($connection->websocketCurrentFrameLength === $recv_len) {
-            self::decode($buffer, $connection);
+            static::decode($buffer, $connection);
             $connection->consumeRecvBuffer($connection->websocketCurrentFrameLength);
             $connection->websocketCurrentFrameLength = 0;
             return 0;
         } // The length of the received data is greater than the length of a frame.
         elseif ($connection->websocketCurrentFrameLength < $recv_len) {
-            self::decode(substr($buffer, 0, $connection->websocketCurrentFrameLength), $connection);
+            static::decode(substr($buffer, 0, $connection->websocketCurrentFrameLength), $connection);
             $connection->consumeRecvBuffer($connection->websocketCurrentFrameLength);
             $current_frame_length                    = $connection->websocketCurrentFrameLength;
             $connection->websocketCurrentFrameLength = 0;
             // Continue to read next frame.
-            return self::input(substr($buffer, $current_frame_length), $connection);
+            return static::input(substr($buffer, $current_frame_length), $connection);
         } // The length of the received data is less than the length of a frame.
         else {
             return 0;
@@ -205,9 +221,12 @@ class Websocket implements \Workerman\Protocols\ProtocolInterface
      */
     public static function encode($buffer, ConnectionInterface $connection)
     {
+        if (!is_scalar($buffer)) {
+            throw new \Exception("You can't send(" . gettype($buffer) . ") to client, you need to convert it to a string. ");
+        }
         $len = strlen($buffer);
         if (empty($connection->websocketType)) {
-            $connection->websocketType = self::BINARY_TYPE_BLOB;
+            $connection->websocketType = static::BINARY_TYPE_BLOB;
         }
 
         $first_byte = $connection->websocketType;
@@ -227,7 +246,37 @@ class Websocket implements \Workerman\Protocols\ProtocolInterface
             if (empty($connection->tmpWebsocketData)) {
                 $connection->tmpWebsocketData = '';
             }
+            // If buffer has already full then discard the current package.
+            if (strlen($connection->tmpWebsocketData) > $connection->maxSendBufferSize) {
+                if ($connection->onError) {
+                    try {
+                        call_user_func($connection->onError, $connection, WORKERMAN_SEND_FAIL, 'send buffer full and drop package');
+                    } catch (\Exception $e) {
+                        Worker::log($e);
+                        exit(250);
+                    } catch (\Error $e) {
+                        Worker::log($e);
+                        exit(250);
+                    }
+                }
+                return '';
+            }
             $connection->tmpWebsocketData .= $encode_buffer;
+            // Check buffer is full.
+            if ($connection->maxSendBufferSize <= strlen($connection->tmpWebsocketData)) {
+                if ($connection->onBufferFull) {
+                    try {
+                        call_user_func($connection->onBufferFull, $connection);
+                    } catch (\Exception $e) {
+                        Worker::log($e);
+                        exit(250);
+                    } catch (\Error $e) {
+                        Worker::log($e);
+                        exit(250);
+                    }
+                }
+            }
+
             // Return empty string.
             return '';
         }
@@ -244,7 +293,7 @@ class Websocket implements \Workerman\Protocols\ProtocolInterface
      */
     public static function decode($buffer, ConnectionInterface $connection)
     {
-        $len = $masks = $data = $decoded = null;
+        $masks = $data = $decoded = null;
         $len = ord($buffer[1]) & 127;
         if ($len === 126) {
             $masks = substr($buffer, 4, 4);
@@ -296,7 +345,7 @@ class Websocket implements \Workerman\Protocols\ProtocolInterface
             if (preg_match("/Sec-WebSocket-Key: *(.*?)\r\n/i", $buffer, $match)) {
                 $Sec_WebSocket_Key = $match[1];
             } else {
-                $connection->send("HTTP/1.1 400 Bad Request\r\n\r\n<b>400 Bad Request</b><br>Sec-WebSocket-Key not found.<br>This is a WebSocket service and can not be accessed via HTTP.",
+                $connection->send("HTTP/1.1 400 Bad Request\r\n\r\n<b>400 Bad Request</b><br>Sec-WebSocket-Key not found.<br>This is a WebSocket service and can not be accessed via HTTP.<br>See <a href=\"http://wiki.workerman.net/Error1\">http://wiki.workerman.net/Error1</a> for detail.",
                     true);
                 $connection->close();
                 return 0;
@@ -308,6 +357,7 @@ class Websocket implements \Workerman\Protocols\ProtocolInterface
             $handshake_message .= "Upgrade: websocket\r\n";
             $handshake_message .= "Sec-WebSocket-Version: 13\r\n";
             $handshake_message .= "Connection: Upgrade\r\n";
+            $handshake_message .= "Server: workerman/".Worker::VERSION."\r\n";
             $handshake_message .= "Sec-WebSocket-Accept: " . $new_key . "\r\n\r\n";
             // Mark handshake complete..
             $connection->websocketHandshake = true;
@@ -329,21 +379,27 @@ class Websocket implements \Workerman\Protocols\ProtocolInterface
             }
             // blob or arraybuffer
             if (empty($connection->websocketType)) {
-                $connection->websocketType = self::BINARY_TYPE_BLOB;
+                $connection->websocketType = static::BINARY_TYPE_BLOB;
             }
             // Try to emit onWebSocketConnect callback.
-            if (isset($connection->onWebSocketConnect)) {
-                self::parseHttpHeader($buffer);
+            if (isset($connection->onWebSocketConnect) || isset($connection->worker->onWebSocketConnect)) {
+                static::parseHttpHeader($buffer);
                 try {
-                    call_user_func($connection->onWebSocketConnect, $connection, $buffer);
+                    call_user_func(isset($connection->onWebSocketConnect)?$connection->onWebSocketConnect:$connection->worker->onWebSocketConnect, $connection, $buffer);
                 } catch (\Exception $e) {
-                    echo $e;
+                    Worker::log($e);
+                    exit(250);
+                } catch (\Error $e) {
+                    Worker::log($e);
                     exit(250);
                 }
-                $_GET = $_COOKIE = $_SERVER = array();
+                if (!empty($_SESSION) && class_exists('\GatewayWorker\Lib\Context')) {
+                    $connection->session = \GatewayWorker\Lib\Context::sessionEncode($_SESSION);
+                }
+                $_GET = $_SERVER = $_SESSION = $_COOKIE = array();
             }
             if (strlen($buffer) > $header_length) {
-                return self::input(substr($buffer, $header_length), $connection);
+                return static::input(substr($buffer, $header_length), $connection);
             } 
             return 0;
         } // Is flash policy-file-request.
@@ -354,7 +410,7 @@ class Websocket implements \Workerman\Protocols\ProtocolInterface
             return 0;
         }
         // Bad websocket handshake request.
-        $connection->send("HTTP/1.1 400 Bad Request\r\n\r\n<b>400 Bad Request</b><br>Invalid handshake data for websocket. ",
+        $connection->send("HTTP/1.1 400 Bad Request\r\n\r\n<b>400 Bad Request</b><br>Invalid handshake data for websocket. <br> See <a href=\"http://wiki.workerman.net/Error1\">http://wiki.workerman.net/Error1</a> for detail.",
             true);
         $connection->close();
         return 0;
@@ -368,45 +424,40 @@ class Websocket implements \Workerman\Protocols\ProtocolInterface
      */
     protected static function parseHttpHeader($buffer)
     {
-        $header_data = explode("\r\n", $buffer);
-        $_SERVER     = array();
+        // Parse headers.
+        list($http_header, ) = explode("\r\n\r\n", $buffer, 2);
+        $header_data = explode("\r\n", $http_header);
+
+        if ($_SERVER) {
+            $_SERVER = array();
+        }
+
         list($_SERVER['REQUEST_METHOD'], $_SERVER['REQUEST_URI'], $_SERVER['SERVER_PROTOCOL']) = explode(' ',
             $header_data[0]);
+
         unset($header_data[0]);
         foreach ($header_data as $content) {
             // \r\n\r\n
             if (empty($content)) {
                 continue;
             }
-            list($key, $value) = explode(':', $content, 2);
-            $key   = strtolower($key);
-            $value = trim($value);
+            list($key, $value)       = explode(':', $content, 2);
+            $key                     = str_replace('-', '_', strtoupper($key));
+            $value                   = trim($value);
+            $_SERVER['HTTP_' . $key] = $value;
             switch ($key) {
                 // HTTP_HOST
-                case 'host':
-                    $_SERVER['HTTP_HOST']   = $value;
+                case 'HOST':
                     $tmp                    = explode(':', $value);
                     $_SERVER['SERVER_NAME'] = $tmp[0];
                     if (isset($tmp[1])) {
                         $_SERVER['SERVER_PORT'] = $tmp[1];
                     }
                     break;
-                // HTTP_COOKIE
-                case 'cookie':
-                    $_SERVER['HTTP_COOKIE'] = $value;
+                // cookie
+                case 'COOKIE':
                     parse_str(str_replace('; ', '&', $_SERVER['HTTP_COOKIE']), $_COOKIE);
                     break;
-                // HTTP_USER_AGENT
-                case 'user-agent':
-                    $_SERVER['HTTP_USER_AGENT'] = $value;
-                    break;
-                // HTTP_REFERER
-                case 'referer':
-                    $_SERVER['HTTP_REFERER'] = $value;
-                    break;
-                case 'origin':
-                    $_SERVER['HTTP_ORIGIN'] = $value;
-                    break;
             }
         }
 

Protocols/Ws.php

@@ -1,19 +1,28 @@
 <?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\Protocols;
 
+use Workerman\Worker;
+use Workerman\Lib\Timer;
+use Workerman\Connection\TcpConnection;
+
 /**
  * Websocket protocol for client.
  */
 class Ws
 {
     /**
-     * Minimum head length of websocket protocol.
-     *
-     * @var int
-     */
-    const MIN_HEAD_LEN = 2;
-
-    /**
      * Websocket blob type.
      *
      * @var string
@@ -37,7 +46,7 @@ class Ws
     public static function input($buffer, $connection)
     {
         if (empty($connection->handshakeStep)) {
-            echo "recv data before handshake\n";
+            echo "recv data before handshake. Buffer:" . bin2hex($buffer) . "\n";
             return false;
         }
         // Recv handshake response
@@ -45,7 +54,7 @@ class Ws
             return self::dealHandshake($buffer, $connection);
         }
         $recv_len = strlen($buffer);
-        if ($recv_len < self::MIN_HEAD_LEN) {
+        if ($recv_len < 2) {
             return 0;
         }
         // Buffer websocket frame data.
@@ -56,10 +65,14 @@ class Ws
                 return 0;
             }
         } else {
-            $data_len     = ord($buffer[1]) & 127;
+
             $firstbyte    = ord($buffer[0]);
+            $secondbyte   = ord($buffer[1]);
+            $data_len     = $secondbyte & 127;
             $is_fin_frame = $firstbyte >> 7;
+            $masked       = $secondbyte >> 7;
             $opcode       = $firstbyte & 0xf;
+
             switch ($opcode) {
                 case 0x0:
                     break;
@@ -76,7 +89,10 @@ class Ws
                         try {
                             call_user_func($connection->onWebSocketClose, $connection);
                         } catch (\Exception $e) {
-                            echo $e;
+                            Worker::log($e);
+                            exit(250);
+                        } catch (\Error $e) {
+                            Worker::log($e);
                             exit(250);
                         }
                     } // Close connection.
@@ -91,7 +107,10 @@ class Ws
                         try {
                             call_user_func($connection->onWebSocketPing, $connection);
                         } catch (\Exception $e) {
-                            echo $e;
+                            Worker::log($e);
+                            exit(250);
+                        } catch (\Error $e) {
+                            Worker::log($e);
                             exit(250);
                         }
                     } // Send pong package to client.
@@ -100,9 +119,10 @@ class Ws
                     }
                     // Consume data from receive buffer.
                     if (!$data_len) {
-                        $connection->consumeRecvBuffer(self::MIN_HEAD_LEN);
-                        if ($recv_len > self::MIN_HEAD_LEN) {
-                            return self::input(substr($buffer, self::MIN_HEAD_LEN), $connection);
+                        $head_len = $masked ? 6 : 2;
+                        $connection->consumeRecvBuffer($head_len);
+                        if ($recv_len > $head_len) {
+                            return self::input(substr($buffer, $head_len), $connection);
                         }
                         return 0;
                     }
@@ -114,22 +134,26 @@ class Ws
                         try {
                             call_user_func($connection->onWebSocketPong, $connection);
                         } catch (\Exception $e) {
-                            echo $e;
+                            Worker::log($e);
+                            exit(250);
+                        } catch (\Error $e) {
+                            Worker::log($e);
                             exit(250);
                         }
                     }
                     //  Consume data from receive buffer.
                     if (!$data_len) {
-                        $connection->consumeRecvBuffer(self::MIN_HEAD_LEN);
-                        if ($recv_len > self::MIN_HEAD_LEN) {
-                            return self::input(substr($buffer, self::MIN_HEAD_LEN), $connection);
+                        $head_len = $masked ? 6 : 2;
+                        $connection->consumeRecvBuffer($head_len);
+                        if ($recv_len > $head_len) {
+                            return self::input(substr($buffer, $head_len), $connection);
                         }
                         return 0;
                     }
                     break;
                 // Wrong opcode. 
                 default :
-                    echo "error opcode $opcode and close websocket connection\n";
+                    echo "error opcode $opcode and close websocket connection. Buffer:" . $buffer . "\n";
                     $connection->close();
                     return 0;
             }
@@ -149,6 +173,14 @@ class Ws
             } else {
                 $current_frame_length = $data_len + 2;
             }
+
+            $total_package_size = strlen($connection->websocketDataBuffer) + $current_frame_length;
+            if ($total_package_size > TcpConnection::$maxPackageSize) {
+                echo "error package. package_length=$total_package_size\n";
+                $connection->close();
+                return 0;
+            }
+
             if ($is_fin_frame) {
                 return $current_frame_length;
             } else {
@@ -184,6 +216,9 @@ class Ws
      */
     public static function encode($payload, $connection)
     {
+        if (empty($connection->websocketType)) {
+            $connection->websocketType = self::BINARY_TYPE_BLOB;
+        }
         $payload = (string)$payload;
         if (empty($connection->handshakeStep)) {
             self::sendHandshake($connection);
@@ -194,7 +229,7 @@ class Ws
         $pack = '';
         $length = $length_flag = strlen($payload);
         if (65535 < $length) {
-            $pack   = pack('NN', ($length & 0xFFFFFFFF00000000) >> 0b100000, $length & 0x00000000FFFFFFFF);
+            $pack   = pack('NN', ($length & 0xFFFFFFFF00000000) >> 32, $length & 0x00000000FFFFFFFF);
             $length_flag = 127;
         } else if (125 < $length) {
             $pack   = pack('n*', $length);
@@ -210,7 +245,36 @@ class Ws
             $frame .= $payload[$i] ^ $mask_key[$i % 4];
         }
         if ($connection->handshakeStep === 1) {
-            $connection->tmpWebsocketData = isset($connection->tmpWebsocketData) ? $connection->tmpWebsocketData . $frame : $frame;
+            // If buffer has already full then discard the current package.
+            if (strlen($connection->tmpWebsocketData) > $connection->maxSendBufferSize) {
+                if ($connection->onError) {
+                    try {
+                        call_user_func($connection->onError, $connection, WORKERMAN_SEND_FAIL, 'send buffer full and drop package');
+                    } catch (\Exception $e) {
+                        Worker::log($e);
+                        exit(250);
+                    } catch (\Error $e) {
+                        Worker::log($e);
+                        exit(250);
+                    }
+                }
+                return '';
+            }
+            $connection->tmpWebsocketData = $connection->tmpWebsocketData . $frame;
+            // Check buffer is full.
+            if ($connection->maxSendBufferSize <= strlen($connection->tmpWebsocketData)) {
+                if ($connection->onBufferFull) {
+                    try {
+                        call_user_func($connection->onBufferFull, $connection);
+                    } catch (\Exception $e) {
+                        Worker::log($e);
+                        exit(250);
+                    } catch (\Error $e) {
+                        Worker::log($e);
+                        exit(250);
+                    }
+                }
+            }
             return '';
         }
         return $frame;
@@ -225,7 +289,7 @@ class Ws
      */
     public static function decode($bytes, $connection)
     {
-        $masked = $bytes[1] >> 7;
+        $masked = ord($bytes[1]) >> 7;
         $data_length = $masked ? ord($bytes[1]) & 127 : ord($bytes[1]);
         $decoded_data = '';
         if ($masked === true) {
@@ -264,6 +328,33 @@ class Ws
     }
 
     /**
+     * Send websocket handshake data.
+     *
+     * @return void
+     */
+    public static function onConnect($connection)
+    {
+        self::sendHandshake($connection);
+    }
+
+    /**
+     * Clean
+     *
+     * @param $connection
+     */
+    public static function onClose($connection)
+    {
+        $connection->handshakeStep               = null;
+        $connection->websocketCurrentFrameLength = 0;
+        $connection->tmpWebsocketData            = '';
+        $connection->websocketDataBuffer         = '';
+        if (!empty($connection->websocketPingTimer)) {
+            Timer::del($connection->websocketPingTimer);
+            $connection->websocketPingTimer = null;
+        }
+    }
+
+    /**
      * Send websocket handshake.
      *
      * @param \Workerman\Connection\TcpConnection $connection
@@ -271,24 +362,26 @@ class Ws
      */
     public static function sendHandshake($connection)
     {
+        if (!empty($connection->handshakeStep)) {
+            return;
+        }
         // Get Host.
         $port = $connection->getRemotePort();
         $host = $port === 80 ? $connection->getRemoteHost() : $connection->getRemoteHost() . ':' . $port;
         // Handshake header.
-        $header = "GET / HTTP/1.1\r\n".
+        $header = 'GET ' . $connection->getRemoteURI() . " HTTP/1.1\r\n".
         "Host: $host\r\n".
         "Connection: Upgrade\r\n".
         "Upgrade: websocket\r\n".
         "Origin: ". (isset($connection->websocketOrigin) ? $connection->websocketOrigin : '*') ."\r\n".
+	(isset($connection->WSClientProtocol)?"Sec-WebSocket-Protocol: ".$connection->WSClientProtocol."\r\n":'').
         "Sec-WebSocket-Version: 13\r\n".
-        "Sec-WebSocket-Key: ".base64_encode(sha1(uniqid(mt_rand(), true), true))."\r\n\r\n";
+        "Sec-WebSocket-Key: " . base64_encode(md5(mt_rand(), true)) . "\r\n\r\n";
         $connection->send($header, true);
         $connection->handshakeStep               = 1;
         $connection->websocketCurrentFrameLength = 0;
-        $connection->websocketDataBuffer  = '';
-        if (empty($connection->websocketType)) {
-            $connection->websocketType = self::BINARY_TYPE_BLOB;
-        }
+        $connection->websocketDataBuffer         = '';
+        $connection->tmpWebsocketData            = '';
     }
 
     /**
@@ -303,34 +396,58 @@ class Ws
         $pos = strpos($buffer, "\r\n\r\n");
         if ($pos) {
             // handshake complete
+
+	    // Get WebSocket subprotocol (if specified by server)
+    	    $header = explode("\r\n", substr($buffer, 0, $pos));
+	    foreach ($header as $hrow) {
+		if (preg_match("#^(.+?)\:(.+?)$#", $hrow, $m) && ($m[1] == "Sec-WebSocket-Protocol")) {
+		    $connection->WSServerProtocol = trim($m[2]);
+		}
+
+	    }
+
             $connection->handshakeStep = 2;
-            $handshake_respnse_length = $pos + 4;
+            $handshake_response_length = $pos + 4;
             // Try to emit onWebSocketConnect callback.
             if (isset($connection->onWebSocketConnect)) {
                 try {
-                    call_user_func($connection->onWebSocketConnect, $connection, substr($buffer, 0, $handshake_respnse_length));
+                    call_user_func($connection->onWebSocketConnect, $connection, substr($buffer, 0, $handshake_response_length));
                 } catch (\Exception $e) {
-                    echo $e;
+                    Worker::log($e);
+                    exit(250);
+                } catch (\Error $e) {
+                    Worker::log($e);
                     exit(250);
                 }
             }
             // Headbeat.
             if (!empty($connection->websocketPingInterval)) {
-                $connection->websocketPingTimer = \Workerman\Lib\Timer::add($connection->websocketPingInterval, function() use ($connection){
-                    if (false === $connection->send(pack('H*', '8900'), true)) {
-                        \Workerman\Lib\Timer::del($connection->websocketPingTimer);
+                $connection->websocketPingTimer = Timer::add($connection->websocketPingInterval, function() use ($connection){
+                    if (false === $connection->send(pack('H*', '898000000000'), true)) {
+                        Timer::del($connection->websocketPingTimer);
+                        $connection->websocketPingTimer = null;
                     }
                 });
             }
 
-            $connection->consumeRecvBuffer($handshake_respnse_length);
+            $connection->consumeRecvBuffer($handshake_response_length);
             if (!empty($connection->tmpWebsocketData)) {
                 $connection->send($connection->tmpWebsocketData, true);
+                $connection->tmpWebsocketData = '';
             }
-            if (strlen($buffer > $handshake_respnse_length)) {
-                return self::input(substr($buffer, $handshake_respnse_length));
+            if (strlen($buffer) > $handshake_response_length) {
+                return self::input(substr($buffer, $handshake_response_length), $connection);
             }
         }
         return 0;
     }
+
+    public static function WSSetProtocol($connection, $params) {
+	$connection->WSClientProtocol = $params[0];
+    }
+
+    public static function WSGetServerProtocol($connection) {
+	return (property_exists($connection, 'WSServerProtocol')?$connection->WSServerProtocol:null);
+    }
+
 }

README.md

@@ -1,11 +1,15 @@
 # Workerman
 [![Gitter](https://badges.gitter.im/walkor/Workerman.svg)](https://gitter.im/walkor/Workerman?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=body_badge)
+[![Latest Stable Version](https://poser.pugx.org/workerman/workerman/v/stable)](https://packagist.org/packages/workerman/workerman)
+[![Total Downloads](https://poser.pugx.org/workerman/workerman/downloads)](https://packagist.org/packages/workerman/workerman)
+[![Monthly Downloads](https://poser.pugx.org/workerman/workerman/d/monthly)](https://packagist.org/packages/workerman/workerman)
+[![Daily Downloads](https://poser.pugx.org/workerman/workerman/d/daily)](https://packagist.org/packages/workerman/workerman)
+[![License](https://poser.pugx.org/workerman/workerman/license)](https://packagist.org/packages/workerman/workerman)
 
 ## What is it
-Workerman is a library for event-driven programming in PHP. It has a huge number of features. Each worker is able to handle thousands of connections.
+Workerman is an asynchronous event driven PHP framework with high performance for easily building fast, scalable network applications. Supports HTTP, Websocket, SSL and other custom protocols. Supports libevent, [HHVM](https://github.com/facebook/hhvm) , [ReactPHP](https://github.com/reactphp/react).
 
 ## Requires
-
 PHP 5.3 or Higher  
 A POSIX compatible operating system (Linux, OSX, BSD)  
 POSIX and PCNTL extensions for PHP  
@@ -19,11 +23,10 @@ composer require workerman/workerman
 ## Basic Usage
 
 ### A websocket server 
-test.php
 ```php
 <?php
+require_once __DIR__ . '/vendor/autoload.php';
 use Workerman\Worker;
-require_once './Workerman/Autoloader.php';
 
 // Create a Websocket server
 $ws_worker = new Worker("websocket://0.0.0.0:2346");
@@ -55,9 +58,8 @@ Worker::runAll();
 ```
 
 ### An http server
-test.php
 ```php
-require_once './Workerman/Autoloader.php';
+require_once __DIR__ . '/vendor/autoload.php';
 use Workerman\Worker;
 
 // #### http worker ####
@@ -80,10 +82,10 @@ Worker::runAll();
 ```
 
 ### A WebServer
-test.php
 ```php
-require_once './Workerman/Autoloader.php';
+require_once __DIR__ . '/vendor/autoload.php';
 use Workerman\WebServer;
+use Workerman\Worker;
 
 // WebServer
 $web = new WebServer("http://0.0.0.0:80");
@@ -99,9 +101,8 @@ Worker::runAll();
 ```
 
 ### A tcp server
-test.php
 ```php
-require_once './Workerman/Autoloader.php';
+require_once __DIR__ . '/vendor/autoload.php';
 use Workerman\Worker;
 
 // #### create socket and listen 1234 port ####
@@ -132,6 +133,37 @@ $tcp_worker->onClose = function($connection)
 Worker::runAll();
 ```
 
+### Enable SSL
+```php
+<?php
+require_once __DIR__ . '/vendor/autoload.php';
+use Workerman\Worker;
+
+// SSL context.
+$context = array(
+    'ssl' => array(
+        'local_cert'  => '/your/path/of/server.pem',
+        'local_pk'    => '/your/path/of/server.key',
+        'verify_peer' => false,
+    )
+);
+
+// Create a Websocket server with ssl context.
+$ws_worker = new Worker("websocket://0.0.0.0:2346", $context);
+
+// Enable SSL. WebSocket+SSL means that Secure WebSocket (wss://). 
+// The similar approaches for Https etc.
+$ws_worker->transport = 'ssl';
+
+$ws_worker->onMessage = function($connection, $data)
+{
+    // Send hello $data
+    $connection->send('hello ' . $data);
+};
+
+Worker::runAll();
+```
+
 ### Custom protocol
 Protocols/MyTextProtocol.php
 ```php
@@ -167,9 +199,8 @@ class MyTextProtocol
 }
 ```
 
-test.php
 ```php
-require_once './Workerman/Autoloader.php';
+require_once __DIR__ . '/vendor/autoload.php';
 use Workerman\Worker;
 
 // #### MyTextProtocol worker ####
@@ -196,9 +227,8 @@ Worker::runAll();
 ```
 
 ### Timer
-test.php
 ```php
-require_once './Workerman/Autoloader.php';
+require_once __DIR__ . '/vendor/autoload.php';
 use Workerman\Worker;
 use Workerman\Lib\Timer;
 
@@ -219,25 +249,268 @@ $task->onWorkerStart = function($task)
 Worker::runAll();
 ```
 
-run with:
+### AsyncTcpConnection (tcp/ws/text/frame etc...)
+```php
+require_once __DIR__ . '/vendor/autoload.php';
+use Workerman\Worker;
+use Workerman\Connection\AsyncTcpConnection;
+
+$worker = new Worker();
+$worker->onWorkerStart = function()
+{
+    // Websocket protocol for client.
+    $ws_connection = new AsyncTcpConnection("ws://echo.websocket.org:80");
+    $ws_connection->onConnect = function($connection){
+        $connection->send('hello');
+    };
+    $ws_connection->onMessage = function($connection, $data){
+        echo "recv: $data\n";
+    };
+    $ws_connection->onError = function($connection, $code, $msg){
+        echo "error: $msg\n";
+    };
+    $ws_connection->onClose = function($connection){
+        echo "connection closed\n";
+    };
+    $ws_connection->connect();
+};
+Worker::runAll();
+```
+
+### Async Mysql of ReactPHP
+```
+composer require react/mysql
+```
+
+```php
+<?php
+require_once __DIR__ . '/vendor/autoload.php';
+use Workerman\Worker;
+
+$worker = new Worker('tcp://0.0.0.0:6161');
+$worker->onWorkerStart = function() {
+    global $mysql;
+    $loop  = Worker::getEventLoop();
+    $mysql = new React\MySQL\Connection($loop, array(
+        'host'   => '127.0.0.1',
+        'dbname' => 'dbname',
+        'user'   => 'user',
+        'passwd' => 'passwd',
+    ));
+    $mysql->on('error', function($e){
+        echo $e;
+    });
+    $mysql->connect(function ($e) {
+        if($e) {
+            echo $e;
+        } else {
+            echo "connect success\n";
+        }
+    });
+};
+$worker->onMessage = function($connection, $data) {
+    global $mysql;
+    $mysql->query('show databases' /*trim($data)*/, function ($command, $mysql) use ($connection) {
+        if ($command->hasError()) {
+            $error = $command->getError();
+        } else {
+            $results = $command->resultRows;
+            $fields  = $command->resultFields;
+            $connection->send(json_encode($results));
+        }
+    });
+};
+Worker::runAll();
+```
+
+### Async Redis of ReactPHP
+```
+composer require clue/redis-react
+```
+
+```php
+<?php
+require_once __DIR__ . '/vendor/autoload.php';
+use Clue\React\Redis\Factory;
+use Clue\React\Redis\Client;
+use Workerman\Worker;
+
+$worker = new Worker('tcp://0.0.0.0:6161');
+
+$worker->onWorkerStart = function() {
+    global $factory;
+    $loop    = Worker::getEventLoop();
+    $factory = new Factory($loop);
+};
+
+$worker->onMessage = function($connection, $data) {
+    global $factory;
+    $factory->createClient('localhost:6379')->then(function (Client $client) use ($connection) {
+        $client->set('greeting', 'Hello world');
+        $client->append('greeting', '!');
+
+        $client->get('greeting')->then(function ($greeting) use ($connection){
+            // Hello world!
+            echo $greeting . PHP_EOL;
+            $connection->send($greeting);
+        });
+
+        $client->incr('invocation')->then(function ($n) use ($connection){
+            echo 'This is invocation #' . $n . PHP_EOL;
+            $connection->send($n);
+        });
+    });
+};
+
+Worker::runAll();
+```
+
+### Aysnc dns of ReactPHP
+```
+composer require react/dns
+```
+
+```php
+require_once __DIR__ . '/vendor/autoload.php';
+use Workerman\Worker;
+$worker = new Worker('tcp://0.0.0.0:6161');
+$worker->onWorkerStart = function() {
+    global   $dns;
+    // Get event-loop.
+    $loop    = Worker::getEventLoop();
+    $factory = new React\Dns\Resolver\Factory();
+    $dns     = $factory->create('8.8.8.8', $loop);
+};
+$worker->onMessage = function($connection, $host) {
+    global $dns;
+    $host = trim($host);
+    $dns->resolve($host)->then(function($ip) use($host, $connection) {
+        $connection->send("$host: $ip");
+    },function($e) use($host, $connection){
+        $connection->send("$host: {$e->getMessage()}");
+    });
+};
+
+Worker::runAll();
+```
+
+### Http client of ReactPHP
+```
+composer require react/http-client
+```
+
+```php
+<?php
+require_once __DIR__ . '/vendor/autoload.php';
+use Workerman\Worker;
+
+$worker = new Worker('tcp://0.0.0.0:6161');
+
+$worker->onWorkerStart = function() {
+    global   $client;
+    $loop    = Worker::getEventLoop();
+    $factory = new React\Dns\Resolver\Factory();
+    $dns     = $factory->createCached('8.8.8.8', $loop);
+    $factory = new React\HttpClient\Factory();
+    $client = $factory->create($loop, $dns);
+};
+
+$worker->onMessage = function($connection, $host) {
+    global     $client;
+    $request = $client->request('GET', trim($host));
+    $request->on('error', function(Exception $e) use ($connection) {
+        $connection->send($e);
+    });
+    $request->on('response', function ($response) use ($connection) {
+        $response->on('data', function ($data, $response) use ($connection) {
+            $connection->send($data);
+        });
+    });
+    $request->end();
+};
+
+Worker::runAll();
+```
+
+### ZMQ of ReactPHP
+```
+composer require react/zmq
+```
+
+```php
+<?php
+require_once __DIR__ . '/vendor/autoload.php';
+use Workerman\Worker;
+
+$worker = new Worker('text://0.0.0.0:6161');
+
+$worker->onWorkerStart = function() {
+    global   $pull;
+    $loop    = Worker::getEventLoop();
+    $context = new React\ZMQ\Context($loop);
+    $pull    = $context->getSocket(ZMQ::SOCKET_PULL);
+    $pull->bind('tcp://127.0.0.1:5555');
+
+    $pull->on('error', function ($e) {
+        var_dump($e->getMessage());
+    });
+
+    $pull->on('message', function ($msg) {
+        echo "Received: $msg\n";
+    });
+};
+
+Worker::runAll();
+```
+
+### STOMP of ReactPHP
+```
+composer require react/stomp
+```
+
+```php
+<?php
+require_once __DIR__ . '/vendor/autoload.php';
+use Workerman\Worker;
+
+$worker = new Worker('text://0.0.0.0:6161');
+
+$worker->onWorkerStart = function() {
+    global   $client;
+    $loop    = Worker::getEventLoop();
+    $factory = new React\Stomp\Factory($loop);
+    $client  = $factory->createClient(array('vhost' => '/', 'login' => 'guest', 'passcode' => 'guest'));
+
+    $client
+        ->connect()
+        ->then(function ($client) use ($loop) {
+            $client->subscribe('/topic/foo', function ($frame) {
+                echo "Message received: {$frame->body}\n";
+            });
+        });
+};
+
+Worker::runAll();
+```
+
 
-```php test.php start```
 
 ## Available commands
-```php test.php start  ```  
-```php test.php start -d  ```  
+```php start.php start  ```  
+```php start.php start -d  ```  
 ![workerman start](http://www.workerman.net/img/workerman-start.png)  
-```php test.php status  ```  
-![workerman satus](http://www.workerman.net/img/workerman-status.png?a=123)
-```php test.php stop  ```  
-```php test.php restart  ```  
-```php test.php reload  ```  
+```php start.php status  ```  
+![workerman satus](http://www.workerman.net/img/workerman-status.png?a=123)  
+```php start.php connections```  
+```php start.php stop  ```  
+```php start.php restart  ```  
+```php start.php reload  ```  
 
 ## Documentation
 
 中文主页:[http://www.workerman.net](http://www.workerman.net)
 
-中文文档: [http://doc3.workerman.net](http://doc3.workerman.net)
+中文文档: [http://doc.workerman.net](http://doc.workerman.net)
 
 Documentation:[https://github.com/walkor/workerman-manual](https://github.com/walkor/workerman-manual/blob/master/english/src/SUMMARY.md)
 
@@ -324,75 +597,14 @@ Percentage of the requests served within a certain time (ms)
 ```
 
 
-# Demos
-
-## [tadpole](http://kedou.workerman.net/)  
-[Live demo](http://kedou.workerman.net/)  
-[Source code](https://github.com/walkor/workerman)  
-![workerman todpole](http://www.workerman.net/img/workerman-todpole.png)  
-
-## [BrowserQuest](http://www.workerman.net/demos/browserquest/)   
-[Live demo](http://www.workerman.net/demos/browserquest/)  
-[Source code](https://github.com/walkor/BrowserQuest-PHP)  
-![BrowserQuest width workerman](http://www.workerman.net/img/browserquest.jpg) 
-
-## [web vmstat](http://www.workerman.net/demos/vmstat/)   
-[Live demo](http://www.workerman.net/demos/vmstat/)  
-[Source code](https://github.com/walkor/workerman-vmstat)  
-![web vmstat](http://www.workerman.net/img/workerman-vmstat.png)   
-
-## [live-ascii-camera](https://github.com/walkor/live-ascii-camera)   
-[Live demo camera page](http://www.workerman.net/demos/live-ascii-camera/camera.html)  
-[Live demo receive page](http://www.workerman.net/demos/live-ascii-camera/)  
-[Source code](https://github.com/walkor/live-ascii-camera)  
-![live-ascii-camera](http://www.workerman.net/img/live-ascii-camera.png)   
-
-## [live-camera](https://github.com/walkor/live-camera)   
-[Live demo camera page](http://www.workerman.net/demos/live-camera/camera.html)  
-[Live demo receive page](http://www.workerman.net/demos/live-camera/)  
-[Source code](https://github.com/walkor/live-camera)  
-![live-camera](http://www.workerman.net/img/live-camera.jpg)  
-
-## [chat room](http://chat.workerman.net/)  
-[Live demo](http://chat.workerman.net/)  
-[Source code](https://github.com/walkor/workerman-chat)  
-![workerman-chat](http://www.workerman.net/img/workerman-chat.png)  
-
-## [PHPSocket.IO](https://github.com/walkor/phpsocket.io)  
-[Live demo](http://www.workerman.net/demos/phpsocketio-chat/)  
-[Source code](https://github.com/walkor/phpsocket.io)  
-![phpsocket.io](http://www.workerman.net/img/socket.io.png)  
-
-## [statistics](http://www.workerman.net:55757/)  
-[Live demo](http://www.workerman.net:55757/)  
-[Source code](https://github.com/walkor/workerman-statistics)  
-![workerman-statistics](http://www.workerman.net/img/workerman-statistics.png)  
-
-## [flappybird](http://workerman.net/demos/flappy-bird/)  
-[Live demo](http://workerman.net/demos/flappy-bird/)  
-[Source code](https://github.com/walkor/workerman-flappy-bird)  
-![workerman-statistics](http://www.workerman.net/img/workerman-flappy-bird.png)  
-
-## [jsonRpc](https://github.com/walkor/workerman-JsonRpc)  
-[Source code](https://github.com/walkor/workerman-JsonRpc)  
-![workerman-jsonRpc](http://www.workerman.net/img/workerman-json-rpc.png)  
-
-## [thriftRpc](https://github.com/walkor/workerman-thrift)  
-[Source code](https://github.com/walkor/workerman-thrift)  
-![workerman-thriftRpc](http://www.workerman.net/img/workerman-thrift.png)  
-
-## [web-msg-sender](https://github.com/walkor/web-msg-sender)  
-[Live demo send page](http://workerman.net:3333/)  
-[Live demo receive page](http://workerman.net/web-msg-sender.html)  
-[Source code](https://github.com/walkor/web-msg-sender)  
-![web-msg-sender](http://www.workerman.net/img/web-msg-sender.png)  
+## Other links with workerman
 
-## [shadowsocks-php](https://github.com/walkor/shadowsocks-php)
-[Source code](https://github.com/walkor/shadowsocks-php)  
-![shadowsocks-php](http://www.workerman.net/img/shadowsocks-php.png)  
+[PHPSocket.IO](https://github.com/walkor/phpsocket.io)   
+[php-socks5](https://github.com/walkor/php-socks5)  
+[php-http-proxy](https://github.com/walkor/php-http-proxy)  
 
-## [queue](https://github.com/walkor/workerman-queue)
-[Source code](https://github.com/walkor/workerman-queue)  
+## Donate
+<a href="https://www.paypal.com/cgi-bin/webscr?cmd=_s-xclick&hosted_button_id=UQGGS9UB35WWG"><img src="http://donate.workerman.net/img/donate.png"></a>
 
 ## LICENSE
 

WebServer.php

@@ -22,13 +22,6 @@ use Workerman\Protocols\HttpCache;
 class WebServer extends Worker
 {
     /**
-     * Mime.
-     *
-     * @var string
-     */
-    protected static $defaultMimeType = 'text/html; charset=utf-8';
-
-    /**
      * Virtual host to path mapping.
      *
      * @var array ['workerman.net'=>'/home', 'www.workerman.net'=>'home/www']
@@ -96,10 +89,10 @@ class WebServer extends Worker
     public function onWorkerStart()
     {
         if (empty($this->serverRoot)) {
-            throw new \Exception('server root not set, please use WebServer::addRoot($domain, $root_path) to set server root path');
+            echo new \Exception('server root not set, please use WebServer::addRoot($domain, $root_path) to set server root path');
+            exit(250);
         }
-        // Init HttpCache.
-        HttpCache::init();
+
         // Init mimeMap.
         $this->initMimeTypeMap();
 
@@ -108,7 +101,10 @@ class WebServer extends Worker
             try {
                 call_user_func($this->_onWorkerStart, $this);
             } catch (\Exception $e) {
-                echo $e;
+                self::log($e);
+                exit(250);
+            } catch (\Error $e) {
+                self::log($e);
                 exit(250);
             }
         }
@@ -159,7 +155,7 @@ class WebServer extends Worker
             return;
         }
 
-        $workerman_path = $workerman_url_info['path'];
+        $workerman_path = isset($workerman_url_info['path']) ? $workerman_url_info['path'] : '/';
 
         $workerman_path_info      = pathinfo($workerman_path);
         $workerman_file_extension = isset($workerman_path_info['extension']) ? $workerman_path_info['extension'] : '';
@@ -213,7 +209,11 @@ class WebServer extends Worker
                 }
                 $content = ob_get_clean();
                 ini_set('display_errors', 'on');
-                $connection->close($content);
+                if (strtolower($_SERVER['HTTP_CONNECTION']) === "keep-alive") {
+                    $connection->send($content);
+                } else {
+                    $connection->close($content);
+                }
                 chdir($workerman_cwd);
                 return;
             }
@@ -228,11 +228,11 @@ class WebServer extends Worker
         }
     }
 
-    public static function sendFile($connection, $file_name)
+    public static function sendFile($connection, $file_path)
     {
         // Check 304.
-        $info = stat($file_name);
-        $modified_time = $info ? date('D, d M Y H:i:s', $info['mtime']) . ' GMT' : '';
+        $info = stat($file_path);
+        $modified_time = $info ? date('D, d M Y H:i:s', $info['mtime']) . ' ' . date_default_timezone_get() : '';
         if (!empty($_SERVER['HTTP_IF_MODIFIED_SINCE']) && $info) {
             // Http 304.
             if ($modified_time === $_SERVER['HTTP_IF_MODIFIED_SINCE']) {
@@ -248,22 +248,28 @@ class WebServer extends Worker
         if ($modified_time) {
             $modified_time = "Last-Modified: $modified_time\r\n";
         }
-        $file_size = filesize($file_name);
-        $extension = pathinfo($file_name, PATHINFO_EXTENSION);
-        $content_type = isset(self::$mimeTypeMap[$extension]) ? self::$mimeTypeMap[$extension] : self::$defaultMimeType;
+        $file_size = filesize($file_path);
+        $file_info = pathinfo($file_path);
+        $extension = isset($file_info['extension']) ? $file_info['extension'] : '';
+        $file_name = isset($file_info['filename']) ? $file_info['filename'] : '';
         $header = "HTTP/1.1 200 OK\r\n";
-        $header .= "Content-Type: $content_type\r\n";
+        if (isset(self::$mimeTypeMap[$extension])) {
+            $header .= "Content-Type: " . self::$mimeTypeMap[$extension] . "\r\n";
+        } else {
+            $header .= "Content-Type: application/octet-stream\r\n";
+            $header .= "Content-Disposition: attachment; filename=\"$file_name\"\r\n";
+        }
         $header .= "Connection: keep-alive\r\n";
         $header .= $modified_time;
         $header .= "Content-Length: $file_size\r\n\r\n";
         $trunk_limit_size = 1024*1024;
         if ($file_size < $trunk_limit_size) {
-            return $connection->send($header.file_get_contents($file_name), true);
+            return $connection->send($header.file_get_contents($file_path), true);
         }
         $connection->send($header, true);
 
         // Read file content from disk piece by piece and send to client.
-        $connection->fileHandler = fopen($file_name, 'r');
+        $connection->fileHandler = fopen($file_path, 'r');
         $do_write = function()use($connection)
         {
             // Send buffer not full.

composer.json

@@ -1,33 +1,38 @@
 {
-    "name"  : "workerman/workerman",
-    "type"  : "project",
-    "keywords": ["event-loop", "asynchronous"],
+    "name": "workerman/workerman",
+    "type": "library",
+    "keywords": [
+        "event-loop",
+        "asynchronous"
+    ],
     "homepage": "http://www.workerman.net",
-    "license" : "MIT",
+    "license": "MIT",
     "description": "An asynchronous event driven PHP framework for easily building fast, scalable network applications.",
-    "authors" : [
+    "authors": [
         {
-            "name"      : "walkor",
-            "email"     : "walkor@workerman.net",
-            "homepage"  : "http://www.workerman.net",	
+            "name": "walkor",
+            "email": "walkor@workerman.net",
+            "homepage": "http://www.workerman.net",
             "role": "Developer"
         }
     ],
-    "support" : {
-        "email" : "walkor@workerman.net",
+    "support": {
+        "email": "walkor@workerman.net",
         "issues": "https://github.com/walkor/workerman/issues",
-        "forum" : "http://wenda.workerman.net/",
-        "wiki"  : "http://doc3.workerman.net/index.html",
+        "forum": "http://wenda.workerman.net/",
+        "wiki": "http://doc3.workerman.net/index.html",
         "source": "https://github.com/walkor/workerman"
     },
     "require": {
         "php": ">=5.3"
     },
     "suggest": {
-	"ext-libevent": "For better performance."
+        "ext-event": "For better performance. "
     },
     "autoload": {
-        "psr-4": {"Workerman\\": "./"}
+        "psr-4": {
+            "Workerman\\": "./"
+        }
     },
-    "minimum-stability":"dev"
+    "minimum-stability": "dev"
 }