不指定 position 获取事件。canal 会记住此 client 最新的 position。 <br/>
如果是第一次 fetch，则会从 canal 中保存的最老一条数据开始输出。

<pre>
注意： meta获取和数据的获取需要保证顺序性，优先拿到meta的，一定也会是优先拿到数据，所以需要加同步. (不能出现先拿到meta，拿到第二批数据，这样就会导致数据顺序性出现问题)
</pre>
355946
@Override
    public Message getWithoutAck(ClientIdentity clientIdentity, int batchSize) throws CanalServerException {
        return getWithoutAck(clientIdentity, batchSize, null, null);
    }
467463
function router(req, res) {

	let parse = req.parse = url.parse(req.url, true);
	let method = req.method = req.method.toUpperCase();

	let _postData = '';
	req.setEncoding('utf8');

	setClientInfo(req);

	if (clientDisAccessable(req.client.USERAGENT)) { // 客户端被拒绝，返回403
		res.writeHead(403);
		res.end();
		return;
	} else {
		req.on('error', onError)
			.on('data', function (chunk) {
				// TODO 现在这里只处理 post 上来的字符串，二进制格式要怎么弄还要再研究
				_postData += chunk;
			}).on("end", function () {
				switch (method) {
					case HttpRequestMethod.PUT:
					case HttpRequestMethod.POST:
						Object.addAll(qs.parse(_postData), parse.query);
						// 因为都要调用 request 方法，所以这里 switch 贯穿掉
					case HttpRequestMethod.GET:
						request(req, res);
						break;
					case HttpRequestMethod.DELETE:
					case HttpRequestMethod.HEAD: // TODO 下面这些暂时不做实现
					case HttpRequestMethod.CONNECT:
					case HttpRequestMethod.OPTIONS:
					case HttpRequestMethod.TRACE:
					case HttpRequestMethod.PATCH:
						break;
					default:
						unsupportedOperation(method);
				}
			}).setTimeout(TIMEOUT, function () {
				if (developMode) return; // 开发模式的情况下，无视各种超时
				// req 请求超时，网络不稳定
				// 408 Request Timeout
				Coralian.logger.err('request error code : ' + HttpStatusCode.REQUEST_TIMEOUT);
				req.url = formatString(ERROR_ROUTE_FORMAT, HttpStatusCode.REQUEST_TIMEOUT);
				req.parse = url.parse(req.url, true);
				request(req, res);
			});
		res.on('error', onError)
			.setTimeout(TIMEOUT, function () {
				if (developMode) return; // 开发模式的情况下，无视各种超时
				// res 响应超时，服务器无应答
				// 504 Gateway Timeout
				Coralian.logger.err('response error code : ' + HttpStatusCode.GATEWAY_TIMEOUT);
				req.url = formatString(ERROR_ROUTE_FORMAT, HttpStatusCode.GATEWAY_TIMEOUT);
				parse = url.parse(req.url, true);
				request(req, res);
			});
	}

	/*
	 * 错误相应函数
	 * 发生错误的时候，调用这个错误来进行捕获
	 * 但这个函数已经是服务器层级的捕获，虽然能捕获相关错误，但并不解决错误
	 * 只是将错误显示在最终页面上，返回给用户
	 */
	function onError(err) {
		// 请求（或响应）发生错误的时候，如果内部任何一层都未能捕获，则在最外层捕获
		Coralian.logger.err("HTTP req error : ");
		Coralian.logger.err(err.code);
		Coralian.logger.err(err.stack);

		// 如果发生 write after end ，则只将错误信息记录到 log 中去
		if (String.contains(err.message, 'write after end')) return;

		req.url = formatString(ERROR_ROUTE_FORMAT, (err.code || HttpStatusCode.INTERNAL_SERVER_ERROR));
		req.parse = url.parse(req.url, true);
		request(req, res);
	}
}
547647
function wrapCallback(resolver, promise, callback) {
        return function (data) {

            function work() {
                // 可能抛异常
                var res = callback(data);
                // 返回结果与当前的promise对象相同
                // 抛异常
                // see # 2.3.1
                if (res === promise) {
                    throw new TypeError();
                }

                resolve(resolver, res);
            }

            if (captureException) {
                try {
                    work();
                }
                catch (e) {
                    emitExceptionEvent(e);
                    resolver.reject(e);
                }
            }
            else {
                work();
            }

        };
    }
