public function onConnectedEvent($connSocket, $events, $arg)
 {
     $connId = $arg[0];
     Debug::netEvent(get_class($this) . '::' . __METHOD__ . '(' . $connId . ') invoked. ');
     //处理两种状态，一种是直接连接成功，一种是异步通知
     if (isset($this->checkConnEvPool[$connId])) {
         // 异步通知
         // 因为 注册 EV_WRITE 事件是非持久模式的，所以这里不用 delete, 只需要 unset pool 即可
         unset($this->checkConnSocketPool[$connId]);
         unset($this->checkConnEvPool[$connId]);
     }
     $evBuf = event_buffer_new($connSocket, array($this, 'onEvBufReadEvent'), array($this, 'onEvBufWriteEvent'), array($this, 'onEventEvent'), array($connId));
     event_buffer_base_set($evBuf, daemon::$eventBase);
     event_buffer_priority_set($evBuf, 10);
     event_buffer_watermark_set($evBuf, EV_READ, $this->evBufLowMark, $this->evBufHighMark);
     if (!event_buffer_enable($evBuf, EV_READ | EV_WRITE | EV_PERSIST)) {
         Debug::netErrorEvent(get_class($this) . '::' . __METHOD__ . ': can not set base of buffer. #' . $connId);
         //            socket_close($connSocket);
         fclose($connSocket);
         return;
     }
     // 调试这里时，浪费了很多时间，必须注意的是，以上 event 所用的变量如果在函数中，如果没有交给其他的变量引用，在函数结束时就会销毁，
     // 造成连接直接断或者bufferevent 不能触发。晕啊晕。
     $this->connSocketPool[$connId] = $connSocket;
     $this->connEvBufPool[$connId] = $evBuf;
     list($ip, $port) = explode(':', stream_socket_get_name($connSocket, true));
     $this->updateLastContact($connId);
     $this->onConnected($connId, $ip, $port);
 }

 /**
  * Sets up the stream connection
  *
  * @throws PhpAmqpLib_Exception_AMQPRuntimeException
  * @throws Exception
  */
 public function connect()
 {
     $errstr = $errno = null;
     $remote = sprintf('%s://%s:%s', $this->protocol, $this->host, $this->port);
     set_error_handler(array($this, 'error_handler'));
     $this->sock = stream_socket_client($remote, $errno, $errstr, $this->connection_timeout, STREAM_CLIENT_CONNECT, $this->context);
     restore_error_handler();
     if (false === $this->sock) {
         throw new PhpAmqpLib_Exception_AMQPRuntimeException(sprintf('Error Connecting to server(%s): %s ', $errno, $errstr), $errno);
     }
     if (false === stream_socket_get_name($this->sock, true)) {
         throw new PhpAmqpLib_Exception_AMQPRuntimeException(sprintf('Connection refused: %s ', $remote));
     }
     list($sec, $uSec) = PhpAmqpLib_Helper_MiscHelper::splitSecondsMicroseconds($this->read_write_timeout);
     if (!stream_set_timeout($this->sock, $sec, $uSec)) {
         throw new PhpAmqpLib_Exception_AMQPIOException('Timeout could not be set');
     }
     // php cannot capture signals while streams are blocking
     if ($this->canDispatchPcntlSignal) {
         stream_set_blocking($this->sock, 0);
         stream_set_write_buffer($this->sock, 0);
         if (function_exists('stream_set_read_buffer')) {
             stream_set_read_buffer($this->sock, 0);
         }
     } else {
         stream_set_blocking($this->sock, 1);
     }
     if ($this->keepalive) {
         $this->enable_keepalive();
     }
 }

 protected function onMessage($connectionId, $data, $type)
 {
     //вызывается при получении сообщения от клиента
     if (!strlen($data)) {
         return;
     }
     //антифлуд:
     $source = explode(':', stream_socket_get_name($this->clients[$connectionId], true));
     $ip = $source[0];
     $time = time();
     if (isset($this->ips[$ip]) && $this->ips[$ip] == $time) {
         return;
     } else {
         $this->ips[$ip] = $time;
     }
     if ($login = array_search($connectionId, $this->logins)) {
         $message = $login . ': ' . strip_tags($data);
         $this->sendPacketToMaster('message', $message);
         $this->sendPacketToClients('message', $message);
     } else {
         if (preg_match('/^[a-zA-Z0-9]{1,10}$/', $data, $match)) {
             if (isset($this->logins[$match[0]])) {
                 $this->sendPacketToClient($connectionId, 'message', 'Система: выбранное вами имя занято, попробуйте другое.');
             } else {
                 $this->logins[$match[0]] = -1;
                 $this->sendPacketToMaster('login', array('login' => $match[0], 'clientId' => $connectionId));
             }
         } else {
             $this->sendPacketToClient($connectionId, 'message', 'Система: ошибка при выборе имени. В имени можно использовать английские буквы и цифры. Имя не должно превышать 10 символов.');
         }
     }
     //var_export($data);
     //шлем всем сообщение, о том, что пишет один из клиентов
     //echo $data . "\n";
 }

 public function loop()
 {
     $this->in_loop = true;
     while ($this->in_loop) {
         $conn = false;
         $read = array($this->socket);
         $write = null;
         $except = null;
         declare (ticks=1) {
             // stream_socket_accept() doesn't block on some(?) of the ARM systems
             // so, wrapping it into stream_select() which works always
             // see https://bugs.php.net/bug.php?id=62816
             if (1 === @stream_select($read, $write, $except, null)) {
                 $conn = @stream_socket_accept($this->socket, 0);
             }
         }
         if (false !== $conn) {
             $remote_addr = stream_socket_get_name($conn, true);
             if (false === $remote_addr) {
                 $remote_addr = null;
             }
             call_user_func($this->callback, $conn, $remote_addr);
         }
         pcntl_signal_dispatch();
     }
 }

 public function getRemoteAddress()
 {
     if ($this->socket !== false) {
         return stream_socket_get_name($this->socket, true);
     }
     return null;
 }

 public function onAcceptEvent($bindSocket, $events, $arg)
 {
     $bindSocketId = $arg[0];
     Debug::netEvent(get_class($this) . '::' . __METHOD__ . '(' . $bindSocketId . ') invoked.');
     // add to accept next event
     // why not use EV_PERSIST
     event_add($this->bindSocketEvPool[$bindSocketId]);
     $connSocket = stream_socket_accept($this->bindSocketPool[$bindSocketId]);
     if (!$connSocket) {
         Debug::netErrorEvent(get_class($this) . ': can not accept new TCP-socket');
         return;
     }
     stream_set_blocking($connSocket, 0);
     list($ip, $port) = explode(':', stream_socket_get_name($connSocket, true));
     $connId = daemon::getNextConnId();
     $evBuf = event_buffer_new($connSocket, array($this, 'onEvBufReadEvent'), array($this, 'onEvBufWriteEvent'), array($this, 'onEventEvent'), array($connId));
     event_buffer_base_set($evBuf, daemon::$eventBase);
     event_buffer_priority_set($evBuf, 10);
     event_buffer_watermark_set($evBuf, EV_READ, $this->evBufLowMark, $this->evBufHighMark);
     if (!event_buffer_enable($evBuf, EV_READ | EV_WRITE | EV_PERSIST)) {
         Debug::netErrorEvent(get_class($this) . '::' . __METHOD__ . ': can not set base of buffer. #' . $connId);
         //close socket
         stream_socket_shutdown($connSocket, STREAM_SHUT_RDWR);
         fclose($connSocket);
         return;
     }
     // 调试这里时，浪费了很多时间，必须注意的是，以上 event 所用的变量如果在函数中，如果没有交给其他的变量引用，在函数结束时就会销毁，
     // 造成连接直接断或者bufferevent 不能触发。晕啊晕。
     $this->connSocketPool[$connId] = $connSocket;
     $this->connEvBufPool[$connId] = $evBuf;
     $this->updateLastContact($connId);
     $this->onAccepted($connId, $ip, $port);
 }

