Abort any latches when a client receives an invalid message (#13)

src/main/java/io/playpen/core/coordinator/client/Client.java

@@ -23,6 +23,7 @@
 import io.playpen.core.protocol.Coordinator;
 import io.playpen.core.protocol.P3;
 import io.playpen.core.protocol.Protocol;
+import io.playpen.core.utils.AbortableCountDownLatch;
 import io.playpen.core.utils.AuthUtils;
 import lombok.Getter;
 import lombok.Setter;
@@ -78,7 +79,7 @@ public static Client get() {
 
     private AttachInputListenThread ailThread = null;
 
-    private CountDownLatch latch = null;
+    private AbortableCountDownLatch latch = null;
 
     private int acks = 0;
 
@@ -268,6 +269,11 @@ public boolean receive(Protocol.AuthenticatedMessage auth, Channel from) {
             log.error("Invalid hash on message");
             System.err.println("Received an invalid hash on a message from the network coordinator.");
             System.err.println("This is likely due to us having an invalid UUID or secret key. Please check your local.json!");
+            from.close();
+
+            if (latch != null)
+                latch.abort();
+
             return false;
         }
 
@@ -283,6 +289,11 @@ public boolean receive(Protocol.AuthenticatedMessage auth, Channel from) {
             log.error("Unable to read transaction from message", e);
             System.err.println("Received an unreadable message from the network coordinator.");
             System.err.println("This is likely due to us having an invalid UUID or secret key. Please check your local.json!");
+            from.close();
+
+            if (latch != null)
+                latch.abort();
+
             return false;
         }
 
@@ -506,7 +517,7 @@ protected void runDeprovisionCommand(String[] arguments) {
         }
 
         System.out.println("Operation completed, waiting for ack...");
-        latch = new CountDownLatch(count - acks);
+        latch = new AbortableCountDownLatch(count - acks);
         try {
             latch.await();
         }
@@ -535,7 +546,7 @@ protected void runShutdownCommand(String[] arguments) {
             return;
         }
 
-        latch = new CountDownLatch(1 - acks);
+        latch = new AbortableCountDownLatch(1 - acks);
         try {
             latch.await();
         }
@@ -570,7 +581,7 @@ protected void runPromoteCommand(String[] arguments) {
             return;
         }
 
-        latch = new CountDownLatch(1 - acks);
+        latch = new AbortableCountDownLatch(1 - acks);
         try {
             latch.await();
         }
@@ -645,7 +656,7 @@ protected void runSendCommand(String[] arguments) {
         }
 
         System.out.println("Operation completed, waiting for ack...");
-        latch = new CountDownLatch(count - acks);
+        latch = new AbortableCountDownLatch(count - acks);
         try {
             latch.await();
         }
@@ -728,7 +739,7 @@ protected void runFreezeCommand(String[] arguments) {
         }
 
         System.out.println("Operation completed, waiting for ack...");
-        latch = new CountDownLatch(count - acks);
+        latch = new AbortableCountDownLatch(count - acks);
         try {
             latch.await();
         }
@@ -794,7 +805,7 @@ protected void runUploadCommand(String[] arguments) {
         }
 
         System.out.println("Operation completed, waiting for ack...");
-        latch = new CountDownLatch(count - acks);
+        latch = new AbortableCountDownLatch(count - acks);
         try {
             latch.await();
         } catch (InterruptedException e) {
@@ -1338,13 +1349,13 @@ protected boolean blockUntilCoordList() {
             do {
                 Thread.sleep(1000);
             }
-            while(coordList == null);
+            while(coordList == null && channel.isActive());
         }
         catch(InterruptedException e) {
             return false;
         }
 
-        return true;
+        return channel.isActive();
     }
 
     protected Map<String, List<String>> getServersFromList(Pattern coordPattern, Pattern serverPattern) {

src/main/java/io/playpen/core/utils/AbortableCountDownLatch.java

@@ -0,0 +1,54 @@
+package io.playpen.core.utils;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+// credit to https://stackoverflow.com/a/10455821/646180
+public class AbortableCountDownLatch extends CountDownLatch {
+    protected boolean aborted = false;
+
+    public AbortableCountDownLatch(int count) {
+        super(count);
+    }
+
+
+    /**
+     * Unblocks all threads waiting on this latch and cause them to receive an
+     * AbortedException.  If the latch has already counted all the way down,
+     * this method does nothing.
+     */
+    public void abort() {
+        if( getCount()==0 )
+            return;
+
+        this.aborted = true;
+        while(getCount()>0)
+            countDown();
+    }
+
+
+    @Override
+    public boolean await(long timeout, TimeUnit unit) throws InterruptedException {
+        final boolean rtrn = super.await(timeout,unit);
+        if (aborted)
+            throw new AbortedException();
+        return rtrn;
+    }
+
+    @Override
+    public void await() throws InterruptedException {
+        super.await();
+        if (aborted)
+            throw new AbortedException();
+    }
+
+
+    public static class AbortedException extends InterruptedException {
+        public AbortedException() {
+        }
+
+        public AbortedException(String detailMessage) {
+            super(detailMessage);
+        }
+    }
+}