Added event-merger overload

Author: akarnokd

rxjava-core/src/main/java/rx/Observable.java

@@ -3901,6 +3901,23 @@ public <U, R> Observable<R> mergeMapIterable(Func1<? super T, ? extends Iterable
         return mergeMap(OperationFlatMap.flatMapIterableFunc(collectionSelector), resultSelector);
     }
 
+    /**
+     * Create an Observable that projects the notification of an observable sequence to an observable
+     * sequence and merges the results into one.
+     * @param <R> the result type
+     * @param onNext function returning a collection to merge for each onNext event of the source
+     * @param onError function returning a collection to merge for an onError event
+     * @param onCompleted function returning a collection to merge for an onCompleted event
+     * @return an Observable that projects the notification of an observable sequence to an observable
+     * sequence and merges the results into one.
+     */
+    public <R> Observable<R> mergeMap(
+            Func1<? super T, ? extends Observable<? extends R>> onNext, 
+            Func1<? super Throwable, ? extends Observable<? extends R>> onError, 
+            Func0<? extends Observable<? extends R>> onCompleted) {
+        return create(OperationFlatMap.flatMap(this, onNext, onError, onCompleted));
+    }
+    
     /**
      * Creates a new Observable by applying a function that you supply to each
      * item emitted by the source Observable, where that function returns an

rxjava-core/src/main/java/rx/operators/OperationFlatMap.java

@@ -22,6 +22,7 @@
 import rx.Subscription;
 import rx.subscriptions.CompositeSubscription;
 import rx.subscriptions.SerialSubscription;
+import rx.util.functions.Func0;
 import rx.util.functions.Func1;
 import rx.util.functions.Func2;
 
@@ -205,4 +206,165 @@ public void onCompleted() {
             }
         };
     }
+    
+    /**
+     * Projects the notification of an observable sequence to an observable
+     * sequence and merges the results into one.
+     */
+    public static <T, R> OnSubscribeFunc<R> flatMap(Observable<? extends T> source, 
+            Func1<? super T, ? extends Observable<? extends R>> onNext, 
+            Func1<? super Throwable, ? extends Observable<? extends R>> onError, 
+            Func0<? extends Observable<? extends R>> onCompleted) {
+        return new FlatMapTransform<T, R>(source, onNext, onError, onCompleted);
+    }
+    
+    /**
+     * Projects the notification of an observable sequence to an observable
+     * sequence and merges the results into one.
+     * @param <T> the source value type
+     * @param <R> the result value type
+     */
+    private static final class FlatMapTransform<T, R> implements OnSubscribeFunc<R> {
+        final Observable<? extends T> source;
+        final Func1<? super T, ? extends Observable<? extends R>> onNext;
+        final Func1<? super Throwable, ? extends Observable<? extends R>> onError;
+        final Func0<? extends Observable<? extends R>> onCompleted;
+
+        public FlatMapTransform(Observable<? extends T> source, Func1<? super T, ? extends Observable<? extends R>> onNext, Func1<? super Throwable, ? extends Observable<? extends R>> onError, Func0<? extends Observable<? extends R>> onCompleted) {
+            this.source = source;
+            this.onNext = onNext;
+            this.onError = onError;
+            this.onCompleted = onCompleted;
+        }
+
+        @Override
+        public Subscription onSubscribe(Observer<? super R> t1) {
+            CompositeSubscription csub = new CompositeSubscription();
+            
+            csub.add(source.subscribe(new SourceObserver<T, R>(t1, onNext, onError, onCompleted, csub)));
+            
+            return csub;
+        }
+        /**
+         * Observe the source and merge the values.
+         * @param <T> the source value type
+         * @param <R> the result value type
+         */
+        private static final class SourceObserver<T, R> implements Observer<T> {
+            final Observer<? super R> observer;
+            final Func1<? super T, ? extends Observable<? extends R>> onNext;
+            final Func1<? super Throwable, ? extends Observable<? extends R>> onError;
+            final Func0<? extends Observable<? extends R>> onCompleted;
+            final CompositeSubscription csub;
+            final AtomicInteger wip;
+            volatile boolean done;
+            final Object guard;
+
+            public SourceObserver(Observer<? super R> observer, Func1<? super T, ? extends Observable<? extends R>> onNext, Func1<? super Throwable, ? extends Observable<? extends R>> onError, Func0<? extends Observable<? extends R>> onCompleted, CompositeSubscription csub) {
+                this.observer = observer;
+                this.onNext = onNext;
+                this.onError = onError;
+                this.onCompleted = onCompleted;
+                this.csub = csub;
+                this.guard = new Object();
+                this.wip = new AtomicInteger(1);
+            }
+
+            @Override
+            public void onNext(T args) {
+                Observable<? extends R> o;
+                try {
+                    o = onNext.call(args);
+                } catch (Throwable t) {
+                    synchronized (guard) {
+                        observer.onError(t);
+                    }
+                    csub.unsubscribe();
+                    return;
+                }
+                subscribeInner(o);
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                Observable<? extends R> o;
+                try {
+                    o = onError.call(e);
+                } catch (Throwable t) {
+                    synchronized (guard) {
+                        observer.onError(t);
+                    }
+                    csub.unsubscribe();
+                    return;
+                }
+                subscribeInner(o);
+                done = true;
+                finish();
+            }
+
+            @Override
+            public void onCompleted() {
+                Observable<? extends R> o;
+                try {
+                    o = onCompleted.call();
+                } catch (Throwable t) {
+                    synchronized (guard) {
+                        observer.onError(t);
+                    }
+                    csub.unsubscribe();
+                    return;
+                }
+                subscribeInner(o);
+                done = true;
+                finish();
+            }
+            
+            void subscribeInner(Observable<? extends R> o) {
+                SerialSubscription ssub = new SerialSubscription();
+                wip.incrementAndGet();
+                csub.add(ssub);
+                
+                ssub.set(o.subscribe(new CollectionObserver<T, R>(this, ssub)));
+            }
+            void finish() {
+                if (wip.decrementAndGet() == 0) {
+                    synchronized (guard) {
+                        observer.onCompleted();
+                    }
+                    csub.unsubscribe();
+                }
+            }
+        }
+        /** Observes the collections. */
+        private static final class CollectionObserver<T, R> implements Observer<R> {
+            final SourceObserver<T, R> parent;
+            final Subscription cancel;
+
+            public CollectionObserver(SourceObserver<T, R> parent, Subscription cancel) {
+                this.parent = parent;
+                this.cancel = cancel;
+            }
+
+            @Override
+            public void onNext(R args) {
+                synchronized (parent.guard) {
+                    parent.observer.onNext(args);
+                }
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                synchronized (parent.guard) {
+                    parent.observer.onError(e);
+                }
+                parent.csub.unsubscribe();
+            }
+
+            @Override
+            public void onCompleted() {
+                parent.csub.remove(cancel);
+                parent.finish();
+            }
+        }
+    }
 }

