Poll
@param  callable $cb
@callback $cb ( )
@return void
199553
protected function poll($cb = null)
    {
        $this->appInstance->subscribe('s2c:' . $this->sessId, [$this, 's2c'], function ($redis) use ($cb) {
            $this->appInstance->publish('poll:' . $this->sessId, json_encode($this->pollMode),
                function ($redis) use ($cb) {
                    if (!$redis) {
                        $cb === null || $cb();
                        return;
                    }
                    if ($redis->result > 0) {
                        $cb === null || $cb();
                        return;
                    }
                    $this->appInstance->setnx('sess:' . $this->sessId, $this->attrs->server['REQUEST_URI'],
                        function ($redis) use ($cb) {
                            if (!$redis || $redis->result === 0) {
                                $this->error(3000);
                                $cb === null || $cb();
                                return;
                            }
                            $this->appInstance->expire('sess:' . $this->sessId,
                                $this->appInstance->config->deadsessiontimeout->value, function ($redis) use ($cb) {
                                    if (!$redis || $redis->result === 0) {
                                        $this->error(3000);
                                        $cb === null || $cb();
                                        return;
                                    }
                                    $this->appInstance->subscribe('state:' . $this->sessId,
                                        function ($redis) use ($cb) {
                                            if (!$redis) {
                                                return;
                                            }
                                            list(, $chan, $state) = $redis->result;
                                            if ($state === 'started') {
                                                $this->sendFrame('o');
                                                if (!in_array('stream', $this->pollMode)) {
                                                    $this->finish();
                                                    return;
                                                }
                                            }
                                            $this->appInstance->publish('poll:' . $this->sessId,
                                                json_encode($this->pollMode), function ($redis) use ($cb) {
                                                    if (!$redis || $redis->result === 0) {
                                                        $this->error(3000);
                                                        $cb === null || $cb();
                                                        return;
                                                    }
                                                    $cb === null || $cb();
                                                });
                                        }, function ($redis) use ($cb) {
                                            if (!$this->appInstance->beginSession($this->path, $this->sessId,
                                                $this->attrs->server)
                                            ) {
                                                $this->header('404 Not Found');
                                                $this->finish();
                                                $this->unsubscribeReal('state:' . $this->sessId);
                                            }
                                            $cb === null || $cb();
                                        });
                                });
                        });
                });
        });
    }
1170940
function filter(callable $cb) {
		$rowset = clone $this;
		$rowset->index = 0;
		$rowset->rows = array_filter($this->rows, $cb);
		return $rowset;
	}
458680
public function execute(callable $callback)
    {
        $commands = $this->commands;
        $command  = $this->client->createCommand('EXEC');

        $this->client->executeCommand($command, function ($responses, $client) use ($commands, $callback) {
            $size = count($responses);
            $processed = [];

            for ($i = 0; $i < $size; $i++) {
                $command  = $commands->dequeue();
                $response = $responses[$i];

                unset($responses[$i]);

                if (!$response instanceof ResponseInterface) {
                    $response = $command->parseResponse($response);
                }

                $processed[$i] = $response;
            }

            call_user_func($callback, $processed, $client);
        });
    }
586873
public static function call($cb, array $args = [])
    {
        if (!$cb) {
            return true;
        }

        if (\is_array($cb)) {
            [$obj, $mhd] = $cb;

            return \is_object($obj) ? $obj->$mhd(...$args) : $obj::$mhd(...$args);
        }

        if (\is_string($cb)) {
            if (\function_exists($cb)) {
                return $cb(...$args);
            }

            // a class name
            if (\class_exists($cb)) {
                $cb = new $cb;
            }
        }

        // a \Closure or Object implement '__invoke'
        if (\is_object($cb) && \method_exists($cb, '__invoke')) {
            return $cb(...$args);
        }

        throw new \InvalidArgumentException('the callback handler is not callable!');
    }
114094
public function aggregate($p, $cb)
    {
        $this->_params($p);

        $e = explode('.', $p['col'], 2);
        $query = [
            'aggregate' => $e[1]
        ];

        if (isset($p['rp'])) {
            $query['$readPreference'] = $p['rp'];
            unset($p['rp']);
        }
        foreach ($p as $k => $v) {
            if (substr($k, 0, 1) === '$' || $k === 'pipeline') {
                $query[$k] = $v;
            }
        }
        $cb = CallbackWrapper::wrap($cb);
        try {
            $this->request(
                self::OP_QUERY,
                pack('V', $p['opts']) . $e[0] . '.$cmd' . "\x00" . pack('VV', $p['offset'], $p['limit'])
                . bson_encode($query) . (isset($p['fields']) ? bson_encode($p['fields']) : ''),
                true,
                null,
                function ($conn, $reqId = null) use ($p, $cb) {
                    if (!$conn) {
                        !$cb || $cb(['$err' => 'Connection error.']);
                        return;
                    }
                    $conn->requests[$reqId] = [$p['col'], $cb, false];
                }
            );
        } catch (\MongoException $e) {
            Daemon::log('MongoClient exception: ' . $e->getMessage() . ': ' . $e->getTraceAsString());
            if ($cb !== null) {
                $cb([
                    '$err' => $e->getMessage(),
                    '$query' => $query,
                    '$fields' => isset($p['fields']) ? $p['fields'] : null
                ]);
            }
        }
    }
