callback);
+
+ /**
+ * 搜索功能
+ * 构建自己的SearchRequest,然后获取SearchResponse
+ * 示例:
+ *
+ *
+ * SearchQuery searchQuery = new SearchQuery();
+ * TermQuery termQuery = new TermQuery();
+ * termQuery.setFieldName("user_name");
+ * termQuery.setTerm("jay");
+ * searchQuery.setQuery(termQuery);
+ * SearchRequest searchRequest = new SearchRequest(tableName, indexName, searchQuery);
+ * SearchResponse resp = ots.search(searchRequest);
+ *
+ *
+ * @param request 进行搜索所需的参数,详见{@link SearchRequest}
+ * @param callback 请求完成后调用的回调函数,可以为null,则代表不需要执行回调函数
+ * @return 搜索结果,详见{@link SearchResponse}
+ * @throws TableStoreException TableStore服务返回的异常
+ * @throws ClientException 请求的返回结果无效、或遇到网络异常
+ */
+ public Future search(
+ SearchRequest request, TableStoreCallback callback);
+
+
+ /**
+ * 转换成同步接口的Client。
+ * @return 同步Client
+ */
+ public SyncClientInterface asSyncClient();
+
+ /**
+ * 释放资源。
+ * 请确保在所有请求执行完毕之后释放资源。释放资源之后将不能再发送请求,正在执行的请求可能无法返回结果。
+ */
+ public void shutdown();
+
+ /**
+ * Switch CredentialsProvider。
+ *
+ * @param newCrdsProvider new CredentialsProvider, see {@link com.alicloud.openservices.tablestore.core.auth.CredentialsProviderFactory}.
+ */
+ public void switchCredentialsProvider(CredentialsProvider newCrdsProvider);
+}
diff --git a/src/main/java/com/alicloud/openservices/tablestore/ClientConfiguration.java b/src/main/java/com/alicloud/openservices/tablestore/ClientConfiguration.java
new file mode 100755
index 0000000..c461a84
--- /dev/null
+++ b/src/main/java/com/alicloud/openservices/tablestore/ClientConfiguration.java
@@ -0,0 +1,406 @@
+package com.alicloud.openservices.tablestore;
+
+import com.alicloud.openservices.tablestore.core.utils.Preconditions;
+import com.alicloud.openservices.tablestore.model.DefaultRetryStrategy;
+import com.alicloud.openservices.tablestore.model.RetryStrategy;
+
+/**
+ * 访问阿里云服务的客户端配置。
+ */
+public class ClientConfiguration {
+ private static final int AVAILABLE_PROCESSORS = Runtime.getRuntime().availableProcessors();
+
+ private int ioThreadCount = AVAILABLE_PROCESSORS;
+ private int maxConnections = 300;
+ private int socketTimeoutInMillisecond = 30 * 1000;
+ private int connectionTimeoutInMillisecond = 30 * 1000;
+ private int retryThreadCount = 1;
+
+ private boolean enableRequestCompression = false;
+ private boolean enableResponseCompression = false;
+ private boolean enableResponseValidation = true;
+ private boolean enableResponseContentMD5Checking = false;
+ private RetryStrategy retryStrategy = new DefaultRetryStrategy();
+ private int timeThresholdOfServerTracer = 500;
+ private int timeThresholdOfTraceLogger = 1000;
+
+ private String proxyHost;
+ private int proxyPort;
+ private String proxyUsername;
+ private String proxyPassword;
+ private String proxyDomain;
+ private String proxyWorkstation;
+
+ private long syncClientWaitFutureTimeoutInMillis = 60 * 1000;
+
+ /**
+ * 构造新实例。
+ */
+ public ClientConfiguration() {
+ }
+
+
+ /**
+ * 设置是否需要对请求数据进行压缩
+ * @param enableRequestCompression
+ * 是否需要压缩(true/false)
+ */
+ public void setEnableRequestCompression(boolean enableRequestCompression) {
+ this.enableRequestCompression = enableRequestCompression;
+ }
+
+ /**
+ * 返回是否需要对请求数据进行压缩
+ * @return 是否需要对request进行压缩
+ */
+ public boolean isEnableRequestCompression() {
+ return enableRequestCompression;
+ }
+
+ /**
+ * 是否需要告知TableStore对返回的响应内容进行压缩
+ * @param enableResponseCompression
+ * 是否告知TableStore对返回的响应内容进行压缩
+ */
+ public void setEnableResponseCompression(boolean enableResponseCompression) {
+ this.enableResponseCompression = enableResponseCompression;
+ }
+
+ /**
+ * 返回是否需要告知TableStore对返回的响应内容进行压缩
+ * @return
+ * 是否需要告知TableStore对返回的响应内容进行压缩
+ */
+ public boolean isEnableResponseCompression() {
+ return enableResponseCompression;
+ }
+
+ /**
+ * 返回是否需要对响应进行验证
+ * @return
+ * 是否需要对响应进行验证
+ */
+ public boolean isEnableResponseValidation() {
+ return enableResponseValidation;
+ }
+
+ /**
+ * 是否需要对响应进行验证, 如果需要验证,
+ * Client会验证头信息完整性、结果是否过期、授权信息是否正确
+ * @param enableResponseValidation
+ * 是否需要对响应进行验证
+ */
+ public void setEnableResponseValidation(boolean enableResponseValidation) {
+ this.enableResponseValidation = enableResponseValidation;
+ }
+
+ /**
+ * 返回HttpAsyncClient的IOReactor的线程数。
+ *
+ * @return IOReactor的线程数
+ */
+ public int getIoThreadCount() {
+ return ioThreadCount;
+ }
+
+ /**
+ * 设置HttpAsyncClient的IOReactor的线程数(因为采用的是异步IO,所以不需要配置大量线程,每个线程都能提供大量并发)。
+ *
+ * @param ioThreadCount IOReactor的线程数
+ */
+ public void setIoThreadCount(int ioThreadCount) {
+ Preconditions.checkArgument(ioThreadCount > 0, "The io thread count must be greater than 0.");
+ this.ioThreadCount = ioThreadCount;
+ }
+
+ /**
+ * 返回允许打开的最大HTTP连接数。
+ *
+ * @return 最大HTTP连接数。
+ */
+ public int getMaxConnections() {
+ return maxConnections;
+ }
+
+ /**
+ * 设置允许打开的最大HTTP连接数(连接数影响并发度,需要根据QPS、单个请求的时间、机器配置等因素来确定一个合理的值)。
+ *
+ * @param maxConnections 最大HTTP连接数。
+ */
+ public void setMaxConnections(int maxConnections) {
+ Preconditions.checkArgument(maxConnections > 0, "The max connection must be greater than 0.");
+ this.maxConnections = maxConnections;
+ }
+
+ /**
+ * 返回通过打开的连接传输数据的超时时间(单位:毫秒)。 0表示无限等待(但不推荐使用)。受系统超时时间的限制。
+ *
+ * @return 通过打开的连接传输数据的超时时间(单位:毫秒)。
+ */
+ public int getSocketTimeoutInMillisecond() {
+ return socketTimeoutInMillisecond;
+ }
+
+ /**
+ * 设置通过打开的连接传输数据的超时时间(单位:毫秒)。 0表示无限等待(但不推荐使用)。受系统超时时间的限制。
+ *
+ * @param socketTimeoutInMillisecond 通过打开的连接传输数据的超时时间(单位:毫秒)。
+ */
+ public void setSocketTimeoutInMillisecond(int socketTimeoutInMillisecond) {
+ Preconditions.checkArgument(socketTimeoutInMillisecond > 0, "The socket timeout must be greater than 0.");
+ this.socketTimeoutInMillisecond = socketTimeoutInMillisecond;
+ }
+
+ /**
+ * 返回建立连接的超时时间(单位:毫秒)。0表示无限等待(但不推荐使用)。受系统超时时间的限制。
+ *
+ * @return 建立连接的超时时间(单位:毫秒)。
+ */
+ public int getConnectionTimeoutInMillisecond() {
+ return connectionTimeoutInMillisecond;
+ }
+
+ /**
+ * 设置建立连接的超时时间(单位:毫秒)。0表示无限等待(但不推荐使用)。受系统超时时间的限制。
+ *
+ * @param connectionTimeoutInMillisecond 建立连接的超时时间(单位:毫秒)。
+ */
+ public void setConnectionTimeoutInMillisecond(int connectionTimeoutInMillisecond) {
+ Preconditions.checkArgument(connectionTimeoutInMillisecond > 0, "The connection timeout must be greater than 0.");
+ this.connectionTimeoutInMillisecond = connectionTimeoutInMillisecond;
+ }
+
+ /**
+ * 返回用于执行错误重试的线程池的线程的个数。该线程池为一个ScheduledExecutorService。
+ *
+ * @return 用于执行错误重试的线程池的线程的个数。
+ */
+ public int getRetryThreadCount() {
+ return retryThreadCount;
+ }
+
+ /**
+ * 设置用于执行错误重试的线程池的线程的个数。该线程池为一个ScheduledExecutorService。
+ *
+ * @param retryThreadCount 用于执行错误重试的线程池的线程的个数。
+ */
+ public void setRetryThreadCount(int retryThreadCount) {
+ Preconditions.checkArgument(retryThreadCount > 0, "The retry thread count must be greater than 0.");
+ this.retryThreadCount = retryThreadCount;
+ }
+
+ /**
+ * 返回是否需要对响应内容做MD5校验
+ *
+ * @return 是否需要对MD5内容做MD5校验
+ */
+ public boolean isEnableResponseContentMD5Checking() {
+ return enableResponseContentMD5Checking;
+ }
+
+ /**
+ * 是否需要对响应的内容做MD5校验, 如果需要校验,
+ * Client会计算响应数据的MD5值并与返回的响应头中的x-ots-contentmd5头的值进行比对
+ *
+ * @param enableResponseContentMD5Checking 是否需要校验响应数据MD5
+ */
+ public void setEnableResponseContentMD5Checking(
+ boolean enableResponseContentMD5Checking) {
+ this.enableResponseContentMD5Checking = enableResponseContentMD5Checking;
+ }
+
+ /**
+ * 返回TableStore的请求重试策略
+ *
+ * @return 请求重试策略
+ */
+ public RetryStrategy getRetryStrategy() {
+ return retryStrategy;
+ }
+
+ /**
+ * 设置TableStore的请求重试策略
+ *
+ * @param retryStrategy TableStore的请求重试策略
+ */
+ public void setRetryStrategy(RetryStrategy retryStrategy) {
+ Preconditions.checkArgument(retryStrategy != null, "The retry strategy should not be null.");
+ this.retryStrategy = retryStrategy;
+ }
+
+ /**
+ * 返回当前设置的时间阈值(单位:毫秒)。
+ * 当一个请求的总执行时间(包含重试占用的时间)超过该阈值时,SDK会记录一条WARN级别的日志。
+ * 该功能依赖于日志相关的配置。
+ *
+ * @return 时间阈值
+ */
+ public int getTimeThresholdOfTraceLogger() {
+ return timeThresholdOfTraceLogger;
+ }
+
+ /**
+ * 设置一个时间阈值(单位:毫秒)。
+ * 当一个请求的总执行时间(包含重试占用的时间)超过该阈值时,SDK会记录一条WARN级别的日志。
+ * 该功能依赖于日志相关的配置。
+ *
+ * @param timeThresholdOfTraceLogger trace logger
+ */
+ public void setTimeThresholdOfTraceLogger(int timeThresholdOfTraceLogger) {
+ Preconditions.checkArgument(timeThresholdOfTraceLogger > 0, "The time threshold of trace logger must be greater than 0.");
+ this.timeThresholdOfTraceLogger = timeThresholdOfTraceLogger;
+ }
+
+ /**
+ * 返回当前设置的服务端Tracer时间阈值(单位:毫秒)。
+ * 当一个请求在服务端的执行时间超过该阈值时,SDK会收到服务端的tracer信息并记录。
+ * 该功能依赖于服务端相关配置。
+ *
+ * @return 时间阈值
+ */
+ public int getTimeThresholdOfServerTracer() {
+ return timeThresholdOfServerTracer;
+ }
+
+ /**
+ * 设置服务端Tracer的时间阈值(单位:毫秒)。
+ * 当一个请求在服务端的执行时间超过该阈值时,SDK会收到服务端的tracer信息并记录。
+ * 该功能依赖于服务端相关配置
+ *
+ * @param timeThresholdOfServerTracer trace logger
+ */
+ public void setTimeThresholdOfServerTracer(int timeThresholdOfServerTracer) {
+ Preconditions.checkArgument(timeThresholdOfServerTracer > 0, "The time threshold of server tracer must be greater than 0.");
+ this.timeThresholdOfServerTracer = timeThresholdOfServerTracer;
+ }
+
+ /**
+ * 返回代理服务器主机地址。
+ *
+ * @return 代理服务器主机地址。
+ */
+ public String getProxyHost() {
+ return proxyHost;
+ }
+
+ /**
+ * 设置代理服务器主机地址。
+ *
+ * @param proxyHost
+ * 代理服务器主机地址。
+ */
+ public void setProxyHost(String proxyHost) {
+ this.proxyHost = proxyHost;
+ }
+
+ /**
+ * 返回代理服务器端口。
+ *
+ * @return 代理服务器端口。
+ */
+ public int getProxyPort() {
+ return proxyPort;
+ }
+
+ /**
+ * 设置代理服务器端口。
+ *
+ * @param proxyPort
+ * 代理服务器端口。
+ */
+ public void setProxyPort(int proxyPort) {
+ this.proxyPort = proxyPort;
+ }
+
+ /**
+ * 返回代理服务器验证的用户名。
+ *
+ * @return 用户名。
+ */
+ public String getProxyUsername() {
+ return proxyUsername;
+ }
+
+ /**
+ * 设置代理服务器验证的用户名。
+ *
+ * @param proxyUsername
+ * 用户名。
+ */
+ public void setProxyUsername(String proxyUsername) {
+ this.proxyUsername = proxyUsername;
+ }
+
+ /**
+ * 返回代理服务器验证的密码。
+ *
+ * @return 密码。
+ */
+ public String getProxyPassword() {
+ return proxyPassword;
+ }
+
+ /**
+ * 设置代理服务器验证的密码。
+ *
+ * @param proxyPassword
+ * 密码。
+ */
+ public void setProxyPassword(String proxyPassword) {
+ this.proxyPassword = proxyPassword;
+ }
+
+ /**
+ * 返回访问NTLM验证的代理服务器的Windows域名(可选)。
+ *
+ * @return 域名。
+ */
+ public String getProxyDomain() {
+ return proxyDomain;
+ }
+
+ /**
+ * 设置访问NTLM验证的代理服务器的Windows域名(可选)。
+ *
+ * @param proxyDomain
+ * 域名。
+ */
+ public void setProxyDomain(String proxyDomain) {
+ this.proxyDomain = proxyDomain;
+ }
+
+ /**
+ * 返回NTLM代理服务器的Windows工作站名称。
+ *
+ * @return NTLM代理服务器的Windows工作站名称。
+ */
+ public String getProxyWorkstation() {
+ return proxyWorkstation;
+ }
+
+ /**
+ * 设置NTLM代理服务器的Windows工作站名称。 (可选,如果代理服务器非NTLM,不需要设置该参数)。
+ *
+ * @param proxyWorkstation
+ * NTLM代理服务器的Windows工作站名称。
+ */
+ public void setProxyWorkstation(String proxyWorkstation) {
+ this.proxyWorkstation = proxyWorkstation;
+ }
+
+ /**
+ * 获取同步Client内等待异步调用返回的最大超时时间。
+ * @return
+ */
+ public long getSyncClientWaitFutureTimeoutInMillis() {
+ return syncClientWaitFutureTimeoutInMillis;
+ }
+
+ /**
+ * 设置同步Client内等待异步调用返回的最大超时时间。
+ * @param syncClientWaitFutureTimeoutInMillis
+ */
+ public void setSyncClientWaitFutureTimeoutInMillis(long syncClientWaitFutureTimeoutInMillis) {
+ this.syncClientWaitFutureTimeoutInMillis = syncClientWaitFutureTimeoutInMillis;
+ }
+}
diff --git a/src/main/java/com/alicloud/openservices/tablestore/ClientException.java b/src/main/java/com/alicloud/openservices/tablestore/ClientException.java
new file mode 100755
index 0000000..39d2f5e
--- /dev/null
+++ b/src/main/java/com/alicloud/openservices/tablestore/ClientException.java
@@ -0,0 +1,87 @@
+package com.alicloud.openservices.tablestore;
+
+/**
+ * {@link ClientException} 发生在请求的返回结果无效或遇到网络异常。
+ * TraceId: 在SDK内部提供的日志中会追踪某个请求的各个阶段的执行信息,可以通过这个ID来追踪日志排查问题。
+ */
+public class ClientException extends RuntimeException {
+
+ private static final long serialVersionUID = 1870835486798448798L;
+
+ /**
+ * 在SDK内部提供的日志中会追踪某个请求的各个阶段的执行信息,可以通过这个ID来追踪日志排查问题。
+ */
+ private String traceId;
+
+ /**
+ * 构造新实例。
+ */
+ public ClientException() {
+ super();
+ }
+
+ /**
+ * 构造函数。
+ *
+ * @param message 异常信息
+ */
+ public ClientException(String message) {
+ super(message);
+ }
+
+ /**
+ * 构造函数。
+ *
+ * @param cause 异常原因
+ */
+ public ClientException(Throwable cause) {
+ super(cause);
+ }
+
+ /**
+ * 构造函数。
+ *
+ * @param message 异常信息
+ * @param traceId TraceId
+ */
+ public ClientException(String message, String traceId) {
+ super(message);
+ this.traceId = traceId;
+ }
+
+ /**
+ * 构造函数。
+ *
+ * @param message 异常信息
+ * @param cause 异常原因
+ */
+ public ClientException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ /**
+ * 构造函数。
+ *
+ * @param message 异常信息
+ * @param cause 异常原因
+ * @param traceId TraceId
+ */
+ public ClientException(String message, Throwable cause, String traceId) {
+ super(message, cause);
+ this.traceId = traceId;
+ }
+
+ /**
+ * 返回TraceId。
+ *
+ * @return TraceId。
+ */
+ public String getTraceId() {
+ return traceId;
+ }
+
+ public void setTraceId(String traceId) {
+ this.traceId = traceId;
+ }
+
+}
diff --git a/src/main/java/com/alicloud/openservices/tablestore/DefaultTableStoreWriter.java b/src/main/java/com/alicloud/openservices/tablestore/DefaultTableStoreWriter.java
new file mode 100644
index 0000000..d52adc2
--- /dev/null
+++ b/src/main/java/com/alicloud/openservices/tablestore/DefaultTableStoreWriter.java
@@ -0,0 +1,273 @@
+package com.alicloud.openservices.tablestore;
+
+import com.alicloud.openservices.tablestore.model.*;
+import com.alicloud.openservices.tablestore.core.utils.ParamChecker;
+import com.alicloud.openservices.tablestore.core.utils.Preconditions;
+import com.alicloud.openservices.tablestore.writer.*;
+import com.lmax.disruptor.InsufficientCapacityException;
+import com.lmax.disruptor.RingBuffer;
+import com.lmax.disruptor.dsl.Disruptor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.Timer;
+import java.util.TimerTask;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
+
+public class DefaultTableStoreWriter implements TableStoreWriter {
+ private Logger logger = LoggerFactory.getLogger(TableStoreWriter.class);
+
+ private AsyncClientInterface ots;
+
+ private Executor executor;
+
+ private WriterConfig writerConfig;
+
+ private TableStoreCallback callback;
+
+ private TableStoreCallback resultCallback;
+
+ private String tableName;
+
+ private TableMeta tableMeta;
+
+ private Timer flushTimer;
+
+ private Disruptor disruptor;
+
+ private RingBuffer ringBuffer;
+
+ private RowChangeEventHandler eventHandler;
+
+ private ExecutorService disruptorExecutor;
+
+ private DefaultWriterStatistics writerStatistics;
+
+ private AtomicBoolean closed = new AtomicBoolean(false);
+
+ public DefaultTableStoreWriter(AsyncClientInterface ots, String tableName, WriterConfig config, TableStoreCallback callback, Executor executor) {
+ Preconditions.checkNotNull(ots, "The ots client can not be null.");
+ Preconditions.checkArgument(tableName != null && !tableName.isEmpty(), "The table name can not be null or empty.");
+ Preconditions.checkNotNull(executor, "The executor service can not be null.");
+ this.writerStatistics = new DefaultWriterStatistics();
+ this.ots = ots;
+ this.tableName = tableName;
+ this.writerConfig = config;
+ this.callback = callback;
+ this.resultCallback = createResultCallback(callback);
+ this.executor = executor;
+ flushTimer = new Timer();
+
+ initialize();
+ closed.set(false);
+ }
+
+ private TableStoreCallback createResultCallback(final TableStoreCallback callback) {
+ if (callback != null) {
+ return new TableStoreCallback() {
+ @Override
+ public void onCompleted(RowChange req, RowWriteResult res) {
+ callback.onCompleted(req, res.getConsumedCapacity());
+ }
+
+ @Override
+ public void onFailed(RowChange req, Exception ex) {
+ callback.onFailed(req, ex);
+ }
+ };
+ } else {
+ return null;
+ }
+ }
+
+ private void initialize() {
+ logger.info("Start initialize ots writer, table name: {}.", tableName);
+ DescribeTableRequest request = new DescribeTableRequest();
+ request.setTableName(tableName);
+ Future result = ots.describeTable(request, null);
+ DescribeTableResponse res = null;
+ try {
+ res = result.get();
+ } catch (Exception e) {
+ throw new ClientException(e);
+ }
+ this.tableMeta = res.getTableMeta();
+ logger.info("End initialize with table meta: {}.", tableMeta);
+
+ RowChangeEvent.RowChangeEventFactory factory = new RowChangeEvent.RowChangeEventFactory();
+
+ // start flush thread, we only need one event handler, so we just set a thread pool with fixed size 1.
+ disruptorExecutor = Executors.newFixedThreadPool(1);
+ disruptor = new Disruptor(factory, writerConfig.getBufferSize(), disruptorExecutor);
+ ringBuffer = disruptor.getRingBuffer();
+ eventHandler = new RowChangeEventHandler(ots, writerConfig, resultCallback, executor, writerStatistics);
+ disruptor.handleEventsWith(eventHandler);
+ disruptor.start();
+
+ // start flush timer
+ startFlushTimer(writerConfig.getFlushInterval());
+ }
+
+ public void startFlushTimer(int flushInterval) {
+
+ this.flushTimer.cancel();
+
+ this.flushTimer = new Timer();
+ this.flushTimer.scheduleAtFixedRate(new TimerTask() {
+ @Override
+ public void run() {
+ triggerFlush();
+ }
+ }, flushInterval, flushInterval);
+ }
+
+ @Override
+ public void addRowChange(RowChange rowChange) {
+ if (writerConfig.isEnableSchemaCheck()) {
+ ParamChecker.checkRowChange(tableMeta, rowChange, writerConfig);
+ }
+
+ while (true) {
+ if (!addRowChangeInternal(rowChange)) {
+ try {
+ Thread.sleep(1);
+ } catch (InterruptedException exp) {
+ }
+ } else {
+ break;
+ }
+ }
+ }
+
+ @Override
+ public boolean tryAddRowChange(RowChange rowChange) {
+ if (writerConfig.isEnableSchemaCheck()) {
+ ParamChecker.checkRowChange(tableMeta, rowChange, writerConfig);
+ }
+
+ return addRowChangeInternal(rowChange);
+ }
+
+ public boolean addRowChangeInternal(RowChange rowChange) {
+ if (closed.get()) {
+ throw new ClientException("The writer has been closed.");
+ }
+
+ try {
+ long sequence = ringBuffer.tryNext();
+ RowChangeEvent event = ringBuffer.get(sequence);
+ event.setValue(rowChange);
+ ringBuffer.publish(sequence);
+ return true;
+ } catch (InsufficientCapacityException e) {
+ return false;
+ }
+ }
+
+ private void addSignal(CountDownLatch latch) {
+ while (true) {
+ try {
+ long sequence = ringBuffer.tryNext();
+ RowChangeEvent event = ringBuffer.get(sequence);
+ event.setValue(latch);
+ ringBuffer.publish(sequence);
+ return;
+ } catch (InsufficientCapacityException e) {
+ try {
+ Thread.sleep(1);
+ } catch (InterruptedException exp) {
+ }
+ }
+ }
+ }
+
+ @Override
+ public void addRowChange(List rowChanges, List dirtyRows) throws ClientException {
+ dirtyRows.clear();
+ for (RowChange rowChange : rowChanges) {
+ try {
+ addRowChange(rowChange);
+ } catch (ClientException e) {
+ dirtyRows.add(rowChange);
+ }
+ }
+ if (!dirtyRows.isEmpty()) {
+ throw new ClientException("There is dirty rows.");
+ }
+ }
+
+ @Override
+ public void setCallback(final TableStoreCallback callback) {
+ this.callback = callback;
+ this.resultCallback = createResultCallback(callback);
+ eventHandler.setCallback(resultCallback);
+ }
+
+ @Override
+ public void setResultCallback(TableStoreCallback resultCallback) {
+ this.callback = null;
+ this.resultCallback = resultCallback;
+ eventHandler.setCallback(resultCallback);
+ }
+
+ @Override
+ public TableStoreCallback getCallback() {
+ return this.callback;
+ }
+
+ @Override
+ public TableStoreCallback getResultCallback() {
+ return this.resultCallback;
+ }
+
+
+ @Override
+ public WriterConfig getWriterConfig() {
+ return writerConfig;
+ }
+
+ @Override
+ public WriterStatistics getWriterStatistics() {
+ return writerStatistics;
+ }
+
+ private CountDownLatch triggerFlush() {
+ CountDownLatch latch = new CountDownLatch(1);
+ addSignal(latch);
+ return latch;
+ }
+
+ @Override
+ public synchronized void flush() throws ClientException {
+ logger.debug("trigger flush and waiting.");
+ if (closed.get()) {
+ throw new ClientException("The writer has been closed.");
+ }
+
+ CountDownLatch latch = triggerFlush();
+ try {
+ latch.await();
+ } catch (InterruptedException e) {
+ throw new ClientException(e);
+ }
+ logger.debug("user trigger flush finished.");
+ }
+
+ @Override
+ public synchronized void close() {
+ if (closed.get()) {
+ throw new ClientException("The writer has already been closed.");
+ }
+
+ flushTimer.cancel();
+ flush();
+ disruptor.shutdown();
+ disruptorExecutor.shutdown();
+
+ closed.set(true);
+ }
+}
diff --git a/src/main/java/com/alicloud/openservices/tablestore/InternalClient.java b/src/main/java/com/alicloud/openservices/tablestore/InternalClient.java
new file mode 100644
index 0000000..ac4a235
--- /dev/null
+++ b/src/main/java/com/alicloud/openservices/tablestore/InternalClient.java
@@ -0,0 +1,1212 @@
+package com.alicloud.openservices.tablestore;
+
+import java.util.Map;
+import java.util.Random;
+import java.util.UUID;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.alicloud.openservices.tablestore.core.auth.CredentialsProvider;
+import com.alicloud.openservices.tablestore.core.auth.CredentialsProviderFactory;
+import com.alicloud.openservices.tablestore.core.utils.Preconditions;
+import com.alicloud.openservices.tablestore.core.auth.ServiceCredentials;
+import com.alicloud.openservices.tablestore.core.http.AsyncServiceClient;
+import com.alicloud.openservices.tablestore.core.*;
+import com.alicloud.openservices.tablestore.model.*;
+import com.alicloud.openservices.tablestore.model.search.*;
+import com.alicloud.openservices.tablestore.model.tunnel.CreateTunnelRequest;
+import com.alicloud.openservices.tablestore.model.tunnel.CreateTunnelResponse;
+import com.alicloud.openservices.tablestore.model.tunnel.DeleteTunnelRequest;
+import com.alicloud.openservices.tablestore.model.tunnel.DeleteTunnelResponse;
+import com.alicloud.openservices.tablestore.model.tunnel.DescribeTunnelRequest;
+import com.alicloud.openservices.tablestore.model.tunnel.DescribeTunnelResponse;
+import com.alicloud.openservices.tablestore.model.tunnel.ListTunnelRequest;
+import com.alicloud.openservices.tablestore.model.tunnel.ListTunnelResponse;
+import com.alicloud.openservices.tablestore.model.tunnel.internal.CheckpointRequest;
+import com.alicloud.openservices.tablestore.model.tunnel.internal.CheckpointResponse;
+import com.alicloud.openservices.tablestore.model.tunnel.internal.ConnectTunnelRequest;
+import com.alicloud.openservices.tablestore.model.tunnel.internal.ConnectTunnelResponse;
+import com.alicloud.openservices.tablestore.model.tunnel.internal.GetCheckpointRequest;
+import com.alicloud.openservices.tablestore.model.tunnel.internal.GetCheckpointResponse;
+import com.alicloud.openservices.tablestore.model.tunnel.internal.HeartbeatRequest;
+import com.alicloud.openservices.tablestore.model.tunnel.internal.HeartbeatResponse;
+import com.alicloud.openservices.tablestore.model.tunnel.internal.ReadRecordsRequest;
+import com.alicloud.openservices.tablestore.model.tunnel.internal.ReadRecordsResponse;
+import com.alicloud.openservices.tablestore.model.tunnel.internal.ShutdownTunnelRequest;
+import com.alicloud.openservices.tablestore.model.tunnel.internal.ShutdownTunnelResponse;
+
+public class InternalClient {
+ private static int AVAILABLE_PROCESSORS = Runtime.getRuntime().availableProcessors();
+ private String endpoint; // TableStore endpoint
+ private String instanceName; // 实例的名称
+ private CredentialsProvider crdsProvider; // 用户身份信息。
+ private AsyncServiceClient httpClient;
+ private ScheduledExecutorService retryExecutor;
+ private ExecutorService callbackExecutor; // 用于执行Callback
+ private ClientConfiguration clientConfig;
+ private RetryStrategy retryStrategy;
+ private LauncherFactory launcherFactory;
+ private Random random = new Random();
+
+ /**
+ * 使用指定的TableStore Endpoint和默认配置构造一个新的{@link AsyncClient}实例。
+ *
+ * @param endpoint
+ * TableStore服务的endpoint。
+ * @param accessKeyId
+ * 访问TableStore服务的Access ID。
+ * @param accessKeySecret
+ * 访问TableStore服务的Access Key。
+ * @param instanceName
+ * 访问TableStore服务的实例名称。
+ */
+ public InternalClient(String endpoint, String accessKeyId, String accessKeySecret, String instanceName) {
+ this(endpoint, accessKeyId, accessKeySecret, instanceName, null);
+ }
+
+ /**
+ * 使用指定的TableStore Endpoint和默认配置构造一个新的{@link AsyncClient}实例。
+ *
+ * @param endpoint
+ * TableStore服务的endpoint。
+ * @param accessKeyId
+ * 访问TableStore服务的Access ID。
+ * @param accessKeySecret
+ * 访问TableStore服务的Access Key。
+ * @param instanceName
+ * 访问TableStore服务的实例名称。
+ * @param config
+ * 客户端配置信息({@link ClientConfiguration})。 如果传入null则使用默认配置。
+ */
+ public InternalClient(String endpoint, String accessKeyId, String accessKeySecret, String instanceName,
+ ClientConfiguration config) {
+ this(endpoint, accessKeyId, accessKeySecret, instanceName, config, null);
+ }
+
+ /**
+ * 使用指定的TableStore Endpoint和默认配置构造一个新的{@link AsyncClient}实例。
+ *
+ * @param endpoint
+ * TableStore服务的endpoint。
+ * @param accessKeyId
+ * 访问TableStore服务的Access ID。
+ * @param accessKeySecret
+ * 访问TableStore服务的Access Key。
+ * @param instanceName
+ * 访问TableStore服务的实例名称。
+ * @param config
+ * 客户端配置信息({@link ClientConfiguration})。 如果传入null则使用默认配置。
+ * @param callbackExecutor
+ * 用于执行用户在调用异步接口时传入的Callback。如果传入null则使用默认配置( 线程数与CPU核数相同的线程池)。
+ */
+ public InternalClient(String endpoint, String accessKeyId, String accessKeySecret, String instanceName,
+ ClientConfiguration config, ExecutorService callbackExecutor) {
+ this(endpoint, accessKeyId, accessKeySecret, instanceName, config, callbackExecutor, null);
+ }
+
+ /**
+ * 使用指定的TableStore Endpoint和默认配置构造一个新的{@link AsyncClient}实例。
+ *
+ * @param endpoint
+ * TableStore服务的endpoint。
+ * @param accessKeyId
+ * 访问TableStore服务的Access ID。
+ * @param accessKeySecret
+ * 访问TableStore服务的Access Key。
+ * @param instanceName
+ * 访问TableStore服务的实例名称。
+ * @param config
+ * 客户端配置信息({@link ClientConfiguration})。 如果传入null则使用默认配置。
+ * @param callbackExecutor
+ * 用于执行用户在调用异步接口时传入的Callback。如果传入null则使用默认配置( 线程数与CPU核数相同的线程池)。
+ * @param stsToken
+ * Sts Token.
+ */
+ public InternalClient(String endpoint, String accessKeyId, String accessKeySecret, String instanceName,
+ ClientConfiguration config, ExecutorService callbackExecutor, String stsToken) {
+ this(endpoint, CredentialsProviderFactory.newDefaultCredentialProvider(accessKeyId, accessKeySecret, stsToken),
+ instanceName, config, callbackExecutor);
+
+
+ }
+
+ public InternalClient(String endpoint, CredentialsProvider credsProvider, String instanceName,
+ ClientConfiguration config, ExecutorService callbackExecutor) {
+ Preconditions.checkArgument(endpoint != null && !endpoint.isEmpty(),
+ "The end point should not be null or empty.");
+ Preconditions.checkArgument(instanceName != null && !instanceName.isEmpty(),
+ "The name of instance should not be null or empty.");
+ Preconditions.checkArgument(instanceName.length() == instanceName.getBytes(Constants.UTF8_CHARSET).length,
+ "InstanceName should not have multibyte character.");
+
+ if (!endpoint.startsWith("http://") && !endpoint.startsWith("https://")) {
+ throw new IllegalArgumentException("the endpoint must start with \"http://\" or \"https://\".");
+ }
+ this.endpoint = endpoint;
+
+ this.crdsProvider = credsProvider;
+
+ if (config == null) {
+ config = new ClientConfiguration();
+ }
+
+ this.clientConfig = config;
+
+ this.httpClient = new AsyncServiceClient(config);
+
+ this.retryExecutor = Executors.newScheduledThreadPool(config.getRetryThreadCount(),
+ new ThreadFactory() {
+ private final AtomicInteger counter = new AtomicInteger(1);
+ @Override
+ public Thread newThread(Runnable r) {
+ return new Thread(r, "tablestore-retry-scheduled-" + counter.getAndIncrement());
+ }
+ });
+
+ this.retryStrategy = config.getRetryStrategy();
+
+ this.instanceName = instanceName;
+
+ if (callbackExecutor != null) {
+ this.callbackExecutor = callbackExecutor;
+ } else {
+ this.callbackExecutor = Executors.newFixedThreadPool(AVAILABLE_PROCESSORS,
+ new ThreadFactory() {
+ private final AtomicInteger counter = new AtomicInteger(1);
+ @Override
+ public Thread newThread(Runnable r) {
+ return new Thread(r, "tablestore-callback-" + counter.getAndIncrement());
+ }
+ });
+ }
+
+ this.launcherFactory = new LauncherFactory(endpoint, instanceName, httpClient, crdsProvider, config);
+ }
+
+ public void setExtraHeaders(Map extraHeaders) {
+ this.httpClient.setExtraHeaders(extraHeaders);
+ }
+
+ /**
+ * 返回访问的TableStore Endpoint。
+ *
+ * @return TableStore Endpoint。
+ */
+ public String getEndpoint() {
+ return endpoint;
+ }
+
+ /**
+ * 返回访问的实例的名称
+ *
+ * @return instance name
+ */
+ public String getInstanceName() {
+ return instanceName;
+ }
+
+
+ /**
+ * 返回Client配置。
+ * @return
+ */
+ public ClientConfiguration getClientConfig() {
+ return clientConfig;
+ }
+
+ private TraceLogger getTraceLogger() {
+ String traceId = new UUID(random.nextLong(), (new Random()).nextLong()).toString();
+ return new TraceLogger(traceId, this.clientConfig.getTimeThresholdOfTraceLogger());
+ }
+
+ public Future listTable(TableStoreCallback callback) {
+ ListTableRequest request = new ListTableRequest();
+ TraceLogger tracer = getTraceLogger();
+ RetryStrategy retry = this.retryStrategy.clone();
+ ListTableLauncher launcher = launcherFactory.listTable(tracer, retry, request);
+
+ AsyncCompletion completion;
+ completion = new AsyncCompletion(launcher, request, tracer,
+ callbackExecutor, retry, retryExecutor);
+ CallbackImpledFuture f = new CallbackImpledFuture();
+ completion.watchBy(f);
+ if (callback != null) {
+ // user callback must be triggered after completion of the return
+ // future.
+ f.watchBy(callback);
+ }
+
+ launcher.fire(request, completion);
+
+ return f;
+ }
+
+ public Future createTable(CreateTableRequest request,
+ TableStoreCallback callback) {
+ Preconditions.checkNotNull(request);
+
+ TraceLogger tracer = getTraceLogger();
+ RetryStrategy retry = this.retryStrategy.clone();
+ CreateTableLauncher launcher = launcherFactory.createTable(tracer, retry, request);
+
+ AsyncCompletion completion = new AsyncCompletion(
+ launcher, request, tracer, callbackExecutor, retry, retryExecutor);
+ CallbackImpledFuture f = new CallbackImpledFuture();
+ completion.watchBy(f);
+ if (callback != null) {
+ // user callback must be triggered after completion of the return
+ // future.
+ f.watchBy(callback);
+ }
+
+ launcher.fire(request, completion);
+
+ return f;
+ }
+
+ public Future describeTable(DescribeTableRequest request,
+ TableStoreCallback callback) {
+ Preconditions.checkNotNull(request);
+
+ TraceLogger tracer = getTraceLogger();
+ RetryStrategy retry = this.retryStrategy.clone();
+ DescribeTableLauncher launcher = launcherFactory.describeTable(tracer, retry, request);
+
+ AsyncCompletion completion;
+ completion = new AsyncCompletion(launcher, request, tracer,
+ callbackExecutor, retry, retryExecutor);
+ CallbackImpledFuture f = new CallbackImpledFuture();
+ completion.watchBy(f);
+ if (callback != null) {
+ // user callback must be triggered after completion of the return
+ // future.
+ f.watchBy(callback);
+ }
+
+ launcher.fire(request, completion);
+
+ return f;
+ }
+
+ public Future deleteTable(DeleteTableRequest request,
+ TableStoreCallback callback) {
+ Preconditions.checkNotNull(request);
+
+ TraceLogger tracer = getTraceLogger();
+ RetryStrategy retry = this.retryStrategy.clone();
+ DeleteTableLauncher launcher = launcherFactory.deleteTable(tracer, retry, request);
+
+ AsyncCompletion completion;
+ completion = new AsyncCompletion(launcher, request, tracer,
+ callbackExecutor, retry, retryExecutor);
+ CallbackImpledFuture f = new CallbackImpledFuture();
+ completion.watchBy(f);
+ if (callback != null) {
+ // user callback must be triggered after completion of the return
+ // future.
+ f.watchBy(callback);
+ }
+
+ launcher.fire(request, completion);
+
+ return f;
+ }
+
+ public Future updateTable(UpdateTableRequest request,
+ TableStoreCallback callback) {
+ Preconditions.checkNotNull(request);
+
+ TraceLogger tracer = getTraceLogger();
+ RetryStrategy retry = this.retryStrategy.clone();
+ UpdateTableLauncher launcher = launcherFactory.updateTable(tracer, retry, request);
+
+ AsyncCompletion completion;
+ completion = new AsyncCompletion(launcher, request, tracer,
+ callbackExecutor, retry, retryExecutor);
+ CallbackImpledFuture f = new CallbackImpledFuture();
+ completion.watchBy(f);
+ if (callback != null) {
+ // user callback must be triggered after completion of the return
+ // future.
+ f.watchBy(callback);
+ }
+
+ launcher.fire(request, completion);
+
+ return f;
+ }
+
+ public Future createIndex(CreateIndexRequest request,
+ TableStoreCallback callback) {
+ Preconditions.checkNotNull(request);
+
+ TraceLogger tracer = getTraceLogger();
+ RetryStrategy retry = this.retryStrategy.clone();
+ CreateIndexLauncher launcher = launcherFactory.createIndex(tracer, retry, request);
+
+ AsyncCompletion completion = new AsyncCompletion(
+ launcher, request, tracer, callbackExecutor, retry, retryExecutor);
+ CallbackImpledFuture f = new CallbackImpledFuture();
+ completion.watchBy(f);
+ if (callback != null) {
+ // user callback must be triggered after completion of the return
+ // future.
+ f.watchBy(callback);
+ }
+
+ launcher.fire(request, completion);
+
+ return f;
+ }
+
+ public Future deleteIndex(DeleteIndexRequest request,
+ TableStoreCallback callback) {
+ Preconditions.checkNotNull(request);
+
+ TraceLogger tracer = getTraceLogger();
+ RetryStrategy retry = this.retryStrategy.clone();
+ DeleteIndexLauncher launcher = launcherFactory.deleteIndex(tracer, retry, request);
+
+ AsyncCompletion completion = new AsyncCompletion(
+ launcher, request, tracer, callbackExecutor, retry, retryExecutor);
+ CallbackImpledFuture f = new CallbackImpledFuture();
+ completion.watchBy(f);
+ if (callback != null) {
+ // user callback must be triggered after completion of the return
+ // future.
+ f.watchBy(callback);
+ }
+
+ launcher.fire(request, completion);
+
+ return f;
+ }
+
+ public Future getRowInternal(GetRowRequest request,
+ TableStoreCallback callback) {
+ Preconditions.checkNotNull(request);
+
+ TraceLogger tracer = getTraceLogger();
+ RetryStrategy retry = this.retryStrategy.clone();
+ GetRowLauncher launcher = launcherFactory.getRow(tracer, retry, request);
+
+ AsyncCompletion completion = new AsyncCompletion(
+ launcher, request, tracer, callbackExecutor, retry, retryExecutor);
+ CallbackImpledFuture f = new CallbackImpledFuture();
+ completion.watchBy(f);
+ if (callback != null) {
+ // user callback must be triggered after completion of the return
+ // future.
+ f.watchBy(callback);
+ }
+
+ launcher.fire(request, completion);
+
+ return f;
+ }
+
+ public Future getRow(GetRowRequest request,
+ TableStoreCallback callback) {
+ Preconditions.checkNotNull(request);
+
+ TraceLogger tracer = getTraceLogger();
+ RetryStrategy retry = this.retryStrategy.clone();
+ GetRowLauncher launcher = launcherFactory.getRow(tracer, retry, request);
+
+ AsyncGetRowCompletion completion = new AsyncGetRowCompletion(launcher, request, tracer, callbackExecutor, retry,
+ retryExecutor);
+ CallbackImpledFuture f = new CallbackImpledFuture();
+ completion.watchBy(f);
+ if (callback != null) {
+ // user callback must be triggered after completion of the return
+ // future.
+ f.watchBy(callback);
+ }
+
+ launcher.fire(request, completion);
+
+ return f;
+ }
+
+ public Future putRow(PutRowRequest request,
+ TableStoreCallback callback) {
+ Preconditions.checkNotNull(request);
+
+ TraceLogger tracer = getTraceLogger();
+ RetryStrategy retry = this.retryStrategy.clone();
+ PutRowLauncher launcher = launcherFactory.putRow(tracer, retry, request);
+
+ AsyncCompletion completion = new AsyncCompletion(
+ launcher, request, tracer, callbackExecutor, retry, retryExecutor);
+ CallbackImpledFuture f = new CallbackImpledFuture();
+ completion.watchBy(f);
+ if (callback != null) {
+ // user callback must be triggered after completion of the return
+ // future.
+ f.watchBy(callback);
+ }
+
+ launcher.fire(request, completion);
+
+ return f;
+ }
+
+ public Future updateRow(UpdateRowRequest request,
+ TableStoreCallback callback) {
+ Preconditions.checkNotNull(request);
+
+ TraceLogger tracer = getTraceLogger();
+ RetryStrategy retry = this.retryStrategy.clone();
+ UpdateRowLauncher launcher = launcherFactory.updateRow(tracer, retry, request);
+
+ AsyncCompletion completion = new AsyncCompletion(
+ launcher, request, tracer, callbackExecutor, retry, retryExecutor);
+ CallbackImpledFuture f = new CallbackImpledFuture();
+ completion.watchBy(f);
+ if (callback != null) {
+ // user callback must be triggered after completion of the return
+ // future.
+ f.watchBy(callback);
+ }
+
+ launcher.fire(request, completion);
+
+ return f;
+ }
+
+ public Future deleteRow(DeleteRowRequest request,
+ TableStoreCallback callback) {
+ Preconditions.checkNotNull(request);
+
+ TraceLogger tracer = getTraceLogger();
+ RetryStrategy retry = this.retryStrategy.clone();
+ DeleteRowLauncher launcher = launcherFactory.deleteRow(tracer, retry, request);
+
+ AsyncCompletion completion = new AsyncCompletion(
+ launcher, request, tracer, callbackExecutor, retry, retryExecutor);
+ CallbackImpledFuture f = new CallbackImpledFuture();
+ completion.watchBy(f);
+ if (callback != null) {
+ // user callback must be triggered after completion of the return
+ // future.
+ f.watchBy(callback);
+ }
+
+ launcher.fire(request, completion);
+
+ return f;
+ }
+
+ public Future batchGetRowInternal(BatchGetRowRequest request,
+ TableStoreCallback callback) {
+ Preconditions.checkNotNull(request);
+
+ TraceLogger tracer = getTraceLogger();
+ RetryStrategy retry = this.retryStrategy.clone();
+ BatchGetRowLauncher launcher = launcherFactory.batchGetRow(tracer, retry, request);
+
+ AsyncCompletion completion = new AsyncCompletion(
+ launcher, request, tracer, callbackExecutor, retry, retryExecutor);
+ CallbackImpledFuture f = new CallbackImpledFuture();
+ completion.watchBy(f);
+ if (callback != null) {
+ // user callback must be triggered after completion of the return
+ // future.
+ f.watchBy(callback);
+ }
+
+ launcher.fire(request, completion);
+
+ return f;
+ }
+
+ public Future batchGetRow(BatchGetRowRequest request,
+ TableStoreCallback callback) {
+ Preconditions.checkNotNull(request);
+
+ TraceLogger tracer = getTraceLogger();
+ RetryStrategy retry = this.retryStrategy.clone();
+ BatchGetRowLauncher launcher = launcherFactory.batchGetRow(tracer, retry, request);
+
+ AsyncBatchGetRowCompletion completion = new AsyncBatchGetRowCompletion(launcher, request, tracer,
+ callbackExecutor, retry, retryExecutor);
+ CallbackImpledFuture f = new CallbackImpledFuture();
+ completion.watchBy(f);
+ if (callback != null) {
+ // user callback must be triggered after completion of the return
+ // future.
+ f.watchBy(callback);
+ }
+
+ launcher.fire(request, completion);
+
+ return f;
+ }
+
+ public Future batchWriteRow(BatchWriteRowRequest request,
+ TableStoreCallback callback) {
+ Preconditions.checkNotNull(request);
+
+ TraceLogger tracer = getTraceLogger();
+ RetryStrategy retry = this.retryStrategy.clone();
+ BatchWriteRowLauncher launcher = launcherFactory.batchWriteRow(tracer, retry, request);
+
+ AsyncCompletion completion = new AsyncCompletion(
+ launcher, request, tracer, callbackExecutor, retry, retryExecutor);
+ CallbackImpledFuture f = new CallbackImpledFuture();
+ completion.watchBy(f);
+ if (callback != null) {
+ // user callback must be triggered after completion of the return
+ // future.
+ f.watchBy(callback);
+ }
+
+ launcher.fire(request, completion);
+
+ return f;
+ }
+
+ public Future getRangeInternal(GetRangeRequest request,
+ TableStoreCallback callback) {
+ Preconditions.checkNotNull(request);
+
+ TraceLogger tracer = getTraceLogger();
+ RetryStrategy retry = this.retryStrategy.clone();
+ GetRangeLauncher launcher = launcherFactory.getRange(tracer, retry, request);
+
+ AsyncCompletion completion = new AsyncCompletion(
+ launcher, request, tracer, callbackExecutor, retry, retryExecutor);
+ CallbackImpledFuture f = new CallbackImpledFuture();
+ completion.watchBy(f);
+ if (callback != null) {
+ // user callback must be triggered after completion of the return
+ // future.
+ f.watchBy(callback);
+ }
+
+ launcher.fire(request, completion);
+
+ return f;
+ }
+
+ public Future getRange(GetRangeRequest request,
+ TableStoreCallback callback) {
+ Preconditions.checkNotNull(request);
+
+ TraceLogger tracer = getTraceLogger();
+ RetryStrategy retry = this.retryStrategy.clone();
+ GetRangeLauncher launcher = launcherFactory.getRange(tracer, retry, request);
+
+ AsyncGetRangeCompletion completion = new AsyncGetRangeCompletion(launcher, request, tracer, callbackExecutor,
+ retry, retryExecutor);
+ CallbackImpledFuture f = new CallbackImpledFuture();
+ completion.watchBy(f);
+ if (callback != null) {
+ // user callback must be triggered after completion of the return
+ // future.
+ f.watchBy(callback);
+ }
+
+ launcher.fire(request, completion);
+
+ return f;
+ }
+
+ public Future computeSplitsBySize(ComputeSplitsBySizeRequest request,
+ TableStoreCallback callback) {
+ Preconditions.checkNotNull(request);
+ Preconditions.checkStringNotNullAndEmpty(request.getTableName(),
+ "The table name for ComputeSplitsBySize should not be null or empty.");
+
+ TraceLogger tracer = getTraceLogger();
+ RetryStrategy retry = this.retryStrategy.clone();
+ ComputeSplitsBySizeLauncher launcher = launcherFactory.computeSplitsBySize(tracer, retry, request);
+
+ AsyncCompletion completion = new AsyncCompletion(
+ launcher, request, tracer, callbackExecutor, retry, retryExecutor);
+ CallbackImpledFuture f = new CallbackImpledFuture();
+ completion.watchBy(f);
+ if (callback != null) {
+ // user callback must be triggered after completion of the return
+ // future.
+ f.watchBy(callback);
+ }
+
+ launcher.fire(request, completion);
+
+ return f;
+ }
+
+ public void shutdown() {
+ this.retryExecutor.shutdownNow();
+ this.callbackExecutor.shutdownNow();
+ this.httpClient.shutdown();
+ }
+
+ public Future listStream(ListStreamRequest request,
+ TableStoreCallback callback) {
+ Preconditions.checkNotNull(request);
+
+ TraceLogger tracer = getTraceLogger();
+ RetryStrategy retry = this.retryStrategy.clone();
+ ListStreamLauncher launcher = launcherFactory.listStream(tracer, retry, request);
+
+ AsyncCompletion completion = new AsyncCompletion(
+ launcher, request, tracer, callbackExecutor, retry, retryExecutor);
+ CallbackImpledFuture f = new CallbackImpledFuture();
+ completion.watchBy(f);
+ if (callback != null) {
+ // user callback must be triggered after completion of the return
+ // future.
+ f.watchBy(callback);
+ }
+
+ launcher.fire(request, completion);
+
+ return f;
+ }
+
+ public Future describeStream(DescribeStreamRequest request,
+ TableStoreCallback callback) {
+ Preconditions.checkNotNull(request);
+
+ TraceLogger tracer = getTraceLogger();
+ RetryStrategy retry = this.retryStrategy.clone();
+ DescribeStreamLauncher launcher = launcherFactory.describeStream(tracer, retry, request);
+
+ AsyncCompletion completion = new AsyncCompletion(
+ launcher, request, tracer, callbackExecutor, retry, retryExecutor);
+ CallbackImpledFuture f = new CallbackImpledFuture();
+ completion.watchBy(f);
+ if (callback != null) {
+ // user callback must be triggered after completion of the return
+ // future.
+ f.watchBy(callback);
+ }
+
+ launcher.fire(request, completion);
+
+ return f;
+ }
+
+ public Future getShardIterator(GetShardIteratorRequest request,
+ TableStoreCallback callback) {
+ Preconditions.checkNotNull(request);
+
+ TraceLogger tracer = getTraceLogger();
+ RetryStrategy retry = this.retryStrategy.clone();
+ GetShardIteratorLauncher launcher = launcherFactory.getShardIterator(tracer, retry, request);
+
+ AsyncCompletion completion = new AsyncCompletion(
+ launcher, request, tracer, callbackExecutor, retry, retryExecutor);
+ CallbackImpledFuture f = new CallbackImpledFuture();
+ completion.watchBy(f);
+ if (callback != null) {
+ // user callback must be triggered after completion of the return
+ // future.
+ f.watchBy(callback);
+ }
+
+ launcher.fire(request, completion);
+
+ return f;
+ }
+
+ public Future getStreamRecord(GetStreamRecordRequest request,
+ TableStoreCallback callback) {
+ Preconditions.checkNotNull(request);
+
+ TraceLogger tracer = getTraceLogger();
+ RetryStrategy retry = this.retryStrategy.clone();
+ GetStreamRecordLauncher launcher = launcherFactory.getStreamRecord(tracer, retry, request);
+
+ AsyncCompletion completion = new AsyncCompletion(
+ launcher, request, tracer, callbackExecutor, retry, retryExecutor);
+ CallbackImpledFuture f = new CallbackImpledFuture();
+ completion.watchBy(f);
+ if (callback != null) {
+ // user callback must be triggered after completion of the return
+ // future.
+ f.watchBy(callback);
+ }
+
+ launcher.fire(request, completion);
+
+ return f;
+ }
+
+ public Future startLocalTransaction(StartLocalTransactionRequest request,
+ TableStoreCallback