TimeDatumStream extends MBStream<Datum> { public List<Datum> drainDuration(long duration) { int i = 0; long startTime = getDatumTime(i); while (i < output.size() && getDatumTime(i) - startTime < duration) { i ++; } return drain(i); } TimeDatumStream(int timeColumn);  TimeDatumStream(List<Datum> data, int timeColumn); List<Datum> drainDuration(long duration); Datum peek();  }
@Test public void testDrainDuration() { conf.set(MacroBaseConf.TIME_COLUMN, 0); for (int i = 0; i < 100; ++i) { Datum d = TestUtils.createTimeDatum(2 * i, 100 - i); data.add(d); } stream = new TimeDatumStream(data, 0); List<Datum> result = stream.drainDuration(1); assertTrue(result.size() == 1); result = stream.drainDuration(100); assertTrue(result.size() == 50); result = stream.drainDuration(200); assertTrue(result.size() == 49); }