回滚到未进行 {@link #ack} 的地方，下次fetch的时候，可以从最后一个没有 {@link #ack} 的地方开始拿
513221
@Override
    public void rollback(ClientIdentity clientIdentity) throws CanalServerException {
        checkStart(clientIdentity.getDestination());
        CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
        // 因为存在第一次链接时自动rollback的情况，所以需要忽略未订阅
        boolean hasSubscribe = canalInstance.getMetaManager().hasSubscribe(clientIdentity);
        if (!hasSubscribe) {
            return;
        }

        synchronized (canalInstance) {
            // 清除batch信息
            canalInstance.getMetaManager().clearAllBatchs(clientIdentity);
            // rollback eventStore中的状态信息
            canalInstance.getEventStore().rollback();
            logger.info("rollback successfully, clientId:{}", new Object[] { clientIdentity.getClientId() });
        }
    }
906274
public void check() throws ThriftConnectionPoolException {
		if (connectTimeout <= 0) {
			throw new ThriftConnectionPoolException("连接超时时间必须大于0");
		}

		// 判断是否是单接口模式 如果是则只检测单接口
		if (getThriftServiceType() == ThriftServiceType.SINGLE_INTERFACE) {
			if (clientClass == null) {
				throw new ThriftConnectionPoolException("thrift客户端实现类未设置");
			}
			// 检测ping方法
			try {
				clientClass.getMethod("ping");
			} catch (NoSuchMethodException e) {
				throw new ThriftConnectionPoolException("Thrift客户端实现类必须带有ping()方法用于检测连接");
			}
		} else if (getThriftServiceType() == ThriftServiceType.MULTIPLEXED_INTERFACE) {
			if (clientClasses.size() == 0) {
				throw new ThriftConnectionPoolException("多服务thrift客户端实现类未设置");
			}
			// 检测所有接口
			List<String> toRemoveClasses = new ArrayList<>();
			Iterator<Entry<String, Class<? extends TServiceClient>>> iterator = clientClasses.entrySet().iterator();
			while (iterator.hasNext()) {
				Entry<String, Class<? extends TServiceClient>> entry = iterator.next();
				Class<? extends TServiceClient> clazz = entry.getValue();
				try {
					clazz.getMethod("ping");
				} catch (NoSuchMethodException e) {
					toRemoveClasses.add(entry.getKey());
					logger.warn("接口：" + entry.getKey() + " 没有实现ping方法 无法创建对应的服务客户端");
				}
			}
			for (String toRemoveClass : toRemoveClasses) {
				clientClasses.remove(toRemoveClass);
			}
			Iterator<String> servicesNameIterator = clientClasses.keySet().iterator();
			while (servicesNameIterator.hasNext()) {
				logger.info("注册服务客户端：" + servicesNameIterator.next());
			}
		}

		if (maxConnectionPerServer <= 0) {
			throw new ThriftConnectionPoolException("每台服务器的最大连接数必须大于0");
		}

		if (minConnectionPerServer < 0) {
			throw new ThriftConnectionPoolException("每台服务器最小连接数不能小于0");
		}

		if (minConnectionPerServer > maxConnectionPerServer) {
			throw new ThriftConnectionPoolException("每台服务器的最小连接数不能超过最大连接数配置");
		}

		if (idleConnectionTestPeriodInSeconds <= 0) {
			throw new ThriftConnectionPoolException("检测时间周期必须大于0秒");
		}

		if (idleMaxAgeInSeconds <= 0) {
			throw new ThriftConnectionPoolException("未使用连接关闭时间必须大于0秒");
		}

		if (maxConnectionAgeInSeconds <= 0) {
			throw new ThriftConnectionPoolException("连接最大生存时间必须大于0秒");
		}

		if (acquireRetryAttempts < 0) {
			throw new ThriftConnectionPoolException("获取连接重试次数不能小于0");
		}

		if (acquireRetryDelayInMs < 0) {
			throw new ThriftConnectionPoolException("获取连接重试等待时间不能小于0毫秒");
		}

		if (connectionTimeoutInMs < 0) {
			throw new ThriftConnectionPoolException("获取连接等待时间不能小于0毫秒");
		}

		if (acquireIncrement <= 0) {
			throw new ThriftConnectionPoolException("每次创建原始连接的数量必须大于0个");
		}
	}
990230
private void getRets(HashMap<String, Return> all, Method method,
			Class<?> clz) throws SecurityException {
		if (!clz.getSuperclass().equals(Object.class))
			getRets(all, method, clz.getSuperclass());

		Method m = null;

		try {
			m = clz.getDeclaredMethod(method.getName(),
					method.getParameterTypes());
		} catch (NoSuchMethodException e) {
		}
		if (m != null) {
			Return[] rets = m.getAnnotationsByType(Return.class);
			if (rets != null)
				for (Return ret : rets)
					all.put(ret.value(), ret);
		}
	}

	public ReturnDefList(Method method) {
		Return[] rets = null;

		if (method.getDeclaringClass().getSuperclass().equals(Object.class)) { // 没有父类
			rets = method.getAnnotationsByType(Return.class);
		} else { // 有父类，需要检查合并父类上声明的Return
			HashMap<String, Return> allRets = new HashMap<String, Return>();

			try {
				getRets(allRets, method, method.getDeclaringClass());
			} catch (Exception e) {
				e.printStackTrace();
			}

			if (allRets.size() > 0)
				rets = allRets.values().toArray(new Return[allRets.size()]);
		}

		if (rets != null)
			for (Return ret : rets) {
				ReturnDef rd = new ReturnDef(ret);
				if (rd.hasValue())
					getRetDefList(rd.getValue()).add(rd);
			}
	}

	public ArrayList<ReturnDef> getReturnDef(String code) {
		ArrayList<ReturnDef> list = htRetDefs.get(code);
		if (list == null || list.size() == 0)
			return htRetDefs.get(Return.NOT_SPECIFIED);

		return list;
	}
}
628046
private void fetchNext() {
		if (pendingUtc != null || done) {
			return;
		}

		DateValue dUtc = generateInstance();

		//check the exit condition
		if (dUtc == null || !condition.apply(dUtc)) {
			done = true;
			return;
		}

		pendingUtc = dUtc;
		yearGenerator.workDone();
	}
