InfluxDBSink extends AbstractRecordHandler { static boolean checkHistoryAndInsertIfNeeded(Map<String,HistoryObject> historyMap, String alias, Value value, long time){ HistoryObject historyObject = historyMap.get(alias); if(historyObject==null){ historyObject = new HistoryObject(time,value); historyMap.put(alias,historyObject); return true; } if(historyObject.getTime()>time){ return false; } if(historyObject.getValue().equals(value)){ if (historyObject.getTime() <= time - RESEND_DURATION) { historyObject.setTime(time); historyMap.put(alias,historyObject); return true; } return false; } else{ historyObject.setTime(time); historyObject.setValue(value); historyMap.put(alias,historyObject); return true; } } InfluxDBSink(InfluxFactory factory, String measurement); @Override void init(); @Override void apply(MRecord record); @Override Set<String> getChannelIdentifiers(); static final int RESEND_DURATION; }
@Test public void testHistory() { Map<String, InfluxDBSink.HistoryObject> historyObjectMap = new HashMap<>(); String varName = "testVar"; String varName2 = "testVar2"; Value value = Value.of(123.45); long timestamp = 123456L; assertTrue(historyObjectMap.isEmpty()); InfluxDBSink.checkHistoryAndInsertIfNeeded(historyObjectMap,varName,value,timestamp); assertFalse(historyObjectMap.isEmpty()); assertEquals(historyObjectMap.get(varName).getValue(),value); assertEquals(historyObjectMap.get(varName).getTime(),timestamp); value = Value.of(123.46); timestamp = 123457L; InfluxDBSink.checkHistoryAndInsertIfNeeded(historyObjectMap,varName,value,timestamp); assertFalse(historyObjectMap.isEmpty()); assertEquals(historyObjectMap.get(varName).getValue(),value); assertEquals(historyObjectMap.get(varName).getTime(),timestamp); value = Value.of(123.47); timestamp = 123456L; InfluxDBSink.checkHistoryAndInsertIfNeeded(historyObjectMap,varName,value,timestamp); assertFalse(historyObjectMap.isEmpty()); assertNotEquals(historyObjectMap.get(varName).getValue(),value); assertNotEquals(historyObjectMap.get(varName).getTime(),timestamp); value = Value.of(123.46); timestamp = 124457L; InfluxDBSink.checkHistoryAndInsertIfNeeded(historyObjectMap,varName,value,timestamp); assertFalse(historyObjectMap.isEmpty()); assertEquals(historyObjectMap.get(varName).getValue(),value); assertNotEquals(historyObjectMap.get(varName).getTime(),timestamp); value = Value.of(123.47); timestamp = 134456L; InfluxDBSink.checkHistoryAndInsertIfNeeded(historyObjectMap,varName,value,timestamp); assertFalse(historyObjectMap.isEmpty()); assertEquals(historyObjectMap.get(varName).getValue(),value); assertEquals(historyObjectMap.get(varName).getTime(),timestamp); value = Value.of(123.47); timestamp = 134456L; InfluxDBSink.checkHistoryAndInsertIfNeeded(historyObjectMap,varName2,value,timestamp); assertEquals(2,historyObjectMap.size()); assertEquals(historyObjectMap.get(varName2).getValue(),value); assertEquals(historyObjectMap.get(varName2).getTime(),timestamp); }