end-to-end testing

Author: kahing

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

@@ -45,6 +45,12 @@ private void init() throws IOException {
                 });
     }
 
+    public void stop() {
+        synchronized (topics) {
+            topics.values().forEach(q -> q.stop());
+        }
+    }
+
     @Path("/{topic}/{username}")
     @POST
     public Response subscribe(@PathParam("topic") String topic,

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

@@ -182,6 +182,7 @@ public void gc(long needLSN) {
                 long lsn = Long.parseLong(name, 16);
                 if (lsn < baseLSN) {
                     try {
+                        logger.info("gc deleting {}", p);
                         Files.delete(p);
                     } catch (IOException e) {
                         e.printStackTrace();

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

@@ -16,10 +16,12 @@
 public class Main {
     private static Logger logger = LoggerFactory.getLogger(Main.class);
     final HttpServer server;
+    final Api api;
 
     Main(int port, Path dir) throws IOException {
         ResourceConfig rc = new ResourceConfig();
-        rc.registerInstances(new Api(dir, 4096));
+        api = new Api(dir, 4096);
+        rc.registerInstances(api);
         if (logger.isDebugEnabled()) {
             rc.register(new LoggingFilter(java.util.logging.Logger.getGlobal(), false));
         }
@@ -30,6 +32,15 @@ void start() throws IOException {
         server.start();
     }
 
+    public int getPort() {
+        return server.getListeners().stream().findAny().map(n -> n.getPort()).orElse(0);
+    }
+
+    void stop() {
+        api.stop();
+        server.shutdownNow();
+    }
+
     public static void main(String[] args) throws IOException {
         if (args.length != 2) {
             System.err.println("Usage: Main <port> <dir>");

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

@@ -172,6 +172,7 @@ void spawnGCThread() {
     void stop() {
         shutdown = true;
         try {
+            gcThread.interrupt();
             gcThread.join();
         } catch (InterruptedException e) {
             e.printStackTrace();

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

@@ -0,0 +1,103 @@
+package us.hxbc.clusterhq.queue;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import javax.ws.rs.client.Client;
+import javax.ws.rs.client.ClientBuilder;
+import javax.ws.rs.client.WebTarget;
+import javax.ws.rs.core.Response;
+import java.nio.file.Path;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class MainTest {
+    @Rule
+    public TemporaryFolder folder= new TemporaryFolder();
+    private Path dir;
+    private Main main;
+    private int port;
+    private WebTarget target;
+    private Client c;
+
+    @Before
+    public void setUp() throws Exception {
+        dir = folder.newFolder().toPath();
+        initClient();
+    }
+
+    private void initClient() throws Exception {
+        main = new Main(0, dir);
+        main.start();
+        port = main.getPort();
+        if (c != null) {
+            c.close();
+        }
+        c = ClientBuilder.newClient();
+        target = c.target("http://127.0.0.1:" + port);
+    }
+
+    @After
+    public void tearDown() throws Exception {
+        main.stop();
+    }
+
+    @Test
+    public void testSubscribe() throws Exception {
+        Response resp = target.path("/topic1/user1").request().post(null);
+        assertThat(resp.getStatus()).isEqualTo(200);
+        resp = target.path("/topic1/user1").request().post(null);
+        assertThat(resp.getStatus()).isEqualTo(200);
+    }
+
+    @Test
+    public void testUnsubscribe() throws Exception {
+        Response resp;
+        // unsubscribe without subscribing
+        resp = target.path("/topic1/user1").request().delete();
+        assertThat(resp.getStatus()).isEqualTo(404);
+
+        // unsubscribe after subscribing
+        resp = target.path("/topic1/user1").request().post(null);
+        assertThat(resp.getStatus()).isEqualTo(200);
+        resp = target.path("/topic1/user1").request().delete();
+        assertThat(resp.getStatus()).isEqualTo(200);
+
+        // unsubscribe again
+        resp = target.path("/topic1/user1").request().delete();
+        assertThat(resp.getStatus()).isEqualTo(404);
+    }
+
+    @Test
+    public void testMessage() throws Exception {
+        Response resp;
+        // subscription doesn't exist
+        resp = target.path("/topic1/user1").request().get();
+        assertThat(resp.getStatus()).isEqualTo(404);
+
+        // no message available
+        resp = target.path("/topic1/user1").request().post(null);
+        assertThat(resp.getStatus()).isEqualTo(200);
+        resp = target.path("/topic1/user1").request().get();
+        assertThat(resp.getStatus()).isEqualTo(204);
+
+        resp = target.path("/topic1").request().post(null);
+        assertThat(resp.getStatus()).isEqualTo(200);
+        resp = target.path("/topic1/user1").request().get();
+        assertThat(resp.getStatus()).isEqualTo(200);
+    }
+
+    @Test
+    public void testRestart() throws Exception {
+        Response resp;
+        resp = target.path("/topic1/user1").request().post(null);
+        assertThat(resp.getStatus()).isEqualTo(200);
+        main.stop();
+        initClient();
+        resp = target.path("/topic1/user1").request().get();
+        assertThat(resp.getStatus()).isEqualTo(204);
+    }
+}