@Test public void testTerminationNotCalled() { AtomicBoolean terminated = new AtomicBoolean(); AssertSubscriber<Integer> subscriber = Multi.createFrom().<Integer> emitter(e -> { e.onTermination(() -> terminated.set(true)); e.emit(1).emit(2).emit(3); }).subscribe().withSubscriber(AssertSubscriber.create(3)); subscriber.assertSubscribed() .assertReceived(1, 2, 3) .assertHasNotCompleted() .assertHasNotFailed(); assertThat(terminated).isFalse(); subscriber.cancel(); assertThat(terminated).isTrue(); }
@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);  }