CassandraLoadBalancerStore implements LoadBalancerStore { @Override public List<JobLoadBalancerState> getAssociations() { return loadBalancerStateMap.entrySet().stream() .map(JobLoadBalancerState::from) .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 testRetrieveLoadBalancers() throws Exception { TestData data = generateTestData(10, 20, 1); loadTestData(data); CassandraLoadBalancerStore store = getInitdStore(); checkDataSetExists(store, data.getAssociations()); }
@Test public void testAssociationStateIsCaseInsensitive() throws Exception { Session session = cassandraCQLUnit.getSession(); PreparedStatement insertStmt = session.prepare("INSERT INTO load_balancer_jobs(job_id, load_balancer_id, state) VALUES(?, ?, ?);"); ResultSet rs1 = session.execute(insertStmt.bind("job-1", "lb-1", "Associated")); assertThat(rs1.isExhausted()).isTrue(); assertThat(rs1.wasApplied()).isTrue(); ResultSet rs2 = session.execute(insertStmt.bind("job-2", "lb-2", "Dissociated")); assertThat(rs2.isExhausted()).isTrue(); assertThat(rs2.wasApplied()).isTrue(); ResultSet rs3 = session.execute(insertStmt.bind("job-3", "lb-3", "aSsOcIaTeD")); assertThat(rs3.isExhausted()).isTrue(); assertThat(rs3.wasApplied()).isTrue(); CassandraLoadBalancerStore store = getInitdStore(); assertThat(store.getAssociations()).containsExactlyInAnyOrder( new JobLoadBalancerState(new JobLoadBalancer("job-1", "lb-1"), ASSOCIATED), new JobLoadBalancerState(new JobLoadBalancer("job-2", "lb-2"), DISSOCIATED), new JobLoadBalancerState(new JobLoadBalancer("job-3", "lb-3"), ASSOCIATED) ); }