Merge https://github.com/walkor/Workerman into shl_feature_useing

.gitignore

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

Connection/AsyncTcpConnection.php

@@ -52,6 +52,13 @@ class AsyncTcpConnection extends TcpConnection
     protected $_remoteHost = '';
 
     /**
+     * Remote port.
+     *
+     * @var int
+     */
+    protected $_remotePort = 80;
+
+    /**
      * Connect start time.
      *
      * @var string
@@ -124,11 +131,15 @@ class AsyncTcpConnection extends TcpConnection
             }
             $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             = self::$_idRecorder++;
+        $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);
@@ -145,8 +156,9 @@ class AsyncTcpConnection extends TcpConnection
 
         // For statistics.
         self::$statistics['connection_count']++;
-        $this->maxSendBufferSize = self::$defaultMaxSendBufferSize;
-        $this->_contextOption    = $context_option;
+        $this->maxSendBufferSize        = self::$defaultMaxSendBufferSize;
+        $this->_contextOption           = $context_option;
+        static::$connections[$this->id] = $this;
     }
 
     /**
@@ -162,11 +174,16 @@ class AsyncTcpConnection extends TcpConnection
         }
         $this->_status           = self::STATUS_CONNECTING;
         $this->_connectStartTime = microtime(true);
-        // Open socket connection asynchronously.
-        if ($this->_contextOption) {
-            $context = stream_context_create($this->_contextOption);
-            $this->_socket = stream_socket_client("{$this->transport}://{$this->_remoteAddress}", $errno, $errstr, 0,
-                STREAM_CLIENT_ASYNC_CONNECT, $context);
+        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);
@@ -285,7 +302,7 @@ class AsyncTcpConnection extends TcpConnection
             if ($this->_sendBuffer) {
                 Worker::$globalEvent->add($socket, EventInterface::EV_WRITE, array($this, 'baseWrite'));
             }
-            $this->_status                = self::STATUS_ESTABLISH;
+            $this->_status                = self::STATUS_ESTABLISHED;
             $this->_remoteAddress         = $address;
             $this->_sslHandshakeCompleted = true;
 

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

@@ -48,7 +48,7 @@ class TcpConnection extends ConnectionInterface
      *
      * @var int
      */
-    const STATUS_ESTABLISH = 2;
+    const STATUS_ESTABLISHED = 2;
 
     /**
      * Status closing.
@@ -122,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
@@ -197,7 +211,7 @@ class TcpConnection extends ConnectionInterface
      *
      * @var int
      */