function handshake($connect)
{
    //Функция рукопожатия
    $info = array();
    $line = fgets($connect);
    $header = explode(' ', $line);
    $info['method'] = $header[0];
    $info['uri'] = $header[1];
    //считываем заголовки из соединения
    while ($line = rtrim(fgets($connect))) {
        if (preg_match('/\\A(\\S+): (.*)\\z/', $line, $matches)) {
            $info[$matches[1]] = $matches[2];
        } else {
            break;
        }
    }
    $address = explode(':', stream_socket_get_name($connect, true));
    //получаем адрес клиента
    $info['ip'] = $address[0];
    $info['port'] = $address[1];
    if (empty($info['Sec-WebSocket-Key'])) {
        return false;
    }
    //отправляем заголовок согласно протоколу вебсокета
    $SecWebSocketAccept = base64_encode(pack('H*', sha1($info['Sec-WebSocket-Key'] . '258EAFA5-E914-47DA-95CA-C5AB0DC85B11')));
    $upgrade = "HTTP/1.1 101 Web Socket Protocol Handshake\r\n" . "Upgrade: websocket\r\n" . "Connection: Upgrade\r\n" . "Sec-WebSocket-Accept:" . $SecWebSocketAccept . "\r\n\r\n";
    fwrite($connect, $upgrade);
    return $info;
}

 /** @internal */
 public function checkConnectedSocket($socket)
 {
     // The following hack looks like the only way to
     // detect connection refused errors with PHP's stream sockets.
     if (false === stream_socket_get_name($socket, true)) {
         return Promise\reject(new ConnectionException('Connection refused'));
     }
     return Promise\resolve($socket);
 }

 function __construct($sock, $ssl = false)
 {
     $this->socket = $sock;
     $this->ssl = $ssl;
     $ip = stream_socket_get_name($this->socket, true);
     $c = strrpos($ip, ":");
     $this->ip = substr($ip, 0, $c);
     $this->lastping = $this->lastpong = time();
 }

 public function __construct($client)
 {
     if (!is_resource($client)) {
         return;
     }
     $this->_socket = $client;
     $this->name = stream_socket_get_name($client, TRUE);
     $this->initialized = TRUE;
 }

 public function checkConnectedSocket($socket) : PromiseInterface
 {
     // The following hack looks like the only way to
     // detect connection refused errors with PHP's stream sockets.
     if (false === stream_socket_get_name($socket, true)) {
         return new RejectedPromise(new ConnectionException('Connection refused'));
     }
     return new FulfilledPromise($socket);
 }

 public function __construct(EventHttpServer $Server, $Socket)
 {
     $this->Config = $Server->getConfig();
     $this->BaseEvent = $Server->getBaseEvent();
     $this->Socket = $Socket;
     $this->Name = stream_socket_get_name($Socket, true);
     $this->Expire = time() + $this->Config['ClientTransTimeout'];
     $this->Server = $Server;
     $this->InitSocket();
 }

 public function __construct(ServerManager $manager, $identifier, $socket)
 {
     $this->manager = $manager;
     $this->identifier = $identifier;
     $this->socket = $socket;
     $addr = stream_socket_get_name($this->socket, true);
     $final = strrpos($addr, ":");
     $this->port = (int) substr($addr, $final + 1);
     $this->address = substr($addr, 0, $final);
 }

 function read_requests($client, $options = [])
 {
     $options['server'] = stream_socket_get_name($client, true);
     $dt = CY_Util_Stream::http_read($client, $options);
     if ($dt['errno'] !== 0) {
         return $dt;
     }
     if (empty($dt['data'])) {
         return array('errno' => CYE_DATA_EMPTY);
     }
     $o = new http\Message($dt['data']);
     $this->version = $o->getHttpVersion();
     $this->method = $o->getRequestMethod();
     $headers = $o->getHeaders();
     if (!empty($headers['Content-Length'])) {
         if (isset($headers['Expect'])) {
             return array('errno' => CYE_EXPECT_FAIL);
         }
         if (empty($o->getBody())) {
             mp_log(CYE_ERROR, "Bad req:" . $options['server'] . " " . str_replace("\r\n", "\\r\\n", $dt['data']));
         }
     }
     $this->keepalive = isset($headers['Connection']) && strcasecmp($headers['Connection'], 'keep-alive') == 0;
     $this->compress = isset($headers['Accept-Encoding']) && strpos($headers['Accept-Encoding'], 'gzip') !== false;
     if (empty($headers['Host'])) {
         return array('errno' => CYE_ACCESS_DENIED);
     }
     $parts = parse_url('http://' . $headers['Host'] . $o->getRequestUrl());
     $_SERVER['REQUEST_URI'] = $o->getRequestUrl();
     $_SERVER['QUERY_STRING'] = '';
     if (!empty($parts['query'])) {
         $_SERVER['QUERY_STRING'] = $query = $parts['query'];
         parse_str($query, $_GET);
     }
     if (!empty($o->getBody())) {
         if (isset($headers['Content-Type']) && strpos($headers['Content-Type'], 'multipart/form-data') !== false) {
             // grab multipart boundary from content type header
             preg_match('/boundary=(.*)$/', $headers['Content-Type'], $matches);
             // content type is probably regular form-encoded
             if (count($matches)) {
                 $boundary = $matches[1];
                 $_POST = cy_parse_http_multipart($o->getBody(), $boundary);
             }
         }
         if (!isset($boundary)) {
             parse_str($o->getBody(), $_POST);
         }
     }
     if (isset($headers['Cookie'])) {
         $c = new http\Cookie($headers['Cookie']);
         $_COOKIE = $c->getCookies();
     }
     $_REQUEST = array_merge($_GET, $_POST, $_COOKIE);
     return array('errno' => 0);
 }

 /**
  * Returns local endpoint
  *
  * @return  peer.SocketEndpoint
  * @throws  peer.SocketException
  */
 public function localEndpoint()
 {
     if (is_resource($this->_sock)) {
         if (false === ($addr = stream_socket_get_name($this->_sock, false))) {
             throw new SocketException('Cannot get socket name on ' . $this->_sock);
         }
         return SocketEndpoint::valueOf($addr);
     }
     return null;
     // Not connected
 }

 /**
  * @param $socket
  * @param ObserverInterface $observer
  */
 public function onConnected($socket, $observer)
 {
     $this->loop->removeWriteStream($socket);
     if (false === stream_socket_get_name($socket, true)) {
         $observer->onError(new ConnectionException('Connection refused'));
         $observer->onCompleted();
         return;
     }
     $observer->onNext(new ConnectorEvent("/connector/connected", new Stream($socket, $this->loop), $this->labels));
     //$observer->onCompleted();
 }

 public function onEventAccept($socket, $event)
 {
     $conn = $this->acceptSocket();
     $remote_addr = stream_socket_get_name($conn, true);
     if (false === $remote_addr) {
         $remote_addr = null;
     }
     self::log('Socket', 'callback');
     call_user_func($this->callback, $conn, $remote_addr);
     pcntl_signal_dispatch();
 }

 public function __construct($server, $socket)
 {
     $this->server = $server;
     $this->socket = $socket;
     // set some client-information:
     $socketName = stream_socket_get_name($socket, true);
     $tmp = explode(':', $socketName);
     $this->ip = $tmp[0];
     $this->port = $tmp[1];
     $this->connectionId = md5($this->ip . $this->port . spl_object_hash($this));
     $this->log('Connected');
 }

 public function setUp()
 {
     $host = "127.0.0.1";
     $this->serverSocket = stream_socket_server("udp://{$host}:0", $errNo, $errMsg, $flags = STREAM_SERVER_BIND);
     if (!$this->serverSocket) {
         throw new \RuntimeException("Failed to create test-server-socket");
     }
     // get random port
     $socketName = stream_socket_get_name($this->serverSocket, $peerName = false);
     list(, $port) = explode(":", $socketName);
     $this->socketClient = new StreamSocketClient('udp', $host, $port);
 }

 public function getPeerName(&$ip, &$port)
 {
     $ip = 'N/A';
     $port = -1;
     $name = stream_socket_get_name($this->getHandle(), true);
     $pos = strpos($name, ':');
     if ($pos === false) {
         $ip = $name;
     } else {
         $ip = substr($name, 0, $pos);
         $port = substr($name, $pos + 1);
     }
 }