SlidingWindowPolicy implements Serializable { public long higherFrameTs(long timestamp) { long tsPlusFrame = timestamp + frameSize; return sumHadOverflow(timestamp, frameSize, tsPlusFrame) ? addClamped(floorFrameTs(timestamp), frameSize) : floorFrameTs(tsPlusFrame); } SlidingWindowPolicy(long frameSize, long frameOffset, long framesPerWindow); long frameSize(); long frameOffset(); long windowSize(); boolean isTumbling(); long floorFrameTs(long timestamp); long higherFrameTs(long timestamp); SlidingWindowPolicy withOffset(long offset); SlidingWindowPolicy toTumblingByFrame(); static SlidingWindowPolicy slidingWinPolicy(long windowSize, long slideBy); static SlidingWindowPolicy tumblingWinPolicy(long windowSize);  }
@Test public void when_higherOutOfRange_then_maxValue() { definition = new SlidingWindowPolicy(4, 2, 10); assertEquals(Long.MAX_VALUE, definition.higherFrameTs(Long.MAX_VALUE - 1)); assertEquals(Long.MIN_VALUE + 2, definition.higherFrameTs(Long.MIN_VALUE)); }