LongTimeSeriesAggregator extends KorypheBinaryOperator<LongTimeSeries> { @Override protected LongTimeSeries _apply(final LongTimeSeries a, final LongTimeSeries b) { if (!b.getTimeBucket().equals(a.getTimeBucket())) { throw new RuntimeException("Can't aggregate two LongTimeSeries with different time buckets: " + "a had bucket " + a.getTimeBucket() + ", b had bucket " + b.getTimeBucket()); } for (final Map.Entry<Instant, Long> entry : b.getTimeSeries().entrySet()) { a.upsert(entry.getKey(), entry.getValue()); } return a; }    }
@Test public void testAggregate() { final LongTimeSeries timeSeries1 = new LongTimeSeries(CommonTimeUtil.TimeBucket.SECOND); timeSeries1.put(Instant.ofEpochMilli(1_000L), 100L); timeSeries1.put(Instant.ofEpochMilli(10_000L), 200L); final LongTimeSeries timeSeries2 = new LongTimeSeries(CommonTimeUtil.TimeBucket.SECOND); timeSeries2.put(Instant.ofEpochMilli(1_000L), 100L); timeSeries2.put(Instant.ofEpochMilli(10_000L), 200L); timeSeries2.put(Instant.ofEpochMilli(100_000L), 500L); final LongTimeSeries aggregated = LONG_TIME_SERIES_AGGREGATOR ._apply(timeSeries1, timeSeries2); final LongTimeSeries expected = new LongTimeSeries(CommonTimeUtil.TimeBucket.SECOND); expected.put(Instant.ofEpochMilli(1_000L), 200L); expected.put(Instant.ofEpochMilli(10_000L), 400L); expected.put(Instant.ofEpochMilli(100_000L), 500L); assertEquals(expected, aggregated); }
@Test public void testCantMergeIfDifferentTimeBucket() { try { final LongTimeSeries timeSeries1 = new LongTimeSeries(CommonTimeUtil.TimeBucket.SECOND); final LongTimeSeries timeSeries2 = new LongTimeSeries(CommonTimeUtil.TimeBucket.MINUTE); LONG_TIME_SERIES_AGGREGATOR._apply(timeSeries1, timeSeries2); } catch (final RuntimeException e) { } }