@Test public void shouldThrowWhenCapacityExceeded() throws Exception { eventSubject.accept("a"); eventSubject.accept("b"); eventSubject.accept("c"); assertThatThrownBy(() -> eventSubject.accept("noo, too many things")) .isInstanceOf(IllegalStateException.class); }
@Override public void accept(E value) { Consumer<E> consumerToInvoke = null; synchronized (this) { switch (state) { case NO_SUBSCRIBER: queue.add(value); break; case SUBSCRIBED: consumerToInvoke = subscriber; break; } } if (consumerToInvoke != null) { consumerToInvoke.accept(value); } }
QueueingEventSubject implements EventSource<E>, Consumer<E> { @Override public void accept(E value) { Consumer<E> consumerToInvoke = null; synchronized (this) { switch (state) { case NO_SUBSCRIBER: queue.add(value); break; case SUBSCRIBED: consumerToInvoke = subscriber; break; } } if (consumerToInvoke != null) { consumerToInvoke.accept(value); } } }
QueueingEventSubject implements EventSource<E>, Consumer<E> { @Override public void accept(E value) { Consumer<E> consumerToInvoke = null; synchronized (this) { switch (state) { case NO_SUBSCRIBER: queue.add(value); break; case SUBSCRIBED: consumerToInvoke = subscriber; break; } } if (consumerToInvoke != null) { consumerToInvoke.accept(value); } } QueueingEventSubject(int capacity); }
QueueingEventSubject implements EventSource<E>, Consumer<E> { @Override public void accept(E value) { Consumer<E> consumerToInvoke = null; synchronized (this) { switch (state) { case NO_SUBSCRIBER: queue.add(value); break; case SUBSCRIBED: consumerToInvoke = subscriber; break; } } if (consumerToInvoke != null) { consumerToInvoke.accept(value); } } QueueingEventSubject(int capacity); @Nonnull @Override Disposable subscribe(Consumer<E> eventConsumer); @Override void accept(E value); }
QueueingEventSubject implements EventSource<E>, Consumer<E> { @Override public void accept(E value) { Consumer<E> consumerToInvoke = null; synchronized (this) { switch (state) { case NO_SUBSCRIBER: queue.add(value); break; case SUBSCRIBED: consumerToInvoke = subscriber; break; } } if (consumerToInvoke != null) { consumerToInvoke.accept(value); } } QueueingEventSubject(int capacity); @Nonnull @Override Disposable subscribe(Consumer<E> eventConsumer); @Override void accept(E value);  }