@Test public void testSerializedWithConcurrentEmissionsAndFailure() { AtomicReference<MultiEmitter<? super Integer>> reference = new AtomicReference<>(); AssertSubscriber<Integer> subscriber = Multi.createFrom().<Integer> emitter(reference::set).subscribe() .withSubscriber(AssertSubscriber.create(Long.MAX_VALUE)); await().until(() -> reference.get() != null); CountDownLatch latch = new CountDownLatch(2); Runnable r1 = () -> { latch.countDown(); try { latch.await(); } catch (InterruptedException e) { } for (int i = 0; i < 1000; i++) { reference.get().emit(i); } }; Runnable r2 = () -> { latch.countDown(); try { latch.await(); } catch (InterruptedException e) { } for (int i = 0; i < 200; i++) { reference.get().emit(i); } reference.get().fail(new Exception("boom")); }; ExecutorService service = Executors.newFixedThreadPool(2); service.submit(r1); service.submit(r2); subscriber.await().assertHasFailedWith(Exception.class, "boom"); service.shutdown(); }
@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);  }