InfluxDBImpl implements DBInterface, EventListener<AbstractWriteEvent> { @Override public void executeStore(TSData tsData) { long startTime = 0; if (LOG.isDebugEnabled()) { startTime = System.currentTimeMillis(); } synchronized (m_busyMutex) { while (m_busy) { try { LOG.error("Waiting"); m_busyMutex.wait(); LOG.error("Waiting has an end."); } catch (InterruptedException ex) { LOG.error("Waiting for busy mutex interrupted.", ex); return; } } m_busy = true; } long numberOfPoints = 0; for (TSPoint p : tsData.getTSPoints()) { Point point = Point .measurement(p.m_measurement) .time((long)p.m_timestamp, WritePrecision.S); p.m_tags.keySet().forEach((key) -> { point.addTag(key, p.m_tags.get(key)); }); if (p.m_fields == null) { continue; } p.m_fields.keySet().forEach((String key) -> { Object val = p.m_fields.get(key); if (val instanceof Long) { point.addField(key, (Long)val); } else if (val instanceof Double) { point.addField(key, (Double)val); } else if (val instanceof Float) { point.addField(key, (Number)val); } else if (val instanceof String) { point.addField(key, (String)val); } else if (val instanceof Boolean) { point.addField(key, (Boolean)val); } else if (val != null) { LOG.error("Type of field not supported: {}", val.getClass()); } }); m_writeApi.writePoint(m_bucketId, m_organisation, point); numberOfPoints++; } m_writeApi.flush(); if (LOG.isDebugEnabled()) { long elapsedTime = System.currentTimeMillis() - startTime; LOG.debug("Added {} points in {} milliseconds.", numberOfPoints, 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 test1ExecuteStore() { System.out.println("executeStore"); TSData tsData = fillTSData(); instance.executeStore(tsData); }