@Test public void putDataTimeRangeTest() throws Exception { long time = System.currentTimeMillis(); Map<String, String> value = new HashMap<String, String>(); value.put("value", "1.3"); value.put("status", "ok"); Map<Long, Map<String, String>> feedData1 = new HashMap<Long, Map<String, String>>(); feedData1.put(time, value); Map<String, String> value2 = new HashMap<String, String>(); value2.put("value", "1.4"); value2.put("status", "ok"); Map<Long, Map<String, String>> feedData2 = new HashMap<Long, Map<String, String>>(); feedData2.put(time + 100, value2); Map<String, Map<Long, Map<String, String>>> data = new HashMap<String, Map<Long, Map<String, String>>>(); data.put(testFeedID1, feedData1); data.put(testFeedID2, feedData2); codataBuffer.putData(data, TimeUnit.MILLISECONDS, null); long nanotime = TimeUnit.NANOSECONDS.convert(time, TimeUnit.MILLISECONDS); long nanotime2 = TimeUnit.NANOSECONDS.convert(time + 100, TimeUnit.MILLISECONDS); Assert.assertEquals(codataBuffer.metaDataBuffer.getStartTimestamp(0, testFeedID1), nanotime); Assert.assertEquals(codataBuffer.metaDataBuffer.getEndTimestamp(0, testFeedID1), nanotime); Assert.assertEquals(codataBuffer.metaDataBuffer.getStartTimestamp(0, testFeedID2), nanotime2); Assert.assertEquals(codataBuffer.metaDataBuffer.getEndTimestamp(0, testFeedID2), nanotime2); }
@Override public void putData(String feedID, TimeUnit timeUnit, Map<Long, Map<String, String>> entries) throws BufferFullException { synchronized (movePartitionLock) { if (reset) return; } final ElapsedTimer timer = new ElapsedTimer(); timer.startInterval(); Map<String, Map<Long, Map<String, String>>> feedDataToPut = new HashMap<String, Map<Long,Map<String,String>>>(); feedDataToPut.put(feedID, entries); int i = this.currentParition.getBufferEnv().getCurrentBufferPartition(); int startPartition = i; do { PartitionDataBuffer partitionBuffer = this.partitionDataBuffers[i].get(); if (partitionBuffer == null || !partitionBuffer.isActive()) { break; } LOGGER.debug("Putting in partition {}", i); Map<String, PartitionTimestamps> timeStamps = putData(partitionBuffer, feedDataToPut, timeUnit); if (timeStamps != null) { metaDataBuffer.updatePartitionMetaData(partitionBuffer.getBufferEnv().getCurrentBufferPartition(), timeStamps); } i = (i + 1) % this.currentParition.getBufferEnv().getNumOfBufferPartitions(); } while (i != startPartition); timer.stopInterval(); PERF_LOGGER.debug("Time to save data for feed {}: {}", feedID, timer.getIntervalInMillis()); }
CODataBuffer extends DataBuffer implements DataArchive, DataProvider { @Override public void putData(String feedID, TimeUnit timeUnit, Map<Long, Map<String, String>> entries) throws BufferFullException { synchronized (movePartitionLock) { if (reset) return; } final ElapsedTimer timer = new ElapsedTimer(); timer.startInterval(); Map<String, Map<Long, Map<String, String>>> feedDataToPut = new HashMap<String, Map<Long,Map<String,String>>>(); feedDataToPut.put(feedID, entries); int i = this.currentParition.getBufferEnv().getCurrentBufferPartition(); int startPartition = i; do { PartitionDataBuffer partitionBuffer = this.partitionDataBuffers[i].get(); if (partitionBuffer == null || !partitionBuffer.isActive()) { break; } LOGGER.debug("Putting in partition {}", i); Map<String, PartitionTimestamps> timeStamps = putData(partitionBuffer, feedDataToPut, timeUnit); if (timeStamps != null) { metaDataBuffer.updatePartitionMetaData(partitionBuffer.getBufferEnv().getCurrentBufferPartition(), timeStamps); } i = (i + 1) % this.currentParition.getBufferEnv().getNumOfBufferPartitions(); } while (i != startPartition); timer.stopInterval(); PERF_LOGGER.debug("Time to save data for feed {}: {}", feedID, timer.getIntervalInMillis()); } }
CODataBuffer extends DataBuffer implements DataArchive, DataProvider { @Override public void putData(String feedID, TimeUnit timeUnit, Map<Long, Map<String, String>> entries) throws BufferFullException { synchronized (movePartitionLock) { if (reset) return; } final ElapsedTimer timer = new ElapsedTimer(); timer.startInterval(); Map<String, Map<Long, Map<String, String>>> feedDataToPut = new HashMap<String, Map<Long,Map<String,String>>>(); feedDataToPut.put(feedID, entries); int i = this.currentParition.getBufferEnv().getCurrentBufferPartition(); int startPartition = i; do { PartitionDataBuffer partitionBuffer = this.partitionDataBuffers[i].get(); if (partitionBuffer == null || !partitionBuffer.isActive()) { break; } LOGGER.debug("Putting in partition {}", i); Map<String, PartitionTimestamps> timeStamps = putData(partitionBuffer, feedDataToPut, timeUnit); if (timeStamps != null) { metaDataBuffer.updatePartitionMetaData(partitionBuffer.getBufferEnv().getCurrentBufferPartition(), timeStamps); } i = (i + 1) % this.currentParition.getBufferEnv().getNumOfBufferPartitions(); } while (i != startPartition); timer.stopInterval(); PERF_LOGGER.debug("Time to save data for feed {}: {}", feedID, timer.getIntervalInMillis()); } CODataBuffer(DataBufferEnv env, DataBufferHelper partitionBufferFactory); }
CODataBuffer extends DataBuffer implements DataArchive, DataProvider { @Override public void putData(String feedID, TimeUnit timeUnit, Map<Long, Map<String, String>> entries) throws BufferFullException { synchronized (movePartitionLock) { if (reset) return; } final ElapsedTimer timer = new ElapsedTimer(); timer.startInterval(); Map<String, Map<Long, Map<String, String>>> feedDataToPut = new HashMap<String, Map<Long,Map<String,String>>>(); feedDataToPut.put(feedID, entries); int i = this.currentParition.getBufferEnv().getCurrentBufferPartition(); int startPartition = i; do { PartitionDataBuffer partitionBuffer = this.partitionDataBuffers[i].get(); if (partitionBuffer == null || !partitionBuffer.isActive()) { break; } LOGGER.debug("Putting in partition {}", i); Map<String, PartitionTimestamps> timeStamps = putData(partitionBuffer, feedDataToPut, timeUnit); if (timeStamps != null) { metaDataBuffer.updatePartitionMetaData(partitionBuffer.getBufferEnv().getCurrentBufferPartition(), timeStamps); } i = (i + 1) % this.currentParition.getBufferEnv().getNumOfBufferPartitions(); } while (i != startPartition); timer.stopInterval(); PERF_LOGGER.debug("Time to save data for feed {}: {}", feedID, timer.getIntervalInMillis()); } CODataBuffer(DataBufferEnv env, DataBufferHelper partitionBufferFactory); @Override Map<String, SortedMap<Long, Map<String, String>>> getData(Set<String> feedIDs, long startTime, long endTime,
            TimeUnit timeUnit); @Override Map<String, List<Map<String, String>>> getData(Set<String> feedIDs, TimeUnit timeUnit, long startTime,
            long endTime); @Override void putData(String feedID, TimeUnit timeUnit, Map<Long, Map<String, String>> entries); @Override void putData(Map<String, Map<Long, Map<String, String>>> value, TimeUnit timeUnit,
                    Runnable callback); @Override void putData(String feedID, TimeUnit timeUnit, long time, Map<String, String> value); }
