import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
/**
 * Executes M tasks every N timeunits
 * @author eolivelli@gmail.com
 */
public class Throttler {
    private static class Task implements Delayed {
        long trigger;
        Task(long i) {
            trigger = System.nanoTime() + i;
        }
        public int compareTo(Delayed y) {
            long i = trigger;
            long j = ((Task) y).trigger;
            if (i < j) {
                return -1;
            }
            if (i > j) {
                return 1;
            }
            return 0;
        }
        public boolean equals(Object other) {
            return ((Task) other).trigger == trigger;
        }
        public long getDelay(TimeUnit unit) {
            long n = trigger - System.nanoTime();
            return unit.convert(n, TimeUnit.NANOSECONDS);
        }
    }
    private DelayQueue window = new DelayQueue();
    private int numTasks;
    private int timeUnits;
    private TimeUnit timeUnit;
    private ExecutorService executor;
    private LinkedBlockingQueue tasks = new LinkedBlockingQueue();
    private Thread thread;
    private boolean requestedstop;
    public Throttler(int numTasks, int timeUnits, TimeUnit timeUnit, ExecutorService executor) {
        this.numTasks = numTasks;
        this.timeUnits = timeUnits;
        this.timeUnit = timeUnit;
        this.executor = executor;
        for (int i = 0; i < numTasks; i++) {
            window.add(new Task(0));
        }
        thread = new Thread(new Runner());
        thread.setDaemon(false);
        thread.start();
    }
    public void submit(Runnable task) {
        tasks.add(task);
    }
    public void stop() {
        requestedstop = true;
    }
    
    
    private class Runner implements Runnable {
        @Override
        public void run() {
            while (!requestedstop) {
                try {
//                    System.out.println("Waiting for a task..." + System.currentTimeMillis() + " " + window);
                    window.take();
//                    System.out.println("poll a task.."+tasks);
                    Runnable r = tasks.poll();
                    if (r != null) {
                        //System.out.println("executing task " + r);
                        executor.submit(r);
                    }
                    long delay = TimeUnit.NANOSECONDS.convert(timeUnits, timeUnit);
//                    System.out.println("delay=" + delay);
                    window.add(new Task(delay));
                } catch (InterruptedException ex) {
                    return;
                }
            }
        }
    }
}