PHP基于libevent的webSocket连接实例

由于libevent不支持php7,本demo采用的php版本是：php5.6.30
上文讲解php7基于event的socket的demo,本文讲解的是libevent的socket实例。
libevent管理类

/**
 * 基于libevent的webSocket连接,不支持php7
 */
class MyLibEvent
{
    protected $eventBase;
    protected $allEvents = [];

    public function __construct()
    {
        if (!extension_loaded('libevent')) {
            echo 'libevent extension is require' . PHP_EOL;
            exit(250);
        }
        $this->eventBase = event_base_new();
    }

    public function add($fd, $flag, $func, $args = array())
    {
        $fd_key = (int)$fd;
        $event = event_new();
        if (!event_set($event, $fd, $flag | EV_PERSIST, $func, null)) {
            return false;
        }
        if (!event_base_set($event, $this->eventBase)) {
            return false;
        }
        if (!event_add($event)) {
            return false;
        }

        $this->allEvents[$fd_key][$flag] = $event;
        return true;
    }

    public function del($fd, $flag)
    {
        $fd_key = (int)$fd;
        if (isset($this->allEvents[$fd_key][$flag])) {
            event_del($this->allEvents[$fd_key][$flag]);
            unset($this->allEvents[$fd_key][$flag]);
        }
        if (empty($this->allEvents[$fd_key])) {
            unset($this->allEvents[$fd_key]);
        }
    }

    public function loop()
    {
        event_base_loop($this->eventBase);
    }
}

socket管理类

class Socket
{
    const READ_BUFFER_SIZE = 65535;
    protected $mainSocket;
    protected $context;
    protected $socketName;
    protected static $connectPools = [];
    protected $sendBuffer;
    public $reusePort = false;
    protected $eventBase;
    protected $event;


    public function __construct($socketName, $context = [])
    {
        $this->socketName = $socketName;
        $this->context = stream_context_create($context);
        $this->event = new MyLibEvent();
    }

    public function start()
    {
        if ($this->reusePort) {
            stream_context_set_option($this->context, 'socket', 'so_reuseport', 1);
        }

        $local_socket = $this->socketName;
        $errorNo = 0;
        $errorMsg = '';
        $this->mainSocket = stream_socket_server($local_socket, $errorNo, $errorMsg, STREAM_SERVER_BIND | STREAM_SERVER_LISTEN, $this->context);

        if (function_exists('socket_import_stream')) {
            $socket = socket_import_stream($this->mainSocket);
            @socket_set_option($socket, SOL_SOCKET, SO_KEEPALIVE, 1);
            @socket_set_option($socket, SOL_TCP, TCP_NODELAY, 1);
        }

        stream_set_blocking($this->mainSocket, 0);

        if (function_exists('stream_set_read_buffer')) {
            stream_set_read_buffer($this->mainSocket, 0);
        }

        $flags = EV_READ;

        $this->event->add($this->mainSocket, $flags, [$this, 'accept']);
        $this->event->loop();
    }


    public function baseRead($socket)
    {
        $this->connect($socket);

        $buffer = fread($socket, self::READ_BUFFER_SIZE);

        if ($buffer === '' || $buffer === false) {
            $this->disconnect($socket);
            return;
        }
        if (false === self::$connectPools[(int)$socket]['handshake']) {
            self::$connectPools[(int)$socket]['handshake'] = $this->toHandshake($socket, $buffer);
        } else {
            $buffer = $this->decode($buffer);
            $this->send($socket, $buffer);
        }
    }


    public function accept($_socket)
    {
        $socket = @stream_socket_accept($_socket, 0, $remote_address);

        if (!$socket) {
            return;
        }
        stream_set_blocking($socket, 0);

        if (function_exists('stream_set_read_buffer')) {
            stream_set_read_buffer($socket, 0);
        }

        $flags = EV_READ;
        $this->event->add($socket, $flags, [$this, 'baseRead']);
    }

    //打包函数 返回帧处理
    protected function frame($buffer)
    {
        $len = strlen($buffer);
        if ($len <= 125) {

            return "\x81" . chr($len) . $buffer;
        } else if ($len <= 65535) {

            return "\x81" . chr(126) . pack("n", $len) . $buffer;
        } else {

            return "\x81" . char(127) . pack("xxxxN", $len) . $buffer;
        }
    }


    //解码 解析数据帧
    protected function decode($buffer)
    {
        $masks = $data = $decoded = null;
        $len = ord($buffer[1]) & 127;

        if ($len === 126) {
            $masks = substr($buffer, 4, 4);
            $data = substr($buffer, 8);
        } else if ($len === 127) {
            $masks = substr($buffer, 10, 4);
            $data = substr($buffer, 14);
        } else {
            $masks = substr($buffer, 2, 4);
            $data = substr($buffer, 6);
        }
        for ($index = 0; $index < strlen($data); $index++) {
            $decoded .= $data[$index] ^ $masks[$index % 4];
        }
        return $decoded;
    }

    protected function toHandshake($socket, $buffer)
    {
        list($resource, $host, $origin, $key) = $this->getHeaders($buffer);
        $upgrade = "HTTP/1.1 101 Switching Protocol\r\n" .
            "Upgrade: websocket\r\n" .
            "Connection: Upgrade\r\n" .
            "Sec-WebSocket-Accept: " . $this->calcKey($key) . "\r\n\r\n";  //必须以两个回车结尾

        $this->send($socket, $upgrade,false);
        return true;
    }

    public function send($socket, $buffer,$frame = true)
    {
        if($frame){
            $buffer = $this->frame($buffer);
        }
        $this->sendBuffer = $buffer;
        $this->event->add($socket, EV_WRITE, [$this, 'baseWrite']);
    }

    public function baseWrite($socket)
    {
        $len = @fwrite($socket, $this->sendBuffer, 8192);
        if ($len === strlen($this->sendBuffer)) {
            $this->event->del($socket, EV_WRITE);
        }
    }


    protected function getHeaders($req)
    {
        $r = $h = $o = $key = null;
        if (preg_match("/GET (.*) HTTP/", $req, $match)) {
            $r = $match[1];
        }
        if (preg_match("/Host: (.*)\r\n/", $req, $match)) {
            $h = $match[1];
        }
        if (preg_match("/Origin: (.*)\r\n/", $req, $match)) {
            $o = $match[1];
        }
        if (preg_match("/Sec-WebSocket-Key: (.*)\r\n/", $req, $match)) {
            $key = $match[1];
        }
        return [$r, $h, $o, $key];
    }

    protected function calcKey($key)
    {
        //基于websocket version 13
        $accept = base64_encode(sha1($key . '258EAFA5-E914-47DA-95CA-C5AB0DC85B11', true));
        return $accept;
    }

    protected function connect($socket)
    {
        $fd_key = (int)$socket;
        if (!isset(self::$connectPools[$fd_key])) {
            $connect['handshake'] = false;
            $connect['fd'] = $socket;
            self::$connectPools[$fd_key] = $connect;
        }
    }

    protected function disconnect($socket)
    {
        $fd_key = (int)$socket;
        if (isset(self::$connectPools[$fd_key])) {
            unset(self::$connectPools[$fd_key]);
            $this->event->del($socket, EV_READ);
            $this->event->del($socket, EV_WRITE);
            @fclose($socket);
        }

    }
}

运行

<?php
$connect = 'tcp://0.0.0.0:8952';
$socket = new Socket($connect);
$socket->start();

结果如下