/**
  * 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 run()
 {
     $client = $this->socket;
     fwrite($client, "Hello.Bye.\n");
     stream_socket_shutdown($client, STREAM_SHUT_RDWR);
     fclose($client);
 }

 /**
  * Command Channel event callback
  * @param resorec $Socket
  * @param int $Events
  */
 public function evcb_CommandEvent($Socket, $Events)
 {
     $ClientSocketForCommand = @stream_socket_accept($this->CommandSocket);
     $Data = json_decode($aaa = @fread($ClientSocketForCommand, 4096), true);
     stream_set_blocking($ClientSocketForCommand, 0);
     @stream_socket_shutdown($ClientSocketForCommand, STREAM_SHUT_RDWR);
     @fclose($ClientSocketForCommand);
     switch ($Data['Command']) {
         case 'Report':
             file_put_contents('/tmp/xdebug/' . $this->ServerIndex, $this->ServerIndex . ':' . time() . ':' . $this->CurrentConnected . "\n");
             break;
         case 'GC':
             $this->ClientGC();
             break;
         case 'Exit':
             echo "Start Killall Client Socket!\n";
             foreach ($this->ClientData as $SockName => $Data) {
                 $this->ClientShutDown($SockName);
             }
             echo "Killed all Client Socket!\n";
             event_base_loopbreak($this->BaseEvent);
             shmop_close($this->SHM);
             break;
         case 'Push':
             $this->ServerPush($Data['Event']);
             break;
     }
 }

 public function handleClose()
 {
     if (is_resource($this->stream)) {
         stream_socket_shutdown($this->stream, STREAM_SHUT_RDWR);
         fclose($this->stream);
     }
 }

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

 /**
  * Handles the request by calling the IPC server.
  *
  * This function will connect to the IPC server, send the request, and copy what the server sends back to the response.
  *
  * @param HTTPRequestInterface 		$httpRequest		The request to handle (if null, an instance of HTTPRequestGlobal)
  * @param HTTPResponseInterface 	$httpResponse		The response where to write the output (if null, an instance of HTTPResponseGlobal)
  */
 public function handle(HTTPRequestInterface $httpRequest = null, HTTPResponseInterface $httpResponse = null)
 {
     if (!$httpRequest) {
         $httpRequest = new HTTPRequestGlobal();
     }
     if (!$httpResponse) {
         $httpResponse = new HTTPResponseGlobal();
     }
     if (($this->socket = stream_socket_client($this->bindAddress, $errNo, $errString)) === false) {
         throw new \RuntimeException('Could not create client socket: ' . $errString);
     }
     // writing request line
     $toWrite = $httpRequest->getMethod() . ' ' . $httpRequest->getURL() . ' HTTP/1.1' . "\r\n";
     // TODO: HTTP version
     fwrite($this->socket, $toWrite);
     // writing headers
     foreach ($httpRequest->getHeadersList() as $header => $value) {
         $toWrite = $header . ': ' . $value . "\r\n";
         fwrite($this->socket, $toWrite);
     }
     // writing data
     $toWrite = "\r\n" . $httpRequest->getRawData();
     fwrite($this->socket, $toWrite);
     stream_socket_shutdown($this->socket, STREAM_SHUT_WR);
     fflush($this->socket);
     // reading response
     $response = HTTPResponseStream::build($httpResponse, true);
     $response->fwrite(stream_get_contents($this->socket));
     $response->fflush();
     unset($response);
     stream_socket_shutdown($this->socket, STREAM_SHUT_RDWR);
     $httpResponse->flush();
 }

 public function disconnect($quitMessage = null)
 {
     $this->bot->removeConnection($this);
     if ($this->socket !== null) {
         stream_socket_shutdown($this->socket, STREAM_SHUT_RDWR);
     }
     $this->socket = null;
 }

 public function close()
 {
     $this->log->info('closing connection: ' . $this->nodeUrl);
     stream_socket_shutdown($this->stream, STREAM_SHUT_RDWR);
     $this->stream = null;
     $this->log->info('connection closed: ' . $this->nodeUrl);
     return true;
 }

 public function close()
 {
     if ($this->_socket !== null) {
         $this->executeCommand('QUIT');
         stream_socket_shutdown($this->_socket, STREAM_SHUT_RDWR);
         $this->_socket = null;
     }
 }

 public function handleClose()
 {
     if (is_resource($this->stream)) {
         // http://chat.stackoverflow.com/transcript/message/7727858#7727858
         stream_socket_shutdown($this->stream, STREAM_SHUT_RDWR);
         stream_set_blocking($this->stream, false);
         fclose($this->stream);
     }
 }

 /**
  * Close the connection.
  */
 public function close()
 {
     $meta_data = stream_get_meta_data($this->resource);
     $this->extra = $meta_data['wrapper_data'];
     stream_socket_shutdown($this->resource, STREAM_SHUT_RDWR);
     // Don't allow any more reads/writes
     stream_get_contents($this->resource);
     // Clear any stuff on the socket, otherwise it will stay open
     fclose($this->resource);
     $this->resource = null;
 }

 public function shutdown($read, $write)
 {
     $r = null;
     if (function_exists("stream_socket_shutdown")) {
         $rw = $read && $write ? 2 : ($write ? 1 : ($read ? 0 : 2));
         $r = stream_socket_shutdown($this->__s, $rw);
     } else {
         $r = fclose($this->__s);
     }
     sys_net_Socket::checkError($r, 0, "Unable to Shutdown");
 }

 public function disconnect()
 {
     // TODO Check
     $id = $this->idCount++;
     $this->sendRaw("<presence id='{$id}' type='unavailable'></presence></stream:stream>");
     \stream_socket_shutdown($this->socket, \STREAM_SHUT_RDWR);
     // TODO Check error
     \fclose($this->socket);
     // TODO Check error
     $this->socket = null;
     $this->stanzaParser = null;
 }

 public function close()
 {
     echo "closed!\n";
     $this->closed = true;
     if (is_resource($this->getProperSocket())) {
         stream_socket_shutdown($this->getProperSocket(), STREAM_SHUT_RDWR);
         //This causes weird crashes. Let's just not close them.
         //@fclose($this->getProperSocket());
     }
     $this->socket = false;
     $this->rawSocket = false;
 }

 public function run()
 {
     if (!($ipcSocket = @stream_socket_client($this->ipcUri, $errno, $errstr, 5))) {
         throw new \RuntimeException(sprintf("Failed connecting to IPC server: (%d) %s", $errno, $errstr));
     }
     stream_set_write_buffer($ipcSocket, 0);
     stream_socket_shutdown($ipcSocket, STREAM_SHUT_RD);
     $openMsg = $this->getThreadId() . "\n";
     if (@fwrite($ipcSocket, $openMsg) !== strlen($openMsg)) {
         throw new \RuntimeException("Failed writing open message to IPC server");
     }
     $this->ipcSocket = $ipcSocket;
 }

 public function run()
 {
     #print __CLASS__.'->'.__FUNCTION__.''."\n";
     $readHandles = array();
     $writeHandles = array();
     $exceptHandles = array();
     if ($this->isListening()) {
         $readHandles[] = $this->getHandle();
         foreach ($this->getClients() as $client) {
             $readHandles[] = $client['handle'];
         }
         #print __CLASS__.'->'.__FUNCTION__.': isListening ('.count($readHandles).')'."\n";
     } elseif ($this->isConnected()) {
         #print __CLASS__.'->'.__FUNCTION__.': isConnected'."\n";
         $readHandles[] = $this->getHandle();
     }
     if (count($readHandles)) {
         $handlesChangedNum = stream_select($readHandles, $writeHandles, $exceptHandles, 0);
         #print __CLASS__.'->'.__FUNCTION__.': handlesChangedNum ('.$handlesChangedNum.')'."\n";
         if ($handlesChangedNum) {
             foreach ($readHandles as $readableHandle) {
                 if ($this->isListening() && $readableHandle == $this->getHandle()) {
                     // Server
                     #print __CLASS__.'->'.__FUNCTION__.': accept'."\n";
                     $handle = @stream_socket_accept($this->getHandle(), 2);
                     $client = $this->clientAdd($handle);
                     $this->execOnClientConnectFunction($client);
                 } else {
                     // Client
                     if (feof($readableHandle)) {
                         #print __CLASS__.'->'.__FUNCTION__.': feof'."\n";
                         if ($this->isListening()) {
                             $client = $this->clientGetByHandle($readableHandle);
                             if ($client) {
                                 stream_socket_shutdown($client['handle'], STREAM_SHUT_RDWR);
                                 $this->clientRemove($client);
                             }
                         } else {
                             stream_socket_shutdown($this->getHandle(), STREAM_SHUT_RDWR);
                             $this->isConnected(false);
                         }
                     } else {
                         #print __CLASS__.'->'.__FUNCTION__.': recvfrom'."\n";
                         $this->handleDataRecv($readableHandle);
                     }
                 }
             }
         }
     }
 }

 public function onRequest($clientSocket, $events, $arg)
 {
     try {
         //$transport = new TBufferedTransport(new TNonblockingSocket($clientSocket));
         $transport = new TNonblockingSocket($clientSocket);
         call_user_func($this->callback, $transport);
     } catch (Exception $e) {
         \event_del($arg[0]);
         \event_free($arg[0]);
         // close socket
         @stream_socket_shutdown($clientSocket, STREAM_SHUT_RDWR);
         @fclose($clientSocket);
         return;
     }
 }

 /**
   Bi-directional UDP communication
   @param $msg the message
   @param $port integer port
   @return the response or an empty string
 
   Whatever program is listening on the other
   end cannot respond on the same port. It must
   send the response on (port+1).
 */
 public static function udpPoke($msg, $port = 9450)
 {
     $socket = stream_socket_server("udp://127.0.0.1:" . ($port + 1), $errno, $errstr, STREAM_SERVER_BIND);
     self::udpSend($msg, $port);
     $read = array($socket);
     $write = null;
     $except = null;
     $ready = stream_select($read, $write, $except, 0, 500);
     $buf = "";
     if ($ready > 0) {
         $buf = stream_socket_recvfrom($socket, 1024, 0, $peer);
     }
     stream_socket_shutdown($socket, STREAM_SHUT_RDWR);
     return $buf;
 }

 public function disconnectClient($id)
 {
     $this->log("disconnectClient({$id})");
     $client = $this->clients[$id];
     if (!$client['request']->fired('end')) {
         $client['request']->fire('end');
     }
     if (!$client['disconnect'] && !$client['request']->fired('close')) {
         $client['request']->fire('close', $id);
     }
     ///
     $this->clients[$id]['response']->connection = null;
     $this->clients[$id]['response']->ev_buffer = null;
     ///
     stream_socket_shutdown($client['socket'], STREAM_SHUT_RDWR);
     event_buffer_disable($client['buffer'], EV_READ | EV_WRITE);
     event_buffer_free($client['buffer']);
     unset($this->clients[$id]);
 }

 public function __construct(DuplexStreamInterface $stream, Response $response, Request $request)
 {
     $this->_stream = $stream;
     $this->response = $response;
     $this->request = $request;
     $stream->on('data', function ($data) {
         $this->handleData($data);
     });
     $stream->on('end', function (DuplexStreamInterface $stream) {
         if (is_resource($stream->stream)) {
             stream_socket_shutdown($stream->stream, STREAM_SHUT_RDWR);
             stream_set_blocking($stream->stream, false);
         }
     });
     $stream->on('close', function () {
         $this->emit('close', [$this]);
     });
     $stream->on('error', function ($error) {
         $this->emit('error', [$error, $this]);
     });
 }