Properly dispose schedulers on shutdown

With this commit all schedulers are disposed on shutdown so the jvm can
exit gracefully.

Author: t-beckmann

spring-boot-admin-server/src/main/java/de/codecentric/boot/admin/server/notify/NotificationTrigger.java

@@ -20,21 +20,39 @@
 import de.codecentric.boot.admin.server.services.AbstractEventHandler;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Scheduler;
 import reactor.core.scheduler.Schedulers;
 
 import org.reactivestreams.Publisher;
 
 public class NotificationTrigger extends AbstractEventHandler<InstanceEvent> {
     private final Notifier notifier;
+    private Scheduler scheduler;
 
     public NotificationTrigger(Notifier notifier, Publisher<InstanceEvent> publisher) {
         super(publisher, InstanceEvent.class);
         this.notifier = notifier;
     }
 
+    @Override
+    public void start() {
+        assert scheduler == null;
+        scheduler = Schedulers.newSingle("notifications");
+        super.start();
+    }
+
+    @Override
+    public void stop() {
+        super.stop();
+        if (scheduler != null) {
+            scheduler.dispose();
+            scheduler = null;
+        }
+    }
+
     @Override
     protected Publisher<Void> handle(Flux<InstanceEvent> publisher) {
-        return publisher.subscribeOn(Schedulers.newSingle("notifications")).flatMap(this::sendNotifications);
+        return publisher.subscribeOn(scheduler).flatMap(this::sendNotifications);
     }
 
     protected Mono<Void> sendNotifications(InstanceEvent event) {

spring-boot-admin-server/src/main/java/de/codecentric/boot/admin/server/notify/RemindingNotifier.java

@@ -25,6 +25,7 @@
 import reactor.core.Disposable;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Scheduler;
 import reactor.core.scheduler.Schedulers;
 
 import java.time.Duration;
@@ -51,6 +52,7 @@ public class RemindingNotifier extends AbstractEventNotifier {
     private String[] reminderStatuses = {"DOWN", "OFFLINE"};
     @Nullable
     private Disposable subscription;
+    private Scheduler scheduler;
 
     public RemindingNotifier(Notifier delegate, InstanceRepository repository) {
         super(repository);
@@ -69,9 +71,10 @@ public Mono<Void> doNotify(InstanceEvent event, Instance instance) {
         }));
     }
 
-
     public void start() {
-        this.subscription = Flux.interval(this.checkReminderInverval, Schedulers.newSingle("reminders"))
+        assert scheduler == null;
+        scheduler = Schedulers.newSingle("reminders");
+        this.subscription = Flux.interval(this.checkReminderInverval, scheduler)
                                 .log(log.getName(), Level.FINEST)
                                 .doOnSubscribe(s -> log.debug("Started reminders"))
                                 .flatMap(i -> this.sendReminders())
@@ -87,6 +90,10 @@ public void stop() {
             log.debug("stopped reminders");
             this.subscription.dispose();
         }
+        if (scheduler != null) {
+            scheduler.dispose();
+            scheduler = null;
+        }
     }
 
     protected Mono<Void> sendReminders() {

spring-boot-admin-server/src/main/java/de/codecentric/boot/admin/server/services/EndpointDetectionTrigger.java

@@ -21,21 +21,39 @@
 import de.codecentric.boot.admin.server.domain.events.InstanceStatusChangedEvent;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Scheduler;
 import reactor.core.scheduler.Schedulers;
 
 import org.reactivestreams.Publisher;
 
 public class EndpointDetectionTrigger extends AbstractEventHandler<InstanceEvent> {
     private final EndpointDetector endpointDetector;
+    private Scheduler scheduler;
 
     public EndpointDetectionTrigger(EndpointDetector endpointDetector, Publisher<InstanceEvent> publisher) {
         super(publisher, InstanceEvent.class);
         this.endpointDetector = endpointDetector;
     }
 
+    @Override
+    public void start() {
+        assert scheduler == null;
+        scheduler = Schedulers.newSingle("endpoint-detector");
+        super.start();
+    }
+
+    @Override
+    public void stop() {
+        super.stop();
+        if (scheduler != null) {
+            scheduler.dispose();
+            scheduler = null;
+        }
+    }
+
     @Override
     protected Publisher<Void> handle(Flux<InstanceEvent> publisher) {
-        return publisher.subscribeOn(Schedulers.newSingle("endpoint-detector"))
+        return publisher.subscribeOn(scheduler)
                         .filter(event -> event instanceof InstanceStatusChangedEvent ||
                                          event instanceof InstanceRegistrationUpdatedEvent)
                         .flatMap(this::detectEndpoints);

spring-boot-admin-server/src/main/java/de/codecentric/boot/admin/server/services/InfoUpdateTrigger.java

@@ -22,21 +22,39 @@
 import de.codecentric.boot.admin.server.domain.events.InstanceStatusChangedEvent;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Scheduler;
 import reactor.core.scheduler.Schedulers;
 
 import org.reactivestreams.Publisher;
 
 public class InfoUpdateTrigger extends AbstractEventHandler<InstanceEvent> {
     private final InfoUpdater infoUpdater;
+    private Scheduler scheduler;
 
     public InfoUpdateTrigger(InfoUpdater infoUpdater, Publisher<InstanceEvent> publisher) {
         super(publisher, InstanceEvent.class);
         this.infoUpdater = infoUpdater;
     }
 
+    @Override
+    public void start() {
+        assert scheduler == null;
+        scheduler = Schedulers.newSingle("info-updater");
+        super.start();
+    }
+
+    @Override
+    public void stop() {
+        super.stop();
+        if (scheduler != null) {
+            scheduler.dispose();
+            scheduler = null;
+        }
+    }
+
     @Override
     protected Publisher<Void> handle(Flux<InstanceEvent> publisher) {
-        return publisher.subscribeOn(Schedulers.newSingle("info-updater"))
+        return publisher.subscribeOn(scheduler)
                         .filter(event -> event instanceof InstanceEndpointsDetectedEvent ||
                                          event instanceof InstanceStatusChangedEvent ||
                                          event instanceof InstanceRegistrationUpdatedEvent)

spring-boot-admin-server/src/main/java/de/codecentric/boot/admin/server/services/StatusUpdateTrigger.java

@@ -23,6 +23,7 @@
 import reactor.core.Disposable;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Scheduler;
 import reactor.core.scheduler.Schedulers;
 
 import java.time.Duration;
@@ -43,7 +44,8 @@ public class StatusUpdateTrigger extends AbstractEventHandler<InstanceEvent> {
     private Duration statusLifetime = Duration.ofSeconds(10);
     @Nullable
     private Disposable intervalSubscription;
-
+    private Scheduler statusMonitorScheduler;
+    private Scheduler statusUpdaterScheduler;
 
     public StatusUpdateTrigger(StatusUpdater statusUpdater, Publisher<InstanceEvent> publisher) {
         super(publisher, InstanceEvent.class);
@@ -52,11 +54,15 @@ public StatusUpdateTrigger(StatusUpdater statusUpdater, Publisher<InstanceEvent>
 
     @Override
     public void start() {
+        assert statusMonitorScheduler == null;
+        statusMonitorScheduler = Schedulers.newSingle("status-monitor");
+        assert statusUpdaterScheduler == null;
+        statusUpdaterScheduler = Schedulers.newSingle("status-updater");
         super.start();
         intervalSubscription = Flux.interval(updateInterval)
                                    .doOnSubscribe(s -> log.debug("Scheduled status update every {}", updateInterval))
                                    .log(log.getName(), Level.FINEST)
-                                   .subscribeOn(Schedulers.newSingle("status-monitor"))
+                                   .subscribeOn(statusMonitorScheduler)
                                    .concatMap(i -> this.updateStatusForAllInstances())
                                    .onErrorContinue((ex, value) -> log.warn("Unexpected error while updating statuses",
                                        ex
@@ -66,7 +72,7 @@ public void start() {
 
     @Override
     protected Publisher<Void> handle(Flux<InstanceEvent> publisher) {
-        return publisher.subscribeOn(Schedulers.newSingle("status-updater"))
+        return publisher.subscribeOn(statusUpdaterScheduler)
                         .filter(event -> event instanceof InstanceRegisteredEvent ||
                                          event instanceof InstanceRegistrationUpdatedEvent)
                         .flatMap(event -> updateStatus(event.getInstance()));
@@ -78,6 +84,14 @@ public void stop() {
         if (intervalSubscription != null) {
             intervalSubscription.dispose();
         }
+        if (statusUpdaterScheduler != null) {
+            statusUpdaterScheduler.dispose();
+            statusUpdaterScheduler = null;
+        }
+        if (statusMonitorScheduler != null) {
+            statusMonitorScheduler.dispose();
+            statusMonitorScheduler = null;
+        }
     }
 
     protected Mono<Void> updateStatusForAllInstances() {