@Test public void delayError() { TestSubscriber<Object> ts = new TestSubscriber<>(Long.MAX_VALUE); Multi.concat(Multi.range(1, 1000), Multi.error(new IOException())) .observeOn(executor, Flow.defaultBufferSize(), true) .subscribe(ts); ts.awaitDone(5, TimeUnit.SECONDS) .assertItemCount(1000) .assertError(IOException.class); }
@Override public void subscribe(Flow.Subscriber<? super T> subscriber) { source.subscribe(new ObserveOnSubscriber<>(subscriber, executor, bufferSize, delayError)); }
MultiObserveOn implements Multi<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { source.subscribe(new ObserveOnSubscriber<>(subscriber, executor, bufferSize, delayError)); } }
MultiObserveOn implements Multi<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { source.subscribe(new ObserveOnSubscriber<>(subscriber, executor, bufferSize, delayError)); } MultiObserveOn(Multi<T> source, Executor executor, int bufferSize, boolean delayError); }
MultiObserveOn implements Multi<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { source.subscribe(new ObserveOnSubscriber<>(subscriber, executor, bufferSize, delayError)); } MultiObserveOn(Multi<T> source, Executor executor, int bufferSize, boolean delayError); @Override void subscribe(Flow.Subscriber<? super T> subscriber); }
MultiObserveOn implements Multi<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { source.subscribe(new ObserveOnSubscriber<>(subscriber, executor, bufferSize, delayError)); } MultiObserveOn(Multi<T> source, Executor executor, int bufferSize, boolean delayError); @Override void subscribe(Flow.Subscriber<? super T> subscriber);  }