902528
@Override
	protected ThriftConnection<T> getConnectionInternal() throws ThriftConnectionPoolException {
		ThriftConnection<T> result = pollConnection();
		// 如果立即获取连接失败 则换一个分区继续获取
		// TODO 设置当连接获取超时返回null？
		if (result == null) {
			if (this.pool.getThriftServerCount() == 0) {
				throw new ThriftConnectionPoolException("当前没有可用的服务器  无法获取连接");
			}
			int partition = (int) (Thread.currentThread().getId() % this.pool.thriftServerCount);

			ThriftConnectionPartition<T> thriftConnectionPartition = this.pool.partitions.get(partition);

			try {
				result = thriftConnectionPartition.poolFreeConnection(this.pool.connectionTimeoutInMs,
						TimeUnit.MILLISECONDS);
				if (result == null) {

					throw new ThriftConnectionPoolException("Timed out waiting for a free available connection.");
				}
			} catch (InterruptedException e) {
				throw new ThriftConnectionPoolException(e);
			}
		}
		return result;
	}
1021760
public static HttpCallContext get() throws IllegalStateException
	{
		final HttpCallContext ctx = peek();

		if (ctx != null)
			return ctx;
		else
			throw new IllegalStateException("Not in an HttpCallContext!");
	}
421282
@Override
    public CompletableFuture<Integer> forceCloseWebSocket(Serializable userid, @RpcTargetAddress InetSocketAddress targetAddress) {
        //不能从sncpNodeAddresses中移除，因为engine.forceCloseWebSocket 会调用到disconnect
        if (logger.isLoggable(Level.FINEST)) logger.finest(WebSocketNodeService.class.getSimpleName() + ".event: " + userid + " forceCloseWebSocket from " + targetAddress);
        if (localEngine == null) return CompletableFuture.completedFuture(0);
        return CompletableFuture.completedFuture(localEngine.forceCloseLocalWebSocket(userid));
    }
