@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)); }
@Override public Flux<ContainerHealthEvent> events(boolean snapshot) { return snapshot ? healthStatuses.compose(ReactorExt.head(() -> Collections.singletonList(buildCurrentSnapshot()))) : healthStatuses; }
AggregatingContainerHealthService implements ContainerHealthService { @Override public Flux<ContainerHealthEvent> events(boolean snapshot) { return snapshot ? healthStatuses.compose(ReactorExt.head(() -> Collections.singletonList(buildCurrentSnapshot()))) : healthStatuses; } }
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); }
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); }
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; }