/**
  * 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]);
                         }
                     }
                 }
             }
         }
     }
 }

 protected function execute(InputInterface $input, OutputInterface $output)
 {
     $host = $this->getConfig('connecthost');
     $port = $this->getConfig('connectport');
     $timeout = $this->getConfig('connecttimeout');
     $conn = $host . ':' . $port;
     $prompt = 'php-resque ' . $conn . '> ';
     $output->writeln('<comment>Connecting to ' . $conn . '...</comment>');
     if (!($fh = @fsockopen('tcp://' . $host, $port, $errno, $errstr, $timeout))) {
         $output->writeln('<error>[' . $errno . '] ' . $errstr . ' host ' . $conn . '</error>');
         return;
     }
     // Set socket timeout to 200ms
     stream_set_timeout($fh, 0, 200 * 1000);
     $stdin = fopen('php://stdin', 'r');
     $prompting = false;
     Resque\Socket\Server::fwrite($fh, 'shell');
     while (true) {
         if (feof($fh)) {
             $output->writeln('<comment>Connection to ' . $conn . ' closed.</comment>');
             break;
         }
         $read = array($fh, $stdin);
         $write = null;
         $except = null;
         $selected = @stream_select($read, $write, $except, 0);
         if ($selected > 0) {
             foreach ($read as $r) {
                 if ($r == $stdin) {
                     $input = trim(fgets($stdin));
                     if (empty($input)) {
                         $output->write($prompt);
                         $prompting = true;
                     } else {
                         Resque\Socket\Server::fwrite($fh, $input);
                         $prompting = false;
                     }
                 } elseif ($r == $fh) {
                     $input = '';
                     while (($buffer = fgets($fh, 1024)) !== false) {
                         $input .= $buffer;
                     }
                     if ($prompting) {
                         $output->writeln('');
                     }
                     $output->writeln('<pop>' . trim($input) . '</pop>');
                     if (!feof($fh)) {
                         $output->write($prompt);
                         $prompting = true;
                     }
                 }
             }
         }
         // Sleep for 10ms to stop CPU spiking
         usleep(10 * 1000);
     }
     fclose($fh);
 }

function getch_nonblock($timeout)
{
    $read = array(STDIN);
    $null = null;
    if (stream_select($read, $null, $null, floor($timeout / 1000000), $timeout % 1000000) != 1) {
        return null;
    }
    return ncurses_getch();
}

function test($name, $fd, $return_value)
{
    echo "\n------ {$name}: -------\n";
    $data = stream_get_meta_data($fd);
    $data['wrapper_data']->return_value = $return_value;
    $r = array($fd);
    $w = $e = null;
    var_dump(stream_select($r, $w, $e, 0) !== false);
}

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

function execute($cmd)
{
    $descriptors = array(0 => array('pipe', 'r'), 1 => array('pipe', 'w'), 2 => array('pipe', 'w'));
    $stdout = '';
    $proc = proc_open(escapeshellcmd($cmd), $descriptors, $pipes);
    if (!is_resource($proc)) {
        throw new \Exception('Could not execute process');
    }
    // Set the stdout stream to none-blocking.
    stream_set_blocking($pipes[1], 0);
    $timeout = 3000;
    // miliseconds.
    $forceKill = true;
    while ($timeout > 0) {
        $start = round(microtime(true) * 1000);
        // Wait until we have output or the timer expired.
        $status = proc_get_status($proc);
        $read = array($pipes[1]);
        stream_select($read, $other, $other, 0, $timeout);
        $stdout .= stream_get_contents($pipes[1]);
        if (!$status['running']) {
            // Break from this loop if the process exited before the timeout.
            $forceKill = false;
            break;
        }
        // Subtract the number of microseconds that we waited.
        $timeout -= round(microtime(true) * 1000) - $start;
    }
    if ($forceKill == true) {
        proc_terminate($proc, 9);
    }
    return $stdout;
}

 public function run()
 {
     if ($this->readline) {
         readline_callback_handler_install("CS> ", [$this, "readline_callback"]);
         $this->logger->setConsoleCallback("readline_redisplay");
     }
     while (!$this->shutdown) {
         $r = [$this->stdin];
         $w = null;
         $e = null;
         if (stream_select($r, $w, $e, 0, 200000) > 0) {
             // PHP on Windows sucks
             if (feof($this->stdin)) {
                 if (Utils::getOS() == "win") {
                     $this->stdin = fopen("php://stdin", "r");
                     if (!is_resource($this->stdin)) {
                         break;
                     }
                 } else {
                     break;
                 }
             }
             $this->readLine();
         }
     }
     if ($this->readline) {
         $this->logger->setConsoleCallback(null);
         readline_callback_handler_remove();
     }
 }

 private function selectActionableStreams($timeout)
 {
     $r = $this->readStreams;
     $w = $this->writeStreams;
     $e = NULL;
     if ($timeout <= 0) {
         $sec = 0;
         $usec = 0;
     } else {
         $sec = floor($timeout);
         $usec = ($timeout - $sec) * $this->microsecondResolution;
     }
     if (stream_select($r, $w, $e, $sec, $usec)) {
         foreach ($r as $readableStream) {
             $streamId = (int) $readableStream;
             foreach ($this->readCallbacks[$streamId] as $watcherId => $callback) {
                 $callback($watcherId, $readableStream, $this);
             }
         }
         foreach ($w as $writableStream) {
             $streamId = (int) $writableStream;
             foreach ($this->writeCallbacks[$streamId] as $watcherId => $callback) {
                 $callback($watcherId, $writableStream, $this);
             }
         }
     }
 }

 /**
  * Send data to whois server
  *
  * @throws WriteErrorException
  * @throws ReadErrorException
  * @param  object $query
  * @param  array $config
  * @return string
  */
 public function call($query, $config)
 {
     $this->disconnect();
     $this->connect($config);
     stream_set_blocking($this->sock, 1);
     if (isset($query->tld) && !isset($query->idnFqdn)) {
         $lookupString = str_replace('%domain%', $query->tld, $config['format']);
     } elseif (isset($query->ip)) {
         $lookupString = str_replace('%domain%', $query->ip, $config['format']);
     } elseif (isset($query->asn)) {
         $lookupString = str_replace('%domain%', $query->asn, $config['format']);
     } else {
         $lookupString = str_replace('%domain%', $query->idnFqdn, $config['format']);
     }
     $send = fwrite($this->sock, $lookupString . "\r\n");
     if ($send != strlen($lookupString . "\r\n")) {
         throw \WhoisParser\AbstractException::factory('WriteError', 'Error while sending data (' . $send . '/' . strlen($lookupString . "\r\n") . ')');
     }
     $read = $write = array($this->sock);
     $except = null;
     $rawdata = '';
     do {
         if (stream_select($read, $write, $except, 30) === false) {
             break;
         }
         $recv = stream_get_contents($this->sock);
         if ($recv === false) {
             throw \WhoisParser\AbstractException::factory('ReadError', 'Could not read from socket.');
         }
         $rawdata .= $recv;
     } while (!feof($this->sock));
     return str_replace("\r", '', $rawdata);
 }

 public function start()
 {
     $this->stop = false;
     $fd = inotify_init();
     $wd = inotify_add_watch($fd, $this->path, IN_ALL_EVENTS);
     $read = [$fd];
     $write = null;
     $except = null;
     stream_select($read, $write, $except, 0);
     stream_set_blocking($fd, 0);
     while (true) {
         if ($this->stop) {
             inotify_rm_watch($fd, $wd);
             return fclose($fd);
         }
         if ($events = inotify_read($fd)) {
             foreach ($events as $details) {
                 if ($details['name']) {
                     $target = rtrim($this->path, '/') . '/' . $details['name'];
                 } else {
                     $target = $this->path;
                 }
                 $file = new \SplFileInfo($target);
                 switch (true) {
                     case $details['mask'] & IN_MODIFY:
                         $this->modify($file);
                         break;
                 }
                 $this->all($file);
             }
         }
     }
 }

 public function run()
 {
     if (!$this->_connect()) {
         Logger::warning('Failed to connect.');
         return false;
     }
     Logger::debug('Connected :-)');
     if ($pcntl = extension_loaded('pcntl')) {
         Logger::debug('Trapping signals...');
         $this->_trapSignals();
     }
     Logger::debug('Entering loop.');
     $start = time();
     while (is_resource($this->socket->resource())) {
         if ($pcntl) {
             pcntl_signal_dispatch();
         }
         if ($start < time() - 60 * 30) {
             foreach ($this->_plugins['poll'] as $class) {
                 $this->_respond($this->_channels, $class->poll());
             }
             $start = time();
         }
         $r = [$this->socket->resource()];
         $w = [];
         $e = [];
         if (stream_select($r, $w, $e, 2) !== 1) {
             continue;
         }
         $this->_process($this->_read());
     }
     $this->_disconnect();
 }

 public function select($sec, $usec)
 {
     $read = array($this->sock);
     $write = null;
     $except = null;
     return stream_select($read, $write, $except, $sec, $usec);
 }

 public function execute($command)
 {
     $descriptors = array(0 => array("pipe", "r"), 1 => array("pipe", "w"), 2 => array("pipe", "w"));
     //    $command = 'start ' . $command;
     $process = proc_open($command, $descriptors, $pipes);
     if (!is_resource($process)) {
         throw new LogicException('Process cannot be spawned');
     }
     $stdoutDone = null;
     $stderrDone = null;
     while (true) {
         $rx = array();
         // The program's stdout/stderr
         if (!$stdoutDone) {
             $rx[] = $pipes[1];
         }
         if (!$stderrDone) {
             $rx[] = $pipes[2];
         }
         //      echo "stream_select: " . stream_select($rx, $tx = array(), $ex = array($rx[0]), 0) . "\n";
         $tx = array();
         $ex = array();
         stream_select($rx, $tx, $ex, 10);
         foreach ($rx as $r) {
             if ($r == $pipes[1]) {
                 $res = fgets($pipes[1]);
                 $this->output .= $res;
                 if (!$this->redirectOutput) {
                     echo $res;
                 }
                 if (!$stdoutDone && feof($pipes[1])) {
                     fclose($pipes[1]);
                     $stdoutDone = true;
                 }
             }
             if ($r == $pipes[2]) {
                 $res = fgets($pipes[2]);
                 $this->error .= $res;
                 if (!$this->redirectOutput) {
                     echo $res;
                 }
                 if (!$stderrDone && feof($pipes[2])) {
                     fclose($pipes[2]);
                     $stderrDone = true;
                 }
             }
         }
         if ($stdoutDone && $stderrDone) {
             break;
         }
     }
     $this->returnCode = proc_close($process);
     //    if(0 !== $this->returnCode) {
     //      throw new LogicException(sprintf(
     //        '[nbShell::execute] Command "%s" exited with error code %s',
     //        $command, $this->returnCode
     //      ));
     //    }
     return $this->returnCode === 0;
 }

 /**
  *
  * Exec the command and return code
  *
  * @param string $cmd
  * @param string $stdout
  * @param string $stderr
  * @param int    $timeout
  * @return int|null
  */
 public static function exec($cmd, &$stdout, &$stderr, $timeout = 3600)
 {
     if ($timeout <= 0) {
         $timeout = 3600;
     }
     $descriptors = array(1 => array("pipe", "w"), 2 => array("pipe", "w"));
     $stdout = $stderr = $status = null;
     $process = proc_open($cmd, $descriptors, $pipes);
     $time_end = time() + $timeout;
     if (is_resource($process)) {
         do {
             $time_left = $time_end - time();
             $read = array($pipes[1]);
             stream_select($read, $null, $null, $time_left, NULL);
             $stdout .= fread($pipes[1], 2048);
         } while (!feof($pipes[1]) && $time_left > 0);
         fclose($pipes[1]);
         if ($time_left <= 0) {
             proc_terminate($process);
             $stderr = 'process terminated for timeout.';
             return -1;
         }
         while (!feof($pipes[2])) {
             $stderr .= fread($pipes[2], 2048);
         }
         fclose($pipes[2]);
         $status = proc_close($process);
     }
     return $status;
 }

 public function read($length = 1024)
 {
     $data = '';
     while (true) {
         $read = array($this->socket);
         $write = array();
         $except = array();
         $r = stream_select($read, $write, $except, 0, self::TIMEOUT_USEC);
         if (false === $r) {
             throw new PhpBuf_RPC_Socket_Exception('socket select fail');
         }
         if (0 === $r) {
             continue;
         }
         $result = @fread($this->socket, $length);
         if (false === $result) {
             throw new PhpBuf_RPC_Socket_Exception('socket read error');
         }
         if (empty($result)) {
             break;
         }
         $data .= $result;
     }
     return $data;
 }

 private function _fillbuffer($pipe_num)
 {
     $buffer = [];
     $pipes = [Arr::get($this->pipes, $pipe_num)];
     if (feof(Arr::get($pipes, 0))) {
         return FALSE;
     }
     $ready = stream_select($pipes, $write, $ex, 1, 0);
     if ($ready === FALSE) {
         return FALSE;
     } elseif ($ready === 0) {
         return $buffer;
         // will be empty
     }
     $status = ['unread_bytes' => 1];
     $read = TRUE;
     while ($status['unread_bytes'] > 0) {
         $read = fread(Arr::get($pipes, 0), Kohana_Process::READ_LENGTH);
         if ($read !== FALSE) {
             $buffer[] = trim($read);
         }
         $status = stream_get_meta_data(Arr::get($pipes, 0));
     }
     return $buffer;
 }

 function poll()
 {
     $readfds = $this->readfds;
     $writefds = $this->writefds;
     if (!count($readfds) && !count($writefds)) {
         return NULL;
     }
     $null = array();
     $this->current_time = microtime(true);
     if ($this->next_alarm_time > $this->current_time) {
         $timediff = $this->next_alarm_time - $this->current_time;
         $timediff_sec = floor($timediff);
         $timediff_usec = $timediff * 1000000 % 1000000;
     } else {
         $timediff_sec = NULL;
         $timediff_usec = NULL;
     }
     $num = stream_select($readfds, $writefds, $null, $timediff_sec, $timediff_usec);
     if ($num === FALSE) {
         return FALSE;
     }
     if ($num === 0) {
         return TRUE;
     }
     foreach ($readfds as $readfd) {
         $this->returns[] = array('r', $readfd);
     }
     foreach ($writefds as $writefd) {
         $this->returns[] = array('w', $writefd);
     }
     if (count($this->returns)) {
         return $this->current = array_shift($this->returns);
     }
     return FALSE;
 }

 public function ready_for_read()
 {
     $read = array($this->socket);
     $write = array();
     $except = array();
     return (bool) stream_select($read, $write, $except, 0);
 }

 /**
  * Parse options that are passed to us from stdin.Not sure what the format
  * should be yet. To start with, we probably need options to specify the
  * location of the wordpress installation to run.
  *
  * Eventually it would be nice to be able to start a development version
  * of wordpress using sqlite (with minimal setup) and just a wp-content
  * folder, with the core being loaded from /usr/share (or similar)
  *
  * @return Array An array of options (probably)
  */
 public function get_options()
 {
     //
     // Sort out options.
     // Because a routing script can't read command arguments, we need to read
     // this from stdin (on the first request) or from the temporary settings
     // file (on subsequent requests)
     //
     if (file_exists(sys_get_temp_dir() . "/.whippet-arguments")) {
         $options = file_get_contents(sys_get_temp_dir() . "/.whippet-arguments");
     } else {
         $read = array(fopen('php://stdin', 'r'));
         $write = null;
         $except = null;
         $options = '';
         if (stream_select($read, $write, $except, 0) > 0) {
             $options = stream_get_contents($read[0]);
             fclose($read[0]);
         }
         if (file_put_contents(sys_get_temp_dir() . "/.whippet-arguments", $options) === false) {
             $this->message(Colours::fg('bold_red') . "Error: " . Colours::off() . "Unable to write options file. This is a serious error. You should probably give up and report a bug.");
         }
     }
     if (!$options) {
         $this->message(Colours::fg('bold_red') . "Error: " . Colours::off() . "Unable to locate options on stdin or on the disk. This is a serious error. You should probably give up and report a bug.");
     }
     // There's no need to validate here because the bootstrap script has done that.
     return unserialize($options);
 }