605334
@Override
	protected final void initChannel(Channel ch) throws Exception {
		ChannelPipeline pipeline=ch.pipeline();
		if (sslCtx != null) {
			pipeline.addLast(sslCtx.newHandler(ch.alloc(),host,port));
        }
		pipeline.addLast(new HttpClientCodec());
		pipeline.addLast(new ChunkedWriteHandler());
		pipeline.addLast(new HttpObjectAggregator(64*1024));
		pipeline.addLast(new WebSocketClientProtocolHandler(WebSocketClientHandshakerFactory.newHandshaker(webSocketURL, WebSocketVersion.V13, subprotocol, false, new DefaultHttpHeaders())));
        pipeline.addLast(new WebSocketConnectedClientHandler());//连接成功监听handler
	}
843434
public static String doPost(String url, Map<String, String> header, Map<String, String> param)
            throws KeyManagementException, NoSuchAlgorithmException, IOException {

        String result = null;
        HttpClient httpClient = new SSLClient();
        httpClient.getParams().setParameter(CoreConnectionPNames.CONNECTION_TIMEOUT, Constant.DEFAULT_CONN_TIMEOUT);
        httpClient.getParams().setParameter(CoreConnectionPNames.SO_TIMEOUT, Constant.DEFAULT_READ_TIMEOUT);
        HttpPost httpPost = new HttpPost(url);
        // 设置Header
        if (header != null) {
            Set<String> keys = header.keySet();
            for (String key : keys) httpPost.setHeader(key, header.get(key));
        }
        //设置参数
        List<NameValuePair> list = new ArrayList<NameValuePair>();
        Iterator iterator = param.entrySet().iterator();
        while (iterator.hasNext()) {
            Entry<String, String> elem = (Entry<String, String>) iterator.next();
            list.add(new BasicNameValuePair(elem.getKey(), elem.getValue()));
        }
        if (list.size() > 0) {
            UrlEncodedFormEntity entity = new UrlEncodedFormEntity(list, Constant.DEFAULT_CHARSET);
            httpPost.setEntity(entity);
        }
        HttpResponse response = httpClient.execute(httpPost);
        if (response != null) {
            HttpEntity resEntity = response.getEntity();
            if (resEntity != null) {
                result = EntityUtils.toString(resEntity, Constant.DEFAULT_CHARSET);
            }
        }
        return result;
    }
897674
public static BatchGetResult batchGet(String accessToken, String postJson) {
		HttpUriRequest httpUriRequest = RequestBuilder
				.post()
				.setHeader(jsonHeader)
				.setUri(BASE_URI + "/card/batchget")
				.addParameter(PARAM_ACCESS_TOKEN, API.accessToken(accessToken))
				.setEntity(new StringEntity(postJson, Charset.forName("utf-8")))
				.build();
		return LocalHttpClient.executeJsonResult(httpUriRequest,
				BatchGetResult.class);
	}
