@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(); }
@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; }