/**
  * Runs the server.
  * This function will block forever and never return, except if an exception is thrown during the startup of the server.
  */
 public function run()
 {
     if (ini_get('max_execution_time') != 0) {
         throw new \LogicException('PHP must have no max_execution_time in order to start a server');
     }
     if (($this->socket = stream_socket_server($this->bindAddress, $errNo, $errString)) === false) {
         throw new \RuntimeException('Could not create listening socket: ' . $errString);
     }
     do {
         $readableSockets = [$this->socket];
         $write = null;
         $except = null;
         stream_select($readableSockets, $write, $except, 5);
         foreach ($readableSockets as $socket) {
             if ($socket === $this->socket) {
                 $newSocket = stream_socket_accept($this->socket);
                 try {
                     $request = new HTTPRequestFromStream($newSocket);
                     $response = new HTTPResponseToStream($newSocket);
                     $response->setHeader('Server', 'Niysu IPC server');
                     $response->setHeader('Connection', 'close');
                     $this->niysuServer->handle($request, $response);
                     stream_socket_shutdown($newSocket, STREAM_SHUT_RDWR);
                 } catch (\Exception $e) {
                     fwrite($newSocket, 'HTTP/1.1 500 Internal Server Error' . "\r\n");
                     fwrite($newSocket, 'Server: Niysu IPC server' . "\r\n\r\n");
                     fflush($newSocket);
                     stream_socket_shutdown($newSocket, STREAM_SHUT_RDWR);
                 }
             }
         }
     } while (true);
     stream_socket_shutdown($this->socket, STREAM_SHUT_RDWR);
 }

 public function listen()
 {
     echo "Starting server at http://{$this->address}:{$this->port}...\n";
     $this->socket = stream_socket_server($this->address . ':' . $this->port, $errNo, $errStr);
     if (!$this->socket) {
         throw new Exception("Can't connect socket: [{$errNo}] {$errStr}");
     }
     $this->connections[] = $this->socket;
     while (1) {
         echo "connections:";
         var_dump($this->connections);
         $reads = $this->connections;
         $writes = NULL;
         $excepts = NULL;
         $modified = stream_select($reads, $writes, $excepts, 5);
         if ($modified === false) {
             break;
         }
         echo "modified fd:";
         var_dump($modified);
         echo "reads:";
         var_dump($reads);
         foreach ($reads as $modifiedRead) {
             if ($modifiedRead === $this->socket) {
                 $conn = stream_socket_accept($this->socket);
                 fwrite($conn, "Hello! The time is " . date("n/j/Y g:i a") . "\n");
                 $this->connections[] = $conn;
             } else {
                 $data = fread($modifiedRead, 1024);
                 var_dump($data);
                 if (strlen($data) === 0) {
                     // connection closed
                     $idx = array_search($modifiedRead, $this->connections, TRUE);
                     fclose($modifiedRead);
                     if ($idx != -1) {
                         unset($this->connections[$idx]);
                     }
                 } else {
                     if ($data === FALSE) {
                         echo "Something bad happened";
                         $idx = array_search($modifiedRead, $this->connections, TRUE);
                         if ($idx != -1) {
                             unset($this->connections[$idx]);
                         }
                     } else {
                         echo "The client has sent :";
                         var_dump($data);
                         fwrite($modifiedRead, "You have sent :[" . $data . "]\n");
                         fclose($modifiedRead);
                         $idx = array_search($modifiedRead, $this->connections, TRUE);
                         if ($idx != -1) {
                             unset($this->connections[$idx]);
                         }
                     }
                 }
             }
         }
     }
 }

 public function listen($port, $host = '127.0.0.1', $protocol = 'tcp')
 {
     if (strpos($host, ':') !== false) {
         // enclose IPv6 addresses in square brackets before appending port
         $host = '[' . $host . ']';
     }
     $opts = ['socket' => ['backlog' => 1024, 'so_reuseport' => 1]];
     $flags = $protocol == 'udp' ? STREAM_SERVER_BIND : STREAM_SERVER_BIND | STREAM_SERVER_LISTEN;
     $context = stream_context_create($opts);
     $this->master = @stream_socket_server("{$protocol}://{$host}:{$port}", $errorCode, $errorMessage, $flags, $context);
     if (false === $this->master) {
         $message = "Could not bind to tcp://{$host}:{$port}: {$errorMessage}";
         throw new ConnectionException($message, $errorCode);
     }
     // Try to open keep alive for tcp and disable Nagle algorithm.
     if (function_exists('socket_import_stream') && $protocol == 'tcp') {
         $socket = socket_import_stream($this->master);
         @socket_set_option($socket, SOL_SOCKET, SO_KEEPALIVE, 1);
         @socket_set_option($socket, SOL_TCP, TCP_NODELAY, 1);
     }
     stream_set_blocking($this->master, 0);
     $this->loop->addReadStream($this->master, function ($master) {
         $newSocket = stream_socket_accept($master, 0, $remote_address);
         if (false === $newSocket) {
             $this->notifyError(new \RuntimeException('Error accepting new connection'));
             return;
         }
         $this->handleConnection($newSocket);
     });
     return $this;
 }

 public function getLine($timeout)
 {
     if (!is_resource($this->server)) {
         $this->server = stream_socket_server('tcp://127.0.0.1:' . $this->port, $errno, $errstr);
         if (!is_resource($this->server)) {
             ezcLog::getInstance()->log(sprintf('Error opening socket: %d %s', $errno, $errstr), ezcLog::ERROR);
             return;
         }
     }
     if (!is_resource($this->client)) {
         ezcLog::getInstance()->log('waiting for client connect', ezcLog::DEBUG);
         $this->client = @stream_socket_accept($this->server, $timeout);
         if (is_resource($this->client)) {
             stream_set_blocking($this->client, 0);
             stream_set_timeout($this->client, $timeout);
         }
     }
     if (is_resource($this->client)) {
         ezcLog::getInstance()->log('reading input line from client', ezcLog::DEBUG);
         $line = trim(fgets($this->client));
         if ($line) {
             ezcLog::getInstance()->log('received input line ' . $line, ezcLog::DEBUG);
         } elseif (feof($this->client)) {
             ezcLog::getInstance()->log('client closed connection', ezcLog::INFO);
             $this->close();
         } else {
             //ezcLog::getInstance()->log( 'no client input, sleeping', ezcLog::INFO );
             sleep(1);
         }
         return $line;
     }
 }

 public function onAccept($listener, $what, $arg)
 {
     $socket = stream_socket_accept($listener, 5000);
     if (!$socket) {
         echo "Invalid fd\n";
         return;
     }
     echo "Accept new socket ({$socket})\n";
     //    echo "Peer name=" . $this->peername($socket) . "\n";
     //    echo "Sock name=" . $this->sockname($socket) . "\n";
     stream_set_blocking($socket, 0);
     //    stream_set_timeout($socket, 5000);
     $readEvent = event_new();
     $writeEvent = event_new();
     if (!$readEvent || !$writeEvent) {
         echo "Unable to create read or write event ({$socket})\n";
         $this->close($socket);
     }
     event_set($readEvent, $socket, EV_READ | EV_PERSIST, array($this, 'onRead'), null);
     event_set($writeEvent, $socket, EV_WRITE | EV_PERSIST, array($this, 'onWrite'), null);
     event_base_set($readEvent, $this->base);
     event_base_set($writeEvent, $this->base);
     $this->sockets[(int) $socket] = ['readEvent' => $readEvent, 'writeEvent' => $writeEvent];
     event_add($readEvent, -1);
 }

 /**
  * Accept the incomming connection
  */
 public function accept()
 {
     $socket = stream_socket_accept($this->stream);
     if ($socket) {
         return new Connection($this, $socket);
     }
 }

 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();
     }
 }

 protected function acceptSocket()
 {
     $connection = stream_socket_accept($this->socket, 0);
     stream_set_blocking($this->socket, 0);
     self::log('Socket', 'accepted');
     return $connection;
 }

 function runTest($data, array $commands, array $ini_options = null)
 {
     file_put_contents($this->tmpDir . '/xdebug-dbgp-test.php', $data);
     $i = 1;
     $socket = $this->open();
     if ($socket === false) {
         echo "Could not create socket server - already in use?\n";
         return;
     }
     $php = $this->launchPhp($ppipes, $ini_options);
     $conn = @stream_socket_accept($socket, 3);
     if ($conn === false) {
         echo @file_get_contents($this->tmpDir . '/error-output.txt'), "\n";
         echo @file_get_contents($this->tmpDir . '/remote_log.txt'), "\n";
         proc_close($php);
         return;
     }
     // read header
     $this->doRead($conn);
     foreach ($commands as $command) {
         // inject identifier
         $parts = explode(' ', $command, 2);
         if (count($parts) == 1) {
             $command = $parts[0] . " -i {$i}";
         } else {
             $command = $parts[0] . " -i {$i} " . $parts[1];
         }
         echo "-> ", $command, "\n";
         fwrite($conn, $command . "");
         $this->doRead($conn);
         $i++;
     }
     fclose($conn);
     proc_close($php);
 }

