HAProxyPublisher implements Function<Long, Flowable<HAProxyPublisher.HAProxyAction>> { @Override public Flowable<HAProxyAction> apply(Long n) { LOGGER.debug("Lookup registration actions for HAProxy instances"); try { List<HAProxyAction> actions = new ArrayList<>(); final Set<String> currentIdSet = repository.getHaproxyIds(); Set<String> removedHAProxies = registered.stream() .filter(k -> !currentIdSet.contains(k)) .peek(id -> actions.add(HAProxyAction.unregister(id))) .collect(Collectors.toSet()); registered.removeAll(removedHAProxies); Set<String> addedHAProxies = currentIdSet.stream() .filter(k -> !registered.contains(k)) .peek(id -> actions.add(HAProxyAction.register(id))) .collect(Collectors.toSet()); registered.addAll(addedHAProxies); if (LOGGER.isInfoEnabled()) { if (actions.size() > 0) { actions.forEach(a -> LOGGER.info("registration action on haproxy id={} -> register={}", a.id, a.isRegistration)); } else { LOGGER.debug("no registration action to perform"); } } return Flowable.fromIterable(actions); } catch (Exception e) { LOGGER.error("Unable to lookup HAProxy IDs.", e); return Flowable.empty(); } } HAProxyPublisher(ConsulRepository repository); @Override Flowable<HAProxyAction> apply(Long n);  }
@Test public void should_publish_registration_actions_when_repository_returns_more_elements_than_previous_call() { when(repository.getHaproxyIds()).thenReturn(new HashSet<>(asList("hap1", "hap2"))); Flowable<HAProxyAction> flowable1 = HAProxyPublisher.apply(0L); List<HAProxyAction> result1 = new ArrayList<>(); flowable1.subscribe(result1::add); assertThat(result1).extracting(HAProxyAction::getId).containsExactlyInAnyOrder("hap1", "hap2"); assertThat(result1).extracting(HAProxyAction::isRegistration).containsExactlyInAnyOrder(true, true); when(repository.getHaproxyIds()).thenReturn(new HashSet<>(asList("hap1", "hap2", "hap3", "hap4"))); Flowable<HAProxyAction> flowable2 = HAProxyPublisher.apply(1L); List<HAProxyAction> result2 = new ArrayList<>(); flowable2.subscribe(result2::add); assertThat(result2).extracting(HAProxyAction::getId).containsExactlyInAnyOrder("hap3", "hap4").doesNotContain("hap1", "hap2"); assertThat(result2).extracting(HAProxyAction::isRegistration).containsExactlyInAnyOrder(true, true); }
@Test public void should_publish_unregistration_actions_when_repository_returns_less_elements_than_previous_call() { when(repository.getHaproxyIds()).thenReturn(new HashSet<>(asList("hap1", "hap2", "hap3", "hap4"))); Flowable<HAProxyAction> flowable1 = HAProxyPublisher.apply(0L); List<HAProxyAction> result1 = new ArrayList<>(); flowable1.subscribe(result1::add); assertThat(result1).extracting(HAProxyAction::getId).containsExactlyInAnyOrder("hap1", "hap2", "hap3", "hap4"); assertThat(result1).extracting(HAProxyAction::isRegistration).containsExactlyInAnyOrder(true, true, true, true); when(repository.getHaproxyIds()).thenReturn(new HashSet<>(asList("hap1", "hap2"))); Flowable<HAProxyAction> flowable2 = HAProxyPublisher.apply(1L); List<HAProxyAction> result2 = new ArrayList<>(); flowable2.subscribe(result2::add); assertThat(result2).extracting(HAProxyAction::getId).containsExactlyInAnyOrder("hap3", "hap4").doesNotContain("hap1", "hap2"); assertThat(result2).extracting(HAProxyAction::isRegistration).containsExactlyInAnyOrder(false, false); }
@Test public void should_publish_empty_actions_when_repository_throws_exception() { when(repository.getHaproxyIds()).thenThrow(new RuntimeException()); Flowable<HAProxyAction> flowable = HAProxyPublisher.apply(0L); List<HAProxyAction> result = new ArrayList<>(); flowable.subscribe(result::add); assertThat(result).isEmpty(); }