class CassandraLoadBalancerStore implements LoadBalancerStore {

    @Override
    public Mono<Void> addOrUpdateTargets(Collection<LoadBalancerTargetState> targets) {
        List<Mono<Void>> insertOperations = targets.stream()
                .map(this::addOrUpdateTarget)
                .collect(Collectors.toList());

        int limit = configuration.getLoadBalancerWriteConcurrencyLimit();
        // prefetch does not matter here because operations don't produce any result (they are Mono<Void>)
        return Flux.mergeSequentialDelayError(insertOperations, limit, limit)
                .ignoreElements()
                .doOnSubscribe(ignored -> {
                    Map<String, Long> countPerLoadBalancer = targets.stream().collect(Collectors.groupingBy(
                            target -> target.getLoadBalancerTarget().getLoadBalancerId(),
                            Collectors.counting()
                    ));
                    logger.info("Inserting/updating targets: {}", countPerLoadBalancer);
                    logger.debug("Inserting/updating {} targets. Details: {}", targets.size(), targets);
                });
    }

    @Inject public  CassandraLoadBalancerStore(CassandraStoreConfiguration configuration,
                                      @Named(LOAD_BALANCER_SANITIZER) EntitySanitizer entitySanitizer,
                                      @Named(LOAD_BALANCER_CASSANDRA_SESSION) Session session);

}