@Test public void testInvalidStateWhenOnSubscribeIsCalled() { AbstractUni<Integer> rogue = new AbstractUni<Integer>() { @Override protected void subscribing(UniSerializedSubscriber<? super Integer> subscriber) { } }; UniAssertSubscriber<Integer> subscriber = UniAssertSubscriber.create(); UniSerializedSubscriber<Integer> serialized = new UniSerializedSubscriber<>(rogue, subscriber); serialized.onSubscribe(() -> { }); subscriber .assertSubscribed() .assertFailure(IllegalStateException.class, "Invalid transition"); }
@Override public void onSubscribe(UniSubscription subscription) { ParameterValidation.nonNull(subscription, "subscription"); if (state.compareAndSet(SUBSCRIBED, HAS_SUBSCRIPTION)) { this.subscription = subscription; this.downstream.onSubscribe(this); } else if (state.get() == DONE) { Throwable collected = failure.getAndSet(null); if (collected != null) { this.downstream.onSubscribe(this); this.downstream.onFailure(collected); } } else { EmptyUniSubscription.propagateFailureEvent(this.downstream, new IllegalStateException( "Invalid transition, expected to be in the SUBSCRIBED state but was in " + state)); } }
UniSerializedSubscriber implements UniSubscriber<T>, UniSubscription { @Override public void onSubscribe(UniSubscription subscription) { ParameterValidation.nonNull(subscription, "subscription"); if (state.compareAndSet(SUBSCRIBED, HAS_SUBSCRIPTION)) { this.subscription = subscription; this.downstream.onSubscribe(this); } else if (state.get() == DONE) { Throwable collected = failure.getAndSet(null); if (collected != null) { this.downstream.onSubscribe(this); this.downstream.onFailure(collected); } } else { EmptyUniSubscription.propagateFailureEvent(this.downstream, new IllegalStateException( "Invalid transition, expected to be in the SUBSCRIBED state but was in " + state)); } } }
UniSerializedSubscriber implements UniSubscriber<T>, UniSubscription { @Override public void onSubscribe(UniSubscription subscription) { ParameterValidation.nonNull(subscription, "subscription"); if (state.compareAndSet(SUBSCRIBED, HAS_SUBSCRIPTION)) { this.subscription = subscription; this.downstream.onSubscribe(this); } else if (state.get() == DONE) { Throwable collected = failure.getAndSet(null); if (collected != null) { this.downstream.onSubscribe(this); this.downstream.onFailure(collected); } } else { EmptyUniSubscription.propagateFailureEvent(this.downstream, new IllegalStateException( "Invalid transition, expected to be in the SUBSCRIBED state but was in " + state)); } } UniSerializedSubscriber(AbstractUni<T> upstream, UniSubscriber<? super T> subscriber); }
UniSerializedSubscriber implements UniSubscriber<T>, UniSubscription { @Override public void onSubscribe(UniSubscription subscription) { ParameterValidation.nonNull(subscription, "subscription"); if (state.compareAndSet(SUBSCRIBED, HAS_SUBSCRIPTION)) { this.subscription = subscription; this.downstream.onSubscribe(this); } else if (state.get() == DONE) { Throwable collected = failure.getAndSet(null); if (collected != null) { this.downstream.onSubscribe(this); this.downstream.onFailure(collected); } } else { EmptyUniSubscription.propagateFailureEvent(this.downstream, new IllegalStateException( "Invalid transition, expected to be in the SUBSCRIBED state but was in " + state)); } } UniSerializedSubscriber(AbstractUni<T> upstream, UniSubscriber<? super T> subscriber); static void subscribe(AbstractUni<T> source, UniSubscriber<? super T> subscriber); @Override void onSubscribe(UniSubscription subscription); @Override void onItem(T item); @Override void onFailure(Throwable throwable); @Override void cancel(); synchronized boolean isCancelledOrDone(); }
UniSerializedSubscriber implements UniSubscriber<T>, UniSubscription { @Override public void onSubscribe(UniSubscription subscription) { ParameterValidation.nonNull(subscription, "subscription"); if (state.compareAndSet(SUBSCRIBED, HAS_SUBSCRIPTION)) { this.subscription = subscription; this.downstream.onSubscribe(this); } else if (state.get() == DONE) { Throwable collected = failure.getAndSet(null); if (collected != null) { this.downstream.onSubscribe(this); this.downstream.onFailure(collected); } } else { EmptyUniSubscription.propagateFailureEvent(this.downstream, new IllegalStateException( "Invalid transition, expected to be in the SUBSCRIBED state but was in " + state)); } } UniSerializedSubscriber(AbstractUni<T> upstream, UniSubscriber<? super T> subscriber); static void subscribe(AbstractUni<T> source, UniSubscriber<? super T> subscriber); @Override void onSubscribe(UniSubscription subscription); @Override void onItem(T item); @Override void onFailure(Throwable throwable); @Override void cancel(); synchronized boolean isCancelledOrDone();  }