package edu.uw.zookeeper.clients.common;

import static com.google.common.base.Preconditions.*;

import java.util.concurrent.Callable;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import com.google.common.base.Function;
import com.google.common.base.Optional;
import com.typesafe.config.ConfigValueType;

import edu.uw.zookeeper.common.Configurable;
import edu.uw.zookeeper.common.Configuration;

public class IterationCallable<V> implements Callable<Optional<V>> {

    public static <V> IterationCallable<V> create(
            Configuration configuration,
            Callable<V> callable) {
        int iterations = ConfigurableIterations.get(configuration);
        int logIterations = ConfigurableLogIterations.get(configuration);
        int logInterval;
        if (logIterations > 0) {
            logInterval = iterations / logIterations;
        } else {
            logInterval = 0;
        }
        return create(iterations, logInterval, callable);
    }
    
    public static <V> IterationCallable<V> create(
            int iterations,
            int logInterval,
            Callable<V> callable) {
        return new IterationCallable<V>(
                iterations, logInterval, callable, LogManager.getLogger(IterationCallable.class));
    }

    public static abstract class ConfigurableInt implements Function<Configuration, Integer> {

        @Override
        public Integer apply(Configuration configuration) {
            Configurable configurable = getClass().getAnnotation(Configurable.class);
            return configuration.withConfigurable(configurable)
                        .getConfigOrEmpty(configurable.path())
                            .getInt(configurable.key());
        }
    }
    
    @Configurable(arg="iterations", key="Iterations", value="10", type=ConfigValueType.NUMBER)
    public static class ConfigurableIterations extends ConfigurableInt {

        public static Integer get(Configuration configuration) {
            return new ConfigurableIterations().apply(configuration);
        }
    }
    
    @Configurable(key="LogIterations", value="10", type=ConfigValueType.NUMBER)
    public static class ConfigurableLogIterations extends ConfigurableInt {

        public static Integer get(Configuration configuration) {
            return new ConfigurableLogIterations().apply(configuration);
        }
    }

    protected final Logger logger;
    protected final int logInterval;
    protected final int iterations;
    protected final Callable<V> callable;
    protected int count;
    
    public IterationCallable(
            int iterations,
            int logInterval,
            Callable<V> callable,
            Logger logger) {
        checkArgument(iterations >= 0);
        this.logger = logger;
        this.logInterval = logInterval;
        this.iterations = iterations;
        this.callable = callable;
        this.count = 0;
    }
    
    public int getIterations() {
        return iterations;
    }
    
    public int getCount() {
        return count;
    }
    
    @Override
    public Optional<V> call() throws Exception {
        this.count++;
        checkState(count <= iterations);
        if ((logInterval != 0) && ((count == 1) || (count == iterations) || (count % logInterval == 0))) {
            logger.info("Iteration {}", count);
        }
        V result = callable.call();
        if (count < iterations) {
            return Optional.absent();
        } else {
            return Optional.of(result);
        }
    }
}
@@ -431,16 +431,44 @@
 ation;%0A%0A
+/**%0A * Not thread-safe.%0A */%0A
 public c
@@ -669,26 +669,21 @@
     int 
-iterations
+limit
  = Confi
@@ -878,26 +878,21 @@
 erval = 
-iterations
+limit
  / logIt
@@ -979,26 +979,21 @@
  create(
-iterations
+limit
 , logInt
@@ -1084,34 +1084,29 @@
         int 
-iterations
+limit
 ,%0A          
@@ -1217,26 +1217,21 @@
         
-iterations
+limit
 , logInt
@@ -2398,24 +2398,86 @@
      %7D%0A    %7D
+%0A    %0A    protected static final int INFINITE_ITERATIONS = -1;
 %0A%0A    protec
@@ -2562,26 +2562,21 @@
 nal int 
-iterations
+limit
 ;%0A    pr
@@ -2687,26 +2687,21 @@
     int 
-iterations
+limit
 ,%0A      
@@ -2812,23 +2812,61 @@
 ent(
-iterations %3E= 0
+(limit %3E= 0) %7C%7C (limit == INFINITE_ITERATIONS), limit
 );%0A 
@@ -2951,31 +2951,21 @@
 his.
-iterations = iterations
+limit = limit
 ;%0A  
@@ -3049,26 +3049,21 @@
  int get
-Iterations
+Limit
 () %7B%0A   
@@ -3074,26 +3074,21 @@
  return 
-iterations
+limit
 ;%0A    %7D%0A
@@ -3228,61 +3228,115 @@
     
-this.count++;%0A        checkState(count %3C= iterations)
+if ((limit %3E=0) && (count %3E= limit)) %7B%0A            return Optional.absent();%0A        %7D%0A        this.count++
 ;%0A  
@@ -3394,26 +3394,21 @@
 ount == 
-iterations
+limit
 ) %7C%7C (co
@@ -3545,26 +3545,57 @@
 if (
-count %3C iterations
+(limit == INFINITE_ITERATIONS) %7C%7C (count %3C limit)
 ) %7B%0A

allow infinite iterations