### Question:
AggregatingMetaStoreEventReader implements MetaStoreEventReader { @Override public Optional<MetaStoreEvent> read() { requestMoreMessagesIfNeeded(); while (buffer.isEmpty()) { try { List<MetaStoreEvent> events = null; synchronized (monitor) { events = lastSubmittedTask.get(window, windowUnits); lastSubmittedTask = null; } if (events.isEmpty()) { return Optional.empty(); } buffer.addAll(events); } catch (TimeoutException e) { log.debug("Timeout whilst buffering message. Retrying...", e); } catch (InterruptedException e) { log.warn("Thread was interrupted whilst buffering message. Retrying...", e); } catch (ExecutionException e) { lastSubmittedTask = null; requestMoreMessagesIfNeeded(); Throwable cause = e.getCause(); if (cause != null && RuntimeException.class.isAssignableFrom(cause.getClass())) { throw (RuntimeException) cause; } throw new ShuntingYardException("Delegate MessageReader has failed to read messages", cause); } } return Optional.ofNullable(buffer.poll()); } AggregatingMetaStoreEventReader(MetaStoreEventReader delegate, MetaStoreEventAggregator aggregator);  AggregatingMetaStoreEventReader(
      MetaStoreEventReader delegate,
      MetaStoreEventAggregator aggregator,
      long window,
      TimeUnit windowUnits); @VisibleForTesting  AggregatingMetaStoreEventReader(
      MetaStoreEventReader delegate,
      MetaStoreEventAggregator aggregator,
      long window,
      TimeUnit windowUnits,
      Queue<MetaStoreEvent> buffer); @Override void close(); @Override Optional<MetaStoreEvent> read();  }### Answer:
@Test public void failuresInDelegateArePropagated() { SerDeException e = new SerDeException("oops"); expectedException.expect(is(e)); when(delegate.read()).thenThrow(e); aggregatingMessageReader.read(); }
@Test public void emptyAggregate() { when(delegate.read()).thenAnswer(new Answer<Optional<MetaStoreEvent>>() { @Override public Optional<MetaStoreEvent> answer(InvocationOnMock invocation) throws Throwable { WINDOW_UNITS.sleep(WINDOW + 1); return Optional.empty(); } }); Optional<MetaStoreEvent> event = aggregatingMessageReader.read(); assertThat(event).isEqualTo(Optional.empty()); }
@Test public void readExceedsTheWindow() { List<MetaStoreEvent> events = Arrays.asList(mock(MetaStoreEvent.class)); when(delegate.read()).thenAnswer(new Answer<Optional<MetaStoreEvent>>() { @Override public Optional<MetaStoreEvent> answer(InvocationOnMock invocation) throws Throwable { WINDOW_UNITS.sleep(WINDOW + 1); return Optional.of(events.get(0)); } }); Optional<MetaStoreEvent> next = aggregatingMessageReader.read(); assertThat(next.get()).isEqualTo(events.get(0)); verify(delegate).read(); verify(aggregator).aggregate(events); }
@Test public void multipleReadsBeforeExceedingTheWindow() { MetaStoreEvent[] events = new MetaStoreEvent[] { mock(MetaStoreEvent.class), mock(MetaStoreEvent.class), mock(MetaStoreEvent.class), mock(MetaStoreEvent.class), mock(MetaStoreEvent.class) }; final int counter[] = new int[] { 0 }; when(delegate.read()).thenAnswer(new Answer<Optional<MetaStoreEvent>>() { @Override public Optional<MetaStoreEvent> answer(InvocationOnMock invocation) throws Throwable { WINDOW_UNITS.sleep(counter[0] < events.length - 1 ? 1 : WINDOW); return Optional.of(events[counter[0]++]); } }); Optional<MetaStoreEvent> next = aggregatingMessageReader.read(); assertThat(next.get()).isEqualTo(events[0]); verify(delegate, atLeast(2)).read(); verify(aggregator).aggregate(any()); int numOfEventsCaptured = buffer.size(); assertThat(buffer).containsAll(Arrays.asList(Arrays.copyOfRange(events, 1, numOfEventsCaptured))); }