351743
@Override
    public void onConnect(int rst, HuaweiApiClient client) {

        HMSAgentLog.d("onConnect:" + rst);

        if (client == null || !ApiClientMgr.INST.isConnect(client)) {
            HMSAgentLog.e("client not connted");
            onPMSPayEnd(rst, null);
            return;
        }

        // 调用HMS-SDK pay 接口
        PendingResult<PayResult> payResult = HuaweiPay.HuaweiPayApi.productPay(client, payReq);
        payResult.setResultCallback(new ResultCallback<PayResult>() {
            @Override
            public void onResult(PayResult result) {

                if (result == null) {
                    HMSAgentLog.e("result is null");
                    onPMSPayEnd(HMSAgent.AgentResultCode.RESULT_IS_NULL, null);
                    return;
                }

                Status status = result.getStatus();
                if (status == null) {
                    HMSAgentLog.e("status is null");
                    onPMSPayEnd(HMSAgent.AgentResultCode.STATUS_IS_NULL, null);
                    return;
                }

                int rstCode = status.getStatusCode();
                HMSAgentLog.d("status=" + status);
                // 需要重试的错误码，并且可以重试
                if ((rstCode == CommonCode.ErrorCode.SESSION_INVALID
                        || rstCode == CommonCode.ErrorCode.CLIENT_API_INVALID) && retryTimes > 0) {
                    retryTimes--;
                    connect();
                } else if (rstCode == PayStatusCodes.PAY_STATE_SUCCESS) {
                    // 支付校验完成，取当前activity进行后续支付操作
                    Activity curActivity = ActivityMgr.INST.getLastActivity();
                    if (curActivity == null) {
                        HMSAgentLog.e("activity is null");
                        onPMSPayEnd(HMSAgent.AgentResultCode.NO_ACTIVITY_FOR_USE, null);
                        return;
                    }

                    if (statusForPay != null) {
                        HMSAgentLog.e("has already a pay to dispose");
                        onPMSPayEnd(HMSAgent.AgentResultCode.REQUEST_REPEATED, null);
                        return;
                    }

                    //启动支付流程
                    try {
                        statusForPay = status;
                        Intent intent = new Intent(curActivity, HMSPMSPayAgentActivity.class);
                        intent.putExtra(BaseAgentActivity.EXTRA_IS_FULLSCREEN, UIUtils.isActivityFullscreen(curActivity));
                        curActivity.startActivity(intent);
                    } catch (Exception e) {
                        HMSAgentLog.e("start HMSPayAgentActivity error:" + e.getMessage());
                        onPMSPayEnd(HMSAgent.AgentResultCode.START_ACTIVITY_ERROR, null);
                    }
                }else {
                    onPMSPayEnd(rstCode, null);
                }
            }
        });
    }
808134
public FetchRet fetch(String url, String bucket, String key) throws QiniuException {
        String resource = UrlSafeBase64.encodeToString(url);
        String to = encodedEntry(bucket, key);
        String path = String.format("/fetch/%s/to/%s", resource, to);
        Response res = ioPost(bucket, path);
        if (!res.isOK()) {
            throw new QiniuException(res);
        }
        FetchRet fetchRet = res.jsonToObject(FetchRet.class);
        res.close();
        return fetchRet;
    }
192535
private void refreshStreamPosition() throws IOException {
    if (currentEditLogInputStreamPosition != -1) {
      // stream was reopened
      currentEditLogInputStream.refresh(currentEditLogInputStreamPosition,
          mostRecentlyReadTransactionTxId);
    } else {
      // freshly opened stream
      currentEditLogInputStreamPosition = currentEditLogInputStream.getPosition();
    }
  }
