class QueueingEventSubject implements EventSource<E>, Consumer<E> {

  @Nonnull
  @Override
  public Disposable subscribe(Consumer<E> eventConsumer) {
    List<E> queued = new ArrayList<>();

    // avoid calling the consumer from the synchronized section
    synchronized (this) {
      if (state == State.SUBSCRIBED) {
        throw new IllegalStateException(
            "Only a single subscription is supported, previous subscriber is: " + subscriber);
      }

      state = State.SUBSCRIBED;
      subscriber = eventConsumer;
      queue.drainTo(queued);
    }

    for (E event : queued) {
      eventConsumer.accept(event);
    }

    return new Unsubscriber();
  }

  public  QueueingEventSubject(int capacity);

}