@Test public void shouldOnlySupportASingleSubscriber() throws Exception { eventSubject.subscribe(receiver); assertThatThrownBy(() -> eventSubject.subscribe(new RecordingConsumer<>())) .isInstanceOf(IllegalStateException.class); }
@Nonnull @Override public Disposable subscribe(Consumer<E> eventConsumer) { List<E> queued = new ArrayList<>(); 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(); }
QueueingEventSubject implements EventSource<E>, Consumer<E> { @Nonnull @Override public Disposable subscribe(Consumer<E> eventConsumer) { List<E> queued = new ArrayList<>(); 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(); } }
QueueingEventSubject implements EventSource<E>, Consumer<E> { @Nonnull @Override public Disposable subscribe(Consumer<E> eventConsumer) { List<E> queued = new ArrayList<>(); 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(); } QueueingEventSubject(int capacity); }
QueueingEventSubject implements EventSource<E>, Consumer<E> { @Nonnull @Override public Disposable subscribe(Consumer<E> eventConsumer) { List<E> queued = new ArrayList<>(); 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(); } QueueingEventSubject(int capacity); @Nonnull @Override Disposable subscribe(Consumer<E> eventConsumer); @Override void accept(E value); }
QueueingEventSubject implements EventSource<E>, Consumer<E> { @Nonnull @Override public Disposable subscribe(Consumer<E> eventConsumer) { List<E> queued = new ArrayList<>(); 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(); } QueueingEventSubject(int capacity); @Nonnull @Override Disposable subscribe(Consumer<E> eventConsumer); @Override void accept(E value);  }