BoundedTimestampSetAggregator extends KorypheBinaryOperator<BoundedTimestampSet> { @Override protected BoundedTimestampSet _apply(final BoundedTimestampSet a, final BoundedTimestampSet b) { if (!a.getTimeBucket().equals(b.getTimeBucket())) { throw new IllegalArgumentException("Can only merge two BoundedTimestampSet with the same time bucket:" + "a has time bucket " + a.getTimeBucket() + ", b has time bucket " + b.getTimeBucket()); } if (a.getMaxSize() != b.getMaxSize()) { throw new IllegalArgumentException("Can only merge two BoundedTimestampSet with the same maximum size:" + "a has maximum size " + a.getMaxSize() + ", b has maximum size " + b.getMaxSize()); } if (a.getState().equals(BoundedTimestampSet.State.NOT_FULL) && b.getState().equals(BoundedTimestampSet.State.NOT_FULL)) { a.add(b.getTimestamps()); return a; } if (a.getState().equals(BoundedTimestampSet.State.NOT_FULL)) { a.switchToSampleState(); a.add(b.getTimestamps()); } else { a.add(b.getTimestamps()); } return a; }    }
@Test public void testAggregateWhenBothInNotFullState() { final BoundedTimestampSet boundedTimestampSet1 = new BoundedTimestampSet(CommonTimeUtil.TimeBucket.SECOND, 10); boundedTimestampSet1.add(Instant.ofEpochMilli(1000L)); boundedTimestampSet1.add(Instant.ofEpochMilli(1000000L)); final BoundedTimestampSet boundedTimestampSet2 = new BoundedTimestampSet(CommonTimeUtil.TimeBucket.SECOND, 10); boundedTimestampSet2.add(Instant.ofEpochMilli(1000L)); boundedTimestampSet2.add(Instant.ofEpochMilli(2000000L)); final BoundedTimestampSet aggregated = BOUNDED_TIMESTAMP_SET_AGGREGATOR._apply(boundedTimestampSet1, boundedTimestampSet2); final BoundedTimestampSet expected = new BoundedTimestampSet(CommonTimeUtil.TimeBucket.SECOND, 10); expected.add(Instant.ofEpochMilli(1000L)); expected.add(Instant.ofEpochMilli(1000000L)); expected.add(Instant.ofEpochMilli(2000000L)); assertEquals(3, aggregated.getNumberOfTimestamps()); assertEquals(BoundedTimestampSet.State.NOT_FULL, aggregated.getState()); assertEquals(expected.getTimestamps(), aggregated.getTimestamps()); }
@Test public void testAggregateWhenBothInSampleState() { final BoundedTimestampSet boundedTimestampSet1 = new BoundedTimestampSet(CommonTimeUtil.TimeBucket.SECOND, 10); final Set<Instant> instants1 = new HashSet<>(); IntStream.range(0, 100) .forEach(i -> instants1.add(Instant.ofEpochMilli(i * 1000L))); instants1.forEach(boundedTimestampSet1::add); final BoundedTimestampSet boundedTimestampSet2 = new BoundedTimestampSet(CommonTimeUtil.TimeBucket.SECOND, 10); final Set<Instant> instants2 = new HashSet<>(); IntStream.range(50, 150) .forEach(i -> instants2.add(Instant.ofEpochMilli(i * 1000L))); instants2.forEach(boundedTimestampSet2::add); final Set<Instant> allInstants = new HashSet<>(instants1); allInstants.addAll(instants2); final BoundedTimestampSet aggregated = BOUNDED_TIMESTAMP_SET_AGGREGATOR._apply(boundedTimestampSet1, boundedTimestampSet2); assertEquals(10, aggregated.getNumberOfTimestamps()); assertEquals(BoundedTimestampSet.State.SAMPLE, aggregated.getState()); assertTrue(allInstants.containsAll(aggregated.getTimestamps())); }
@Test public void testAggregateWhenAIsInNotFullStateAndBIsInSampleState() { final BoundedTimestampSet boundedTimestampSet1 = new BoundedTimestampSet(CommonTimeUtil.TimeBucket.SECOND, 10); final Set<Instant> instants1 = new HashSet<>(); instants1.add(Instant.ofEpochMilli(1000L)); instants1.add(Instant.ofEpochMilli(1000000L)); instants1.forEach(boundedTimestampSet1::add); final BoundedTimestampSet boundedTimestampSet2 = new BoundedTimestampSet(CommonTimeUtil.TimeBucket.SECOND, 10); final Set<Instant> instants2 = new HashSet<>(); IntStream.range(50, 150) .forEach(i -> instants2.add(Instant.ofEpochMilli(i * 1000L))); instants2.forEach(boundedTimestampSet2::add); final Set<Instant> allInstants = new HashSet<>(instants1); allInstants.addAll(instants2); final BoundedTimestampSet aggregated = BOUNDED_TIMESTAMP_SET_AGGREGATOR._apply(boundedTimestampSet1, boundedTimestampSet2); assertEquals(10, aggregated.getNumberOfTimestamps()); assertEquals(BoundedTimestampSet.State.SAMPLE, aggregated.getState()); assertTrue(allInstants.containsAll(aggregated.getTimestamps())); }
@Test public void testAggregateWhenAIsInSampleStateAndBIsInNotFullState() { final BoundedTimestampSet boundedTimestampSet1 = new BoundedTimestampSet(CommonTimeUtil.TimeBucket.SECOND, 10); final Set<Instant> instants1 = new HashSet<>(); instants1.add(Instant.ofEpochMilli(1000L)); instants1.add(Instant.ofEpochMilli(1000000L)); instants1.forEach(boundedTimestampSet1::add); final BoundedTimestampSet boundedTimestampSet2 = new BoundedTimestampSet(CommonTimeUtil.TimeBucket.SECOND, 10); final Set<Instant> instants2 = new HashSet<>(); IntStream.range(50, 150) .forEach(i -> instants2.add(Instant.ofEpochMilli(i * 1000L))); instants2.forEach(boundedTimestampSet2::add); final Set<Instant> allInstants = new HashSet<>(instants1); allInstants.addAll(instants2); final BoundedTimestampSet aggregated = BOUNDED_TIMESTAMP_SET_AGGREGATOR._apply(boundedTimestampSet2, boundedTimestampSet1); assertEquals(10, aggregated.getNumberOfTimestamps()); assertEquals(BoundedTimestampSet.State.SAMPLE, aggregated.getState()); assertTrue(allInstants.containsAll(aggregated.getTimestamps())); }
@Test public void testCantMergeIfDifferentTimeBucket() { try { final BoundedTimestampSet boundedTimestampSet1 = new BoundedTimestampSet(CommonTimeUtil.TimeBucket.SECOND, 10); final BoundedTimestampSet boundedTimestampSet2 = new BoundedTimestampSet(CommonTimeUtil.TimeBucket.MINUTE, 10); BOUNDED_TIMESTAMP_SET_AGGREGATOR._apply(boundedTimestampSet1, boundedTimestampSet2); } catch (final RuntimeException e) { } }
@Test public void testCantMergeIfDifferentMaxSize() { try { final BoundedTimestampSet boundedTimestampSet1 = new BoundedTimestampSet(CommonTimeUtil.TimeBucket.SECOND, 10); final BoundedTimestampSet boundedTimestampSet2 = new BoundedTimestampSet(CommonTimeUtil.TimeBucket.MINUTE, 11); BOUNDED_TIMESTAMP_SET_AGGREGATOR._apply(boundedTimestampSet1, boundedTimestampSet2); } catch (final RuntimeException e) { } }