@Test public void testThatEmitterCannotPropagateNullAsFailure() { AtomicBoolean terminated = new AtomicBoolean(); Multi.createFrom().<String> emitter(e -> { e.onTermination(() -> terminated.set(true)); e.emit("a"); e.fail(null); }).subscribe().withSubscriber(AssertSubscriber.create(2)) .await() .assertHasFailedWith(NullPointerException.class, "") .assertReceived("a"); assertThat(terminated).isTrue(); Multi.createFrom().<String> emitter(e -> { e.emit("a"); e.fail(null); }, BackPressureStrategy.LATEST) .subscribe().withSubscriber(AssertSubscriber.create(2)) .await() .assertHasFailedWith(NullPointerException.class, "") .assertReceived("a"); Multi.createFrom().<String> emitter(e -> { e.emit("a"); e.fail(null); }, BackPressureStrategy.DROP) .subscribe().withSubscriber(AssertSubscriber.create(2)) .await() .assertHasNotCompleted() .assertHasFailedWith(NullPointerException.class, "") .assertReceived("a"); Multi.createFrom().<String> emitter(e -> { e.emit("a"); e.fail(null); }, BackPressureStrategy.ERROR) .subscribe().withSubscriber(AssertSubscriber.create(2)) .await() .assertHasNotCompleted() .assertHasFailedWith(NullPointerException.class, "") .assertReceived("a"); Multi.createFrom().<String> emitter(e -> { e.emit("a"); e.fail(null); }, BackPressureStrategy.IGNORE) .subscribe().withSubscriber(AssertSubscriber.create(2)) .await() .assertHasNotCompleted() .assertHasFailedWith(NullPointerException.class, "") .assertReceived("a"); }
@Override public void subscribe(MultiSubscriber<? super T> downstream) { BaseMultiEmitter<T> emitter; switch (backpressure) { case DROP: emitter = new DropItemOnOverflowMultiEmitter<>(downstream); break; case ERROR: emitter = new ErrorOnOverflowMultiEmitter<>(downstream); break; case IGNORE: emitter = new IgnoreBackPressureMultiEmitter<>(downstream); break; case LATEST: emitter = new DropLatestOnOverflowMultiEmitter<>(downstream); break; default: emitter = new BufferItemMultiEmitter<>(downstream, 16); break; } downstream.onSubscribe(emitter); try { consumer.accept(emitter.serialize()); } catch (Throwable ex) { emitter.fail(ex); } }
EmitterBasedMulti extends AbstractMulti<T> { @Override public void subscribe(MultiSubscriber<? super T> downstream) { BaseMultiEmitter<T> emitter; switch (backpressure) { case DROP: emitter = new DropItemOnOverflowMultiEmitter<>(downstream); break; case ERROR: emitter = new ErrorOnOverflowMultiEmitter<>(downstream); break; case IGNORE: emitter = new IgnoreBackPressureMultiEmitter<>(downstream); break; case LATEST: emitter = new DropLatestOnOverflowMultiEmitter<>(downstream); break; default: emitter = new BufferItemMultiEmitter<>(downstream, 16); break; } downstream.onSubscribe(emitter); try { consumer.accept(emitter.serialize()); } catch (Throwable ex) { emitter.fail(ex); } } }
EmitterBasedMulti extends AbstractMulti<T> { @Override public void subscribe(MultiSubscriber<? super T> downstream) { BaseMultiEmitter<T> emitter; switch (backpressure) { case DROP: emitter = new DropItemOnOverflowMultiEmitter<>(downstream); break; case ERROR: emitter = new ErrorOnOverflowMultiEmitter<>(downstream); break; case IGNORE: emitter = new IgnoreBackPressureMultiEmitter<>(downstream); break; case LATEST: emitter = new DropLatestOnOverflowMultiEmitter<>(downstream); break; default: emitter = new BufferItemMultiEmitter<>(downstream, 16); break; } downstream.onSubscribe(emitter); try { consumer.accept(emitter.serialize()); } catch (Throwable ex) { emitter.fail(ex); } } EmitterBasedMulti(Consumer<MultiEmitter<? super T>> consumer, BackPressureStrategy backpressure); }
EmitterBasedMulti extends AbstractMulti<T> { @Override public void subscribe(MultiSubscriber<? super T> downstream) { BaseMultiEmitter<T> emitter; switch (backpressure) { case DROP: emitter = new DropItemOnOverflowMultiEmitter<>(downstream); break; case ERROR: emitter = new ErrorOnOverflowMultiEmitter<>(downstream); break; case IGNORE: emitter = new IgnoreBackPressureMultiEmitter<>(downstream); break; case LATEST: emitter = new DropLatestOnOverflowMultiEmitter<>(downstream); break; default: emitter = new BufferItemMultiEmitter<>(downstream, 16); break; } downstream.onSubscribe(emitter); try { consumer.accept(emitter.serialize()); } catch (Throwable ex) { emitter.fail(ex); } } EmitterBasedMulti(Consumer<MultiEmitter<? super T>> consumer, BackPressureStrategy backpressure); @Override void subscribe(MultiSubscriber<? super T> downstream); }
EmitterBasedMulti extends AbstractMulti<T> { @Override public void subscribe(MultiSubscriber<? super T> downstream) { BaseMultiEmitter<T> emitter; switch (backpressure) { case DROP: emitter = new DropItemOnOverflowMultiEmitter<>(downstream); break; case ERROR: emitter = new ErrorOnOverflowMultiEmitter<>(downstream); break; case IGNORE: emitter = new IgnoreBackPressureMultiEmitter<>(downstream); break; case LATEST: emitter = new DropLatestOnOverflowMultiEmitter<>(downstream); break; default: emitter = new BufferItemMultiEmitter<>(downstream, 16); break; } downstream.onSubscribe(emitter); try { consumer.accept(emitter.serialize()); } catch (Throwable ex) { emitter.fail(ex); } } EmitterBasedMulti(Consumer<MultiEmitter<? super T>> consumer, BackPressureStrategy backpressure); @Override void subscribe(MultiSubscriber<? super T> downstream);  }