function httpGet($host, $base_fd)
{
    global $index;
    $fd = stream_socket_client("{$host}", $errno, $errstr, 3, STREAM_CLIENT_ASYNC_CONNECT | STREAM_CLIENT_CONNECT);
    $index[$fd] = 0;
    $event_fd = event_new();
    event_set($event_fd, $fd, EV_WRITE | EV_PERSIST, function ($fd, $events, $arg) use($host) {
        global $times, $limit, $index;
        if (!$index[$fd]) {
            $index[$fd] = 1;
            $out = "GET / HTTP/1.1\r\n";
            $out .= "Host: {$host}\r\n";
            $out .= "Connection: Close\r\n\r\n";
            fwrite($fd, $out);
        } else {
            $str = fread($fd, 4096);
            echo $str, PHP_EOL;
            if (feof($fd)) {
                fclose($fd);
                $times++;
                echo "done\n";
                if ($times == $limit - 1) {
                    event_base_loopexit($arg[1]);
                }
            }
        }
    }, array($event_fd, $base_fd));
    event_base_set($event_fd, $base_fd);
    event_add($event_fd);
}

 /**
  * 关闭服务器程序
  * @return unknown_type
  */
 function shutdown()
 {
     //关闭服务器端
     sw_socket_close($this->server_sock, $this->server_event);
     //关闭事件循环
     event_base_loopexit($this->base_event);
     $this->protocol->onShutdown();
 }

 function print_line($fd, $events, $arg)
 {
     echo "callback fired" . PHP_EOL;
     var_dump($arg[0]->recv());
     $arg[0]->send("got msg " . ZmqController::$msgs);
     if (ZmqController::$msgs++ >= 10) {
         event_base_loopexit($arg[1]);
     }
 }

function foo2($buf, $arg)
{
    static $i;
    $i++;
    if ($i == 10) {
        event_base_loopexit($arg);
    }
    var_dump($buf);
    var_dump(event_buffer_read($buf, 10));
}

 /**
  * Destroy libevent resource if it is opened
  *
  * @return void
  */
 private function destroyResource()
 {
     foreach ($this->events as $event) {
         $this->removeEvent($event);
     }
     if ($this->handle) {
         event_base_loopexit($this->handle, 1);
         event_base_free($this->handle);
         $this->handle = null;
     }
 }