function ssl_server($port)
{
    $host = 'ssl://127.0.0.1' . ':' . $port;
    $flags = STREAM_SERVER_BIND | STREAM_SERVER_LISTEN;
    $data = "Sending bug48182\n";
    $pem = dirname(__FILE__) . '/bug46127.pem';
    $ssl_params = array('verify_peer' => false, 'allow_self_signed' => true, 'local_cert' => $pem);
    $ssl = array('ssl' => $ssl_params);
    $context = stream_context_create($ssl);
    $sock = stream_socket_server($host, $errno, $errstr, $flags, $context);
    if (!$sock) {
        return false;
    }
    $link = stream_socket_accept($sock);
    if (!$link) {
        return false;
    }
    // bad link?
    $r = array($link);
    $w = array();
    $e = array();
    if (stream_select($r, $w, $e, 1, 0) != 0) {
        $data .= fread($link, 8192);
    }
    $r = array();
    $w = array($link);
    if (stream_select($r, $w, $e, 1, 0) != 0) {
        $wrote = fwrite($link, $data, strlen($data));
    }
    // close stuff
    fclose($link);
    fclose($sock);
    exit;
}

 function runTest($data, array $commands)
 {
     file_put_contents('/tmp/xdebug-dbgp-test.php', $data);
     $i = 1;
     $socket = $this->open();
     $php = $this->launchPhp($ppipes);
     $conn = stream_socket_accept($socket);
     // read header
     $this->doRead($conn);
     foreach ($commands as $command) {
         // inject identifier
         $parts = explode(' ', $command, 2);
         if (count($parts) == 1) {
             $command = $parts[0] . " -i {$i}";
         } else {
             $command = $parts[0] . " -i {$i} " . $parts[1];
         }
         echo "-> ", $command, "\n";
         fwrite($conn, $command . "");
         $this->doRead($conn);
         $i++;
     }
     fclose($conn);
     proc_close($php);
 }

 /**
  * creates a tcp socket if needed and accepts connection to read
  * @return resource
  */
 public function fetchMessage()
 {
     if ($this->socket === null) {
         $this->socket = stream_socket_server(static::buildFullAddress($this->address));
     }
     return @stream_socket_accept($this->socket);
 }

 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);
 }

 public function listen($port, $host = '127.0.0.1')
 {
     $localSocket = '';
     if (filter_var($host, FILTER_VALIDATE_IP, FILTER_FLAG_IPV4)) {
         $localSocket = 'tcp://' . $host . ':' . $port;
     } elseif (filter_var($host, FILTER_VALIDATE_IP, FILTER_FLAG_IPV6)) {
         // enclose IPv6 addresses in square brackets before appending port
         $localSocket = 'tcp://[' . $host . ']:' . $port;
     } elseif (preg_match('#^unix://#', $host)) {
         $localSocket = $host;
     } else {
         throw new \UnexpectedValueException('"' . $host . '" does not match to a set of supported transports. ' . 'Supported transports are: IPv4, IPv6 and unix:// .', 1433253311);
     }
     $this->master = stream_socket_server($localSocket, $errno, $errstr);
     if (false === $this->master) {
         $message = "Could not bind to {$localSocket} . Error: [{$errno}] {$errstr}";
         throw new ConnectionException($message, $errno);
     }
     stream_set_blocking($this->master, 0);
     $this->loop->addReadStream($this->master, function ($master) {
         $newSocket = stream_socket_accept($master);
         if (false === $newSocket) {
             $this->emit('error', array(new \RuntimeException('Error accepting new connection')));
             return;
         }
         $this->handleConnection($newSocket);
     });
 }

 public function start($count = 1)
 {
     $loop = Factory::create();
     $server = stream_socket_server($this->config->getConnectionString());
     if ($server === false) {
         throw new \RuntimeException("create socket server failed");
     }
     $blocking = stream_set_blocking($server, 0);
     if ($blocking === false) {
         throw new \RuntimeException("stream_set_blocking failed");
     }
     $loop->addReadStream($server, function ($server) use($loop) {
         $dispatcher = $this->config->getDispatcher();
         if ($dispatcher !== null && !$dispatcher->acquire($server, $loop)) {
             return;
         }
         $conn = stream_socket_accept($server);
         $dispatcher->release($server, $loop);
         $loop->addReadStream($conn, function ($conn) use($loop) {
             call_user_func(array($this->handler, 'handle'), $conn, $loop);
         });
     });
     $master = new Master($loop, $count);
     $master->start();
 }

 public function run($force = false)
 {
     if ($force) {
         $this->killProgramsOnDreddPort();
     }
     $socket = sprintf('tcp://%s:%s', $this->host, $this->port);
     $server = stream_socket_server($socket, $errno, $errorMessage);
     if ($server === false) {
         throw new UnexpectedValueException("Server could not bind to socket: {$errorMessage}");
     }
     $buffer = "";
     for (;;) {
         $client = stream_socket_accept($server);
         while ($socketData = stream_socket_recvfrom($client, self::RECV_LENGTH)) {
             $buffer .= $socketData;
             // determine if message terminating character is present.
             if (strpos($buffer, self::MESSAGE_END) === false) {
                 continue;
             }
             $messages = [];
             foreach (explode(self::MESSAGE_END, $buffer) as $data) {
                 $message = json_decode($data);
                 // if not valid json the partial message needs saved
                 if (!$message) {
                     $buffer = $message;
                     continue;
                 }
                 $messages[] = $message;
             }
             foreach ($messages as $message) {
                 $this->processMessage($message, $client);
             }
         }
     }
 }

 public function listen($port, $host = '127.0.0.1', $streamContext = array())
 {
     if (isset($streamContext['ssl']) && PHP_VERSION_ID < 50600) {
         throw new \RuntimeException('Secure connections are not available before PHP 5.6.0');
     }
     if (strpos($host, ':') !== false) {
         // enclose IPv6 addresses in square brackets before appending port
         $host = '[' . $host . ']';
     }
     $this->master = @stream_socket_server("tcp://{$host}:{$port}", $errno, $errstr, STREAM_SERVER_BIND | STREAM_SERVER_LISTEN, stream_context_create($streamContext));
     if (false === $this->master) {
         $message = "Could not bind to tcp://{$host}:{$port}: {$errstr}";
         throw new ConnectionException($message, $errno);
     }
     stream_set_blocking($this->master, 0);
     $that = $this;
     $this->loop->addReadStream($this->master, function ($master) use($that) {
         $newSocket = @stream_socket_accept($master);
         if (false === $newSocket) {
             $that->emit('error', array(new \RuntimeException('Error accepting new connection')));
             return;
         }
         $that->handleConnection($newSocket);
     });
 }

 public function accept()
 {
     echo "cosocket accept\n";
     (yield waitForRead($this->socket));
     echo "cosocket accept waitforead after\n";
     (yield retval(new CoSocket(stream_socket_accept($this->socket, 0))));
     echo "cosocket accept retval after";
 }

 /**
  * Accepts a new connection on the socket
  *
  * @throws ConnectionException
  * @return resource
  */
 public function accept()
 {
     $new = stream_socket_accept($this->socket, $this->options['timeout_accept']);
     if (!$new) {
         throw new ConnectionException(socket_strerror(socket_last_error($new)));
     }
     return $new;
 }

 protected function acceptSocket($socket_num)
 {
     $socket = $this->sockets[$socket_num];
     $connection = stream_socket_accept($socket, 0);
     stream_set_blocking($socket, 0);
     self::log('Socket', $socket_num, 'accepted');
     return $connection;
 }