Add another safety net to catch exceptions from notifiers

hopefully fixes codecentric#1175

Author: joshiste

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

@@ -24,8 +24,11 @@
 import reactor.core.scheduler.Schedulers;
 
 import org.reactivestreams.Publisher;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class NotificationTrigger extends AbstractEventHandler<InstanceEvent> {
+ private static final Logger log = LoggerFactory.getLogger(NotificationTrigger.class);
  private final Notifier notifier;
 
  public NotificationTrigger(Notifier notifier, Publisher<InstanceEvent> publisher) {
@@ -40,6 +43,8 @@ protected Publisher<Void> handle(Flux<InstanceEvent> publisher) {
  }
 
  protected Mono<Void> sendNotifications(InstanceEvent event) {
- return notifier.notify(event);
+ return this.notifier.notify(event)
+ .doOnError(e -> log.warn("Couldn't notify for event {} ", event, e))
+ .onErrorResume(e -> Mono.empty());
  }
 }

spring-boot-admin-server/src/test/java/de/codecentric/boot/admin/server/notify/NotificationTriggerTest.java

@@ -23,15 +23,18 @@
 import de.codecentric.boot.admin.server.domain.values.InstanceId;
 import de.codecentric.boot.admin.server.domain.values.Registration;
 import de.codecentric.boot.admin.server.domain.values.StatusInfo;
+import reactor.core.publisher.Mono;
 import reactor.test.publisher.TestPublisher;
 
 import org.junit.Test;
 
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.clearInvocations;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 public class NotificationTriggerTest {
  private final Instance instance = Instance.create(InstanceId.of("id-1"))
@@ -47,17 +50,44 @@ public void should_notify_on_event() throws InterruptedException {
  Thread.sleep(500L); //wait for subscription
 
  //when registered event is emitted
- InstanceStatusChangedEvent event = new InstanceStatusChangedEvent(instance.getId(), instance.getVersion(),
- StatusInfo.ofDown());
+ InstanceStatusChangedEvent event = new InstanceStatusChangedEvent(this.instance.getId(),
+ this.instance.getVersion(),
+ StatusInfo.ofDown()
+ );
  events.next(event);
  //then should notify
  verify(notifier, times(1)).notify(event);
 
  //when registered event is emitted but the trigger has been stopped
  trigger.stop();
  clearInvocations(notifier);
- events.next(new InstanceRegisteredEvent(instance.getId(), instance.getVersion(), instance.getRegistration()));
+ events.next(new InstanceRegisteredEvent(this.instance.getId(), this.instance.getVersion(), this.instance.getRegistration()));
  //then should not notify
  verify(notifier, never()).notify(event);
  }
+
+ @Test
+ public void should_resume_on_exceptopn() throws InterruptedException {
+ //given
+ Notifier notifier = mock(Notifier.class);
+ TestPublisher<InstanceEvent> events = TestPublisher.create();
+ NotificationTrigger trigger = new NotificationTrigger(notifier, events);
+ trigger.start();
+ Thread.sleep(500L); //wait for subscription
+
+
+ when(notifier.notify(any())).thenReturn(Mono.error(new IllegalStateException("Test"))).thenReturn(Mono.empty());
+
+ //when exception for the first event is thrown and a subsequent event is fired
+ InstanceStatusChangedEvent event = new InstanceStatusChangedEvent(this.instance.getId(),
+ this.instance.getVersion(),
+ StatusInfo.ofDown()
+ );
+ events.next(event);
+ events.next(event);
+
+ //the notifier was after the exception
+ verify(notifier, times(2)).notify(event);
+ }
+
 }