package com.ctrip.ops.sysdev.filters;

import java.util.*;

import com.ctrip.ops.sysdev.baseplugin.BaseFilter;

import lombok.extern.log4j.Log4j2;

@Log4j2
public class Metric extends BaseFilter {


    int windowSize;
    String key;
    String value;
    Map<String, Object> metric;
    Map<String, Object> metricToEmit;
    long lastEmitTime;

    public Metric(Map config) {
        super(config);
    }

    protected void prepare() {
        this.key = (String) config.get("key");
        this.value = (String) config.get("value");
        this.windowSize = (int) config.get("windowSize") * 1000;
        this.processExtraEventsFunc = true;
        this.metric = new HashMap();
        this.metricToEmit = new HashMap();

        this.lastEmitTime = System.currentTimeMillis();
    }

    @Override
    protected Map filter(final Map event) {
        if (System.currentTimeMillis() >= this.windowSize + this.lastEmitTime) {
            this.metricToEmit = this.metric;
            this.metric = new HashMap();
        }
        if (event.containsKey(this.key) && event.containsKey(this.value)) {
            String keyValue = event.get(this.key).toString();
            Object valueValue = event.get(this.value);
            HashMap set = (HashMap) this.metric.get(keyValue);
            if (set == null) {
                set = new HashMap();
                set.put(valueValue, 1);
            } else {
                if (set.containsKey(valueValue)) {
                    int count = (int) set.get(valueValue);
                    set.put(valueValue, count + 1);
                } else {
                    set.put(valueValue, 1);
                }
            }
            this.metric.put(keyValue, set);
        }
        return event;
    }

    @Override
    public void filterExtraEvents(Stack<Map<String, Object>> to_st) {
        if (metricToEmit.size() == 0) {
            return;
        }
        this.metricToEmit.put("@timestamp", this.lastEmitTime);
        this.postProcess(this.metricToEmit, true);
        to_st.add(this.metricToEmit);
        this.metricToEmit = new HashMap();
        this.lastEmitTime = System.currentTimeMillis();
        return;
    }
}

@@ -105,16 +105,67 @@
 ilter;%0A%0A
+import com.ctrip.ops.sysdev.baseplugin.BaseOutput;%0A
 import l
@@ -191,16 +191,16 @@
 Log4j2;%0A
-
 %0A@Log4j2
@@ -1051,32 +1051,273 @@
 = new HashMap();
+%0A%0A            for (BaseFilter f : this.nextFilters%0A            ) %7B%0A                f.process(metricToEmit);%0A            %7D%0A            for (BaseOutput o : this.nextOutputs%0A            ) %7B%0A                o.process(metricToEmit);%0A            %7D
 %0A        %7D%0A     
@@ -2005,32 +2005,32 @@
 set);%0A        %7D%0A
+
         return e
@@ -2044,436 +2044,7 @@
    %7D
-%0A%0A    @Override%0A    public void filterExtraEvents(Stack%3CMap%3CString, Object%3E%3E to_st) %7B%0A        if (metricToEmit.size() == 0) %7B%0A            return;%0A        %7D%0A        this.metricToEmit.put(%22@timestamp%22, this.lastEmitTime);%0A        this.postProcess(this.metricToEmit, true);%0A        to_st.add(this.metricToEmit);%0A        this.metricToEmit = new HashMap();%0A        this.lastEmitTime = System.currentTimeMillis();%0A        return;%0A    %7D
 %0A%7D%0A

use next