-    protected $_status = self::STATUS_ESTABLISH;
+    protected $_status = self::STATUS_ESTABLISHED;
 
     /**
      * Remote address.
@@ -214,13 +228,58 @@ class TcpConnection extends ConnectionInterface
     protected $_isPaused = false;
 
     /**
-     * SSL handshake completed or not
+     * 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
@@ -229,7 +288,10 @@ 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
@@ -239,6 +301,22 @@ class TcpConnection extends ConnectionInterface
         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];
     }
 
     /**
@@ -255,7 +333,7 @@ class TcpConnection extends ConnectionInterface
         }
 
         // 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 === '') {
@@ -263,7 +341,7 @@ class TcpConnection extends ConnectionInterface
             }
         }
 
-        if ($this->_status !== self::STATUS_ESTABLISH ||
+        if ($this->_status !== self::STATUS_ESTABLISHED ||
             ($this->transport === 'ssl' && $this->_sslHandshakeCompleted !== true)
         ) {
             if ($this->_sendBuffer) {
@@ -283,11 +361,13 @@ class TcpConnection extends ConnectionInterface
             $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)) {
@@ -333,7 +413,7 @@ class TcpConnection extends ConnectionInterface
     {
         $pos = strrpos($this->_remoteAddress, ':');
         if ($pos) {
-            return trim(substr($this->_remoteAddress, 0, $pos), '[]');
+            return substr($this->_remoteAddress, 0, $pos);
         }
         return '';
     }
@@ -352,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
@@ -387,8 +563,8 @@ class TcpConnection extends ConnectionInterface
     {
         // SSL handshake.
         if ($this->transport === 'ssl' && $this->_sslHandshakeCompleted !== true) {
-            $ret = stream_socket_enable_crypto($socket, true, STREAM_CRYPTO_METHOD_SSLv2_SERVER |
-                STREAM_CRYPTO_METHOD_SSLv3_SERVER | STREAM_CRYPTO_METHOD_SSLv23_SERVER);
+            $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)) {
@@ -417,7 +593,7 @@ class TcpConnection extends ConnectionInterface
             return;
         }
 
-        $buffer = fread($socket, self::READ_BUFFER_SIZE);
+        $buffer = @fread($socket, self::READ_BUFFER_SIZE);
 
         // Check connection closed.
         if ($buffer === '' || $buffer === false) {
@@ -426,11 +602,12 @@ class TcpConnection extends ConnectionInterface
                 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.
@@ -521,6 +698,7 @@ class TcpConnection extends ConnectionInterface
     {
         $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. 
@@ -541,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']++;
@@ -691,6 +870,7 @@ 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) {
@@ -729,6 +909,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

@@ -99,6 +99,82 @@ class UdpConnection 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);
+    }
+
+    /**
+     * 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;
+    }
+
+    /**
      * Close connection.
      *
      * @param mixed $data

Events/Ev.php

@@ -181,4 +181,14 @@ class Ev implements EventInterface
             $event->stop();
         }
     }
+
+    /**
+     * Get timer count.
+     *
+     * @return integer
+     */
+    public function getTimerCount()
+    {
+        return count($this->_eventTimer);
+    }
 }

Events/Event.php

@@ -196,4 +196,14 @@ class Event implements EventInterface
             $event->del();
         }
     }
+
+    /**
+     * Get timer count.
+     *
+     * @return integer
+     */
+    public function getTimerCount()
+    {
+        return count($this->_eventTimer);
+    }
 }

Events/EventInterface.php

@@ -97,4 +97,11 @@ interface EventInterface
      * @return mixed
      */
     public function destroy();
+
+    /**
+     * Get Timer count.
+     *
+     * @return mixed
+     */
+    public function getTimerCount();
 }

Events/Libevent.php

@@ -213,5 +213,15 @@ class Libevent implements EventInterface
             event_del($event);
         }
     }
+
+    /**
+     * Get timer count.
+     *
+     * @return integer
+     */
+    public function getTimerCount()
+    {
+        return count($this->_eventTimer);
+    }
 }
 

Events/React/ExtEventLoop.php

@@ -64,17 +64,20 @@ class ExtEventLoop extends \React\EventLoop\ExtEventLoop
             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[++$this->_timerIdIndex] = $timer_obj;
-                return $this->_timerIdIndex;
+                $this->_timerIdMap[$timer_id] = $timer_obj;
+                return $timer_id;
             case EventInterface::EV_TIMER_ONCE:
-                $timer_obj = $this->addTimer($fd, function() use ($func, $args) {
+                $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[++$this->_timerIdIndex] = $timer_obj;
-                return $this->_timerIdIndex;
+                $this->_timerIdMap[$timer_id] = $timer_obj;
+                return $timer_id;
         }
         return false;
     }
@@ -170,4 +173,14 @@ class ExtEventLoop extends \React\EventLoop\ExtEventLoop
             $event->del();
         }
     }
+
+    /**
+     * Get timer count.
+     *
+     * @return integer
+     */
+    public function getTimerCount()
+    {
+        return count($this->_timerIdMap);
+    }
 }

Events/React/LibEventLoop.php

@@ -64,17 +64,20 @@ class LibEventLoop extends \React\EventLoop\LibEventLoop
             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[++$this->_timerIdIndex] = $timer_obj;
-                return $this->_timerIdIndex;
+                $this->_timerIdMap[$timer_id] = $timer_obj;
+                return $timer_id;
             case EventInterface::EV_TIMER_ONCE:
-                $timer_obj = $this->addTimer($fd, function() use ($func, $args) {
+                $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[++$this->_timerIdIndex] = $timer_obj;
-                return $this->_timerIdIndex;
+                $this->_timerIdMap[$timer_id] = $timer_obj;
+                return $timer_id;
         }
         return false;
     }
@@ -171,4 +174,14 @@ class LibEventLoop extends \React\EventLoop\LibEventLoop
             event_del($event);
         }
     }
+
+    /**
+     * Get timer count.
+     *
+     * @return integer
+     */
+    public function getTimerCount()
+    {
+        return count($this->_timerIdMap);
+    }
 }

Events/React/StreamSelectLoop.php

@@ -173,4 +173,14 @@ class StreamSelectLoop extends \React\EventLoop\StreamSelectLoop
     {
 
     }
+
+    /**
+     * Get timer count.
+     *
+     * @return integer
+     */
+    public function getTimerCount()
+    {
+        return count($this->_timerIdMap);
+    }
 }

Events/Select.php

@@ -265,7 +265,7 @@ class Select implements EventInterface
 
             $read  = $this->_readFds;
             $write = $this->_writeFds;
-            $except = $this->_writeFds;
+            $except = $this->_exceptFds;
 
             // Waiting read/write/signal/timeout events.
             $ret = @stream_select($read, $write, $except, 0, $this->_selectTimeout);
@@ -319,4 +319,14 @@ class Select implements EventInterface
     {
 
     }
+
+    /**
+     * Get timer count.
+     *
+     * @return integer
+     */
+    public function getTimerCount()
+    {
+        return count($this->_eventTimer);
+    }
 }

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,8 +76,8 @@ 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 int/false

Protocols/Http.php

@@ -168,6 +168,9 @@ class Http
                     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;

Protocols/Websocket.php

@@ -48,7 +48,7 @@ class Websocket implements \Workerman\Protocols\ProtocolInterface
         // Receive length.
         $recv_len = strlen($buffer);
         // We need more data.
