@Test public void testBadHealthDownstreamSourceTerminatesClientSubscription() { StepVerifier.create(healthService.events(false)) .then(() -> { downstream2.breakSubscriptionsWithError(new RuntimeException("Simulated error")); }) .verifyError(RuntimeException.class); }
@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; }