Shut worker down if listening thread goes down

NASA-AMMOS · Oct 9, 2023 · e8ebffb · e8ebffb
1 parent bd0a7c6
commit e8ebffb
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 4 deletions.
diff --git a/merlin-worker/src/main/java/gov/nasa/jpl/aerie/merlin/worker/MerlinWorkerAppDriver.java b/merlin-worker/src/main/java/gov/nasa/jpl/aerie/merlin/worker/MerlinWorkerAppDriver.java
@@ -21,6 +21,7 @@
 import java.time.Instant;
 import java.util.Optional;
 import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
 
 public final class MerlinWorkerAppDriver {
   public static void main(String[] args) throws InterruptedException {
@@ -67,8 +68,9 @@ public static void main(String[] args) throws InterruptedException {
     try (final var app = Javalin.create().start(8080)) {
       app.get("/health", ctx -> ctx.status(200));
 
-      while (true) {
-        final var notification = notificationQueue.take();
+      while (listenThread.isAlive()) {
+        final var notification = notificationQueue.poll(1, TimeUnit.MINUTES);
+        if(notification == null) continue;
         final var planId = new PlanId(notification.planId());
         final var datasetId = notification.datasetId();
 

diff --git a/...er-worker/src/main/java/gov/nasa/jpl/aerie/scheduler/worker/SchedulerWorkerAppDriver.java b/...er-worker/src/main/java/gov/nasa/jpl/aerie/scheduler/worker/SchedulerWorkerAppDriver.java
@@ -5,6 +5,8 @@
 import java.nio.file.Path;
 import java.util.Optional;
 import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
 import com.zaxxer.hikari.HikariConfig;
 import com.zaxxer.hikari.HikariDataSource;
 import gov.nasa.jpl.aerie.scheduler.server.ResultsProtocol;
@@ -76,8 +78,9 @@ public static void main(String[] args) throws Exception {
     try(final var app = Javalin.create().start(8080)) {
       app.get("/health", ctx -> ctx.status(200));
 
-      while (true) {
-        final var notification = notificationQueue.take();
+      while (listenThread.isAlive()) {
+        final var notification = notificationQueue.poll(1, TimeUnit.MINUTES);
+        if (notification == null) continue;
         final var specificationRevision = notification.specificationRevision();
         final var specificationId = new SpecificationId(notification.specificationId());