AggregatingContainerHealthService implements ContainerHealthService { @Override public Flux<ContainerHealthEvent> events(boolean snapshot) { return snapshot ? healthStatuses.compose(ReactorExt.head(() -> Collections.singletonList(buildCurrentSnapshot()))) : healthStatuses; } @Inject  AggregatingContainerHealthService(Set<ContainerHealthService> healthServices,
                                             ReadOnlyJobOperations jobOperations,
                                             TitusRuntime titusRuntime); @Override String getName(); @Override Optional<ContainerHealthStatus> findHealthStatus(String taskId); @Override Flux<ContainerHealthEvent> events(boolean snapshot); static final String NAME; }
@Test public void testSubscriptionWithSnapshot() { downstream1.makeHealthy(taskId1); downstream2.makeHealthy(taskId1); StepVerifier.create(healthService.events(true)) .assertNext(event -> assertContainerHealthSnapshot( event, status -> status.getTaskId().equals(taskId1) && status.getState() == ContainerHealthState.Healthy) ) .then(() -> downstream1.makeUnhealthy(taskId1)) .assertNext(event -> assertContainerHealthEvent(event, taskId1, ContainerHealthState.Unhealthy)) .thenCancel() .verify(Duration.ofSeconds(5)); }
@Test public void testBadHealthDownstreamSourceTerminatesClientSubscription() { StepVerifier.create(healthService.events(false)) .then(() -> { downstream2.breakSubscriptionsWithError(new RuntimeException("Simulated error")); }) .verifyError(RuntimeException.class); }
@Test public void testBadSubscriberIsIsolated() { TitusRxSubscriber<ContainerHealthEvent> goodSubscriber = new TitusRxSubscriber<>(); healthService.events(false).subscribe(goodSubscriber); Disposable badSubscriber = healthService.events(false).subscribe( next -> { throw new RuntimeException("simulated error"); }, e -> { throw new RuntimeException("simulated error"); }, () -> { throw new RuntimeException("simulated error"); } ); downstream1.makeHealthy(taskId1); assertThat(goodSubscriber.isOpen()).isTrue(); assertThat(badSubscriber.isDisposed()).isTrue(); }