@Test(expected = RuntimeException.class) public void testBadJsonInput() { byte[] bytePayload = "{\"key: value}".getBytes(); JsonStringMessageDecoder testDecoder = new JsonStringMessageDecoder(); testDecoder.decode(bytePayload); }
@Override public CamusWrapper<String> decode(byte[] payload) { long timestamp = 0; String payloadString; JsonObject jsonObject; payloadString = new String(payload); try { jsonObject = jsonParser.parse(payloadString.trim()).getAsJsonObject(); } catch (RuntimeException e) { log.error("Caught exception while parsing JSON string '" + payloadString + "'."); throw new RuntimeException(e); } if (jsonObject.has(timestampField)) { if (timestampFormat.equals("unix_seconds") || timestampFormat.equals("unix")) { timestamp = jsonObject.get(timestampField).getAsLong(); timestamp = timestamp * 1000L; } else if (timestampFormat.equals("unix_milliseconds")) { timestamp = jsonObject.get(timestampField).getAsLong(); } else if (timestampFormat.equals("ISO-8601")) { String timestampString = jsonObject.get(timestampField).getAsString(); try { timestamp = new DateTime(timestampString).getMillis(); } catch (IllegalArgumentException e) { log.error("Could not parse timestamp '" + timestampString + "' as ISO-8601 while decoding JSON message."); } } else { String timestampString = jsonObject.get(timestampField).getAsString(); try { timestamp = dateTimeParser.parseDateTime(timestampString).getMillis(); } catch (IllegalArgumentException e) { try { timestamp = new SimpleDateFormat(timestampFormat).parse(timestampString).getTime(); } catch (ParseException pe) { log.error("Could not parse timestamp '" + timestampString + "' while decoding JSON message."); } } catch (Exception ee) { log.error("Could not parse timestamp '" + timestampString + "' while decoding JSON message."); } } } if (timestamp == 0) { log.warn("Couldn't find or parse timestamp field '" + timestampField + "' in JSON message, defaulting to current time."); timestamp = System.currentTimeMillis(); } return new CamusWrapper<String>(payloadString, timestamp); }
JsonStringMessageDecoder extends MessageDecoder<byte[], String> { @Override public CamusWrapper<String> decode(byte[] payload) { long timestamp = 0; String payloadString; JsonObject jsonObject; payloadString = new String(payload); try { jsonObject = jsonParser.parse(payloadString.trim()).getAsJsonObject(); } catch (RuntimeException e) { log.error("Caught exception while parsing JSON string '" + payloadString + "'."); throw new RuntimeException(e); } if (jsonObject.has(timestampField)) { if (timestampFormat.equals("unix_seconds") || timestampFormat.equals("unix")) { timestamp = jsonObject.get(timestampField).getAsLong(); timestamp = timestamp * 1000L; } else if (timestampFormat.equals("unix_milliseconds")) { timestamp = jsonObject.get(timestampField).getAsLong(); } else if (timestampFormat.equals("ISO-8601")) { String timestampString = jsonObject.get(timestampField).getAsString(); try { timestamp = new DateTime(timestampString).getMillis(); } catch (IllegalArgumentException e) { log.error("Could not parse timestamp '" + timestampString + "' as ISO-8601 while decoding JSON message."); } } else { String timestampString = jsonObject.get(timestampField).getAsString(); try { timestamp = dateTimeParser.parseDateTime(timestampString).getMillis(); } catch (IllegalArgumentException e) { try { timestamp = new SimpleDateFormat(timestampFormat).parse(timestampString).getTime(); } catch (ParseException pe) { log.error("Could not parse timestamp '" + timestampString + "' while decoding JSON message."); } } catch (Exception ee) { log.error("Could not parse timestamp '" + timestampString + "' while decoding JSON message."); } } } if (timestamp == 0) { log.warn("Couldn't find or parse timestamp field '" + timestampField + "' in JSON message, defaulting to current time."); timestamp = System.currentTimeMillis(); } return new CamusWrapper<String>(payloadString, timestamp); } }
JsonStringMessageDecoder extends MessageDecoder<byte[], String> { @Override public CamusWrapper<String> decode(byte[] payload) { long timestamp = 0; String payloadString; JsonObject jsonObject; payloadString = new String(payload); try { jsonObject = jsonParser.parse(payloadString.trim()).getAsJsonObject(); } catch (RuntimeException e) { log.error("Caught exception while parsing JSON string '" + payloadString + "'."); throw new RuntimeException(e); } if (jsonObject.has(timestampField)) { if (timestampFormat.equals("unix_seconds") || timestampFormat.equals("unix")) { timestamp = jsonObject.get(timestampField).getAsLong(); timestamp = timestamp * 1000L; } else if (timestampFormat.equals("unix_milliseconds")) { timestamp = jsonObject.get(timestampField).getAsLong(); } else if (timestampFormat.equals("ISO-8601")) { String timestampString = jsonObject.get(timestampField).getAsString(); try { timestamp = new DateTime(timestampString).getMillis(); } catch (IllegalArgumentException e) { log.error("Could not parse timestamp '" + timestampString + "' as ISO-8601 while decoding JSON message."); } } else { String timestampString = jsonObject.get(timestampField).getAsString(); try { timestamp = dateTimeParser.parseDateTime(timestampString).getMillis(); } catch (IllegalArgumentException e) { try { timestamp = new SimpleDateFormat(timestampFormat).parse(timestampString).getTime(); } catch (ParseException pe) { log.error("Could not parse timestamp '" + timestampString + "' while decoding JSON message."); } } catch (Exception ee) { log.error("Could not parse timestamp '" + timestampString + "' while decoding JSON message."); } } } if (timestamp == 0) { log.warn("Couldn't find or parse timestamp field '" + timestampField + "' in JSON message, defaulting to current time."); timestamp = System.currentTimeMillis(); } return new CamusWrapper<String>(payloadString, timestamp); }  }
JsonStringMessageDecoder extends MessageDecoder<byte[], String> { @Override public CamusWrapper<String> decode(byte[] payload) { long timestamp = 0; String payloadString; JsonObject jsonObject; payloadString = new String(payload); try { jsonObject = jsonParser.parse(payloadString.trim()).getAsJsonObject(); } catch (RuntimeException e) { log.error("Caught exception while parsing JSON string '" + payloadString + "'."); throw new RuntimeException(e); } if (jsonObject.has(timestampField)) { if (timestampFormat.equals("unix_seconds") || timestampFormat.equals("unix")) { timestamp = jsonObject.get(timestampField).getAsLong(); timestamp = timestamp * 1000L; } else if (timestampFormat.equals("unix_milliseconds")) { timestamp = jsonObject.get(timestampField).getAsLong(); } else if (timestampFormat.equals("ISO-8601")) { String timestampString = jsonObject.get(timestampField).getAsString(); try { timestamp = new DateTime(timestampString).getMillis(); } catch (IllegalArgumentException e) { log.error("Could not parse timestamp '" + timestampString + "' as ISO-8601 while decoding JSON message."); } } else { String timestampString = jsonObject.get(timestampField).getAsString(); try { timestamp = dateTimeParser.parseDateTime(timestampString).getMillis(); } catch (IllegalArgumentException e) { try { timestamp = new SimpleDateFormat(timestampFormat).parse(timestampString).getTime(); } catch (ParseException pe) { log.error("Could not parse timestamp '" + timestampString + "' while decoding JSON message."); } } catch (Exception ee) { log.error("Could not parse timestamp '" + timestampString + "' while decoding JSON message."); } } } if (timestamp == 0) { log.warn("Couldn't find or parse timestamp field '" + timestampField + "' in JSON message, defaulting to current time."); timestamp = System.currentTimeMillis(); } return new CamusWrapper<String>(payloadString, timestamp); }  @Override void init(Properties props, String topicName); @Override CamusWrapper<String> decode(byte[] payload); }
JsonStringMessageDecoder extends MessageDecoder<byte[], String> { @Override public CamusWrapper<String> decode(byte[] payload) { long timestamp = 0; String payloadString; JsonObject jsonObject; payloadString = new String(payload); try { jsonObject = jsonParser.parse(payloadString.trim()).getAsJsonObject(); } catch (RuntimeException e) { log.error("Caught exception while parsing JSON string '" + payloadString + "'."); throw new RuntimeException(e); } if (jsonObject.has(timestampField)) { if (timestampFormat.equals("unix_seconds") || timestampFormat.equals("unix")) { timestamp = jsonObject.get(timestampField).getAsLong(); timestamp = timestamp * 1000L; } else if (timestampFormat.equals("unix_milliseconds")) { timestamp = jsonObject.get(timestampField).getAsLong(); } else if (timestampFormat.equals("ISO-8601")) { String timestampString = jsonObject.get(timestampField).getAsString(); try { timestamp = new DateTime(timestampString).getMillis(); } catch (IllegalArgumentException e) { log.error("Could not parse timestamp '" + timestampString + "' as ISO-8601 while decoding JSON message."); } } else { String timestampString = jsonObject.get(timestampField).getAsString(); try { timestamp = dateTimeParser.parseDateTime(timestampString).getMillis(); } catch (IllegalArgumentException e) { try { timestamp = new SimpleDateFormat(timestampFormat).parse(timestampString).getTime(); } catch (ParseException pe) { log.error("Could not parse timestamp '" + timestampString + "' while decoding JSON message."); } } catch (Exception ee) { log.error("Could not parse timestamp '" + timestampString + "' while decoding JSON message."); } } } if (timestamp == 0) { log.warn("Couldn't find or parse timestamp field '" + timestampField + "' in JSON message, defaulting to current time."); timestamp = System.currentTimeMillis(); } return new CamusWrapper<String>(payloadString, timestamp); }  @Override void init(Properties props, String topicName); @Override CamusWrapper<String> decode(byte[] payload); static final String CAMUS_MESSAGE_TIMESTAMP_FORMAT; static final String DEFAULT_TIMESTAMP_FORMAT; static final String CAMUS_MESSAGE_TIMESTAMP_FIELD; static final String DEFAULT_TIMESTAMP_FIELD; }