FEATURE: Add CompletableFuture BTree count API

src/main/java/net/spy/memcached/collection/BTreeCount.java

@@ -26,14 +26,17 @@ public class BTreeCount extends CollectionCount {
 
   protected final ElementFlagFilter elementFlagFilter;
 
-  public BTreeCount(long from, long to, ElementFlagFilter elementFlagFilter) {
+  public BTreeCount(String from, String to, ElementFlagFilter elementFlagFilter) {
     this.range = from + ".." + to;
     this.elementFlagFilter = elementFlagFilter;
   }
 
+  public BTreeCount(long from, long to, ElementFlagFilter elementFlagFilter) {
+    this(String.valueOf(from), String.valueOf(to), elementFlagFilter);
+  }
+
   public BTreeCount(byte[] from, byte[] to, ElementFlagFilter elementFlagFilter) {
-    this.range = BTreeUtil.toHex(from) + ".." + BTreeUtil.toHex(to);
-    this.elementFlagFilter = elementFlagFilter;
+    this(BTreeUtil.toHex(from), BTreeUtil.toHex(to), elementFlagFilter);
   }
 
   public String stringify() {

src/main/java/net/spy/memcached/v2/AsyncArcusCommands.java

@@ -34,6 +34,7 @@
 import net.spy.memcached.MemcachedClient;
 import net.spy.memcached.MemcachedNode;
 import net.spy.memcached.collection.BKeyObject;
+import net.spy.memcached.collection.BTreeCount;
 import net.spy.memcached.collection.BTreeCreate;
 import net.spy.memcached.collection.BTreeGet;
 import net.spy.memcached.collection.BTreeGetBulk;
@@ -48,10 +49,12 @@
 import net.spy.memcached.collection.BTreeUpdate;
 import net.spy.memcached.collection.BTreeUpsert;
 import net.spy.memcached.collection.CollectionAttributes;
+import net.spy.memcached.collection.CollectionCount;
 import net.spy.memcached.collection.CollectionCreate;
 import net.spy.memcached.collection.CollectionInsert;
 import net.spy.memcached.collection.CollectionMutate;
 import net.spy.memcached.collection.CollectionUpdate;
+import net.spy.memcached.collection.ElementFlagFilter;
 import net.spy.memcached.collection.ElementValueType;
 import net.spy.memcached.internal.result.GetsResultImpl;
 import net.spy.memcached.ops.APIType;
@@ -1423,4 +1426,44 @@ public void complete() {
 
     return future;
   }
+
+  public ArcusFuture<Long> bopCount(String key, BKey from, BKey to, ElementFlagFilter eFlagFilter) {
+    verifyBKeyRange(from, to);
+
+    AbstractArcusResult<Long> result = new AbstractArcusResult<>(new AtomicReference<>());
+    ArcusFutureImpl<Long> future = new ArcusFutureImpl<>(result);
+    CollectionCount collectionCount = new BTreeCount(from.toString(), to.toString(), eFlagFilter);
+    ArcusClient client = arcusClientSupplier.get();
+
+    OperationCallback cb = new OperationCallback() {
+      @Override
+      public void receivedStatus(OperationStatus status) {
+        switch (status.getStatusCode()) {
+          case SUCCESS:
+            long count = Long.parseLong(status.getMessage());
+            result.set(count);
+            break;
+          case ERR_NOT_FOUND:
+            result.set(null);
+            break;
+          case CANCELLED:
+            future.internalCancel();
+            break;
+          default:
+            /* TYPE_MISMATCH / BKEY_MISMATCH / UNREADABLE / NOT_SUPPORTED or unknown statement */
+            result.addError(key, status);
+        }
+      }
+
+      @Override
+      public void complete() {
+        future.complete();
+      }
+    };
+    Operation op = client.getOpFact().collectionCount(key, collectionCount, cb);
+    future.setOp(op);
+    client.addOp(key, op);
+
+    return future;
+  }
 }

src/main/java/net/spy/memcached/v2/AsyncArcusCommandsIF.java

@@ -22,6 +22,7 @@
 
 import net.spy.memcached.CASValue;
 import net.spy.memcached.collection.CollectionAttributes;
+import net.spy.memcached.collection.ElementFlagFilter;
 import net.spy.memcached.collection.ElementValueType;
 import net.spy.memcached.v2.vo.BKey;
 import net.spy.memcached.v2.vo.BTreeElement;
@@ -432,4 +433,16 @@ ArcusFuture<SMGetElements<T>> bopSortMergeGet(List<String> keys, BKey from, BKey
    *         or {@code initial} if the element did not exist.
    */
   ArcusFuture<Long> bopDecr(String key, BKey bKey, int delta, long initial, byte[] eFlag);
+
+  /**
+   * Count elements in a bKey range from a btree item.
+   *
+   * @param key         key of the btree item
+   * @param from        BKey range start (inclusive)
+   * @param to          BKey range end (inclusive)
+   * @param eFlagFilter eFlag filter condition, or {@code null} to count all elements in the range
+   * @return the number of elements in the range (0 if none exist),
+   * or {@code null} if the key is not found.
+   */
+  ArcusFuture<Long> bopCount(String key, BKey from, BKey to, ElementFlagFilter eFlagFilter);
 }

src/test/java/net/spy/memcached/v2/BTreeAsyncArcusCommandsTest.java

@@ -9,6 +9,7 @@
 
 import net.spy.memcached.collection.CollectionAttributes;
 import net.spy.memcached.collection.CollectionOverflowAction;
+import net.spy.memcached.collection.ElementFlagFilter;
 import net.spy.memcached.collection.ElementFlagUpdate;
 import net.spy.memcached.collection.ElementValueType;
 import net.spy.memcached.ops.OperationException;
@@ -1282,4 +1283,153 @@ void bopDecrBKeyMismatch() throws ExecutionException, InterruptedException, Time
             .toCompletableFuture()
             .get(300L, TimeUnit.MILLISECONDS);
   }
+
+  @Test
+  void bopCountSuccess() throws ExecutionException, InterruptedException, TimeoutException {
+    // given
+    String key = keys.get(0);
+
+    async.bopInsert(key, ELEMENTS.get(0), new CollectionAttributes())
+            .thenCompose(result -> {
+              assertTrue(result);
+              return async.bopInsert(key, ELEMENTS.get(1));
+            })
+            .thenCompose(result -> {
+              assertTrue(result);
+              return async.bopInsert(key, ELEMENTS.get(2));
+            })
+            .thenAccept(Assertions::assertTrue)
+            .toCompletableFuture()
+            .get(300L, TimeUnit.MILLISECONDS);
+
+    // when
+    BKey from = ELEMENTS.get(0).getBkey();
+    BKey to = ELEMENTS.get(2).getBkey();
+
+    async.bopCount(key, from, to, ElementFlagFilter.DO_NOT_FILTER)
+            // then
+            .thenAccept(result -> assertEquals(3L, result))
+            .toCompletableFuture()
+            .get(300L, TimeUnit.MILLISECONDS);
+  }
+
+  @Test
+  void bopCountEFlagFilter() throws ExecutionException, InterruptedException, TimeoutException {
+    // given
+    String key = keys.get(0);
+    byte[] eFlag = new byte[]{0x01};
+    BTreeElement<Object> elementWithFlag1 = new BTreeElement<>(BKey.of(1L), "value1", eFlag);
+    BTreeElement<Object> elementWithFlag2 = new BTreeElement<>(BKey.of(2L), "value2", eFlag);
+    BTreeElement<Object> elementWithoutFlag = new BTreeElement<>(BKey.of(3L), "value3", null);
+
+    async.bopInsert(key, elementWithFlag1, new CollectionAttributes())
+            .thenCompose(result -> {
+              assertTrue(result);
+              return async.bopInsert(key, elementWithFlag2);
+            })
+            .thenCompose(result -> {
+              assertTrue(result);
+              return async.bopInsert(key, elementWithoutFlag);
+            })
+            .thenAccept(Assertions::assertTrue)
+            .toCompletableFuture()
+            .get(300L, TimeUnit.MILLISECONDS);
+
+    // when
+    ElementFlagFilter filter = new ElementFlagFilter(ElementFlagFilter.CompOperands.Equal, eFlag);
+    BKey from = BKey.of(0L);
+    BKey to = BKey.of(10L);
+
+    async.bopCount(key, from, to, filter)
+            // then
+            .thenAccept(result -> assertEquals(2L, result))
+            .toCompletableFuture()
+            .get(300L, TimeUnit.MILLISECONDS);
+  }
+
+  @Test
+  void bopCountZero() throws ExecutionException, InterruptedException, TimeoutException {
+    // given
+    String key = keys.get(0);
+
+    async.bopCreate(key, ElementValueType.STRING, new CollectionAttributes())
+            .thenAccept(Assertions::assertTrue)
+            .toCompletableFuture()
+            .get(300L, TimeUnit.MILLISECONDS);
+
+    // when
+    BKey from = BKey.of(0L);
+    BKey to = BKey.of(100L);
+
+    async.bopCount(key, from, to, ElementFlagFilter.DO_NOT_FILTER)
+            // then
+            .thenAccept(result -> assertEquals(0L, result))
+            .toCompletableFuture()
+            .get(300L, TimeUnit.MILLISECONDS);
+  }
+
+  @Test
+  void bopCountNotFound() throws ExecutionException, InterruptedException, TimeoutException {
+    // given
+    String key = keys.get(0);
+    BKey from = BKey.of(0L);
+    BKey to = BKey.of(10L);
+
+    // when
+    async.bopCount(key, from, to, ElementFlagFilter.DO_NOT_FILTER)
+            // then
+            .thenAccept(Assertions::assertNull)
+            .toCompletableFuture()
+            .get(300L, TimeUnit.MILLISECONDS);
+  }
+
+  @Test
+  void bopCountTypeMismatch() throws ExecutionException, InterruptedException, TimeoutException {
+    // given
+    String key = keys.get(0);
+
+    async.set(key, 60, "invalid-type-value")
+            .thenAccept(Assertions::assertTrue)
+            .toCompletableFuture()
+            .get(300L, TimeUnit.MILLISECONDS);
+
+    // when
+    BKey from = BKey.of(0L);
+    BKey to = BKey.of(10L);
+
+    async.bopCount(key, from, to, ElementFlagFilter.DO_NOT_FILTER)
+            // then
+            .handle((result, ex) -> {
+              assertInstanceOf(OperationException.class, ex);
+              assertTrue(ex.getMessage().contains("TYPE_MISMATCH"));
+              return result;
+            })
+            .toCompletableFuture()
+            .get(300L, TimeUnit.MILLISECONDS);
+  }
+
+  @Test
+  void bopCountBKeyMismatch() throws ExecutionException, InterruptedException, TimeoutException {
+    // given
+    String key = keys.get(0);
+
+    async.bopInsert(key, ELEMENTS.get(0), new CollectionAttributes())
+            .thenAccept(Assertions::assertTrue)
+            .toCompletableFuture()
+            .get(300L, TimeUnit.MILLISECONDS);
+
+    // when
+    BKey from = BKey.of(new byte[]{0x00});
+    BKey to = BKey.of(new byte[]{0x10});
+
+    async.bopCount(key, from, to, ElementFlagFilter.DO_NOT_FILTER)
+            // then
+            .handle((result, ex) -> {
+              assertInstanceOf(OperationException.class, ex);
+              assertTrue(ex.getMessage().contains("BKEY_MISMATCH"));
+              return result;
+            })
+            .toCompletableFuture()
+            .get(300L, TimeUnit.MILLISECONDS);
+  }
 }