### Question:
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);  }### Answer:
@Test public void longSequence() { TestSubscriber<Object> ts = new TestSubscriber<>(Long.MAX_VALUE); Multi.range(1, 1000) .observeOn(executor) .subscribe(ts); ts.awaitDone(5, TimeUnit.SECONDS) .assertItemCount(1000) .assertComplete(); }
@Test public void longSequence2() { TestSubscriber<Object> ts = new TestSubscriber<>(Long.MAX_VALUE); Multi.range(1, 1000) .observeOn(executor, Flow.defaultBufferSize(), false) .subscribe(ts); ts.awaitDone(5, TimeUnit.SECONDS) .assertItemCount(1000) .assertComplete(); }
@Test public void longSequence3() { TestSubscriber<Object> ts = new TestSubscriber<>(Long.MAX_VALUE); Multi.range(1, 1_000_000) .observeOn(executor, Flow.defaultBufferSize(), false) .subscribe(ts); ts.awaitDone(5, TimeUnit.SECONDS) .assertItemCount(1_000_000) .assertComplete(); }
@Test public void oneByOne() { TestSubscriber<Object> ts = new TestSubscriber<>(Long.MAX_VALUE); Multi.range(1, 1000) .observeOn(executor, 1, false) .subscribe(ts); ts.awaitDone(5, TimeUnit.SECONDS) .assertItemCount(1000) .assertComplete(); }
@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); }