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, Integer> 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", parsedDateTime.getYear());
        headers.put("timestampMonth", parsedDateTime.getMonthValue());
        headers.put("timestampDay", parsedDateTime.getDayOfMonth());
        headers.put("timestampHour", 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) {}
    }
}

@@ -682,15 +682,14 @@
 ng, 
-Integer
+String
 %3E he
@@ -1296,24 +1296,39 @@
 stampYear%22, 
+String.valueOf(
 parsedDateTi
@@ -1340,16 +1340,17 @@
 tYear())
+)
 ;%0A      
@@ -1381,16 +1381,31 @@
 Month%22, 
+String.valueOf(
 parsedDa
@@ -1427,16 +1427,17 @@
 Value())
+)
 ;%0A      
@@ -1466,16 +1466,31 @@
 mpDay%22, 
+String.valueOf(
 parsedDa
@@ -1512,16 +1512,17 @@
 Month())
+)
 ;%0A      
@@ -1552,16 +1552,31 @@
 pHour%22, 
+String.valueOf(
 parsedDa
@@ -1592,16 +1592,17 @@
 tHour())
+)
 ;%0A%0A     
@@ -2101,10 +2101,9 @@
 %7D%0A    %7D%0A
+
 %7D
-%0A

convert to String.