691999
public void queryMessages(final int limit, final AVIMMessagesQueryCallback callback) {
    if (limit <= 0 || limit > 1000) {
      if (callback != null) {
        callback.internalDone(null, new AVException(new IllegalArgumentException(
                "limit should be in [1, 1000]")));
      }
    }
    // 如果屏蔽了本地缓存则全部走网络
    if (!AVIMOptions.getGlobalOptions().isMessageQueryCacheEnabled()) {
      queryMessagesFromServer(null, 0, limit, null, 0, new AVIMMessagesQueryCallback() {

        @Override
        public void done(List<AVIMMessage> messages, AVIMException e) {
          if (callback != null) {
            if (e != null) {
              callback.internalDone(e);
            } else {
              callback.internalDone(messages, null);
            }
          }
        }
      });
      return;
    }
    if (!AppConfiguration.getGlobalNetworkingDetector().isConnected()) {
      queryMessagesFromCache(null, 0, limit, callback);
    } else {
      // 选择最后一条有 breakpoint 为 false 的消息做截断，因为是 true 的话，会造成两次查询。
      // 在 queryMessages 还是遇到 breakpoint，再次查询了
      long cacheMessageCount = storage.getMessageCount(conversationId);
      long toTimestamp = 0;
      String toMsgId = null;
      // 如果本地的缓存的量都不够的情况下，应该要去服务器查询，以免第一次查询的时候出现limit跟返回值不一致让用户认为聊天记录已经到头的问题
      if (cacheMessageCount >= limit) {
        final AVIMMessage latestMessage =
                storage.getLatestMessageWithBreakpoint(conversationId, false);

        if (latestMessage != null) {
          toMsgId = latestMessage.getMessageId();
          toTimestamp = latestMessage.getTimestamp();
        }
      }

      // 去服务器查询最新消息，看是否在其它终端产生过消息。为省流量，服务器会截断至 toMsgId 、toTimestamp
      queryMessagesFromServer(null, 0, limit, toMsgId, toTimestamp,
              new AVIMMessagesQueryCallback() {
                @Override
                public void done(List<AVIMMessage> messages, AVIMException e) {
                  if (e != null) {
                    // 如果遇到本地错误或者网络错误，直接返回缓存数据
                    if (e.getCode() == AVIMException.TIMEOUT || e.getCode() == 0 || e.getCode() == 3000) {
                      queryMessagesFromCache(null, 0, limit, callback);
                    } else {
                      if (callback != null) {
                        callback.internalDone(e);
                      }
                    }
                  } else {
                    if (null == messages || messages.size() < 1) {
                      // 这种情况就说明我们的本地消息缓存是最新的
                    } else {
                      /*
                       * 1.messages.size()<=limit && messages.contains(latestMessage)
                       * 这种情况就说明在本地客户端退出后，该用户在其他客户端也产生了聊天记录而没有缓存到本地来,且产生了小于一页的聊天记录
                       * 2.messages==limit && !messages.contains(latestMessage)
                       * 这种情况就说明在本地客户端退出后，该用户在其他客户端也产生了聊天记录而没有缓存到本地来,且产生了大于一页的聊天记录
                       */

                      processContinuousMessages(messages);
                    }
                    queryMessagesFromCache(null, 0, limit, callback);
                  }
                }
              });
    }
  }
50518
@Override
	public void run() {
		int maxNewConnections;

		while (run) {
			maxNewConnections = 0;
			try {
				if (this.lazyInit) { // 无视第一次的信号量
					this.thriftConnectionPartition.getPoolWatchThreadSignalQueue().take();
				}

				maxNewConnections = this.thriftConnectionPartition.getMaxConnections()
						- this.thriftConnectionPartition.getCreatedConnections();

				while (maxNewConnections == 0 || (this.thriftConnectionPartition.getAvailableConnections() * 100
						/ this.thriftConnectionPartition.getMaxConnections() > this.poolAvailabilityThreshold)) {
					if (maxNewConnections == 0) {
						this.thriftConnectionPartition.setUnableToCreateMoreTransactions(true);
					}

					this.thriftConnectionPartition.getPoolWatchThreadSignalQueue().take();
					if (!run) {
						break;
					}
					maxNewConnections = this.thriftConnectionPartition.getMaxConnections()
							- this.thriftConnectionPartition.getCreatedConnections();

				}

				if (maxNewConnections > 0 && !this.thriftConnectionPool.poolShuttingDown) {
					fillConnections(Math.min(maxNewConnections, this.thriftConnectionPartition.getAcquireIncrement()));

					if (this.thriftConnectionPartition.getCreatedConnections() < this.thriftConnectionPartition
							.getMinConnections()) {
						fillConnections(this.thriftConnectionPartition.getMinConnections()
								- this.thriftConnectionPartition.getCreatedConnections());

					}
				}

				if (this.thriftConnectionPool.poolShuttingDown) {
					return;
				}

			} catch (InterruptedException e) {
				// 发生中断事件 停止运行监听器
				if (logger.isDebugEnabled()) {
					logger.debug("Terminating pool watch thread");
				}
				return;
			}
		}
	}