rxjava-core/src/test/java/rx/operators/OperationFlatMapTest.java

@@ -21,6 +21,7 @@
 import static org.mockito.Mockito.*;
 import rx.Observable;
 import rx.Observer;
+import rx.util.functions.Func0;
 import rx.util.functions.Func1;
 import rx.util.functions.Func2;
 
@@ -142,4 +143,153 @@ public Integer call(Integer t1, Integer t2) {
         verify(o, never()).onNext(any());
         verify(o).onError(any(OperationReduceTest.CustomException.class));
     }
+    <T, R> Func1<T, R> just(final R value) {
+        return new Func1<T, R>() {
+
+            @Override
+            public R call(T t1) {
+                return value;
+            }
+        };
+    }
+    <R> Func0<R> just0(final R value) {
+        return new Func0<R>() {
+
+            @Override
+            public R call() {
+                return value;
+            }
+        };
+    }
+    @Test
+    public void testFlatMapTransformsNormal() {
+        Observable<Integer> onNext = Observable.from(Arrays.asList(1, 2, 3));
+        Observable<Integer> onCompleted = Observable.from(Arrays.asList(4));
+        Observable<Integer> onError = Observable.from(Arrays.asList(5));
+        
+        Observable<Integer> source = Observable.from(Arrays.asList(10, 20, 30));
+        
+        @SuppressWarnings("unchecked")
+        Observer<Object> o = mock(Observer.class);
+        
+        source.mergeMap(just(onNext), just(onError), just0(onCompleted)).subscribe(o);
+        
+        verify(o, times(3)).onNext(1);
+        verify(o, times(3)).onNext(2);
+        verify(o, times(3)).onNext(3);
+        verify(o).onNext(4);
+        verify(o).onCompleted();
+        
+        verify(o, never()).onNext(5);
+        verify(o, never()).onError(any(Throwable.class));
+    }
+    @Test
+    public void testFlatMapTransformsException() {
+        Observable<Integer> onNext = Observable.from(Arrays.asList(1, 2, 3));
+        Observable<Integer> onCompleted = Observable.from(Arrays.asList(4));
+        Observable<Integer> onError = Observable.from(Arrays.asList(5));
+        
+        Observable<Integer> source = Observable.concat(
+                Observable.from(Arrays.asList(10, 20, 30))
+                , Observable.<Integer>error(new RuntimeException("Forced failure!"))
+                );
+        
+        @SuppressWarnings("unchecked")
+        Observer<Object> o = mock(Observer.class);
+        
+        source.mergeMap(just(onNext), just(onError), just0(onCompleted)).subscribe(o);
+        
+        verify(o, times(3)).onNext(1);
+        verify(o, times(3)).onNext(2);
+        verify(o, times(3)).onNext(3);
+        verify(o).onNext(5);
+        verify(o).onCompleted();
+        verify(o, never()).onNext(4);
+        
+        verify(o, never()).onError(any(Throwable.class));
+    }
+    <R> Func0<R> funcThrow0(R r) {
+        return new Func0<R>() {
+            @Override
+            public R call() {
+                throw new OperationReduceTest.CustomException();
+            }
+        };
+    }
+    <T, R> Func1<T, R> funcThrow(T t, R r) {
+        return new Func1<T, R>() {
+            @Override
+            public R call(T t) {
+                throw new OperationReduceTest.CustomException();
+            }
+        };
+    }
+    @Test
+    public void testFlatMapTransformsOnNextFuncThrows() {
+        Observable<Integer> onCompleted = Observable.from(Arrays.asList(4));
+        Observable<Integer> onError = Observable.from(Arrays.asList(5));
+        
+        Observable<Integer> source = Observable.from(Arrays.asList(10, 20, 30));
+        
+        @SuppressWarnings("unchecked")
+        Observer<Object> o = mock(Observer.class);
+        
+        source.mergeMap(funcThrow(1, onError), just(onError), just0(onCompleted)).subscribe(o);
+        
+        verify(o).onError(any(OperationReduceTest.CustomException.class));
+        verify(o, never()).onNext(any());
+        verify(o, never()).onCompleted();
+    }
+    @Test
+    public void testFlatMapTransformsOnErrorFuncThrows() {
+        Observable<Integer> onNext = Observable.from(Arrays.asList(1, 2, 3));
+        Observable<Integer> onCompleted = Observable.from(Arrays.asList(4));
+        Observable<Integer> onError = Observable.from(Arrays.asList(5));
+        
+        Observable<Integer> source = Observable.error(new OperationReduceTest.CustomException());
+        
+        @SuppressWarnings("unchecked")
+        Observer<Object> o = mock(Observer.class);
+        
+        source.mergeMap(just(onNext), funcThrow((Throwable)null, onError), just0(onCompleted)).subscribe(o);
+        
+        verify(o).onError(any(OperationReduceTest.CustomException.class));
+        verify(o, never()).onNext(any());
+        verify(o, never()).onCompleted();
+    }
+    
+    @Test
+    public void testFlatMapTransformsOnCompletedFuncThrows() {
+        Observable<Integer> onNext = Observable.from(Arrays.asList(1, 2, 3));
+        Observable<Integer> onCompleted = Observable.from(Arrays.asList(4));
+        Observable<Integer> onError = Observable.from(Arrays.asList(5));
+        
+        Observable<Integer> source = Observable.from(Arrays.<Integer>asList());
+        
+        @SuppressWarnings("unchecked")
+        Observer<Object> o = mock(Observer.class);
+        
+        source.mergeMap(just(onNext), just(onError), funcThrow0(onCompleted)).subscribe(o);
+        
+        verify(o).onError(any(OperationReduceTest.CustomException.class));
+        verify(o, never()).onNext(any());
+        verify(o, never()).onCompleted();
+    }
+    @Test 
+    public void testFlatMapTransformsMergeException() {
+        Observable<Integer> onNext = Observable.error(new OperationReduceTest.CustomException());
+        Observable<Integer> onCompleted = Observable.from(Arrays.asList(4));
+        Observable<Integer> onError = Observable.from(Arrays.asList(5));
+        
+        Observable<Integer> source = Observable.from(Arrays.asList(10, 20, 30));
+        
+        @SuppressWarnings("unchecked")
+        Observer<Object> o = mock(Observer.class);
+        
+        source.mergeMap(just(onNext), just(onError), funcThrow0(onCompleted)).subscribe(o);
+        
+        verify(o).onError(any(OperationReduceTest.CustomException.class));
+        verify(o, never()).onNext(any());
+        verify(o, never()).onCompleted();
+    }
 }