diff --git a/src/main/java/rx/Observable.java b/src/main/java/rx/Observable.java index 8f4cb464ee..42c3406237 100644 --- a/src/main/java/rx/Observable.java +++ b/src/main/java/rx/Observable.java @@ -906,6 +906,33 @@ public static Observable combineLatest(Iterable(sources, combineFunction)); } + /** + * Combines a collection of source Observables by emitting an item that aggregates the latest values of each of + * the source Observables each time an item is received from any of the source Observables, where this + * aggregation is defined by a specified function and delays any error from the sources until + * all source Observables terminate. + * + *
+ *
Scheduler:
+ *
{@code combineLatest} does not operate by default on a particular {@link Scheduler}.
+ *
+ * + * @param + * the common base type of source values + * @param + * the result type + * @param sources + * the collection of source Observables + * @param combineFunction + * the aggregation function used to combine the items emitted by the source Observables + * @return an Observable that emits items that are the result of combining the items emitted by the source + * Observables by means of the given aggregation function + * @see ReactiveX operators documentation: CombineLatest + */ + public static Observable combineLatestDelayError(Iterable> sources, FuncN combineFunction) { + return create(new OnSubscribeCombineLatest(null, sources, combineFunction, RxRingBuffer.SIZE, true)); + } + /** * Returns an Observable that emits the items emitted by each of the Observables emitted by the source * Observable, one after the other, without interleaving them. diff --git a/src/main/java/rx/internal/operators/OnSubscribeCombineLatest.java b/src/main/java/rx/internal/operators/OnSubscribeCombineLatest.java index 152a0831b0..93dcb5de5d 100644 --- a/src/main/java/rx/internal/operators/OnSubscribeCombineLatest.java +++ b/src/main/java/rx/internal/operators/OnSubscribeCombineLatest.java @@ -213,7 +213,7 @@ void combine(Object value, int index) { if (value != null && allSourcesFinished) { queue.offer(combinerSubscriber, latest.clone()); } else - if (value == null && error.get() != null) { + if (value == null && error.get() != null && (o == MISSING || !delayError)) { done = true; // if this source completed without a value } } else { diff --git a/src/test/java/rx/internal/operators/OnSubscribeCombineLatestTest.java b/src/test/java/rx/internal/operators/OnSubscribeCombineLatestTest.java index 67085640e2..840077af59 100644 --- a/src/test/java/rx/internal/operators/OnSubscribeCombineLatestTest.java +++ b/src/test/java/rx/internal/operators/OnSubscribeCombineLatestTest.java @@ -29,6 +29,7 @@ import rx.*; import rx.Observable; import rx.Observer; +import rx.exceptions.*; import rx.functions.*; import rx.internal.util.RxRingBuffer; import rx.observers.TestSubscriber; @@ -954,5 +955,107 @@ public Integer call(Object... args) { throw new RuntimeException(); } - }; + }; + + @SuppressWarnings("unchecked") + @Test + public void firstJustError() { + TestSubscriber ts = TestSubscriber.create(); + + Observable.combineLatestDelayError( + Arrays.asList(Observable.just(1), Observable.error(new TestException())), + new FuncN() { + @Override + public Integer call(Object... args) { + return ((Integer)args[0]) + ((Integer)args[1]); + } + } + ).subscribe(ts); + + ts.assertNoValues(); + ts.assertError(TestException.class); + ts.assertNotCompleted(); + } + + @SuppressWarnings("unchecked") + @Test + public void secondJustError() { + TestSubscriber ts = TestSubscriber.create(); + + Observable.combineLatestDelayError( + Arrays.asList(Observable.error(new TestException()), Observable.just(1)), + new FuncN() { + @Override + public Integer call(Object... args) { + return ((Integer)args[0]) + ((Integer)args[1]); + } + } + ).subscribe(ts); + + ts.assertNoValues(); + ts.assertError(TestException.class); + ts.assertNotCompleted(); + } + + @SuppressWarnings("unchecked") + @Test + public void oneErrors() { + TestSubscriber ts = TestSubscriber.create(); + + Observable.combineLatestDelayError( + Arrays.asList(Observable.just(10).concatWith(Observable.error(new TestException())), Observable.just(1)), + new FuncN() { + @Override + public Integer call(Object... args) { + return ((Integer)args[0]) + ((Integer)args[1]); + } + } + ).subscribe(ts); + + ts.assertValues(11); + ts.assertError(TestException.class); + ts.assertNotCompleted(); + } + + @SuppressWarnings("unchecked") + @Test + public void twoErrors() { + TestSubscriber ts = TestSubscriber.create(); + + Observable.combineLatestDelayError( + Arrays.asList(Observable.just(1), Observable.just(10).concatWith(Observable.error(new TestException()))), + new FuncN() { + @Override + public Integer call(Object... args) { + return ((Integer)args[0]) + ((Integer)args[1]); + } + } + ).subscribe(ts); + + ts.assertValues(11); + ts.assertError(TestException.class); + ts.assertNotCompleted(); + } + + @SuppressWarnings("unchecked") + @Test + public void bothError() { + TestSubscriber ts = TestSubscriber.create(); + + Observable.combineLatestDelayError( + Arrays.asList(Observable.just(1).concatWith(Observable.error(new TestException())), + Observable.just(10).concatWith(Observable.error(new TestException()))), + new FuncN() { + @Override + public Integer call(Object... args) { + return ((Integer)args[0]) + ((Integer)args[1]); + } + } + ).subscribe(ts); + + ts.assertValues(11); + ts.assertError(CompositeException.class); + ts.assertNotCompleted(); + } + }