@Test public void testDroppedExceptionsWhenOnItemThrowsAnException() { AtomicBoolean called = new AtomicBoolean(); AtomicReference<Throwable> captured = new AtomicReference<>(); Infrastructure.setDroppedExceptionHandler(captured::set); AtomicReference<UniEmitter<? super Integer>> reference = new AtomicReference<>(); AbstractUni<Integer> uni = (AbstractUni<Integer>) Uni.createFrom().<Integer> emitter( reference::set); UniSerializedSubscriber.subscribe(uni, new UniSubscriber<Integer>() { @Override public void onSubscribe(UniSubscription subscription) { } @Override public void onItem(Integer item) { throw new IllegalArgumentException("boom"); } @Override public void onFailure(Throwable failure) { called.set(true); } }); try { reference.get().complete(1); fail("Exception expected"); } catch (IllegalArgumentException e) { } assertThat(captured.get()).isNotNull().isInstanceOf(IllegalArgumentException.class).hasMessageContaining("boom"); assertThat(called).isFalse(); }
public static <T> void subscribe(AbstractUni<T> source, UniSubscriber<? super T> subscriber) { UniSubscriber<? super T> actual = Infrastructure.onUniSubscription(source, subscriber); UniSerializedSubscriber<T> wrapped = new UniSerializedSubscriber<>(source, actual); wrapped.subscribe(); }
UniSerializedSubscriber implements UniSubscriber<T>, UniSubscription { public static <T> void subscribe(AbstractUni<T> source, UniSubscriber<? super T> subscriber) { UniSubscriber<? super T> actual = Infrastructure.onUniSubscription(source, subscriber); UniSerializedSubscriber<T> wrapped = new UniSerializedSubscriber<>(source, actual); wrapped.subscribe(); } }
UniSerializedSubscriber implements UniSubscriber<T>, UniSubscription { public static <T> void subscribe(AbstractUni<T> source, UniSubscriber<? super T> subscriber) { UniSubscriber<? super T> actual = Infrastructure.onUniSubscription(source, subscriber); UniSerializedSubscriber<T> wrapped = new UniSerializedSubscriber<>(source, actual); wrapped.subscribe(); } UniSerializedSubscriber(AbstractUni<T> upstream, UniSubscriber<? super T> subscriber); }
UniSerializedSubscriber implements UniSubscriber<T>, UniSubscription { public static <T> void subscribe(AbstractUni<T> source, UniSubscriber<? super T> subscriber) { UniSubscriber<? super T> actual = Infrastructure.onUniSubscription(source, subscriber); UniSerializedSubscriber<T> wrapped = new UniSerializedSubscriber<>(source, actual); wrapped.subscribe(); } 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 { public static <T> void subscribe(AbstractUni<T> source, UniSubscriber<? super T> subscriber) { UniSubscriber<? super T> actual = Infrastructure.onUniSubscription(source, subscriber); UniSerializedSubscriber<T> wrapped = new UniSerializedSubscriber<>(source, actual); wrapped.subscribe(); } 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();  }