Issue-967

modules/api/src/main/java/org/apache/fluo/api/client/SnapshotBase.java

@@ -17,7 +17,9 @@
 
 import java.util.Collection;
 import java.util.Map;
+import java.util.PriorityQueue;
 import java.util.Set;
+import java.util.concurrent.CompletableFuture;
 
 import org.apache.fluo.api.client.scanner.ScannerBuilder;
 import org.apache.fluo.api.data.Bytes;
@@ -165,4 +167,14 @@ Map<String, Map<Column, String>> gets(Collection<? extends CharSequence> rows,
    * @return transactions start timestamp allocated from Oracle.
    */
   long getStartTimestamp();
+
+  default CompletableFuture<String> getsAsync(String row, Column column) {
+    return CompletableFuture.supplyAsync(() -> gets(row, column));
+  }
+
+  default CompletableFuture<String> getsAsync(String row, Column column, String defaultValue) {
+    return CompletableFuture.supplyAsync(() -> gets(row, column, defaultValue));
+  }
+
+
 }

modules/core/src/main/java/org/apache/fluo/core/impl/AsyncReader.java

@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.fluo.core.impl;
+
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.PriorityQueue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import org.apache.fluo.api.data.Bytes;
+import org.apache.fluo.api.data.Column;
+import org.apache.fluo.api.data.RowColumn;
+import org.apache.fluo.core.impl.TransactionImpl;
+
+public class AsyncReader {
+  private PriorityQueue<AsyncGet> getsQueue;
+  private ExecutorService executorService;
+  private TransactionImpl tx;
+
+  public AsyncReader(TransactionImpl tx) {
+    this.tx = tx;
+    getsQueue = new PriorityQueue<>();
+    executorService = Executors.newSingleThreadExecutor();
+  }
+
+  public CompletableFuture<String> gets(String row, Column column) {
+    return gets(row, column, null);
+  }
+
+  public CompletableFuture<String> gets(String row, Column column, String defaultValue) {
+    AsyncGet curAsyncGet = new AsyncGet(row, column, defaultValue);
+    getsQueue.add(curAsyncGet);
+    executorService.submit(() -> {
+      int i = 0;
+      Collection<RowColumn> rowColumns = new HashSet<>();
+      for (AsyncGet asyncGet : getsQueue) {
+        rowColumns.add(asyncGet.rc);
+      }
+      Map<RowColumn, Bytes> getsMap = tx.get(rowColumns);
+
+      while (!getsQueue.isEmpty()) {
+        AsyncGet asyncGet = getsQueue.poll();
+        Bytes bytesRes = getsMap.get(asyncGet.rc);
+        asyncGet.res.complete(bytesRes == null ? defaultValue : bytesRes.toString());
+      }
+    });
+    return curAsyncGet.res;
+  }
+
+  public void close() {
+
+  }
+
+  class AsyncGet {
+    RowColumn rc;
+    CompletableFuture<String> res;
+    String defaultValue;
+
+    public AsyncGet(String row, Column column, String defaultValue) {
+      rc = new RowColumn(row, column);
+      res = new CompletableFuture<>();
+      this.defaultValue = defaultValue;
+    }
+  }
+}

modules/core/src/main/java/org/apache/fluo/core/impl/TransactionImpl.java

@@ -26,6 +26,7 @@
 import java.util.Map.Entry;
 import java.util.Objects;
 import java.util.Set;
+import java.util.concurrent.CompletableFuture;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
@@ -69,6 +70,7 @@
 import org.apache.fluo.core.async.SyncCommitObserver;
 import org.apache.fluo.core.exceptions.AlreadyAcknowledgedException;
 import org.apache.fluo.core.exceptions.StaleScanException;
+import org.apache.fluo.core.impl.AsyncReader;
 import org.apache.fluo.core.impl.scanner.ScannerBuilderImpl;
 import org.apache.fluo.core.oracle.Stamp;
 import org.apache.fluo.core.util.ColumnUtil;
@@ -117,6 +119,8 @@ private static enum TxStatus {
   private TransactorNode tnode = null;
   private TxStatus status = TxStatus.OPEN;
   private boolean commitAttempted = false;
+  private AsyncReader asyncReader = null;
+
 
   // for testing
   private boolean stopAfterPreCommit = false;
@@ -265,6 +269,24 @@ private Map<Column, Bytes> getImpl(Bytes row, Set<Column> columns) {
     return ret;
   }
 
+  @Override
+  public CompletableFuture<String> getsAsync(String row, Column column) {
+    if (asyncReader == null) {
+      asyncReader = new AsyncReader(this);
+    }
+
+    return asyncReader.gets(row, column);
+  }
+
+  @Override
+  public CompletableFuture<String> getsAsync(String row, Column column, String defaultValue) {
+    if (asyncReader == null) {
+      asyncReader = new AsyncReader(this);
+    }
+
+    return asyncReader.gets(row, column, defaultValue);
+  }
+
   @Override
   public ScannerBuilder scanner() {
     checkIfOpen();

modules/core/src/test/java/org/apache/fluo/core/impl/TransactionImplTest.java

@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package org.apache.fluo.core.impl;
+
+import org.apache.fluo.api.client.FluoClient;
+import org.apache.fluo.api.client.FluoFactory;
+import org.apache.fluo.api.client.Transaction;
+import org.apache.fluo.api.config.FluoConfiguration;
+import org.apache.fluo.api.data.Column;
+import org.apache.fluo.core.client.FluoClientImpl;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class TransactionImplTest {
+  @Test
+  public void testBasic() {
+    //   FluoConfiguration fluoConfig = new FluoConfiguration();
+    //   // fluoConfig.setApplicationName(config.getApplicationName());
+    //   // fluoConfig.setInstanceZookeepers(config.getInstanceZookeepers());
+    //   Environment env = new Environment(fluoConfig);
+    //   TransactionBase tx = new TestTransaction(env);
+
+    //   /*try (FluoClient client = FluoFactory.newClient(fluoConfig);
+    //       Transaction tx = client.newTransaction()) {*/
+
+    //     tx.set("row1", new Column("col1"), "val1");
+    //     tx.set("row2", new Column("col2"), "val2");
+    //     tx.set("row3", new Column("col3"), "val3");
+
+    //     tx.commit();
+
+    //     try {
+    //       Assert.assertEquals("val1", tx.getsAsync("row1", new Column("col1")).get());
+    //       Assert.assertEquals("val2", tx.getsAsync("row2", new Column("col2")).get());
+    //       Assert.assertEquals("val3", tx.getsAsync("row3", new Column("col3")).get());
+
+    //       Assert.assertEquals("val4", tx.getsAsync("row4", new Column("col4"), "val4").get());
+    //     } catch (Exception e) {
+    //       System.out.println(e.toString());
+    //     }
+    //   /*}*/
+  }
+}