/**
  * @brief 子进程进入控制调度
  */
 protected function _start()
 {
     swoole_event_add($this->_worker->pipe, function ($pipe) {
         $package = $this->_worker->read();
         $this->_stats($package);
     });
 }

 /**
  * @brief 启动子进程
  * @param \swoole_process $worker
  */
 public function start(\swoole_process $worker)
 {
     $this->_worker = $worker;
     swoole_event_add($worker->pipe, function ($pipe) use($worker) {
         echo $worker->read() . PHP_EOL;
     });
     $this->worker();
 }

 protected function _initPipeEvents($process, $workerType)
 {
     \Daemon\Library\Ipc\Shared::setCurrentTaskTable($process->pid, array('taskNum' => 0, 'workerType' => $workerType));
     $objMaster = \Daemon\Process\Master::getInstance();
     swoole_event_add($process->pipe, function ($pipe) use($process, $objMaster) {
         $objMaster->dispatch($process);
     });
 }

 function init()
 {
     $this->serv->on('WorkerStart', function (\swoole_server $serv, $worker_id) {
         //每1分钟向服务器上报
         $serv->tick(60000, [$this, 'onTimer']);
         swoole_event_add($this->centerSocket->sock, [$this, 'onPacket']);
     });
     $this->log(__CLASS__ . '-' . self::VERSION . ' is running.');
 }

 public function __construct()
 {
     $this->process = new swoole_process(array($this, 'run'), false, true);
     //$this->process->daemon(true,true);
     $this->process->start();
     swoole_event_add($this->process->pipe, function ($pipe) {
         $data = $this->process->read();
         echo "RECV: " . $data . PHP_EOL;
     });
 }

 /**
  * create mysql connection
  */
 protected function createConnection()
 {
     $config = $this->config;
     $db = new \mysqli();
     $db->connect($config['host'], $config['user'], $config['password'], $config['database'], $config['port']);
     $db_sock = swoole_get_mysqli_sock($db);
     swoole_event_add($db_sock, array($this, 'onSQLReady'));
     $this->idle_pool[] = array('object' => $db, 'socket' => $db_sock);
     $this->connection_num++;
 }

 public function onConnect($client)
 {
     fwrite(STDOUT, "Enter Msg:");
     swoole_event_add(STDIN, function ($fp) {
         global $client;
         fwrite(STDOUT, "Enter Msg:");
         $msg = trim(fgets(STDIN));
         $client->send($msg);
     });
 }

 function onStart($serv)
 {
     $this->serv = $serv;
     for ($i = 0; $i < $this->pool_size; $i++) {
         $db = new mysqli();
         $db->connect('172.16.1.19', 'test', 'test', '1188test');
         $db_sock = swoole_get_mysqli_sock($db);
         swoole_event_add($db_sock, array($this, 'onSQLReady'));
         $this->idle_pool[] = array('mysqli' => $db, 'db_sock' => $db_sock, 'fd' => 0);
     }
     echo "Server: start.Swoole version is [" . SWOOLE_VERSION . "]\n";
 }

 public function task_run($worker)
 {
     swoole_event_add($worker->pipe, function ($pipe) use($worker) {
         $data = $worker->read();
         var_dump($worker->pid . ": " . $data);
         if ($data == 'exit') {
             $worker->exit();
             exit;
         }
         sleep(5);
         $worker->write("" . $worker->pid);
     });
 }

function callback_function(swoole_process $worker)
{
    $GLOBALS['worker'] = $worker;
    swoole_event_add($worker->pipe, function ($pipe) {
        $worker = $GLOBALS['worker'];
        $recv = $worker->read();
        if ($recv) {
            $data = call_user_func_array('dosomething', [json_decode($recv, true)]);
            $worker->write($data);
        } else {
            $worker->write('');
        }
    });
}

 public function onConnect($cli)
 {
     fwrite(STDOUT, "Enter your name: ");
     $msg = trim(fgets(STDIN));
     $data = json_encode(array('json' => 'Chat', 'ctrl' => 'Chat', 'method' => 'online', 'name' => $msg));
     $data = pack("Na*", strlen($data), $data);
     $cli->send($data);
     swoole_event_add(STDIN, function ($fp) {
         global $cli;
         $msg = trim(fgets(STDIN));
         $data = json_encode(array('json' => 'Chat', 'ctrl' => 'Chat', 'method' => 'send', 'sendto' => $this->channel, 'msg' => $msg));
         $data = pack("Na*", strlen($data), $data);
         $cli->send($data);
     });
 }

function callback_function_async(swoole_process $worker)
{
    echo "Worker: start. PID=" . $worker->pid . "\n";
    //recv data from master
    $GLOBALS['worker'] = $worker;
    swoole_event_add($worker->pipe, function ($pipe) {
        $worker = $GLOBALS['worker'];
        $recv = $worker->read();
        echo "From Master: {$recv}\n";
        //send data to master
        $worker->write("hello master\n");
        sleep(5);
        $worker->exit(0);
    });
}