function print_line($fd, $events, $arg)
{
    static $max_requests = 0;
    $max_requests++;
    if ($max_requests == 10) {
        // exit loop after 10 writes
        event_base_loopexit($arg[1]);
    }
    // print the line
    echo fgets($fd);
}

 /**
  * 关闭服务器程序
  * @return unknown_type
  */
 function shutdown()
 {
     //关闭所有客户端
     foreach ($this->client_sock as $k => $sock) {
         Stream::close($sock, $this->client_event[$k]);
     }
     //关闭服务器端
     Stream::close($this->server_sock, $this->server_event);
     //关闭事件循环
     event_base_loopexit($this->base_event);
     $this->protocol->onShutdown($this);
 }

 function print_line($fd, $events, $arg)
 {
     static $msgs = 1;
     echo "CALLBACK FIRED" . PHP_EOL;
     if ($arg[0]->getsockopt(ZMQ::SOCKOPT_EVENTS) & ZMQ::POLL_IN) {
         echo "Got incoming data" . PHP_EOL;
         var_dump($arg[0]->recv());
         $arg[0]->send("Got msg {$msgs}");
         if ($msgs++ >= 10) {
             event_base_loopexit($arg[1]);
         }
     }
 }

 function test()
 {
     $count = 0;
     $listener = new ZLogger\Listener('tcp://127.0.0.1:5555');
     ok($listener);
     $listener->recv(function ($e, $arg) use(&$count) {
         $count++;
         event_base_loopexit($arg[1]);
     });
     $queue = new ZMQSocket(new ZMQContext(), ZMQ::SOCKET_REQ, "MySock1");
     ok($queue);
     $queue->connect("tcp://127.0.0.1:5555");
     $encoder = new ZLogger\Encoder();
     // Assign socket 1 to the queue, send and receive
     // var_dump($queue->send("hello there!")->recv());
     $text = $encoder(array('msg' => 'text'));
     // $listener->listen();
     #  		$queue->send($text);
     #  		ok( $count );
 }

 public function stop()
 {
     // @codeCoverageIgnoreStart
     event_base_loopexit($this->base);
     // @codeCoverageIgnoreEnd
 }

 /**
  * @todo description?
  * @param boolean - Hard? If hard, we shouldn't wait for graceful shutdown of the running applications.
  * @return boolean - Ready?
  */
 public function shutdown($hard = FALSE)
 {
     $error = error_get_last();
     if ($error) {
         if ($error['type'] === E_ERROR) {
             Daemon::log('W#' . $this->pid . ' crashed by error \'' . $error['message'] . '\' at ' . $error['file'] . ':' . $error['line']);
         }
     }
     if (Daemon::$config->logevents->value) {
         $this->log('event shutdown(' . ($hard ? 'HARD' : '') . ') invoked.');
     }
     if (Daemon::$config->throwexceptiononshutdown->value) {
         throw new Exception('event shutdown');
     }
     @ob_flush();
     if ($this->terminated === TRUE) {
         if ($hard) {
             exit(0);
         }
         return;
     }
     $this->terminated = TRUE;
     $this->closeSockets();
     $this->setStatus(3);
     if ($hard) {
         exit(0);
     }
     $this->reloadReady = $this->appInstancesReloadReady();
     if ($this->reload === TRUE) {
         $this->reloadReady = $this->reloadReady && microtime(TRUE) > $this->reloadTime;
     }
     if (Daemon::$config->logevents->value) {
         $this->log('reloadReady = ' . Debug::dump($this->reloadReady));
     }
     $n = 0;
     unset(Timer::$list['checkState']);
     Timer::add(function ($event) {
         $self = Daemon::$process;
         $self->reloadReady = $self->appInstancesReloadReady();
         if ($self->reload === TRUE) {
             $self->reloadReady = $self->reloadReady && microtime(TRUE) > $self->reloadTime;
         }
         if (!$self->reloadReady) {
             $event->timeout();
         } else {
             event_base_loopexit($self->eventBase);
         }
     }, 1000000.0, 'checkReloadReady');
     while (!$this->reloadReady) {
         event_base_loop($this->eventBase);
     }
     //FS::waitAllEvents(); // ensure that all I/O events completed before suicide
     posix_kill(posix_getppid(), SIGCHLD);
     // praying to Master
     exit(0);
     // R.I.P.
 }

 function shutdown($timeout = -1)
 {
     event_base_loopexit($this->handle, $timeout);
     return $this;
 }

 public function onWriteFailureEvent($buf, $arg = NULL)
 {
     if (Daemon::$config->logevents->value) {
         Daemon::log(__METHOD__ . '()');
     }
     if ($this->onWriteFailure !== NULL) {
         call_user_func($this->onWriteFailure, $this);
     }
     event_base_loopexit(Daemon::$process->eventBase);
     $this->closeWrite();
 }

 function stop()
 {
     event_base_loopexit($this->base);
 }

 public function unloop()
 {
     event_base_loopexit($this->event_base);
 }

 /**
  * Called when the connection failed
  * @param resource Descriptor
  * @param mixed Attached variable
  * @return void
  */
 public function onFailureEvent($stream, $arg = null)
 {
     $this->close();
     if ($this->finished) {
         return;
     }
     $this->finished = true;
     if ($this->pool) {
         unset($this->pool->list[$this->id]);
     }
     $this->onFinish();
     event_base_loopexit(Daemon::$process->eventBase);
 }

 /**
  * {@inheritDoc}
  */
 public function stop()
 {
     if ($this->state !== self::STOPPED) {
         \event_base_loopexit($this->base);
         \event_base_loopexit($this->keepAliveBase);
         $this->state = self::STOPPING;
     } else {
         throw new \LogicException("Cannot stop(); event reactor not currently active");
     }
 }

 /**
  * 关闭服务器程序
  * @return unknown_type
  */
 public function shutdown()
 {
     //关闭所有客户端
     foreach ($this->client_sock as $k => $sock) {
         $this->_closeSocket($sock, $this->client_event[$k]);
     }
     //关闭服务器端
     $this->_closeSocket($this->server_sock, $this->server_event);
     //关闭事件循环
     \event_base_loopexit($this->base_event);
     $this->client->onShutdown();
 }

 public function onFailureEvent($stream, $arg)
 {
     if (is_int($stream)) {
         $connId = $stream;
     } elseif ($this->directReads) {
         $connId = array_search($stream, Daemon::$worker->pool, TRUE);
     } else {
         $connId = array_search($stream, $this->buf, TRUE);
     }
     if (Daemon::$settings['logevents']) {
         Daemon::log('[WORKER ' . Daemon::$worker->pid . '] event ' . get_class($this) . '::' . __METHOD__ . '(' . $connId . ') invoked.');
     }
     $this->abortRequestsByConnection($connId);
     $this->closeConnection($connId);
     $sess =& $this->sessions[$connId];
     if ($sess) {
         if ($sess->finished) {
             return;
         }
         $sess->finished = TRUE;
         $sess->onFinish();
     }
     event_base_loopexit(Daemon::$worker->eventBase);
 }

 public function onWriteFailureEvent($buf, $arg = NULL)
 {
     if (Daemon::$settings['logevents']) {
         Daemon::log(__METHOD__ . '()');
     }
     if ($this->onWriteFailure !== NULL) {
         call_user_func($this->onWriteFailure, $this);
     }
     event_base_loopexit(Daemon::$worker->eventBase);
     $this->closeWrite();
 }