1049557
public void insert(int line, Vertex vertex, WordNet wordNetAll)
    {
        for (Vertex oldVertex : vertexes[line])
        {
            // 保证唯一性
            if (oldVertex.realWord.length() == vertex.realWord.length()) return;
        }
        vertexes[line].add(vertex);
        ++size;
        // 保证这个词语前面直连
        final int start = Math.max(0, line - 5); // 效率起见，只扫描前4行
        for (int l = line - 1; l > start; --l)
        {
            LinkedList<Vertex> all = wordNetAll.get(l);
            if (all.size() <= vertexes[l].size())
                continue;
            for (Vertex pre : all)
            {
                if (pre.length() + l == line)
                {
                    vertexes[l].add(pre);
                    ++size;
                }
            }
        }
        // 保证这个词语后面直连
        int l = line + vertex.realWord.length();
        LinkedList<Vertex> targetLine = wordNetAll.get(l);
        if (vertexes[l].size() == 0 && targetLine.size() != 0) // 有时候vertexes里面的词语已经经过用户词典合并，造成数量更少
        {
            size += targetLine.size();
            vertexes[l] = targetLine;
        }
    }
582903
@Override
    public void onConnect(final int rst, final HuaweiApiClient client) {
        //需要在子线程中执行获取push状态的操作
        ThreadUtil.INST.excute(new Runnable() {
            @Override
            public void run() {
                if (client == null || !ApiClientMgr.INST.isConnect(client)) {
                    HMSAgentLog.e("client not connted");
                    onGetPushStateResult(rst);
                } else {
                    HuaweiPush.HuaweiPushApi.getPushState(client);
                    onGetPushStateResult(HMSAgent.AgentResultCode.HMSAGENT_SUCCESS);
                }
            }
        });
    }
422158
function finish(requester) {
        var orgHeaders = normal(requester.res.headers);
        // 恢复当前的请求上下文

        var context = requester.ctx;
        if (app.revertContext(context)) {
            // 透传响应头信息
            var res = context.res;
            headers.forEach(function (name) {
                if (orgHeaders[name]) {
                    // 直接覆盖
                    res.set(name, orgHeaders[name]);
                }
            });
        }
        // Dispose
        requester.ctx = context = null;
    }
553673
private long getCurrentPosition() {
        long pos;
        if (!useLoadBuf) {
            return in.position();
        }
        try {
            if (in != null) {
                pos = (channel.position() - in.remaining());
            } else {
                pos = channel.position();
            }
            return pos;
        } catch (Exception e) {
            log.error("Error getCurrentPosition", e);
            return 0;
        }
    }
