Alert publishing and parsing threads that EventHub is closed

So that they terminate after call to EventHub#stopPublishing().

Lower the wait timeout for publishing threads calling EventHub#take
to 2 seconds so that they have time to shutdown nicely before being
interrupted.

Change-Id: Ida6b37b214b6ad914972176b5ac638c48ac9a56c

src/main/java/com/googlesource/gerrit/plugins/eventseiffel/EiffelEventHubImpl.java

diff --git a/src/main/java/com/googlesource/gerrit/plugins/eventseiffel/EiffelEventHubImpl.java b/src/main/java/com/googlesource/gerrit/plugins/eventseiffel/EiffelEventHubImpl.java
index 8a947da..37e96b6 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/eventseiffel/EiffelEventHubImpl.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/eventseiffel/EiffelEventHubImpl.java
@@ -76,13 +76,16 @@
     putLock.lock();
     try {
       try {
-        while (count.get() == MAX_SIZE) {
+        while (open && count.get() == MAX_SIZE) {
           notFull.await(5, TimeUnit.SECONDS);
         }
       } catch (InterruptedException e) {
         notFull.signal();
         throw e;
       }
+      if (!open) {
+        throw new InterruptedException("EventHub is closed");
+      }
       idLookupLock.lock();
       try {
         if (getExistingId(key, true).isPresent()) {
@@ -178,13 +181,11 @@
     final ReentrantLock takeLock = this.takeLock;
     takeLock.lock();
     try {
-      try {
-        while (count.get() == 0 || isTaken()) {
-          readyForTake.await(5, TimeUnit.SECONDS);
-        }
-      } catch (InterruptedException e) {
-        signalReadyForTake();
-        throw e;
+      while (open && (count.get() == 0 || isTaken())) {
+        readyForTake.await(2, TimeUnit.SECONDS);
+      }
+      if (!open) {
+        return null;
       }
       synchronized (eventQueueLock) {
         while (toTake.size() < max && !eventQueue.isEmpty()) {
@@ -243,6 +244,9 @@
 
   @Override
   public void nak(Set<EiffelEvent> events) {
+    if (!open) {
+      return;
+    }
     validateNakAttempt(events);
     final ReentrantLock takeLock = this.takeLock;
     takeLock.lock();

src/main/java/com/googlesource/gerrit/plugins/eventseiffel/PublishEventWorker.java

diff --git a/src/main/java/com/googlesource/gerrit/plugins/eventseiffel/PublishEventWorker.java b/src/main/java/com/googlesource/gerrit/plugins/eventseiffel/PublishEventWorker.java
index 34f4e5d..18e7434 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/eventseiffel/PublishEventWorker.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/eventseiffel/PublishEventWorker.java
@@ -41,7 +41,7 @@
     try {
       while (running) {
         events = eventQueue.take(publisher.maxBatchSize());
-        boolean unpublishedEvents = !events.isEmpty();
+        boolean unpublishedEvents = events != null && !events.isEmpty();
         while (running && unpublishedEvents) {
           unpublishedEvents = !publisher.publish(events);
           if (!unpublishedEvents) {
@@ -85,6 +85,9 @@
     this.running = false;
     try {
       thread.join(5000);
+      if (thread.isAlive()) {
+        thread.interrupt();
+      }
     } catch (InterruptedException e) {
       // Do nothing
     }