added some tests and comments

Author: kahing

clusterhq/queue/pom.xml

@@ -128,4 +128,4 @@
         <jersey.version>2.18</jersey.version>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     </properties>
-</project>
+</project>

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

@@ -19,6 +19,23 @@
 
 import static java.util.Objects.requireNonNull;
 
+/**
+ * DataStore stores messages for our queue. Logically it's a log structured
+ * append only data store. Logs are chunked into different files so old
+ * messages can be garbage collected relatively easily.
+ *
+ * each message has a LSN (log sequence number) which is a 64 bit always
+ * incrementing number. LSN is also the logical address of the position
+ * of the message.
+ *
+ * LSN is translated to physical files base on the configured chunk size.
+ * Suppose chunk size is 4, then LSN 0 would be in a file called "0", LSN
+ * 4 would be in a file called "1", and so on. The file names are hex encoded.
+ *
+ * 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".
+ */
 public class DataStore {
     private Logger logger = LoggerFactory.getLogger(getClass());
     private Path dir;

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

@@ -15,6 +15,12 @@
 
 import static java.util.Objects.requireNonNull;
 
+/**
+ * Queue manages subscriptions. It uses DataStore to store the actual
+ * messages. Each subscription contains the next LSN to start
+ * retrieving messages at. This LSN is updated each time a message is
+ * retrieved.
+ */
 public class Queue {
     private static final long CHUNK_SIZE = 4 * 1024; // 4KB
     private final Path dataDir, subscriptionDir;

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

@@ -95,6 +95,32 @@ public void testGet2() throws Exception {
         assertThatThrownBy(() -> queue.get("foo")).isInstanceOf(ClientErrorException.class);
     }
 
+    @Test
+    public void testGetMany() throws Exception {
+        queue.subscribe("foo");
+        for (int i = 0; i < 10; i++) {
+            queue.post(string2Stream("hello, "));
+            queue.post(string2Stream("world!"));
+        }
+        DataStore.Message m;
+        for (int i = 0; i < 10; i++) {
+            m = queue.get("foo");
+            assertThat(stream2String(m.in)).isEqualTo("hello, ");
+            m = queue.get("foo");
+            assertThat(stream2String(m.in)).isEqualTo("world!");
+        }
+        assertThatThrownBy(() -> queue.get("foo")).isInstanceOf(ClientErrorException.class);
+    }
+
+    @Test
+    public void testResubscribeGet() throws Exception {
+        queue.subscribe("foo");
+        queue.unsubscribe("foo");
+        queue.post(string2Stream("hello"));
+        queue.subscribe("foo");
+        assertThatThrownBy(() -> queue.get("foo")).isInstanceOf(ClientErrorException.class);
+    }
+
     @Test
     public void testUnsubscribe() throws Exception {
         queue.subscribe("foo");