1155287
synchronized void start(Properties properties) throws Throwable {
        String serverMode = CanalController.getProperty(properties, CanalConstants.CANAL_SERVER_MODE);
        if (serverMode.equalsIgnoreCase("kafka")) {
            canalMQProducer = new CanalKafkaProducer();
        } else if (serverMode.equalsIgnoreCase("rocketmq")) {
            canalMQProducer = new CanalRocketMQProducer();
        }

        if (canalMQProducer != null) {
            // disable netty
            System.setProperty(CanalConstants.CANAL_WITHOUT_NETTY, "true");
            String autoScan = CanalController.getProperty(properties, CanalConstants.CANAL_AUTO_SCAN);
            // 设置为raw避免ByteString->Entry的二次解析
            System.setProperty("canal.instance.memory.rawEntry", "false");
            if ("true".equals(autoScan)) {
                String rootDir = CanalController.getProperty(properties, CanalConstants.CANAL_CONF_DIR);
                if (StringUtils.isEmpty(rootDir)) {
                    rootDir = "../conf";
                }
                File rootdir = new File(rootDir);
                if (rootdir.exists()) {
                    File[] instanceDirs = rootdir.listFiles(new FileFilter() {

                        public boolean accept(File pathname) {
                            String filename = pathname.getName();
                            return pathname.isDirectory() && !"spring".equalsIgnoreCase(filename) &&
                                    !"metrics".equalsIgnoreCase(filename);
                        }
                    });
                    if (instanceDirs != null && instanceDirs.length > 0) {
                        List<String> instances = Lists.transform(Arrays.asList(instanceDirs),
                            new Function<File, String>() {

                                @Override
                                public String apply(File instanceDir) {
                                    return instanceDir.getName();
                                }
                            });
                        System.setProperty(CanalConstants.CANAL_DESTINATIONS, Joiner.on(",").join(instances));
                    }
                }
            } else {
                String destinations = CanalController.getProperty(properties, CanalConstants.CANAL_DESTINATIONS);
                System.setProperty(CanalConstants.CANAL_DESTINATIONS, destinations);
            }
        }

        logger.info("## start the canal server.");
        controller = new CanalController(properties);
        controller.start();
        logger.info("## the canal server is running now ......");
        shutdownThread = new Thread() {

            public void run() {
                try {
                    logger.info("## stop the canal server");
                    controller.stop();
                    CanalLauncher.running = false;
                } catch (Throwable e) {
                    logger.warn("##something goes wrong when stopping canal Server:", e);
                } finally {
                    logger.info("## canal server is down.");
                }
            }

        };
        Runtime.getRuntime().addShutdownHook(shutdownThread);

        if (canalMQProducer != null) {
            canalMQStarter = new CanalMQStarter(canalMQProducer);
            MQProperties mqProperties = buildMQProperties(properties);
            canalMQStarter.start(mqProperties);
            controller.setCanalMQStarter(canalMQStarter);
        }
    }
398941
private void watch(WatcherCallback watcherCallback){
        try {
            while (true) {
                final WatchKey key = watchService.take();
                if(key == null){
                    continue;
                }
                for (WatchEvent<?> watchEvent : key.pollEvents()) {
                    final WatchEvent.Kind<?> kind = watchEvent.kind();
                    //忽略无效事件
                    if (kind == StandardWatchEventKinds.OVERFLOW) {
                        continue;
                    }
                    final WatchEvent<Path> watchEventPath = (WatchEvent<Path>) watchEvent;
                    //path是相对路径（相对于监控目录）
                    final Path contextPath = watchEventPath.context();
                    LOGGER.info("contextPath:"+contextPath);
                    //获取监控目录
                    final Path directoryPath = directories.get(key);
                    LOGGER.info("directoryPath:"+directoryPath);
                    //得到绝对路径
                    final Path absolutePath = directoryPath.resolve(contextPath);
                    LOGGER.info("absolutePath:"+absolutePath);
                    LOGGER.info("kind:"+kind);
                    //判断事件类别
                    switch (kind.name()) {
                        case "ENTRY_CREATE":
                            if (Files.isDirectory(absolutePath, LinkOption.NOFOLLOW_LINKS)) {
                                LOGGER.info("新增目录：" + absolutePath);
                                //为新增的目录及其所有子目录注册监控事件
                                registerDirectoryTree(absolutePath);
                            }else{
                                LOGGER.info("新增文件：" + absolutePath);
                            }
                            break;
                        case "ENTRY_DELETE":
                            LOGGER.info("删除：" + absolutePath);
                            break;
                        case "ENTRY_MODIFY":
                            LOGGER.info("修改：" + absolutePath);
                            break;
                    }
                    //业务逻辑
                    watcherCallback.execute(kind, absolutePath.toAbsolutePath().toString());
                }
                boolean valid = key.reset();
                if (!valid) {
                    if(directories.get(key) != null){
                        LOGGER.info("停止监控目录："+directories.get(key));
                        directories.remove(key);
                    }
                }
            }
        } catch (InterruptedException ex) {
            LOGGER.info("监控目录线程退出");
        } finally{
            try {
                watchService.close();
                LOGGER.info("关闭监控目录服务");
            } catch (IOException ex) {
                LOGGER.error("关闭监控目录服务出错", ex);
            }
        }
    }
