ReportingWorker implements Runnable { @Override public void run() { try { Map<AggregationKey, AggregationValue> removedKeysMinute = process(reportingDao.averageAggregator.getMinute(), GraphGranularityType.MINUTE); Map<AggregationKey, AggregationValue> removedKeysHour = process(reportingDao.averageAggregator.getHourly(), GraphGranularityType.HOURLY); Map<AggregationKey, AggregationValue> removedKeysDay = process(reportingDao.averageAggregator.getDaily(), GraphGranularityType.DAILY); reportingDBManager.insertReporting(removedKeysMinute, GraphGranularityType.MINUTE); reportingDBManager.insertReporting(removedKeysHour, GraphGranularityType.HOURLY); reportingDBManager.insertReporting(removedKeysDay, GraphGranularityType.DAILY); reportingDBManager.insertReportingRaw(reportingDao.rawDataProcessor.rawStorage); reportingDBManager.cleanOldReportingRecords(Instant.now()); } catch (Exception e) { log.error("Error during reporting job.", e); } } ReportingWorker(ReportingDiskDao reportingDao,
                           String reportingPath, ReportingDBManager reportingDBManager); @Override void run();  }
@Test public void testFailure() { User user = new User(); user.email = "test"; user.appName = AppNameUtil.BLYNK; ReportingWorker reportingWorker = new ReportingWorker(reportingDaoMock, reportingFolder, new ReportingDBManager(blockingIOProcessor, true)); ConcurrentHashMap<AggregationKey, AggregationValue> map = new ConcurrentHashMap<>(); long ts = getTS() / AverageAggregatorProcessor.HOUR; AggregationKey aggregationKey = new AggregationKey("ddd\0+123@gmail.com", AppNameUtil.BLYNK, 1, 0, PinType.ANALOG, (short) 1, ts); AggregationValue aggregationValue = new AggregationValue(); aggregationValue.update(100); map.put(aggregationKey, aggregationValue); when(averageAggregator.getMinute()).thenReturn(map); reportingWorker.run(); assertTrue(map.isEmpty()); }
@Test public void testStore() { User user = new User(); user.email = "test"; user.appName = AppNameUtil.BLYNK; ReportingWorker reportingWorker = new ReportingWorker(reportingDaoMock, reportingFolder, new ReportingDBManager(blockingIOProcessor, true)); ConcurrentHashMap<AggregationKey, AggregationValue> map = new ConcurrentHashMap<>(); long ts = getTS() / AverageAggregatorProcessor.HOUR; AggregationKey aggregationKey = new AggregationKey("test", AppNameUtil.BLYNK, 1, 0, PinType.ANALOG, (short) 1, ts); AggregationValue aggregationValue = new AggregationValue(); aggregationValue.update(100); AggregationKey aggregationKey2 = new AggregationKey("test", AppNameUtil.BLYNK, 1, 0, PinType.ANALOG, (short) 1, ts - 1); AggregationValue aggregationValue2 = new AggregationValue(); aggregationValue2.update(150.54); AggregationKey aggregationKey3 = new AggregationKey("test2", AppNameUtil.BLYNK, 2, 0, PinType.ANALOG, (short) 2, ts); AggregationValue aggregationValue3 = new AggregationValue(); aggregationValue3.update(200); map.put(aggregationKey, aggregationValue); map.put(aggregationKey2, aggregationValue2); map.put(aggregationKey3, aggregationValue3); when(averageAggregator.getMinute()).thenReturn(new ConcurrentHashMap<>()); when(averageAggregator.getHourly()).thenReturn(map); when(averageAggregator.getDaily()).thenReturn(new ConcurrentHashMap<>()); reportingWorker.run(); assertTrue(Files.exists(Paths.get(reportingFolder, "test", generateFilename(1, 0, PinType.ANALOG, (short) 1, GraphGranularityType.HOURLY)))); assertTrue(Files.exists(Paths.get(reportingFolder, "test2", generateFilename(2, 0, PinType.ANALOG, (short) 2, GraphGranularityType.HOURLY)))); assertTrue(map.isEmpty()); ByteBuffer data = reportingDaoMock.getByteBufferFromDisk(user, 1, 0, PinType.ANALOG, (short) 1, 2, GraphGranularityType.HOURLY, 0); assertNotNull(data); assertEquals(32, data.capacity()); assertEquals(150.54, data.getDouble(), 0.001); assertEquals((ts - 1) * AverageAggregatorProcessor.HOUR, data.getLong()); assertEquals(100.0, data.getDouble(), 0.001); assertEquals(ts * AverageAggregatorProcessor.HOUR, data.getLong()); User user2 = new User(); user2.email = "test2"; user2.appName = AppNameUtil.BLYNK; data = reportingDaoMock.getByteBufferFromDisk(user2, 2, 0, PinType.ANALOG, (short) 2, 1, GraphGranularityType.HOURLY, 0); assertNotNull(data); assertEquals(16, data.capacity()); assertEquals(200.0, data.getDouble(), 0.001); assertEquals(ts * AverageAggregatorProcessor.HOUR, data.getLong()); }
@Test public void testStore2() { ReportingWorker reportingWorker = new ReportingWorker(reportingDaoMock, reportingFolder, new ReportingDBManager(blockingIOProcessor, true)); ConcurrentHashMap<AggregationKey, AggregationValue> map = new ConcurrentHashMap<>(); long ts = getTS() / AverageAggregatorProcessor.HOUR; AggregationKey aggregationKey = new AggregationKey("test", AppNameUtil.BLYNK, 1, 0, PinType.ANALOG, (short) 1, ts); AggregationValue aggregationValue = new AggregationValue(); aggregationValue.update(100); AggregationKey aggregationKey2 = new AggregationKey("test", AppNameUtil.BLYNK, 1, 0, PinType.ANALOG, (short) 1, ts - 1); AggregationValue aggregationValue2 = new AggregationValue(); aggregationValue2.update(150.54); AggregationKey aggregationKey3 = new AggregationKey("test", AppNameUtil.BLYNK, 1, 0, PinType.ANALOG, (short) 1, ts - 2); AggregationValue aggregationValue3 = new AggregationValue(); aggregationValue3.update(200); map.put(aggregationKey, aggregationValue); map.put(aggregationKey2, aggregationValue2); map.put(aggregationKey3, aggregationValue3); when(averageAggregator.getMinute()).thenReturn(new ConcurrentHashMap<>()); when(averageAggregator.getHourly()).thenReturn(map); when(averageAggregator.getDaily()).thenReturn(new ConcurrentHashMap<>()); reportingWorker.run(); assertTrue(Files.exists(Paths.get(reportingFolder, "test", generateFilename(1, 0, PinType.ANALOG, (short) 1, GraphGranularityType.HOURLY)))); assertTrue(map.isEmpty()); User user = new User(); user.email = "test"; user.appName = AppNameUtil.BLYNK; ByteBuffer data = reportingDaoMock.getByteBufferFromDisk(user, 1, 0, PinType.ANALOG, (short) 1, 1, GraphGranularityType.HOURLY, 0); assertNotNull(data); assertEquals(16, data.capacity()); assertEquals(100.0, data.getDouble(), 0.001); assertEquals(ts * AverageAggregatorProcessor.HOUR, data.getLong()); data = reportingDaoMock.getByteBufferFromDisk(user, 1, 0, PinType.ANALOG, (short) 1, 24, GraphGranularityType.HOURLY, 0); assertNotNull(data); assertEquals(48, data.capacity()); assertEquals(200.0, data.getDouble(), 0.001); assertEquals((ts - 2) * AverageAggregatorProcessor.HOUR, data.getLong()); assertEquals(150.54, data.getDouble(), 0.001); assertEquals((ts - 1) * AverageAggregatorProcessor.HOUR, data.getLong()); assertEquals(100.0, data.getDouble(), 0.001); assertEquals(ts * AverageAggregatorProcessor.HOUR, data.getLong()); }
@Test public void testDeleteCommand() { ReportingWorker reportingWorker = new ReportingWorker(reportingDaoMock, reportingFolder, new ReportingDBManager(blockingIOProcessor, true)); ConcurrentHashMap<AggregationKey, AggregationValue> map = new ConcurrentHashMap<>(); long ts = getTS() / AverageAggregatorProcessor.HOUR; AggregationKey aggregationKey = new AggregationKey("test", AppNameUtil.BLYNK, 1, 0, PinType.ANALOG, (short) 1, ts); AggregationValue aggregationValue = new AggregationValue(); aggregationValue.update(100); AggregationKey aggregationKey2 = new AggregationKey("test", AppNameUtil.BLYNK, 1, 0, PinType.ANALOG, (short) 1, ts - 1); AggregationValue aggregationValue2 = new AggregationValue(); aggregationValue2.update(150.54); AggregationKey aggregationKey3 = new AggregationKey("test2", AppNameUtil.BLYNK, 2, 0, PinType.ANALOG, (short) 2, ts); AggregationValue aggregationValue3 = new AggregationValue(); aggregationValue3.update(200); map.put(aggregationKey, aggregationValue); map.put(aggregationKey2, aggregationValue2); map.put(aggregationKey3, aggregationValue3); when(averageAggregator.getMinute()).thenReturn(new ConcurrentHashMap<>()); when(averageAggregator.getHourly()).thenReturn(map); when(averageAggregator.getDaily()).thenReturn(new ConcurrentHashMap<>()); when(properties.getProperty("data.folder")).thenReturn(System.getProperty("java.io.tmpdir")); reportingWorker.run(); assertTrue(Files.exists(Paths.get(reportingFolder, "test", generateFilename(1, 0, PinType.ANALOG, (short) 1, GraphGranularityType.HOURLY)))); assertTrue(Files.exists(Paths.get(reportingFolder, "test2", generateFilename(2, 0, PinType.ANALOG, (short) 2, GraphGranularityType.HOURLY)))); User user = new User(); user.email = "test"; user.appName = AppNameUtil.BLYNK; new ReportingDiskDao(reportingFolder, true).delete(user, 1, 0, PinType.ANALOG, (short) 1); assertFalse(Files.exists(Paths.get(reportingFolder, "test", generateFilename(1, 0, PinType.ANALOG, (short) 1, GraphGranularityType.HOURLY)))); }