CompositeSum extends AbstractCompositeSum<EventCounter> implements EventCounter { @Override public EventCounter merge(EventCounter counter) { if (counter instanceof CompositeSum) { return internalMerge( ((CompositeSum) counter).getEventCounters(), new CompositeSum(getMaxLength(), getMaxChunkLength())); } else { return internalMerge( Arrays.asList(counter), new CompositeSum(getMaxLength(), getMaxChunkLength())); } } CompositeSum(ReadableDuration maxLength, ReadableDuration maxChunkLength);  CompositeSum(ReadableDuration maxLength); @Override EventCounter merge(EventCounter counter);  }
@Test(groups = "fast") public void testMergeComposite() throws Exception { setNow(base); DateTime start = base; CompositeSum counter1 = newCompositeEventCounter(Duration.standardMinutes(60)); CompositeSum counter2 = newCompositeEventCounter(Duration.standardMinutes(60)); counter1.add(100); counter2.add(10); advanceNowMinutes(6); counter1.addEventCounter(windowMinutes(6)); counter2.addEventCounter(windowMinutes(6)); counter1.add(100); counter2.add(10); advanceNowMinutes(6); counter1.addEventCounter(windowMinutes(6)); counter2.addEventCounter(windowMinutes(6)); counter1.add(100); counter2.add(10); EventCounter counter3 = counter1.merge(counter2); Assert.assertEquals(counter3.getStart(), start); Assert.assertEquals(counter3.getEnd(), now.plusMinutes(6)); Assert.assertEquals(counter3.getValue(), 330); }
@Test(groups = "fast") public void testInterleavedMerge() throws Exception { setNow(base); CompositeSum counter1 = newCompositeEventCounter(Duration.standardMinutes(60)); CompositeSum counter2 = newCompositeEventCounter(Duration.standardMinutes(60)); counter2.add(1); advanceNowMinutes(30); counter1.add(10); advanceNowMinutes(30); counter2.add(100); advanceNowMinutes(5); counter1.add(1000); advanceNowMinutes(31); counter2.add(10000); EventCounter counter3 = counter1.merge(counter2); Assert.assertEquals(counter3.getValue(), 11100); }