Make sure events are not duplicated on retry

Author: joshiste

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

@@ -40,23 +40,23 @@ protected AbstractEventHandler(Publisher<InstanceEvent> publisher, Class<T> even
  }
 
  public void start() {
- subscription = Flux.from(publisher)
- .log(log.getName(), Level.FINEST)
- .doOnSubscribe(s -> log.debug("Subscribed to {} events", eventType))
- .ofType(eventType)
- .cast(eventType)
- .transform(this::handle)
- .retryWhen(Retry.any()
- .retryMax(Long.MAX_VALUE)
- .doOnRetry(ctx -> log.warn("Unexpected error", ctx.exception())))
- .subscribe();
+ this.subscription = Flux.from(this.publisher)
+  .log(this.log.getName(), Level.FINEST)
+  .doOnSubscribe(s -> this.log.debug("Subscribed to {} events", this.eventType))
+  .ofType(this.eventType)
+  .cast(this.eventType)
+  .transform(this::handle)
+  .retryWhen(Retry.any()
+  .retryMax(Long.MAX_VALUE)
+  .doOnRetry(ctx -> this.log.warn("Unexpected error", ctx.exception())))
+  .subscribe();
  }
 
  protected abstract Publisher<Void> handle(Flux<T> publisher);
 
  public void stop() {
- if (subscription != null) {
- subscription.dispose();
+ if (this.subscription != null) {
+ this.subscription.dispose();
  }
  }
 }

spring-boot-admin-server/src/test/java/de/codecentric/boot/admin/server/services/AbstractEventHandlerTest.java

@@ -37,17 +37,10 @@
 public class AbstractEventHandlerTest {
  private static final Logger log = LoggerFactory.getLogger(AbstractEventHandlerTest.class);
  private static final Registration registration = Registration.create("foo", "http://health").build();
- private static final InstanceRegisteredEvent event = new InstanceRegisteredEvent(InstanceId.of("id"),
- 0L,
- registration
- );
- private static final InstanceRegisteredEvent errorEvent = new InstanceRegisteredEvent(InstanceId.of("err"),
- 0L,
- registration
- );
- private static final InstanceDeregisteredEvent ignoredEvent = new InstanceDeregisteredEvent(InstanceId.of("id"),
- 1L
- );
+ private static final InstanceRegisteredEvent firstEvent = new InstanceRegisteredEvent(InstanceId.of("id"), 0L, registration);
+ private static final InstanceRegisteredEvent secondEvent = new InstanceRegisteredEvent(InstanceId.of("id"), 1L, registration);
+ private static final InstanceRegisteredEvent errorEvent = new InstanceRegisteredEvent(InstanceId.of("err"), 2L, registration);
+ private static final InstanceDeregisteredEvent ignoredEvent = new InstanceDeregisteredEvent(InstanceId.of("id"), 2L);
 
  @Test
  public void should_resubscribe_after_error() {
@@ -58,14 +51,10 @@ public void should_resubscribe_after_error() {
 
  StepVerifier.create(eventHandler.getFlux())
  .expectSubscription()
- .then(() -> testPublisher.next(event))
- .expectNext(event)
- .then(() -> testPublisher.next(errorEvent))
- .expectNoEvent(Duration.ofMillis(100L))
- .then(() -> testPublisher.next(event))
- .expectNext(event)
+ .then(() -> testPublisher.next(firstEvent, errorEvent, secondEvent))
+ .expectNext(firstEvent, secondEvent)
  .thenCancel()
- .verify(Duration.ofSeconds(5));
+ .verify(Duration.ofSeconds(1));
 
  }
 
@@ -78,12 +67,10 @@ public void should_filter() {
 
  StepVerifier.create(eventHandler.getFlux())
  .expectSubscription()
- .then(() -> testPublisher.next(event))
- .expectNext(event)
- .then(() -> testPublisher.next(ignoredEvent))
- .expectNoEvent(Duration.ofMillis(100L))
+ .then(() -> testPublisher.next(firstEvent, ignoredEvent, secondEvent))
+ .expectNext(firstEvent, secondEvent)
  .thenCancel()
- .verify(Duration.ofSeconds(5));
+ .verify(Duration.ofSeconds(1));
  }