309371
public function poll($redis)
    {
        if (!$redis) {
            return;
        }
        list(, $chan, $msg) = $redis->result;
        $this->pollMode = json_decode($msg, true);

        Timer::setTimeout($this->finishTimer);
        $this->flush();
    }
433557
public function removeMessage($cb, $uid)
    {
        $this->onResponse->push($cb);
        $this->store([self::FLAG_DELETED], $uid, null, '+', true, self::TAG_DELETEMESSAGE);
    }
213117
public function addOnFulfilled(callable $callback): void
    {
        if ($this->getState() === Promise::PENDING) {
            $this->onFulfilled[] = $callback;
        } elseif ($this->getState() === Promise::FULFILLED) {
            $response = call_user_func($callback, $this->responseBuilder->getResponse());
            if ($response instanceof ResponseInterface) {
                $this->responseBuilder->setResponse($response);
            }
        }
    }
547199
public static function create(callable $cb)
    {
        $tid = self::tid();
        return SwCoroutine::create(function () use ($cb, $tid) {
            $id = SwCoroutine::getuid();
            self::$idMap[$id] = $tid;

            PhpHelper::call($cb);
        });
    }
917590
public function onBeforeHandshake($cb)
    {
        if (!method_exists($this->realRoute, 'onBeforeHandshake')) {
            return false;
        }
        $this->realRoute->onBeforeHandshake($cb);
    }
641889
public function publish($chan, $cb, $opcb = null)
    {
        $this->redis->publish($this->config->redisprefix->value . $chan, $cb, $opcb);
    }
453335
public function setCallbackObjectStatus($callbackObjectStatus = FALSE)
    {
        $this->callbackObjectStatus = $callbackObjectStatus;
        $this->debug->debug(__FUNCTION__, 'setCallbackObjectStatus: ' . json_encode($this->callbackObjectStatus));
    }
387131
public function broadcastCall($method, $args = [], $cb = null)
    {
        return Daemon::$process->IPCManager->sendBroadcastCall(
            '\\' . get_class($this) . ($this->name !== '' ? ':' . $this->name : ''),
            $method,
            $args,
            $cb
        );
    }
1139749
public function evaluate($code, $cb)
    {
        $p = [];

        if (!isset($p['offset'])) {
            $p['offset'] = 0;
        }

        if (!isset($p['limit'])) {
            $p['limit'] = -1;
        }

        if (!isset($p['opts'])) {
            $p['opts'] = 0;
        }

        if (!isset($p['db'])) {
            $p['db'] = $this->dbname;
        }

        $cb = CallbackWrapper::wrap($cb);
        try {
            $this->request(
                self::OP_QUERY,
                pack('V', $p['opts']) . $p['db'] . '.$cmd' . "\x00"
                . pack('VV', $p['offset'], $p['limit']) . bson_encode(['$eval' => new \MongoCode($code)])
                . (isset($p['fields']) ? bson_encode($p['fields']) : ''),
                true,
                null,
                function ($conn, $reqId = null) use ($p, $cb) {
                    if (!$conn) {
                        !$cb || $cb(['$err' => 'Connection error.']);
                        return;
                    }
                    $conn->requests[$reqId] = [$p['db'], $cb, true];
                }
            );
        } catch (\MongoException $e) {
            Daemon::log('MongoClient exception: ' . $e->getMessage() . ': ' . $e->getTraceAsString());
            if ($cb !== null) {
                $cb([
                    '$err' => $e->getMessage(),
                    '$query' => $query,
                    '$fields' => isset($p['fields']) ? $p['fields'] : null
                ]);
            }
        }
    }
288227
public function on($name, callable $cb, $replace = false)
    {
        if ($replace || !isset($this->_events[$name])) {
            $this->_events[$name] = $cb;
        }
    }
90728
protected function postFinishHandler($cb = null)
    {
        if (!$this->headers_sent) {
            $this->out('');
        }
        $this->sendfp = null;
        if (isset($this->attrs->files)) {
            foreach ($this->attrs->files as $f) {
                if (isset($f['tmp_name'])) {
                    FileSystem::unlink($f['tmp_name']);
                }
            }
        }
        if (isset($this->attrs->session)) {
            $this->sessionCommit($cb);
        } else {
            $cb === null || $cb();
        }
    }
291275
public function sendPing($host, $cb)
    {
        $this->connect('raw://' . $host, function ($conn) use ($cb) {
            $conn->sendEcho($cb);
        });
    }