InfluxDBImpl implements DBInterface, EventListener<AbstractWriteEvent> { @Override public void executeQuery(String measurement, Instant startTime, Instant stopTime, Map<String, List<String>> filter, List<TSData> results) { long startQueryTime = 0; if (LOG.isDebugEnabled()) { startQueryTime = System.currentTimeMillis(); } StringBuilder query = new StringBuilder() .append("from(bucket: \"").append(m_bucketId).append("\") ") .append("|> range(start: ").append(startTime).append(", stop: ").append(stopTime).append(") ") .append("|> filter(fn: (r) => r._measurement == \"").append(measurement).append("\")"); appendFilter(filter, query); LOG.debug("filter: {}", query.toString()); List<FluxTable> tables = m_dbClient.getQueryApi().query(query.toString(), m_organisation); LOG.debug("result: {} tables", tables.size()); for (FluxTable table : tables) { TSData tsData = new TSData(); for (FluxRecord fluxRecord : table.getRecords()) { TSPoint point = new TSPoint(); point.m_measurement = (String)fluxRecord.getValueByKey("_measurement"); Instant time = fluxRecord.getTime(); if (time != null) { point.m_timestamp = (int)time.getEpochSecond(); } else { point.m_timestamp = 0; LOG.error("Timestamp from DB is null"); } Object val = fluxRecord.getValueByKey("_value"); point.m_fields = new HashMap<>(1); if (val instanceof Long) { point.m_fields.put((String)fluxRecord.getValueByKey("_field"), (Long)val); } else if (val instanceof Double) { point.m_fields.put((String)fluxRecord.getValueByKey("_field"), (Double)val); } else if (val instanceof String) { point.m_fields.put((String)fluxRecord.getValueByKey("_field"), (String)val); } else if (val instanceof Boolean) { point.m_fields.put((String)fluxRecord.getValueByKey("_field"), (Boolean)val); } else if (val != null) { LOG.error("Unsupported value type {}", val.getClass()); } point.m_tags.put("deviceAdapter", (String)fluxRecord.getValueByKey("deviceAdapter")); point.m_tags.put("hostName", (String)fluxRecord.getValueByKey("hostName")); point.m_tags.put("sourceIP", (String)fluxRecord.getValueByKey("sourceIP")); point.m_tags.put("templateID", (String)fluxRecord.getValueByKey("templateID")); point.m_tags.put("observationDomain", (String)fluxRecord.getValueByKey("observationDomain")); tsData.m_tsPoints.add(point); } results.add(tsData); } if (LOG.isDebugEnabled()) { long elapsedTime = System.currentTimeMillis() - startQueryTime; LOG.debug("Query returned in {} milliseconds.", elapsedTime); } } InfluxDBImpl(); @Override void executeStore(TSData tsData); @Override void executeQuery(String measurement, Instant startTime, Instant stopTime,
        Map<String, List<String>> filter, List<TSData> results); @Override boolean openDB(); @Override void shutdownDB(); @Override void onEvent(AbstractWriteEvent value);  }
@Test public void test2ExecuteQuery() { System.out.println("executeQuery"); String measurement = "unittest"; Map<String, List<String>> filter = null; List<TSData> results = new ArrayList<>(); instance.executeQuery(measurement, startTime, stopTime, filter, results); assertEquals(3, results.size()); assertEquals(MEASUREMENTS, results.get(0).m_tsPoints.size()); assertEquals(MEASUREMENTS, results.get(1).m_tsPoints.size()); showResult(results); }