package org.apache.flume.interceptor;

/**
 * Created by koichi.yanagimoto on 2017/05/15.
 */

import java.time.LocalDateTime;
import java.util.*;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.log4j.Logger;
import org.json.simple.JSONObject;
import org.json.simple.parser.JSONParser;

public class JsonBodyTimestampInterceptor implements Interceptor {
    private static final Logger LOG;

    static {
        LOG = Logger.getLogger(JsonBodyTimestampInterceptor.class);
    }

    private JsonBodyTimestampInterceptor() {}

    @Override
    public void initialize() {}

    @Override
    public Event intercept(Event event) {

        Map<String, String> headers = event.getHeaders();

        String body = new String(event.getBody());
        JSONParser jsonParser = new JSONParser();
        JSONObject jsonObject = null;

        try {
            jsonObject = (JSONObject) jsonParser.parse(body);
        } catch (org.json.simple.parser.ParseException e) {
            LOG.info("Received this log message that is not formatted in json: "+body+"\n");
            return event;
        }

        String timestamp = jsonObject.get("@timestamp").toString();
        LocalDateTime parsedDateTime = LocalDateTime.parse(timestamp);

        headers.put("timestampYear", String.valueOf(parsedDateTime.getYear()));
        headers.put("timestampMonth", String.valueOf(parsedDateTime.getMonthValue()));
        headers.put("timestampDay", String.valueOf(parsedDateTime.getDayOfMonth()));
        headers.put("timestampHour", String.valueOf(parsedDateTime.getHour()));

        return event;
    }

    @Override
    public List<Event> intercept(List<Event> events) {
        for (Event event:events) {
            intercept(event);
        }
        return events;
    }

    @Override
    public void close() {}

    public static class Builder implements Interceptor.Builder {
        @Override
        public Interceptor build() {
            return new JsonBodyTimestampInterceptor();
        }

        @Override
        public void configure(Context context) {}
    }
}
@@ -88,16 +88,36 @@
 5.%0A */%0A%0A
+import java.util.*;%0A
 import j
@@ -140,35 +140,16 @@
 teTime;%0A
-import java.util.*;
 %0Aimport 
@@ -237,16 +237,17 @@
 Logger;%0A
+%0A
 import o
@@ -420,17 +420,16 @@
 er LOG;%0A
-%0A
     stat
@@ -524,40 +524,307 @@
 ate 
-JsonBodyTimestampInterceptor() %7B
+String timestampKey;%0A    protected Context context;%0A%0A    public static final String TIMESTAMP_KEY_NAME = %22properties.timestampKeyName%22;%0A%0A    public JsonBodyTimestampInterceptor(Context context) %7B%0A        this.context = context;%0A        this.timestampKey = context.getString(TIMESTAMP_KEY_NAME);%0A    
 %7D%0A%0A 
@@ -925,17 +925,16 @@
 vent) %7B%0A
-%0A
         
@@ -1436,10 +1436,13 @@
 get(
-%22@
+this.
 time
@@ -1446,17 +1446,19 @@
 imestamp
-%22
+Key
 ).toStri
@@ -2109,24 +2109,26 @@
 close() %7B%7D%0A%0A
+%0A%0A
     public s
@@ -2176,24 +2176,53 @@
 r.Builder %7B%0A
+        private Context ctx;%0A
         @Ove
@@ -2316,16 +2316,19 @@
 rceptor(
+ctx
 );%0A     
@@ -2399,16 +2399,57 @@
 ntext) %7B
+%0A            this.ctx = context;%0A        
 %7D%0A    %7D%0A

move timestamp key from code to property