function child_async(swoole_process $worker)
{
    //echo "Worker: start. PID=".$worker->pid."\n";
    //recv data from master
    $GLOBALS['worker'] = $worker;
    global $argv;
    $worker->name("{$argv[0]}: worker");
    swoole_process::signal(SIGTERM, function ($signal_num) use($worker) {
        echo "signal call = {$signal_num}, #{$worker->pid}\n";
    });
    swoole_event_add($worker->pipe, function ($pipe) use($worker) {
        $recv = $worker->read();
        echo "From Master: {$recv}\n";
        $worker->write("hello master\n");
    });
}

 public function __construct()
 {
     $fp = stream_socket_client("tcp://127.0.0.1:9504", $code, $msg, 3);
     $http_request = "GET /index.html HTTP/1.1\r\n\r\n";
     fwrite($fp, $http_request);
     swoole_event_add($fp, function ($fp) {
         echo fread($fp, 8192);
         swoole_event_del($fp);
         fclose($fp);
     });
     swoole_timer_after(2000, function () {
         echo "2000ms timeout\n";
     });
     swoole_timer_tick(1000, function () {
         echo "1000ms interval\n";
     });
 }

 public function onStart($server)
 {
     $pro_vs = $this->process;
     $server = $this->server;
     $application_vm = $this->application;
     swoole_event_add($pro_vs->pipe, function ($pipe) use($pro_vs, $server, $application_vm) {
         $str = $pro_vs->read();
         ob_start();
         $application_vm->execute(array('swoole_socket', 'getfd'), 'vmstat');
         $result = ob_get_contents();
         ob_end_clean();
         $result_fd = json_decode($result, true);
         foreach ($result_fd as $id => $fd) {
             $server->push($fd, $str);
         }
     });
 }

 public function onConnect($cli)
 {
     $this->client->sendHeader($this->client->createHeader());
     fwrite(STDOUT, "Enter your name: ");
     $msg = trim(fgets(STDIN));
     $data = json_encode(array('json' => 'Chat', 'ctrl' => 'Chat', 'method' => 'online', 'name' => $msg));
     $this->client->send($data);
     swoole_event_add(STDIN, function ($fp) {
         $msg = trim(fgets(STDIN));
         if ($msg == 'exit') {
             $data = json_encode(array('json' => 'Chat', 'ctrl' => 'Chat', 'method' => 'offline', 'name' => $msg));
             $this->client->send($data);
             exit;
         }
         $data = json_encode(array('json' => 'Chat', 'ctrl' => 'Chat', 'method' => 'send', 'sendto' => $this->channel, 'msg' => $msg));
         $this->client->send($data);
     });
 }

 /**
  * 创建一个子进程
  * @param $task
  */
 public function create_process($id, $task)
 {
     $this->task = $task;
     $process = new swoole_process(array($this, "run"));
     if (!($pid = $process->start())) {
     }
     //记录当前任务
     Crontab::$task_list[$pid] = array("start" => microtime(true), "id" => $id, "task" => $task, "type" => "crontab", "process" => $process);
     swoole_event_add($process->pipe, function ($pipe) use($process) {
         $task = $process->read();
         list($pid, $sec) = explode(",", $task);
         if (isset(Crontab::$task_list[$pid])) {
             $tasklist = Crontab::$task_list[$pid];
             Crontab::$delay[time() + $sec] = $tasklist["task"];
             $process->write($task);
         }
     });
 }

 public function onConnect($cli)
 {
     fwrite(STDOUT, "Enter cmd:");
     swoole_event_add(STDIN, function ($fp) {
         global $cli;
         $cmd = trim(fgets(STDIN));
         if (is_numeric($cmd)) {
             $filename = "Test.txt";
             $i = intval($cmd);
             if ($i == Client::UPLOAD) {
                 $cli->sendfile($filename);
             } else {
                 if ($i == Client::DOWNLOAD) {
                     $cli->send(Client::DOWNLOAD, $filename);
                 }
             }
         }
     });
 }

 /**
  * [send 兼容Base类封装的send方法，调度器可以不感知client类型]
  * @param  [type] $callback [description]
  * @return [type]           [description]
  */
 public function send(callable $callback)
 {
     if (!isset($this->db)) {
         echo " db not init \n";
         //TODO do callback function to task
         return;
     }
     //TODO conf check
     $config = $this->conf;
     $this->callback = $callback;
     $this->calltime = microtime(true);
     $this->key = md5($this->calltime . $config['host'] . $config['port'] . rand(0, 10000));
     $this->db->connect($config['host'], $config['user'], $config['password'], $config['database'], $config['port']);
     if (!empty($config['charset'])) {
         $this->db->set_charset($config['charset']);
     }
     $db_sock = swoole_get_mysqli_sock($this->db);
     swoole_event_add($db_sock, array($this, 'onSqlReady'));
     $this->doQuery($this->sql);
 }

 /**
  * @param $serverPid
  * @throws NotFound
  */
 function __construct($serverPid)
 {
     $this->pid = $serverPid;
     if (posix_kill($serverPid, 0) === false) {
         throw new NotFound("Process#{$serverPid} not found.");
     }
     $this->inotify = inotify_init();
     $this->events = IN_MODIFY | IN_DELETE | IN_CREATE | IN_MOVE;
     swoole_event_add($this->inotify, function ($ifd) {
         $events = inotify_read($this->inotify);
         if (!$events) {
             return;
         }
         var_dump($events);
         foreach ($events as $ev) {
             if ($ev['mask'] == IN_IGNORED) {
                 continue;
             } else {
                 if ($ev['mask'] == IN_CREATE or $ev['mask'] == IN_DELETE or $ev['mask'] == IN_MODIFY or $ev['mask'] == IN_MOVED_TO or $ev['mask'] == IN_MOVED_FROM) {
                     $fileType = strstr($ev['name'], '.');
                     //非重启类型
                     if (!isset($this->reloadFileTypes[$fileType])) {
                         continue;
                     }
                 }
             }
             //正在reload，不再接受任何事件，冻结10秒
             if (!$this->reloading) {
                 $this->putLog("after 10 seconds reload the server");
                 //有事件发生了，进行重启
                 swoole_timer_after($this->afterNSeconds * 1000, array($this, 'reload'));
                 $this->reloading = true;
             }
         }
     });
 }