@Test public void test() throws IOException { FakedPrioritySeriesReaderByTimestamp reader1 = new FakedPrioritySeriesReaderByTimestamp(100, 200, 5, 11); FakedPrioritySeriesReaderByTimestamp reader2 = new FakedPrioritySeriesReaderByTimestamp(850, 200, 7, 19); FakedPrioritySeriesReaderByTimestamp reader3 = new FakedPrioritySeriesReaderByTimestamp(1080, 200, 13, 31); PriorityMergeReaderByTimestamp priorityReader = new PriorityMergeReaderByTimestamp(); priorityReader.addReaderWithPriority(reader1, 1); priorityReader.addReaderWithPriority(reader2, 2); priorityReader.addReaderWithPriority(reader3, 3); int cnt = 0; Random random = new Random(); for (long time = 4; time < 1080 + 200 * 13 + 600; ) { TsPrimitiveType value = priorityReader.getValueInTimestamp(time); if (time < 100) { Assert.assertNull(value); } else if (time < 850) { if ((time - 100) % 5 == 0) { Assert.assertEquals(time % 11, value.getLong()); } } else if (time < 1080) { if (time >= 850 && (time - 850) % 7 == 0) { Assert.assertEquals(time % 19, value.getLong()); } else if (time < 1100 && (time - 100) % 5 == 0) { Assert.assertEquals(time % 11, value.getLong()); } else { Assert.assertNull(value); } } else if (time < 1080 + 200 * 13) { if (time >= 1080 && (time - 1080) % 13 == 0) { Assert.assertEquals(time % 31, value.getLong()); } else if (time < 850 + 200 * 7 && (time - 850) % 7 == 0) { Assert.assertEquals(time % 19, value.getLong()); } else if (time < 1100 && (time - 100) % 5 == 0) { Assert.assertEquals(time % 11, value.getLong()); } else { Assert.assertNull(value); } } else { Assert.assertNull(value); } time += random.nextInt(50) + 1; } while (priorityReader.hasNext()) { TimeValuePair timeValuePair = priorityReader.next(); long time = timeValuePair.getTimestamp(); long value = timeValuePair.getValue().getLong(); if (time < 850) { Assert.assertEquals(time % 11, value); } else if (time < 1080) { Assert.assertEquals(time % 19, value); } else { Assert.assertEquals(time % 31, value); } cnt++; } }
@Override public TsPrimitiveType getValueInTimestamp(long timestamp) throws IOException { if (hasCachedTimeValuePair) { if (cachedTimeValuePair.getTimestamp() == timestamp) { hasCachedTimeValuePair = false; return cachedTimeValuePair.getValue(); } else if (cachedTimeValuePair.getTimestamp() > timestamp) { return null; } } while (hasNext()) { cachedTimeValuePair = next(); if (cachedTimeValuePair.getTimestamp() == timestamp) { hasCachedTimeValuePair = false; return cachedTimeValuePair.getValue(); } else if (cachedTimeValuePair.getTimestamp() > timestamp) { hasCachedTimeValuePair = true; return null; } } return null; }
PriorityMergeReaderByTimestamp extends PriorityMergeReader implements
    EngineReaderByTimeStamp { @Override public TsPrimitiveType getValueInTimestamp(long timestamp) throws IOException { if (hasCachedTimeValuePair) { if (cachedTimeValuePair.getTimestamp() == timestamp) { hasCachedTimeValuePair = false; return cachedTimeValuePair.getValue(); } else if (cachedTimeValuePair.getTimestamp() > timestamp) { return null; } } while (hasNext()) { cachedTimeValuePair = next(); if (cachedTimeValuePair.getTimestamp() == timestamp) { hasCachedTimeValuePair = false; return cachedTimeValuePair.getValue(); } else if (cachedTimeValuePair.getTimestamp() > timestamp) { hasCachedTimeValuePair = true; return null; } } return null; } }
PriorityMergeReaderByTimestamp extends PriorityMergeReader implements
    EngineReaderByTimeStamp { @Override public TsPrimitiveType getValueInTimestamp(long timestamp) throws IOException { if (hasCachedTimeValuePair) { if (cachedTimeValuePair.getTimestamp() == timestamp) { hasCachedTimeValuePair = false; return cachedTimeValuePair.getValue(); } else if (cachedTimeValuePair.getTimestamp() > timestamp) { return null; } } while (hasNext()) { cachedTimeValuePair = next(); if (cachedTimeValuePair.getTimestamp() == timestamp) { hasCachedTimeValuePair = false; return cachedTimeValuePair.getValue(); } else if (cachedTimeValuePair.getTimestamp() > timestamp) { hasCachedTimeValuePair = true; return null; } } return null; }  }
PriorityMergeReaderByTimestamp extends PriorityMergeReader implements
    EngineReaderByTimeStamp { @Override public TsPrimitiveType getValueInTimestamp(long timestamp) throws IOException { if (hasCachedTimeValuePair) { if (cachedTimeValuePair.getTimestamp() == timestamp) { hasCachedTimeValuePair = false; return cachedTimeValuePair.getValue(); } else if (cachedTimeValuePair.getTimestamp() > timestamp) { return null; } } while (hasNext()) { cachedTimeValuePair = next(); if (cachedTimeValuePair.getTimestamp() == timestamp) { hasCachedTimeValuePair = false; return cachedTimeValuePair.getValue(); } else if (cachedTimeValuePair.getTimestamp() > timestamp) { hasCachedTimeValuePair = true; return null; } } return null; }  @Override TsPrimitiveType getValueInTimestamp(long timestamp); }
PriorityMergeReaderByTimestamp extends PriorityMergeReader implements
    EngineReaderByTimeStamp { @Override public TsPrimitiveType getValueInTimestamp(long timestamp) throws IOException { if (hasCachedTimeValuePair) { if (cachedTimeValuePair.getTimestamp() == timestamp) { hasCachedTimeValuePair = false; return cachedTimeValuePair.getValue(); } else if (cachedTimeValuePair.getTimestamp() > timestamp) { return null; } } while (hasNext()) { cachedTimeValuePair = next(); if (cachedTimeValuePair.getTimestamp() == timestamp) { hasCachedTimeValuePair = false; return cachedTimeValuePair.getValue(); } else if (cachedTimeValuePair.getTimestamp() > timestamp) { hasCachedTimeValuePair = true; return null; } } return null; }  @Override TsPrimitiveType getValueInTimestamp(long timestamp);  }