VPCFlowLogParser { public List<Map<String, Object>> parseFlowLogMap(String json) throws InterceptException { try { List<Map<String, Object>> events = new ArrayList<>(); JsonObject flowLogs = gson.fromJson(json, JsonObject.class); String logGroup = flowLogs.get(LOG_GROUP).getAsString(); String logStream = flowLogs.get(LOG_STREAM).getAsString(); for (JsonElement element : flowLogs.get(LOG_EVENTS).getAsJsonArray()) { Map<String, Object> map = new LinkedHashMap<>(); JsonObject obj = element.getAsJsonObject(); map.put(LOG_GROUP, logGroup); map.put(LOG_STREAM, logStream); map.put(TIMESTAMP, obj.get(TIMESTAMP2).getAsLong()); map.put(ID, obj.get(ID).getAsString()); parseToRecord(map, obj.get(MESSAGE).getAsString()); events.add(map); } return events; } catch (Exception e) { throw new InterceptException(e); } } VPCFlowLogParser(); List<Map<String, Object>> parseFlowLogMap(String json); static void parseToRecord(Map<String, Object> map, String line); List<Event> parseFlowLogJson(EventFactory factory, String json); static void parseToRecord(Event event, String line); static VPCFlowLogRecord parseToRecord(String line);  }
@Test public void testFlowLogParse() throws InterceptException { VPCFlowLogParser parser = new VPCFlowLogParser(); String event = "{\"messageType\":\"DATA_MESSAGE\",\"owner\":\"123456789115\",\"logGroup\":\"vpc-flow-log-group\",\"logStream\"" + ":\"eni-00fv0000-all\",\"subscriptionFilters\":[\"cloudwatch_flowlog_lambda_subscription\"],\"logEvents\":[{\"id\":\"327880164000000002313720044516500000000\",\"timestamp\":1470265507000,\"message\":\"2 123456789115 eni-0625b517 100.100.10.100 100.220.1.1 49162 5938 6 33 1848 1470265507 1470266096 ACCEPT OK\"},{\"id\":\"327880164000000002313720044516500000000\",\"timestamp\":123456789115,\"message\":\"2 686559647175 eni-0000b000 120.220.2.1 102.123.34.145 5938 49162 6 22 1144 1470265507 1470266096 ACCEPT OK\"}]}"; List<Map<String, Object>> result = parser.parseFlowLogMap(event); assertEquals(2, result.size()); }