GC support

Author: kahing

clusterhq/queue/src/main/java/us/hxbc/clusterhq/queue/DataStore.java

@@ -35,6 +35,10 @@
  * LSN marks the beginning of the message, so if LSN 0 is a message that's
  * 5 bytes, it will still be in a file called "0". The next message will get
  * LSN 8 so that it will begin at offset 0 in the file "2".
+ *
+ * Garbage collection works by simply deleting chunks that are no longer in
+ * need. If a chunk may still be needed, we try to be conservative and not
+ * touch it. Behavior is undefined if you request a LSN that is already GC'ed.
  */
 public class DataStore {
     private Logger logger = LoggerFactory.getLogger(getClass());
@@ -138,6 +142,25 @@ public static class Message {
         }
     }
 
+    public void gc(long needLSN) {
+        long baseLSN = getBaseLSN(needLSN);
+        try {
+            Files.list(dir).forEach(p -> {
+                String name = p.getFileName().toString();
+                long lsn = Long.parseLong(name, 16);
+                if (lsn < baseLSN) {
+                    try {
+                        Files.delete(p);
+                    } catch (IOException e) {
+                        e.printStackTrace();
+                    }
+                }
+            });
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+    }
+
     public Message get(long lsn) throws IOException {
         long baseLSN = getBaseLSN(lsn);
         Path chunk = getChunkPath(baseLSN);

clusterhq/queue/src/test/java/us/hxbc/clusterhq/queue/DataStoreTest.java

@@ -109,6 +109,34 @@ public void testPostMany() throws Exception {
         assertThat(i).isEqualTo(10);
     }
 
+    @Test
+    public void testGCAll() throws Exception {
+        testPost2();
+        long lsn = ds.getNextLSN();
+        assertThat(Files.list(dir).count()).isEqualTo(1);
+        ds.gc(lsn);
+        assertThat(Files.list(dir).count()).isEqualTo(0);
+    }
+
+    @Test
+    public void testGCOneChunk() throws Exception {
+        testPost2Chunks();
+        long lsn = ds.getNextLSN();
+        assertThat(Files.list(dir).count()).isEqualTo(2);
+        ds.gc(lsn);
+        assertThat(Files.list(dir).count()).isEqualTo(1);
+    }
+
+    @Test
+    public void testGCNone() throws Exception {
+        long lsn = 0;
+        lsn = post1(new byte[]{9}, lsn);
+        post1(new byte[]{8, 9}, lsn);
+        assertThat(Files.list(dir).count()).isEqualTo(1);
+        ds.gc(lsn);
+        assertThat(Files.list(dir).count()).isEqualTo(1);
+    }
+
     private void dumpFile(Path p) throws IOException {
         byte[] bytes = ByteStreams.toByteArray(Files.newInputStream(p));
         for (int i = 0; i < bytes.length; i++) {