CassandraLoadBalancerStore implements LoadBalancerStore { @Override public List<JobLoadBalancer> getAssociationsPage(int offset, int limit) { return jobToAssociatedLoadBalancersMap.keySet().stream() .flatMap(jobId -> { SortedSet<JobLoadBalancer> jobLoadBalancerSortedSet = jobToAssociatedLoadBalancersMap.getOrDefault(jobId, Collections.emptySortedSet()); return jobLoadBalancerSortedSet.stream(); }) .sorted() .skip(offset) .limit(limit) .collect(Collectors.toList()); } @Inject  CassandraLoadBalancerStore(CassandraStoreConfiguration configuration,
                                      @Named(LOAD_BALANCER_SANITIZER) EntitySanitizer entitySanitizer,
                                      @Named(LOAD_BALANCER_CASSANDRA_SESSION) Session session); @Activator void init(); @Override Observable<JobLoadBalancer> getAssociatedLoadBalancersForJob(String jobId); @Override Set<JobLoadBalancer> getAssociatedLoadBalancersSetForJob(String jobId); @Override List<JobLoadBalancerState> getAssociations(); @Override List<JobLoadBalancer> getAssociationsPage(int offset, int limit); @Override Completable addOrUpdateLoadBalancer(JobLoadBalancer jobLoadBalancer, JobLoadBalancer.State state); @Override Completable removeLoadBalancer(JobLoadBalancer jobLoadBalancer); @Override int getNumLoadBalancersForJob(String jobId); @Override Mono<Void> addOrUpdateTargets(Collection<LoadBalancerTargetState> targets); @Override Mono<Void> removeDeregisteredTargets(Collection<LoadBalancerTarget> toRemove); @Override Flux<LoadBalancerTargetState> getLoadBalancerTargets(String loadBalancerId); static final String LOAD_BALANCER_CASSANDRA_SESSION; }
@Test public void testGetFullPage() throws Exception { loadTestData(generateTestData(10, 20, 1)); CassandraLoadBalancerStore store = getInitdStore(); List<JobLoadBalancer> jobLoadBalancerPage = store.getAssociationsPage(0, (10 * 20) + 1); assertThat(jobLoadBalancerPage.size()).isEqualTo(10 * 20); }