EventTimeClock implements Clock, Serializable { @Override public Optional<Long> currentTimeMillis(JSONObject message) { Long result; if(message != null && message.containsKey(timestampField)) { Object timestamp = message.get(timestampField); result = ConversionUtils.convert(timestamp, Long.class); } else { LOG.debug("message does not contain timestamp field '{}': message will be ignored: message='{}'", timestampField, JSONObject.toJSONString(message)); result = null; } return Optional.ofNullable(result); } EventTimeClock(String timestampField); @Override Optional<Long> currentTimeMillis(JSONObject message);  }
@Test public void testEventTime() { JSONObject message = createMessage(); final Long timestamp = System.currentTimeMillis(); message.put(timestampField, timestamp); EventTimeClock clock = new EventTimeClock(timestampField); Optional<Long> result = clock.currentTimeMillis(message); assertTrue(result.isPresent()); assertEquals(timestamp, result.get()); }
@Test public void testEventTimeWithString() { JSONObject message = createMessage(); final Long timestamp = System.currentTimeMillis(); message.put(timestampField, timestamp.toString()); EventTimeClock clock = new EventTimeClock(timestampField); Optional<Long> result = clock.currentTimeMillis(message); assertTrue(result.isPresent()); assertEquals(timestamp, result.get()); }
@Test public void testMissingTimestampField() { JSONObject message = createMessage(); EventTimeClock clock = new EventTimeClock(timestampField); Optional<Long> result = clock.currentTimeMillis(message); assertFalse(result.isPresent()); }
@Test public void testInvalidValue() { JSONObject message = createMessage(); message.put(timestampField, "invalid-timestamp-value"); EventTimeClock clock = new EventTimeClock(timestampField); Optional<Long> result = clock.currentTimeMillis(message); assertFalse(result.isPresent()); }