diff --git a/src/main/java/net/spy/memcached/collection/BTreeCount.java b/src/main/java/net/spy/memcached/collection/BTreeCount.java index 5e5649e80..dcfbfc565 100644 --- a/src/main/java/net/spy/memcached/collection/BTreeCount.java +++ b/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() { diff --git a/src/main/java/net/spy/memcached/v2/AsyncArcusCommands.java b/src/main/java/net/spy/memcached/v2/AsyncArcusCommands.java index 0693dc6dc..e65389595 100644 --- a/src/main/java/net/spy/memcached/v2/AsyncArcusCommands.java +++ b/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 bopCount(String key, BKey from, BKey to, ElementFlagFilter eFlagFilter) { + verifyBKeyRange(from, to); + + AbstractArcusResult result = new AbstractArcusResult<>(new AtomicReference<>()); + ArcusFutureImpl 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; + } } diff --git a/src/main/java/net/spy/memcached/v2/AsyncArcusCommandsIF.java b/src/main/java/net/spy/memcached/v2/AsyncArcusCommandsIF.java index dd46e7361..1d83a5d34 100644 --- a/src/main/java/net/spy/memcached/v2/AsyncArcusCommandsIF.java +++ b/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> bopSortMergeGet(List keys, BKey from, BKey * or {@code initial} if the element did not exist. */ ArcusFuture 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 bopCount(String key, BKey from, BKey to, ElementFlagFilter eFlagFilter); } diff --git a/src/test/java/net/spy/memcached/v2/BTreeAsyncArcusCommandsTest.java b/src/test/java/net/spy/memcached/v2/BTreeAsyncArcusCommandsTest.java index 52a009c2b..c2b2a2a99 100644 --- a/src/test/java/net/spy/memcached/v2/BTreeAsyncArcusCommandsTest.java +++ b/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 elementWithFlag1 = new BTreeElement<>(BKey.of(1L), "value1", eFlag); + BTreeElement elementWithFlag2 = new BTreeElement<>(BKey.of(2L), "value2", eFlag); + BTreeElement 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); + } }