CODataBuffer extends DataBuffer implements DataArchive, DataProvider { @Override public void putData(String feedID, TimeUnit timeUnit, Map<Long, Map<String, String>> entries) throws BufferFullException { synchronized (movePartitionLock) { if (reset) return; } final ElapsedTimer timer = new ElapsedTimer(); timer.startInterval(); Map<String, Map<Long, Map<String, String>>> feedDataToPut = new HashMap<String, Map<Long,Map<String,String>>>(); feedDataToPut.put(feedID, entries); int i = this.currentParition.getBufferEnv().getCurrentBufferPartition(); int startPartition = i; do { PartitionDataBuffer partitionBuffer = this.partitionDataBuffers[i].get(); if (partitionBuffer == null || !partitionBuffer.isActive()) { break; } LOGGER.debug("Putting in partition {}", i); Map<String, PartitionTimestamps> timeStamps = putData(partitionBuffer, feedDataToPut, timeUnit); if (timeStamps != null) { metaDataBuffer.updatePartitionMetaData(partitionBuffer.getBufferEnv().getCurrentBufferPartition(), timeStamps); } i = (i + 1) % this.currentParition.getBufferEnv().getNumOfBufferPartitions(); } while (i != startPartition); timer.stopInterval(); PERF_LOGGER.debug("Time to save data for feed {}: {}", feedID, timer.getIntervalInMillis()); } CODataBuffer(DataBufferEnv env, DataBufferHelper partitionBufferFactory); @Override Map<String, SortedMap<Long, Map<String, String>>> getData(Set<String> feedIDs, long startTime, long endTime,
            TimeUnit timeUnit); @Override Map<String, List<Map<String, String>>> getData(Set<String> feedIDs, TimeUnit timeUnit, long startTime,
            long endTime); @Override void putData(String feedID, TimeUnit timeUnit, Map<Long, Map<String, String>> entries); @Override void putData(Map<String, Map<Long, Map<String, String>>> value, TimeUnit timeUnit,
                    Runnable callback); @Override void putData(String feedID, TimeUnit timeUnit, long time, Map<String, String> value);  }