Merge pull request #1344 from nats-io/kv-purge-per-message-ttl

KV Purge Per Message TTL

Author: scottf

src/main/java/io/nats/client/KeyValue.java

@@ -182,6 +182,27 @@ public interface KeyValue {
      */
     void purge(String key, long expectedRevision) throws IOException, JetStreamApiException;
 
+    /**
+     * Purge all values/history from the specific key
+     * @param key the key
+     * @param messageTtl the individual ttl for the key
+     * @throws IOException covers various communication issues with the NATS
+     *         server such as timeout or interruption
+     * @throws JetStreamApiException the request had an error related to the data
+     */
+    void purge(String key, MessageTtl messageTtl) throws IOException, JetStreamApiException;
+
+    /**
+     * Purge all values/history from the specific key iff the key exists and its last revision matches the expected
+     * @param key the key
+     * @param expectedRevision the expected last revision
+     * @param messageTtl the individual ttl for the key
+     * @throws IOException covers various communication issues with the NATS
+     *         server such as timeout or interruption
+     * @throws JetStreamApiException the request had an error related to the data
+     */
+    void purge(String key, long expectedRevision, MessageTtl messageTtl) throws IOException, JetStreamApiException;
+
     /**
      * Watch updates for a specific key.
      * @param key the key.

src/main/java/io/nats/client/impl/NatsKeyValue.java

@@ -233,6 +233,22 @@ public void purge(String key, long expectedRevision) throws IOException, JetStre
         _write(key, null, getPurgeHeaders(), getPublishOptions(expectedRevision, null));
     }
 
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public void purge(String key, MessageTtl messageTtl) throws IOException, JetStreamApiException {
+        _write(key, null, getPurgeHeaders(), getPublishOptions(-1, messageTtl));
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public void purge(String key, long expectedRevision, MessageTtl messageTtl) throws IOException, JetStreamApiException {
+        _write(key, null, getPurgeHeaders(), getPublishOptions(expectedRevision, messageTtl));
+    }
+
     private PublishAck _write(String key, byte[] data, Headers h, PublishOptions popts) throws IOException, JetStreamApiException {
         validateNonWildcardKvKeyRequired(key);
         return js.publish(NatsMessage.builder().subject(writeSubject(key)).data(data).headers(h).build(), popts);

src/test/java/io/nats/client/impl/KeyValueTests.java

@@ -1847,59 +1847,115 @@ public void testLimitMarker() throws Exception {
     public void testLimitMarkerAlso() throws Exception {
         jsServer.run(TestBase::atLeast2_11_2, nc -> {
             String bucket = bucket();
-            String key = key();
+            String key1 = key();
+            String key2 = key();
+            String key3 = key();
 
             KeyValueManagement kvm = nc.keyValueManagement();
             KeyValueConfiguration config = KeyValueConfiguration.builder()
                 .name(bucket)
                 .storageType(StorageType.Memory)
-                .limitMarker(Duration.ofSeconds(6))
+                .limitMarker(Duration.ofSeconds(5))
                 .build();
             kvm.create(config);
 
+            Dispatcher d = nc.createDispatcher();
+
             KeyValue kv = nc.keyValue(bucket);
 
-            AtomicInteger puts = new AtomicInteger();
-            AtomicInteger dels = new AtomicInteger();
-            AtomicInteger purge = new AtomicInteger();
-            AtomicInteger eod = new AtomicInteger();
+            AtomicInteger wPuts = new AtomicInteger();
+            AtomicInteger wDels = new AtomicInteger();
+            AtomicInteger wPurges = new AtomicInteger();
+            AtomicInteger wEod = new AtomicInteger();
 
             KeyValueWatcher watcher = new KeyValueWatcher() {
                 @Override
                 public void watch(KeyValueEntry keyValueEntry) {
                     if (keyValueEntry.getOperation() == KeyValueOperation.PUT) {
-                        puts.incrementAndGet();
+                        wPuts.incrementAndGet();
                     }
                     else if (keyValueEntry.getOperation() == KeyValueOperation.DELETE) {
-                        dels.incrementAndGet();
+                        wDels.incrementAndGet();
                     }
                     else if (keyValueEntry.getOperation() == KeyValueOperation.PURGE) {
-                        purge.incrementAndGet();
+                        wPurges.incrementAndGet();
                     }
                 }
 
                 @Override
                 public void endOfData() {
-                    eod.incrementAndGet();
+                    wEod.incrementAndGet();
                 }
             };
 
             NatsKeyValueWatchSubscription watch = kv.watchAll(watcher);
 
-            kv.create(key, dataBytes(), MessageTtl.seconds(2));
-
-            KeyValueEntry kve = kv.get(key);
-            assertNotNull(kve);
-
-            sleep(2100); // longer than the message ttl
-
-            kve = kv.get(key);
-            assertNull(kve);
+            AtomicInteger rMessages = new AtomicInteger();
+            AtomicInteger rPurges = new AtomicInteger();
+            AtomicInteger rMaxAges = new AtomicInteger();
+            AtomicInteger rTtl2 = new AtomicInteger();
+            AtomicInteger rTtl5 = new AtomicInteger();
+
+            MessageHandler rawHandler = msg -> {
+                rMessages.incrementAndGet();
+                if (msg.hasHeaders()) {
+                    String h = msg.getHeaders().getFirst("KV-Operation");
+                    if (h != null && h.equals("PURGE")) {
+                        rPurges.incrementAndGet();
+                    }
+                    h = msg.getHeaders().getFirst("Nats-Marker-Reason");
+                    if (h != null && h.equals("MaxAge")) {
+                        rMaxAges.incrementAndGet();
+                    }
+                    h = msg.getHeaders().getFirst("Nats-TTL");
+                    if (h != null) {
+                        if (h.equals("2s")) {
+                            rTtl2.incrementAndGet();
+                        }
+                        else {
+                            rTtl5.incrementAndGet();
+                        }
+                    }
+                }
+            };
 
-            assertEquals(1, puts.get());
-            assertEquals(0, dels.get());
-            assertEquals(1, purge.get());
-            assertEquals(1, eod.get());
+            JetStreamSubscription sub = nc.jetStream().subscribe(null, d, rawHandler, true,
+                PushSubscribeOptions.builder()
+                    .stream("KV_" + bucket)
+                    .configuration(ConsumerConfiguration.builder().filterSubject(">")
+                        .build())
+                    .build());
+
+            kv.create(key1, dataBytes(), MessageTtl.seconds(2));
+            kv.create(key2, dataBytes());
+            kv.create(key3, dataBytes());
+
+            assertNotNull(kv.get(key1));
+            assertNotNull(kv.get(key2));
+            assertNotNull(kv.get(key3));
+
+            kv.purge(key2, MessageTtl.seconds(2));
+            kv.purge(key3);
+
+            // This section will have to be modified if there are changes
+            // to how purge markers are handled (double purge on ttl purge, fix for no purge of non-ttl purge)
+            sleep(8000); // longer than the message ttl plus the limit marker since double purge plus some extra
+
+            assertNull(kv.get(key1));
+            assertNull(kv.get(key2));
+            assertNull(kv.get(key3));
+
+            // create and put
+            assertEquals(3, wPuts.get());
+            assertEquals(4, wPurges.get()); // the 2 message ttl purge markers, the manual purge and the manual purge's purge.
+            assertEquals(0, wDels.get());
+            assertEquals(1, wEod.get());
+
+            assertEquals(7, rMessages.get());
+            assertEquals(2, rPurges.get());
+            assertEquals(2, rMaxAges.get());
+            assertEquals(2, rTtl2.get());
+            assertEquals(2, rTtl5.get());
         });
     }
 }