171371
public static String get(String url,
                             Map<String, String> header,
                             Map<String, String> param) throws IOException {

        // 参数
        if (param != null) {
            url += "?" + Utils.getUrlParamsByMap(param);
        }

        HttpURLConnection httpUrlConn = httpUrlConn(url);
        httpUrlConn.setDoInput(true);
        httpUrlConn.setRequestMethod(Constant.RequestMethodGet);

        // header
        if (header != null) {
            header(header, httpUrlConn);
        }

        httpUrlConn.connect();

        return readIO(httpUrlConn);
    }
143703
private int checkAck() throws IOException {
        logger.trace( "wait for ack" );
        int b = inputStream.read();
        logger.debug( "ack response: '{}'", b );

        if ( b == 1 || b == 2 ) {
            StringBuilder sb = new StringBuilder();
            int c;
            while ( (c = inputStream.read()) != '\n' ) {
                sb.append( (char) c );
            }
            if ( b == 1 || b == 2 ) {
                throw new IOException( sb.toString() );
            }
        }

        return b;
    }
1299323
public List<String> fetch(int offset, int limit) {
        E.illegalArgumentIf(offset < 0, "offset must not be negative number");
        E.illegalArgumentIf(offset >= lines(), "offset is out of range: " + offset);
        E.illegalArgumentIf(limit < 1, "limit must be at least 1");
        BufferedReader reader = IO.buffered(IO.reader(file));
        try {
            for (int i = 0; i < offset; ++i) {
                if (null == reader.readLine()) {
                    break;
                }
            }
        } catch (IOException e) {
            throw E.ioException(e);
        }
        List<String> lines = new ArrayList<>();
        try {
            for (int i = 0; i < limit; ++i) {
                String line = reader.readLine();
                if (null == line) {
                    break;
                }
                lines.add(line);
            }
        } catch (IOException e) {
            throw E.ioException(e);
        }
        return lines;
    }
197487
public static Callbackip getcallbackip(String access_token){
		HttpUriRequest httpUriRequest = RequestBuilder.get()
				.setUri(BASE_URI + "/cgi-bin/getcallbackip")
				.addParameter(PARAM_ACCESS_TOKEN,API.accessToken(access_token))
				.build();
		return LocalHttpClient.executeJsonResult(httpUriRequest,Callbackip.class);
	}
363121
public static String get(String host, String path, Map<String, String> headers, String param) {

        CloseableHttpClient httpClient = null;
        CloseableHttpResponse response = null;

        try {
            // host校验，一定不能为空，否则无法进行请求
            ExceptionUtil.notNull(host);

            // 创建HttpClient
            httpClient = wrapClient(host);

            // 创建HttpGet
            HttpGet request = new HttpGet(buildUrl(host, path, param));
            defaultHeader(request);

            // header处理
            if (MapUtil.isNotEmpty(headers)) {
                for (Map.Entry<String, String> e : headers.entrySet()) {
                    request.addHeader(e.getKey(), e.getValue());
                }
            }

            // 响应结果
            response = httpClient.execute(request);

            // 响应结果处理响应内容
            String result = null;

            if (response != null) {
                HttpEntity resEntity = response.getEntity();
                if (resEntity != null) {
                    result = EntityUtils.toString(resEntity, Charset.UTF_8);
                }
            }

            return result;
        } catch (IOException | ParseException e) {
            e.printStackTrace();
        } finally {
            close(httpClient, response);
        }
        return null;
    }