Merge pull request #1491 from nats-io/os-purge-on-put-replace

Object Store purge on re-put

Author: scottf

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

@@ -89,6 +89,9 @@ public ObjectInfo put(ObjectMeta meta, InputStream inputStream) throws IOExcepti
             throw OsLinkNotAllowOnPut.instance();
         }
 
+        ObjectInfo newInfo;
+        ObjectInfo oldInfo = getInfo(meta.getObjectName());
+
         String nuid = NUID.nextGlobal();
         String chunkSubject = rawChunkSubject(nuid);
 
@@ -122,7 +125,7 @@ public ObjectInfo put(ObjectMeta meta, InputStream inputStream) throws IOExcepti
                 red = inputStream.read(buffer);
             }
 
-            return publishMeta(ObjectInfo.builder(bucketName, meta)
+            newInfo = publishMeta(ObjectInfo.builder(bucketName, meta)
                 .size(totalSize)
                 .chunks(chunks)
                 .nuid(nuid)
@@ -135,12 +138,20 @@ public ObjectInfo put(ObjectMeta meta, InputStream inputStream) throws IOExcepti
                 jsm.purgeStream(streamName, PurgeOptions.subject(rawChunkSubject(nuid)));
             }
             catch (Exception ignore) {}
-
             throw e;
         }
         finally {
             try { inputStream.close(); } catch (IOException ignore) {}
         }
+
+        if (oldInfo != null) {
+            try {
+                jsm.purgeStream(streamName, PurgeOptions.builder().subject(rawChunkSubject(oldInfo.getNuid())).build());
+            }
+            catch (IOException | JetStreamApiException ignore) {}
+        }
+
+        return newInfo;
     }
 
     /**

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

@@ -238,6 +238,7 @@ private static Object[] getInput(int size) {
             if (f.isFile()) {
                 long flen = f.length();
                 if (flen == size) {
+                    foundLen = flen;
                     found = f;
                     break;
                 }
@@ -506,6 +507,42 @@ public void testCompression() throws Exception {
         });
     }
 
+    @Test
+    public void testOverwrite() throws Exception {
+        jsServer.run(TestBase::atLeast2_10, nc -> {
+            JetStreamManagement jsm = nc.jetStreamManagement();
+            JetStream js = jsm.jetStream();
+
+            String bucket = bucket();
+            String objName = variant();
+
+            ObjectStoreManagement osm = nc.objectStoreManagement();
+            ObjectStoreStatus oss = osm.create(ObjectStoreConfiguration.builder(bucket)
+                .storageType(StorageType.Memory)
+                .build());
+            String realStreamName = oss.getConfiguration().getBackingConfig().getName();
+
+            ObjectStore os = nc.objectStore(bucket);
+
+            Object[] input = getInput(4096 * 10);
+            File file = (File)input[1];
+
+            try (InputStream in = Files.newInputStream(file.toPath())) {
+                os.put(objName, in);
+            }
+            StreamInfo si = jsm.getStreamInfo(realStreamName, StreamInfoOptions.allSubjects());
+            long byteCount1 = si.getStreamState().getByteCount();
+
+            try (InputStream in = Files.newInputStream(file.toPath())) {
+                os.put(objName, in);
+            }
+            si = jsm.getStreamInfo(realStreamName, StreamInfoOptions.allSubjects());
+            long byteCount2 = si.getStreamState().getByteCount();
+
+            assertEquals(byteCount1, byteCount2);
+        });
+    }
+
     static class TestObjectStoreWatcher implements ObjectStoreWatcher {
         public String name;
         public List<ObjectInfo> entries = new ArrayList<>();