EntryPointPublisher implements Function<Long, Flowable<T>> { @Override public Flowable<T> apply(Long n) { try { return Flowable.fromIterable(entryPointRepository.getEntryPointsId()).map(provider); } catch (Exception e) { LOGGER.error("Unable to lookup entry points.", e); return Flowable.empty(); } } EntryPointPublisher(EntryPointRepository entryPointRepository, Function<String, T> provider); @Override Flowable<T> apply(Long n);  }
@Test public void should_build_flowable_with_mapped_list_of_entry_points_from_repository() throws Exception { when(repository.getEntryPointsId()).thenReturn(new HashSet<>(Arrays.asList("ep1", "ep2"))); Flowable<String> flowable = entryPointPublisher.apply(0L); List<String> result = new ArrayList<>(); flowable.subscribe(result::add); assertThat(result).containsExactlyInAnyOrder("EP1", "EP2"); }
@Test public void should_build_empty_flowable_when_repository_throws_exception() throws Exception { when(repository.getEntryPointsId()).thenThrow(new RuntimeException()); Flowable<String> flowable = entryPointPublisher.apply(0L); List<String> result = new ArrayList<>(); flowable.subscribe(result::add); assertThat(result).isEmpty(); }