@Test(timeout = 30_000) public void testAggregation() { Iterator<ReadinessStatus> it = resolver.observeLocalMasterReadinessUpdates().toIterable().iterator(); delegate1Processor.onNext(newStatus(ReadinessState.Enabled)); delegate2Processor.onNext(newStatus(ReadinessState.Disabled)); Assertions.assertThat(it.next().getState()).isEqualTo(ReadinessState.Disabled); delegate2Processor.onNext(newStatus(ReadinessState.Enabled)); Assertions.assertThat(it.next().getState()).isEqualTo(ReadinessState.Enabled); delegate1Processor.onNext(newStatus(ReadinessState.Disabled)); Assertions.assertThat(it.next().getState()).isEqualTo(ReadinessState.Disabled); }
@Override public Flux<ReadinessStatus> observeLocalMasterReadinessUpdates() { return stream; }
AggregatingLocalMasterReadinessResolver implements LocalMasterReadinessResolver { @Override public Flux<ReadinessStatus> observeLocalMasterReadinessUpdates() { return stream; } }
AggregatingLocalMasterReadinessResolver implements LocalMasterReadinessResolver { @Override public Flux<ReadinessStatus> observeLocalMasterReadinessUpdates() { return stream; } AggregatingLocalMasterReadinessResolver(List<LocalMasterReadinessResolver> delegates, TitusRuntime titusRuntime); }
AggregatingLocalMasterReadinessResolver implements LocalMasterReadinessResolver { @Override public Flux<ReadinessStatus> observeLocalMasterReadinessUpdates() { return stream; } AggregatingLocalMasterReadinessResolver(List<LocalMasterReadinessResolver> delegates, TitusRuntime titusRuntime); @Override Flux<ReadinessStatus> observeLocalMasterReadinessUpdates(); }
AggregatingLocalMasterReadinessResolver implements LocalMasterReadinessResolver { @Override public Flux<ReadinessStatus> observeLocalMasterReadinessUpdates() { return stream; } AggregatingLocalMasterReadinessResolver(List<LocalMasterReadinessResolver> delegates, TitusRuntime titusRuntime); @Override Flux<ReadinessStatus> observeLocalMasterReadinessUpdates();  }