132584
@Override
    public void rollback(ClientIdentity clientIdentity, Long batchId) throws CanalServerException {
        checkStart(clientIdentity.getDestination());
        CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());

        // 因为存在第一次链接时自动rollback的情况，所以需要忽略未订阅
        boolean hasSubscribe = canalInstance.getMetaManager().hasSubscribe(clientIdentity);
        if (!hasSubscribe) {
            return;
        }
        synchronized (canalInstance) {
            // 清除batch信息
            PositionRange<LogPosition> positionRanges = canalInstance.getMetaManager().removeBatch(clientIdentity,
                batchId);
            if (positionRanges == null) { // 说明是重复的ack/rollback
                throw new CanalServerException(String.format("rollback error, clientId:%s batchId:%d is not exist , please check",
                    clientIdentity.getClientId(),
                    batchId));
            }

            // lastRollbackPostions.put(clientIdentity,
            // positionRanges.getEnd());// 记录一下最后rollback的位置
            // TODO 后续rollback到指定的batchId位置
            canalInstance.getEventStore().rollback();// rollback
                                                     // eventStore中的状态信息
            logger.info("rollback successfully, clientId:{} batchId:{} position:{}",
                clientIdentity.getClientId(),
                batchId,
                positionRanges);
        }
    }
743091
function bindEvent() {
        // 处理history.back事件
        window.addEventListener('popstate', onPopState, false);
        // 全局接管指定元素点击事件
        layer.addEventListener('click', proxy, true);
        // bigpipe回调事件
        BigPipe.on('pagerendercomplete', onPagerendered, this); // 执行完页面的ready函数后触发

        // 页面数据到达的时候派发事件
        BigPipe.on('pagearrived', onPageArrived, this);

        // 页面内所有的样式JS都被加载完成后触发
        BigPipe.on('onpageloaded', onPageLoaded, this);
    }
1144260
public void releaseConnection(ThriftConnection<T> connection) throws ThriftConnectionPoolException {
		ThriftConnectionHandle<T> handle = (ThriftConnectionHandle<T>) connection;

		// 判断连接池是否是在关闭中 如果不是在关闭中则执行回收操作
		if (!this.poolShuttingDown) {
			internalReleaseConnection(handle);
		}
	}
1331287
public synchronized Position goTo( Double lon, Double lat, Double elev, Double azimuth, boolean animate ) {
        View view = getWwd().getView();
        view.stopAnimations();
        view.stopMovement();

        Position eyePosition;
        if (lon == null || lat == null) {
            Position currentEyePosition = wwd.getView().getCurrentEyePosition();
            if (currentEyePosition != null) {
                lat = currentEyePosition.latitude.degrees;
                lon = currentEyePosition.longitude.degrees;
            } else {
                return null;
            }
        }

        if (elev == null) {
            // use the current
            elev = wwd.getView().getCurrentEyePosition().getAltitude();
        }
        if (Double.isNaN(elev)) {
            if (!Double.isNaN(lastElevation)) {
                elev = lastElevation;
            } else {
                elev = NwwUtilities.DEFAULT_ELEV;
            }
        }
        eyePosition = NwwUtilities.toPosition(lat, lon, elev);
        if (animate) {
            view.goTo(eyePosition, elev);
        } else {
            view.setEyePosition(eyePosition);
        }
        if (azimuth != null) {
            Angle heading = Angle.fromDegrees(azimuth);
            view.setHeading(heading);
        }
        lastElevation = elev;
        return eyePosition;
    }
200032
function emit(resolver) {
        var items = resolver.status === STATUS.FULFILLED
            ? resolver.fulfillList
            : resolver.rejectList;

        emitGlobalEvent(resolver);

        if (!items.length) {
            return;
        }

        // 触发注册的回调函数必须
        // 在状态改变完成后
        // see #2.2.2 #2.2.3 #2.2.4
        Resolver.nextTick(
            function () {
                var item;
                while (item = items.shift()) {
                    item(resolver.data);
                }
            }
        );
    }