package com.github.phantomthief.collection.impl;

import static com.github.phantomthief.collection.impl.SimpleBufferTrigger.TriggerResult.trig;
import static com.google.common.base.Preconditions.checkArgument;
import static java.lang.Math.abs;
import static java.lang.Math.min;
import static java.lang.System.*;

import java.util.Map.Entry;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.concurrent.TimeUnit;

import com.github.phantomthief.collection.impl.SimpleBufferTrigger.TriggerResult;
import com.github.phantomthief.collection.impl.SimpleBufferTrigger.TriggerStrategy;

/**
 * trigger like redis's rdb
 *
 * save 900 1
 * save 300 10
 * save 60 10000
 *
 * @author w.vela
 * Created on 15/07/2016.
 */
public class MultiIntervalTriggerStrategy implements TriggerStrategy {

    private long minTriggerPeriod = Long.MAX_VALUE;
    private final SortedMap<Long, Long> triggerMap = new TreeMap<>();

    public MultiIntervalTriggerStrategy on(long interval, TimeUnit unit, long count) {
        long intervalInMs = unit.toMillis(interval);
        triggerMap.put(intervalInMs, count);
        minTriggerPeriod = checkAndCalcMinPeriod();
        return this;
    }

    long minTriggerPeriod() { // for test case
        return minTriggerPeriod;
    }

    private long checkAndCalcMinPeriod() {
        long minPeriod = Long.MAX_VALUE;
        Long maxTrigChangeCount = null;
        long lastPeriod = 0;

        for (Entry<Long, Long> entry : triggerMap.entrySet()) {
            long period = entry.getKey();
            minPeriod = min(minPeriod, period);
            if (lastPeriod > 0) {
                minPeriod = min(minPeriod, abs(lastPeriod - period));
            }
            long trigChangedCount = entry.getValue();
            if (maxTrigChangeCount == null) {
                maxTrigChangeCount = trigChangedCount;
            } else {
                if (maxTrigChangeCount <= trigChangedCount) {
                    throw new IllegalArgumentException(
                            "found invalid trigger setting:" + triggerMap);
                }
            }
            lastPeriod = period;
        }
        return minPeriod;
    }

    @Override
    public TriggerResult canTrigger(long lastConsumeTimestamp, long changedCount) {
        checkArgument(!triggerMap.isEmpty());

        boolean doConsumer = false;

        long now = currentTimeMillis();

        for (Entry<Long, Long> entry : triggerMap.entrySet()) {
            if (now - lastConsumeTimestamp < entry.getKey()) {
                continue;
            }
            if (changedCount >= entry.getValue()) {
                doConsumer = true;
                break;
            }
        }
        return trig(doConsumer, minTriggerPeriod);
    }
}

@@ -302,17 +302,33 @@
 .System.
-*
+currentTimeMillis
 ;%0A%0Aimpor
@@ -694,16 +694,280 @@
 0000%0A *%0A
+ * %E6%B3%A8%E6%84%8F%EF%BC%8C%E8%BF%99%E9%87%8C%E7%9A%84%E8%A7%A6%E5%8F%91%E6%9D%A1%E4%BB%B6%E5%92%8C Kafka Producer %E7%9A%84 linger %E4%B8%8D%E4%B8%80%E6%A0%B7%EF%BC%8C%E5%B9%B6%E4%B8%8D%E6%98%AF %E6%97%B6%E9%97%B4%E5%92%8C%E6%AC%A1%E6%95%B0%E8%BE%BE%E5%88%B0%E4%B8%80%E4%B8%AA%E5%B0%B1%E8%A7%A6%E5%8F%91%E4%B8%80%E6%AC%A1%0A * %E8%80%8C%E6%98%AF%E6%8C%87%E5%AE%9A%E6%97%B6%E9%97%B4%E5%86%85%E7%B4%AF%E8%AE%A1%E5%88%B0%E7%89%B9%E5%AE%9A%E6%AC%A1%E6%95%B0%E6%89%8D%E8%A7%A6%E5%8F%91%EF%BC%8C%E6%89%80%E4%BB%A5%E4%B8%80%E8%88%AC%E4%BC%9A%E6%9C%89%E4%B8%80%E4%B8%AA%E5%85%9C%E5%BA%95%E7%9A%84 1 %E6%AC%A1%E6%9D%A5%E5%81%9A%E6%97%B6%E9%97%B4%E8%A7%A6%E5%8F%91%0A *%0A * %E6%9B%B4%E8%BF%9B%E4%B8%80%E6%AD%A5%EF%BC%8C%E6%88%91%E4%BB%AC%E7%9A%84%E5%9C%BA%E6%99%AF%E5%A4%A7%E5%A4%9A%E6%95%B0%E6%97%B6%E5%80%99%E4%B8%80%E8%88%AC%E4%BD%BF%E7%94%A8 %7B@link SimpleBufferTriggerBuilder#interval%7D %E8%A7%A6%E5%8F%91%E5%B0%B1%E8%83%BD%E6%BB%A1%E8%B6%B3%E4%B8%9A%E5%8A%A1%E7%9A%84%E5%9F%BA%E6%9C%AC%E9%9C%80%E6%B1%82%0A * %E6%89%80%E4%BB%A5%E7%9B%AE%E5%89%8D %7B@link SimpleBufferTrigger%7D %E6%B2%A1%E6%9C%89%E6%8F%90%E4%BE%9B%E7%B1%BB Kafka Producer %E7%9A%84 linger %E8%A7%A6%E5%8F%91%E6%A8%A1%E5%BC%8F%EF%BC%9B%0A *%0A
  * @auth

add comments for multi interval trigger.