-        if ($recv_len < 2) {
+        if ($recv_len < 6) {
             return 0;
         }
 
@@ -70,6 +70,13 @@ class Websocket implements \Workerman\Protocols\ProtocolInterface
             $data_len     = $secondbyte & 127;
             $is_fin_frame = $firstbyte >> 7;
             $masked       = $secondbyte >> 7;
+
+            if (!$masked) {
+                echo "frame not masked\n";
+                $connection->close();
+                return 0;
+            }
+
             $opcode       = $firstbyte & 0xf;
             switch ($opcode) {
                 case 0x0:
@@ -83,9 +90,9 @@ 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) {
                             Worker::log($e);
                             exit(250);
@@ -101,9 +108,9 @@ 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) {
                             Worker::log($e);
                             exit(250);
@@ -118,7 +125,7 @@ class Websocket implements \Workerman\Protocols\ProtocolInterface
 
                     // Consume data from receive buffer.
                     if (!$data_len) {
-                        $head_len = $masked ? 6 : 2;
+                        $head_len = 6;
                         $connection->consumeRecvBuffer($head_len);
                         if ($recv_len > $head_len) {
                             return static::input(substr($buffer, $head_len), $connection);
@@ -129,9 +136,9 @@ 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) {
                             Worker::log($e);
                             exit(250);
@@ -142,7 +149,7 @@ class Websocket implements \Workerman\Protocols\ProtocolInterface
                     }
                     //  Consume data from receive buffer.
                     if (!$data_len) {
-                        $head_len = $masked ? 6 : 2;
+                        $head_len = 6;
                         $connection->consumeRecvBuffer($head_len);
                         if ($recv_len > $head_len) {
                             return static::input(substr($buffer, $head_len), $connection);
@@ -382,10 +389,10 @@ class Websocket implements \Workerman\Protocols\ProtocolInterface
                 $connection->websocketType = static::BINARY_TYPE_BLOB;
             }
             // Try to emit onWebSocketConnect callback.
-            if (isset($connection->onWebSocketConnect)) {
+            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) {
                     Worker::log($e);
                     exit(250);

Protocols/Ws.php

@@ -71,6 +71,13 @@ class Ws
             $data_len     = $secondbyte & 127;
             $is_fin_frame = $firstbyte >> 7;
             $masked       = $secondbyte >> 7;
+
+            if ($masked) {
+                echo "frame masked\n";
+                $connection->close();
+                return 0;
+            }
+
             $opcode       = $firstbyte & 0xf;
 
             switch ($opcode) {
@@ -119,7 +126,7 @@ class Ws
                     }
                     // Consume data from receive buffer.
                     if (!$data_len) {
-                        $head_len = $masked ? 6 : 2;
+                        $head_len = 2;
                         $connection->consumeRecvBuffer($head_len);
                         if ($recv_len > $head_len) {
                             return self::input(substr($buffer, $head_len), $connection);
@@ -143,7 +150,7 @@ class Ws
                     }
                     //  Consume data from receive buffer.
                     if (!$data_len) {
-                        $head_len = $masked ? 6 : 2;
+                        $head_len = 2;
                         $connection->consumeRecvBuffer($head_len);
                         if ($recv_len > $head_len) {
                             return self::input(substr($buffer, $head_len), $connection);
@@ -151,7 +158,7 @@ class Ws
                         return 0;
                     }
                     break;
-                // Wrong opcode. 
+                // Wrong opcode.
                 default :
                     echo "error opcode $opcode and close websocket connection. Buffer:" . $buffer . "\n";
                     $connection->close();
@@ -159,7 +166,7 @@ class Ws
             }
             // Calculate packet length.
             if ($data_len === 126) {
-                if (strlen($buffer) < 6) {
+                if (strlen($buffer) < 4) {
                     return 0;
                 }
                 $pack = unpack('nn/ntotal_len', $buffer);
@@ -289,31 +296,14 @@ class Ws
      */
     public static function decode($bytes, $connection)
     {
-        $masked = ord($bytes[1]) >> 7;
-        $data_length = $masked ? ord($bytes[1]) & 127 : ord($bytes[1]);
-        $decoded_data = '';
-        if ($masked === true) {
-            if ($data_length === 126) {
-                $mask = substr($bytes, 4, 4);
-                $coded_data = substr($bytes, 8);
-            } else if ($data_length === 127) {
-                $mask = substr($bytes, 10, 4);
-                $coded_data = substr($bytes, 14);
-            } else {
-                $mask = substr($bytes, 2, 4);
-                $coded_data = substr($bytes, 6);
-            }
-            for ($i = 0; $i < strlen($coded_data); $i++) {
-                $decoded_data .= $coded_data[$i] ^ $mask[$i % 4];
-            }
+        $data_length = ord($bytes[1]);
+
+        if ($data_length === 126) {
+            $decoded_data = substr($bytes, 4);
+        } else if ($data_length === 127) {
+            $decoded_data = substr($bytes, 10);
         } else {
-            if ($data_length === 126) {
-                $decoded_data = substr($bytes, 4);
-            } else if ($data_length === 127) {
-                $decoded_data = substr($bytes, 10);
-            } else {
-                $decoded_data = substr($bytes, 2);
-            }
+            $decoded_data = substr($bytes, 2);
         }
         if ($connection->websocketCurrentFrameLength) {
             $connection->websocketDataBuffer .= $decoded_data;
@@ -358,7 +348,7 @@ class Ws
      * Send websocket handshake.
      *
      * @param \Workerman\Connection\TcpConnection $connection
-     * @return void 
+     * @return void
      */
     public static function sendHandshake($connection)
     {
@@ -369,13 +359,26 @@ class Ws
         $port = $connection->getRemotePort();
         $host = $port === 80 ? $connection->getRemoteHost() : $connection->getRemoteHost() . ':' . $port;
         // Handshake header.
+        $connection->websocketSecKey = base64_encode(md5(mt_rand(), true));
+        $userHeader = '';
+        if (!empty($connection->wsHttpHeader)) {
+            if (is_array($connection->wsHttpHeader)){
+                foreach($connection->wsHttpHeader as $k=>$v){
+                    $userHeader .= "$k: $v\r\n";
+                }
+            }else{
+                $userHeader .= $connection->wsHttpHeader;
+            }
+            $userHeader = "\r\n".trim($userHeader);
+        }
         $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(md5(mt_rand(), true)) . "\r\n\r\n";
+        "Sec-WebSocket-Key: " . $connection->websocketSecKey . $userHeader . "\r\n\r\n";
         $connection->send($header, true);
         $connection->handshakeStep               = 1;
         $connection->websocketCurrentFrameLength = 0;
@@ -394,7 +397,26 @@ class Ws
     {
         $pos = strpos($buffer, "\r\n\r\n");
         if ($pos) {
+            //checking Sec-WebSocket-Accept
+            if (preg_match("/Sec-WebSocket-Accept: *(.*?)\r\n/i", $buffer, $match)) {
+                if ($match[1] !== base64_encode(sha1($connection->websocketSecKey . "258EAFA5-E914-47DA-95CA-C5AB0DC85B11", true))) {
+                    echo "Sec-WebSocket-Accept not match. Header:\n" . substr($buffer, 0, $pos) . "\n";
+                    $connection->close();
+                    return 0;
+                }
+            } else {
+                echo "Sec-WebSocket-Accept not found. Header:\n" . substr($buffer, 0, $pos) . "\n";
+                $connection->close();
+                return 0;
+            }
+
             // handshake complete
+
+            // Get WebSocket subprotocol (if specified by server)
+            if (preg_match("/Sec-WebSocket-Protocol: *(.*?)\r\n/i", $buffer, $match)) {
+                $connection->WSServerProtocol = trim($match[1]);
+            }
+
             $connection->handshakeStep = 2;
             $handshake_response_length = $pos + 4;
             // Try to emit onWebSocketConnect callback.
@@ -412,7 +434,7 @@ class Ws
             // Headbeat.
             if (!empty($connection->websocketPingInterval)) {
                 $connection->websocketPingTimer = Timer::add($connection->websocketPingInterval, function() use ($connection){
-                    if (false === $connection->send(pack('H*', '8900'), true)) {
+                    if (false === $connection->send(pack('H*', '898000000000'), true)) {
                         Timer::del($connection->websocketPingTimer);
                         $connection->websocketPingTimer = null;
                     }
@@ -430,4 +452,13 @@ class Ws
         }
         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,5 +1,10 @@
 # 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 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).
@@ -128,7 +133,7 @@ $tcp_worker->onClose = function($connection)
 Worker::runAll();
 ```
 
-### Enable SSL.
+### Enable SSL
 ```php
 <?php
 require_once __DIR__ . '/vendor/autoload.php';
@@ -137,8 +142,9 @@ use Workerman\Worker;
 // SSL context.
 $context = array(
     'ssl' => array(
-        'local_cert' => '/your/path/of/server.pem',
-        'local_pk'   => '/your/path/of/server.key',
+        'local_cert'  => '/your/path/of/server.pem',
+        'local_pk'    => '/your/path/of/server.key',
+        'verify_peer' => false,
     )
 );
 
@@ -490,20 +496,21 @@ Worker::runAll();
 
 
 ## 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)
 
@@ -592,73 +599,12 @@ Percentage of the requests served within a certain time (ms)
 
 ## Other links with workerman
 
-## [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)  
-
-## [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)  
-
-## [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)  
-
-## [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)  
-
-## [queue](https://github.com/walkor/workerman-queue)
-[Source code](https://github.com/walkor/workerman-queue)  
+[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)  
+
+## 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
 

composer.json

@@ -24,9 +24,7 @@
         "source": "https://github.com/walkor/workerman"
     },
     "require": {
-        "php": ">=5.3",
-        "ext-pcntl": "*",
-        "ext-posix": "*"
+        "php": ">=5.3"
     },
     "suggest": {
         "ext-event": "For better performance. "