class FilterJSON extends Filter<LogFeederProps> {

  @Override
  public void apply(String inputStr, InputMarker inputMarker) throws Exception {
    Map<String, Object> jsonMap = null;
    try {
      jsonMap = LogFeederUtil.toJSONObject(inputStr);
    } catch (Exception e) {
      LOG.error(e.getLocalizedMessage());
      throw new LogFeederException("Json parsing failed for inputstr = " + inputStr ,e.getCause());
    }
    Double lineNumberD = (Double) jsonMap.get("line_number");
    if (lineNumberD != null) {
      long lineNumber = lineNumberD.longValue();
      jsonMap.put("line_number", lineNumber);
    }
    String timeStampStr = (String) jsonMap.get("logtime");
    if (timeStampStr != null && !timeStampStr.isEmpty()) {
      String logtime = DateUtil.getDate(timeStampStr);
      jsonMap.put("logtime", logtime);
      jsonMap.put(LogFeederConstants.IN_MEMORY_TIMESTAMP, Long.parseLong(timeStampStr));
    }
    super.apply(jsonMap, inputMarker);
  }

}