RBMBackedTimestampSetAggregator extends KorypheBinaryOperator<RBMBackedTimestampSet> { @Override protected RBMBackedTimestampSet _apply(final RBMBackedTimestampSet a, final RBMBackedTimestampSet b) { if (!b.getTimeBucket().equals(a.getTimeBucket())) { throw new RuntimeException("Can't aggregate two RBMBackedTimestampSet with different time buckets: " + "a had bucket " + a.getTimeBucket() + ", b had bucket " + b.getTimeBucket()); } a.addAll(b); return a; }    }
@Test public void testAggregate() { final RBMBackedTimestampSet rbmBackedTimestampSet1 = new RBMBackedTimestampSet(CommonTimeUtil.TimeBucket.SECOND); rbmBackedTimestampSet1.add(Instant.ofEpochMilli(1000L)); rbmBackedTimestampSet1.add(Instant.ofEpochMilli(1000000L)); final RBMBackedTimestampSet rbmBackedTimestampSet2 = new RBMBackedTimestampSet(CommonTimeUtil.TimeBucket.SECOND); rbmBackedTimestampSet2.add(Instant.ofEpochMilli(1000L)); rbmBackedTimestampSet2.add(Instant.ofEpochMilli(2000000L)); final RBMBackedTimestampSet aggregated = RBM_BACKED_TIMESTAMP_SET_AGGREGATOR._apply(rbmBackedTimestampSet1, rbmBackedTimestampSet2); final RBMBackedTimestampSet expected = new RBMBackedTimestampSet(CommonTimeUtil.TimeBucket.SECOND); expected.add(Instant.ofEpochMilli(1000L)); expected.add(Instant.ofEpochMilli(1000000L)); expected.add(Instant.ofEpochMilli(2000000L)); assertEquals(3, aggregated.getNumberOfTimestamps()); assertEquals(expected, aggregated); }
@Test public void testCantMergeIfDifferentTimeBucket() { try { final RBMBackedTimestampSet rbmBackedTimestampSet1 = new RBMBackedTimestampSet(CommonTimeUtil.TimeBucket.SECOND); final RBMBackedTimestampSet rbmBackedTimestampSet2 = new RBMBackedTimestampSet(CommonTimeUtil.TimeBucket.MINUTE); RBM_BACKED_TIMESTAMP_SET_AGGREGATOR._apply(rbmBackedTimestampSet1, rbmBackedTimestampSet2); } catch (final RuntimeException e) { } }