Merge, MergeDelayError and MergeMaxConcurrent unified and rewritten.

Author: akarnokd

src/main/java/rx/Observable.java

@@ -1663,8 +1663,9 @@ public final static <T> Observable<T> merge(Observable<? extends Observable<? ex
      *             if {@code maxConcurrent} is less than or equal to 0
      * @see <a href="http://reactivex.io/documentation/operators/merge.html">ReactiveX operators documentation: Merge</a>
      */
+    @Experimental
     public final static <T> Observable<T> merge(Observable<? extends Observable<? extends T>> source, int maxConcurrent) {
-        return source.lift(new OperatorMergeMaxConcurrent<T>(maxConcurrent));
+        return source.lift(OperatorMerge.<T>instance(false, maxConcurrent));
     }
 
     /**
@@ -1935,7 +1936,31 @@ public final static <T> Observable<T> merge(Observable<? extends T> t1, Observab
     public final static <T> Observable<T> merge(Observable<? extends T>[] sequences) {
         return merge(from(sequences));
     }
-
+    
+    /**
+     * Flattens an Array of Observables into one Observable, without any transformation, while limiting the
+     * number of concurrent subscriptions to these Observables.
+     * <p>
+     * <img width="640" height="370" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/merge.io.png" alt="">
+     * <p>
+     * You can combine items emitted by multiple Observables so that they appear as a single Observable, by
+     * using the {@code merge} method.
+     * <dl>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code merge} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     * 
+     * @param sequences
+     *            the Array of Observables
+     * @param maxConcurrent
+     *            the maximum number of Observables that may be subscribed to concurrently
+     * @return an Observable that emits all of the items emitted by the Observables in the Array
+     * @see <a href="http://reactivex.io/documentation/operators/merge.html">ReactiveX operators documentation: Merge</a>
+     */
+    @Experimental
+    public final static <T> Observable<T> merge(Observable<? extends T>[] sequences, int maxConcurrent) {
+        return merge(from(sequences), maxConcurrent);
+    }
     /**
      * Flattens an Observable that emits Observables into one Observable, in a way that allows an Observer to
      * receive all successfully emitted items from all of the source Observables without being interrupted by
@@ -1963,6 +1988,37 @@ public final static <T> Observable<T> merge(Observable<? extends T>[] sequences)
     public final static <T> Observable<T> mergeDelayError(Observable<? extends Observable<? extends T>> source) {
         return source.lift(OperatorMerge.<T>instance(true));
     }
+    /**
+     * Flattens an Observable that emits Observables into one Observable, in a way that allows an Observer to
+     * receive all successfully emitted items from all of the source Observables without being interrupted by
+     * an error notification from one of them, while limiting the
+     * number of concurrent subscriptions to these Observables.
+     * <p>
+     * This behaves like {@link #merge(Observable)} except that if any of the merged Observables notify of an
+     * error via {@link Observer#onError onError}, {@code mergeDelayError} will refrain from propagating that
+     * error notification until all of the merged Observables have finished emitting items.
+     * <p>
+     * <img width="640" height="380" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/mergeDelayError.png" alt="">
+     * <p>
+     * Even if multiple merged Observables send {@code onError} notifications, {@code mergeDelayError} will only
+     * invoke the {@code onError} method of its Observers once.
+     * <dl>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code mergeDelayError} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     * 
+     * @param source
+     *            an Observable that emits Observables
+     * @param maxConcurrent
+     *            the maximum number of Observables that may be subscribed to concurrently
+     * @return an Observable that emits all of the items emitted by the Observables emitted by the
+     *         {@code source} Observable
+     * @see <a href="http://reactivex.io/documentation/operators/merge.html">ReactiveX operators documentation: Merge</a>
+     */
+    @Experimental
+    public final static <T> Observable<T> mergeDelayError(Observable<? extends Observable<? extends T>> source, int maxConcurrent) {
+        return source.lift(OperatorMerge.<T>instance(true, maxConcurrent));
+    }
 
     /**
      * Flattens two Observables into one Observable, in a way that allows an Observer to receive all

src/main/java/rx/internal/operators/OperatorAll.java

@@ -46,8 +46,8 @@ public void onNext(T t) {
                     child.onCompleted();
                     unsubscribe();
                 } else {
-                	// if we drop values we must replace them upstream as downstream won't receive and request more
-                	request(1);
+                    // if we drop values we must replace them upstream as downstream won't receive and request more
+                    request(1);
                 }
             }
 

src/main/java/rx/internal/operators/OperatorMapNotification.java

@@ -15,11 +15,12 @@
  */
 package rx.internal.operators;
 
+import java.util.concurrent.atomic.AtomicLong;
+
 import rx.Observable.Operator;
-import rx.Subscriber;
-import rx.exceptions.OnErrorThrowable;
-import rx.functions.Func0;
-import rx.functions.Func1;
+import rx.*;
+import rx.exceptions.*;
+import rx.functions.*;
 
 /**
  * Applies a function of your choosing to every item emitted by an {@code Observable}, and emits the results of
@@ -41,13 +42,18 @@ public OperatorMapNotification(Func1<? super T, ? extends R> onNext, Func1<? sup
 
     @Override
     public Subscriber<? super T> call(final Subscriber<? super R> o) {
-        return new Subscriber<T>(o) {
-
+        Subscriber<T> subscriber = new Subscriber<T>() {
+            SingleEmitter<R> emitter;
+            @Override
+            public void setProducer(Producer producer) {
+                emitter = new SingleEmitter<R>(o, producer, this);
+                o.setProducer(emitter);
+            }
+            
             @Override
             public void onCompleted() {
                 try {
-                    o.onNext(onCompleted.call());
-                    o.onCompleted();
+                    emitter.offerAndComplete(onCompleted.call());
                 } catch (Throwable e) {
                     o.onError(e);
                 }
@@ -56,8 +62,7 @@ public void onCompleted() {
             @Override
             public void onError(Throwable e) {
                 try {
-                    o.onNext(onError.call(e));
-                    o.onCompleted();
+                    emitter.offerAndComplete(onError.call(e));
                 } catch (Throwable e2) {
                     o.onError(e);
                 }
@@ -66,13 +71,154 @@ public void onError(Throwable e) {
             @Override
             public void onNext(T t) {
                 try {
-                    o.onNext(onNext.call(t));
+                    emitter.offer(onNext.call(t));
                 } catch (Throwable e) {
                     o.onError(OnErrorThrowable.addValueAsLastCause(e, t));
                 }
             }
 
         };
+        o.add(subscriber);
+        return subscriber;
     }
-
-}
+    static final class SingleEmitter<T> extends AtomicLong implements Producer, Subscription {
+        /** */
+        private static final long serialVersionUID = -249869671366010660L;
+        final NotificationLite<T> nl;
+        final Subscriber<? super T> child;
+        final Producer producer;
+        final Subscription cancel;
+        volatile Object value;
+        volatile boolean complete;
+        /** Guarded by this. */
+        boolean emitting;
+        /** Guarded by this. */
+        boolean missed;
+        
+        public SingleEmitter(Subscriber<? super T> child, Producer producer, Subscription cancel) {
+            this.child = child;
+            this.producer = producer;
+            this.cancel = cancel;
+            this.nl = NotificationLite.instance();
+        }
+        @Override
+        public void request(long n) {
+            for (;;) {
+                long r = get();
+                if (r < 0) {
+                    return;
+                }
+                long u = r + n;
+                if (u < 0) {
+                    u = Long.MAX_VALUE;
+                }
+                if (compareAndSet(r, u)) {
+                    producer.request(n);
+                    drain();
+                    return;
+                }
+            }
+        }
+        
+        void produced(long n) {
+            for (;;) {
+                long r = get();
+                if (r < 0) {
+                    return;
+                }
+                long u = r - n;
+                if (u < 0) {
+                    throw new IllegalStateException("More produced (" + n + ") than requested (" + r + ")");
+                }
+                if (compareAndSet(r, u)) {
+                    return;
+                }
+            }
+        }
+        
+        public void offer(T value) {
+            if (this.value != null) {
+                child.onError(new MissingBackpressureException());
+                unsubscribe();
+            } else {
+                this.value = nl.next(value);
+                drain();
+            }
+        }
+        public void offerAndComplete(T value) {
+            if (this.value != null) {
+                child.onError(new MissingBackpressureException());
+                unsubscribe();
+            } else {
+                this.value = nl.next(value);
+                this.complete = true;
+                drain();
+            }
+        }
+        
+        void drain() {
+            synchronized (this) {
+                if (emitting) {
+                    missed = true;
+                    return;
+                }
+                emitting = true;
+                missed = false;
+            }
+            boolean skipFinal = false;
+            try {
+                for (;;) {
+                    
+                    long r = get();
+                    boolean c = complete;
+                    Object v = value;
+                    
+                    if (c && v == null) {
+                        child.onCompleted();
+                    } else
+                    if (r > 0 && v != null) {
+                        value = null;
+                        child.onNext(nl.getValue(v));
+                        if (c) {
+                            child.onCompleted();
+                            skipFinal = true;
+                            return;
+                        } else {
+                            produced(1);
+                        }
+                    }
+                    
+                    synchronized (this) {
+                        if (!missed) {
+                            skipFinal = true;
+                            emitting = false;
+                            return;
+                        }
+                        missed = false;
+                    }
+                }
+            } finally {
+                if (!skipFinal) {
+                    synchronized (this) {
+                        emitting = false;
+                    }
+                }
+            }
+        }
+        
+        @Override
+        public boolean isUnsubscribed() {
+            return get() < 0;
+        }
+        @Override
+        public void unsubscribe() {
+            long r = get();
+            if (r != Long.MIN_VALUE) {
+                r = getAndSet(Long.MIN_VALUE);
+                if (r != Long.MIN_VALUE) {
+                    cancel.unsubscribe();
+                }
+            }
+        }
+    }
+}