Polish contribution

Author: joshiste

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2018 the original author or authors.
+ * Copyright 2014-2019 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -27,32 +27,16 @@
 
 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(scheduler).flatMap(this::sendNotifications);
+        Scheduler scheduler = Schedulers.newSingle("notifications");
+        return publisher.subscribeOn(scheduler).flatMap(this::sendNotifications).doFinally(s -> scheduler.dispose());
     }
 
     protected Mono<Void> sendNotifications(InstanceEvent event) {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2018 the original author or authors.
+ * Copyright 2014-2019 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -52,7 +52,6 @@ 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);
@@ -72,16 +71,15 @@ public Mono<Void> doNotify(InstanceEvent event, Instance instance) {
     }
 
     public void start() {
-        assert scheduler == null;
-        scheduler = Schedulers.newSingle("reminders");
+        Scheduler 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())
                                 .onErrorContinue((ex, value) -> log.warn(
                                     "Unexpected error while sending reminders",
                                     ex
-                                ))
+                                )).doFinally(s -> scheduler.dispose())
                                 .subscribe();
     }
 
@@ -90,10 +88,6 @@ 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/AbstractEventHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2018 the original author or authors.
+ * Copyright 2014-2019 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -43,8 +43,7 @@ public void start() {
                            .log(log.getName(), Level.FINEST)
                            .doOnSubscribe(s -> log.debug("Subscribed to {} events", eventType))
                            .ofType(eventType)
-                           .cast(eventType)
-                           .compose(this::handle)
+                           .cast(eventType).transform(this::handle)
                            .onErrorContinue((ex, value) -> log.warn("Unexpected error while handling {}", value, ex))
                            .subscribe();
     }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2018 the original author or authors.
+ * Copyright 2014-2019 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -28,35 +28,20 @@
 
 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) {
+        Scheduler scheduler = Schedulers.newSingle("endpoint-detector");
         return publisher.subscribeOn(scheduler)
                         .filter(event -> event instanceof InstanceStatusChangedEvent ||
                                          event instanceof InstanceRegistrationUpdatedEvent)
-                        .flatMap(this::detectEndpoints);
+                        .flatMap(this::detectEndpoints)
+                        .doFinally(s -> scheduler.dispose());
     }
 
     protected Mono<Void> detectEndpoints(InstanceEvent event) {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2018 the original author or authors.
+ * Copyright 2014-2019 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -29,36 +29,21 @@
 
 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) {
+        Scheduler scheduler = Schedulers.newSingle("info-updater");
         return publisher.subscribeOn(scheduler)
                         .filter(event -> event instanceof InstanceEndpointsDetectedEvent ||
                                          event instanceof InstanceStatusChangedEvent ||
                                          event instanceof InstanceRegistrationUpdatedEvent)
-                        .flatMap(this::updateInfo);
+                        .flatMap(this::updateInfo)
+                        .doFinally(s -> scheduler.dispose());
     }
 
     protected Mono<Void> updateInfo(InstanceEvent event) {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2018 the original author or authors.
+ * Copyright 2014-2019 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -44,8 +44,6 @@ 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);
@@ -54,28 +52,26 @@ 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();
+        Scheduler scheduler = Schedulers.newSingle("status-monitor");
         intervalSubscription = Flux.interval(updateInterval)
                                    .doOnSubscribe(s -> log.debug("Scheduled status update every {}", updateInterval))
-                                   .log(log.getName(), Level.FINEST)
-                                   .subscribeOn(statusMonitorScheduler)
+                                   .log(log.getName(), Level.FINEST).subscribeOn(scheduler)
                                    .concatMap(i -> this.updateStatusForAllInstances())
                                    .onErrorContinue((ex, value) -> log.warn("Unexpected error while updating statuses",
                                        ex
-                                   ))
+                                   )).doFinally(s -> scheduler.dispose())
                                    .subscribe();
     }
 
     @Override
     protected Publisher<Void> handle(Flux<InstanceEvent> publisher) {
-        return publisher.subscribeOn(statusUpdaterScheduler)
+        Scheduler scheduler = Schedulers.newSingle("status-updater");
+        return publisher.subscribeOn(scheduler)
                         .filter(event -> event instanceof InstanceRegisteredEvent ||
                                          event instanceof InstanceRegistrationUpdatedEvent)
-                        .flatMap(event -> updateStatus(event.getInstance()));
+                        .flatMap(event -> updateStatus(event.getInstance()))
+                        .doFinally(s -> scheduler.dispose());
     }
 
     @Override
@@ -84,14 +80,6 @@ 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() {