class 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) {
        // TODO at this point all previously read messages will be lost: this will be addressed in
        // https://github.com/ExpediaGroup/shunting-yard/issues/3
        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());

  }

  public  AggregatingMetaStoreEventReader(MetaStoreEventReader delegate, MetaStoreEventAggregator aggregator);
  public  AggregatingMetaStoreEventReader(
      MetaStoreEventReader delegate,
      MetaStoreEventAggregator aggregator,
      long window,
      TimeUnit windowUnits);
  @VisibleForTesting  AggregatingMetaStoreEventReader(
      MetaStoreEventReader delegate,
      MetaStoreEventAggregator aggregator,
      long window,
      TimeUnit windowUnits,
      Queue<MetaStoreEvent> buffer);

}