Renamed operators to mergeEmptyWith and concatEmptyWith.

Based on discussions with @abersnaze and @benjchristensen these names better suit the intent of these operators.

Also, removed the `swithMap()` + `mergeWith`/`concatWith` implementations with custom operators that are much lighter weight for this case (source never emits an item) vis-a-vis `concat` and `merge`.

Author: Nitesh Kant

src/main/java/rx/Observable.java

@@ -9468,7 +9468,7 @@ public final <T2, R> Observable<R> zipWith(Observable<? extends T2> other, Func2
     }
 
     /**
-     * Returns an Observable that upon completion of the source Observable subscribes to the passed {@code continueWith}
+     * Returns an Observable that upon completion of the source Observable subscribes to the passed {@code other}
      * Observable and then emits all items emitted by that Observable. This function does not expect the source
      * Observable to emit any item, in case, the source Observable, emits any item, an {@link IllegalStateException}
      * is raised.
@@ -9482,23 +9482,23 @@ public final <T2, R> Observable<R> zipWith(Observable<? extends T2> other, Func2
      *
      * <dl>
      *  <dt><b>Scheduler:</b></dt>
-     *  <dd>{@code continueWith} does not operate by default on a particular {@link Scheduler}.</dd>
+     *  <dd>{@code concatEmptyWith} does not operate by default on a particular {@link Scheduler}.</dd>
      * </dl>
      *
-     * @return an Observable that upon completion of the source, starts emitting items from the {@code continueWith}
+     *  <dt><b>Backpressure:</b></dt>
+     *  <dd>{@code concatEmptyWith} does not propagate any demands from the subscriber to the source {@code Observable}
+     *  as it never expects the source to ever emit an item. All demands are sent to the {@code other}
+     *  {@code Observable}.</dd>
+     *
+     * @return an Observable that upon completion of the source, starts emitting items from the {@code other}
      * Observable.
      * @throws IllegalStateException If the source emits any item.
      *
-     * @see #mergeError(Observable)
+     * @see #mergeEmptyWith(Observable)
      */
     @Experimental
-    public final <R> Observable<R> continueWith(Observable<R> continueWith) {
-        return switchMap(new Func1<T, Observable<R>>() {
-            @Override
-            public Observable<R> call(T t) {
-                return Observable.error(new IllegalStateException());
-            }
-        }).concatWith(continueWith);
+    public final <R> Observable<R> concatEmptyWith(Observable<R> other) {
+        return lift(new OperatorConcatEmptyWith<T, R>(other));
     }
 
     /**
@@ -9514,24 +9514,26 @@ public Observable<R> call(T t) {
      *
      * <dl>
      *  <dt><b>Scheduler:</b></dt>
-     *  <dd>{@code continueWith} does not operate by default on a particular {@link Scheduler}.</dd>
+     *  <dd>{@code mergeEmptyWith} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     *
+     * <dl>
+     *  <dt><b>Backpressure:</b></dt>
+     *  <dd>{@code mergeEmptyWith} does not propagate any demands from the subscriber to the source {@code Observable}
+     *  as it never expects the source to ever emit an item. All demands are sent to the {@code other}
+     *  {@code Observable}.</dd>
      * </dl>
      *
      * @return an Observable that only listens for errors from the source and starts emitting items from the
      * {@code other} Observable on subscription.
      * Observable.
      * @throws IllegalStateException If the source emits any item.
      *
-     * @see #continueWith(Observable)
+     * @see #concatEmptyWith(Observable)
      */
     @Experimental
-    public final <R> Observable<R> mergeError(Observable<R> other) {
-        return switchMap(new Func1<T, Observable<R>>() {
-            @Override
-            public Observable<R> call(T t) {
-                return Observable.error(new IllegalStateException());
-            }
-        }).mergeWith(other);
+    public final <R> Observable<R> mergeEmptyWith(Observable<R> other) {
+        return lift(new OperatorMergeEmptyWith<T, R>(other));
     }
 
     /**

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

@@ -0,0 +1,193 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package rx.internal.operators;
+
+import rx.Observable;
+import rx.Observable.Operator;
+import rx.Producer;
+import rx.Subscriber;
+import rx.functions.Action0;
+import rx.internal.producers.ProducerArbiter;
+import rx.observers.SerializedSubscriber;
+import rx.subscriptions.SerialSubscription;
+import rx.subscriptions.Subscriptions;
+
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import java.util.concurrent.atomic.AtomicLongFieldUpdater;
+
+/**
+ * Returns an Observable that emits an error if any item is emitted by the source and emits items from the supplied
+ * alternate {@code Observable} after the source completes.
+ *
+ * @param <T> the source value type
+ * @param <R> the result value type
+ */
+public final class OperatorConcatEmptyWith<T, R> implements Operator<R, T> {
+
+    private final Observable<? extends R> alternate;
+
+    public OperatorConcatEmptyWith(Observable<? extends R> alternate) {
+        this.alternate = alternate;
+    }
+
+    @Override
+    public Subscriber<? super T> call(Subscriber<? super R> child) {
+        final SerialSubscription ssub = new SerialSubscription();
+        final ParentSubscriber parent = new ParentSubscriber(child, ssub, alternate);
+        ssub.set(parent);
+        child.add(ssub);
+        child.setProducer(parent.emptyProducer);
+        return parent;
+    }
+
+    private final class ParentSubscriber extends Subscriber<T> {
+
+        private final Subscriber<? super R> child;
+        private final SerialSubscription ssub;
+        private final EmptyProducer emptyProducer;
+        private final Observable<? extends R> alternate;
+
+        ParentSubscriber(Subscriber<? super R> child, final SerialSubscription ssub, Observable<? extends R> alternate) {
+            this.child = child;
+            this.ssub = ssub;
+            this.emptyProducer = new EmptyProducer();
+            this.alternate = alternate;
+        }
+
+        @Override
+        public void setProducer(final Producer producer) {
+            /*Nothing from this producer will ever be requested as we never expect any items to be emitted from the
+            parent. The only thing w.r.t backpressure that is required is to store the requested count from the
+            child till the time the alternate observable is subscribed. So, this producer is just ignored and the
+            configured EmptyProducer on the child stores any buffered requested items till subscription to alternate.*/
+        }
+
+        @Override
+        public void onCompleted() {
+            if (!child.isUnsubscribed()) {
+                AlternateSubscriber as = new AlternateSubscriber(child, emptyProducer);
+                ssub.set(as);
+                alternate.unsafeSubscribe(as);
+            }
+        }
+
+        @Override
+        public void onError(Throwable e) {
+            child.onError(e);
+        }
+
+        @Override
+        public void onNext(T t) {
+            onError(new IllegalStateException("Concat empty with source emitted an item: " + t));
+        }
+    }
+
+    private final class AlternateSubscriber extends Subscriber<R> {
+
+        private final EmptyProducer emptyProducer;
+        private final Subscriber<? super R> child;
+
+        AlternateSubscriber(Subscriber<? super R> child, EmptyProducer emptyProducer) {
+            this.child = child;
+            this.emptyProducer = emptyProducer;
+        }
+
+        @Override
+        public void setProducer(final Producer producer) {
+            emptyProducer.setAltProducer(producer);
+        }
+
+        @Override
+        public void onCompleted() {
+            child.onCompleted();
+        }
+
+        @Override
+        public void onError(Throwable e) {
+            child.onError(e);
+        }
+
+        @Override
+        public void onNext(R r) {
+            child.onNext(r);
+        }
+    }
+
+    /**
+     * This is a producer implementation that only does the following:
+     * <ul>
+     <li>If the alternate producer has not yet arrived, store the total requested count from downstream.</li>
+     <li>If the alternate producer has arrived, then relay the request demand to it.</li>
+     </ul>
+     *
+     * Since, this is only applicable to this operator, it does not check for emissions from the source, as the source
+     * is never expected to emit any item. Thus it is "lighter" weight than {@link ProducerArbiter}
+     */
+    private static final class EmptyProducer implements Producer {
+
+        /*Total requested items till the time the alternate producer arrives.*/
+        private long requested; /*Guarded by this*/
+        /*Producer from the alternate Observable for this operator*/
+        private Producer altProducer; /*Guarded by this*/
+
+        @Override
+        public void request(long requested) {
+            if (requested < 0) {
+                throw new IllegalArgumentException("Requested items can not be negative.");
+            }
+            if (requested == 0) {
+                return;
+            }
+
+            boolean requestToAlternate = false;
+
+            synchronized (this) {
+                if (null == altProducer) {
+                    /*Accumulate requested till the time an alternate producer arrives.*/
+                    this.requested += requested;
+                } else {
+                    /*If the alternate producer exists, then relay.*/
+                    requestToAlternate = true;
+                }
+            }
+
+            if (requestToAlternate) {
+                altProducer.request(requested);
+            }
+        }
+
+        private void setAltProducer(Producer altProducer) {
+            if (null == altProducer) {
+                throw new IllegalArgumentException("Producer can not be null.");
+            }
+
+            long requestToAlternate = 0;
+
+            synchronized (this) {
+                if (0 != requested) {
+                    /*Something was requested from the source Observable, relay that to the new producer*/
+                    requestToAlternate = requested;
+                }
+                this.altProducer = altProducer;
+            }
+
+            if (0 != requestToAlternate) {
+                this.altProducer.request(requestToAlternate);
+            }
+        }
+    }
+}

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

@@ -0,0 +1,75 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package rx.internal.operators;
+
+import rx.Observable;
+import rx.Observable.Operator;
+import rx.Producer;
+import rx.Subscriber;
+
+/**
+ * Returns an Observable that emits an error if any item is emitted by the source and emits items from the supplied
+ * alternate {@code Observable}. The errors from source are propagated as-is.
+ *
+ * @param <T> the source value type
+ * @param <R> the result value type
+ */
+public final class OperatorMergeEmptyWith<T, R> implements Operator<R, T> {
+
+    private final Observable<? extends R> alternate;
+
+    public OperatorMergeEmptyWith(Observable<? extends R> alternate) {
+        this.alternate = alternate;
+    }
+
+    @Override
+    public Subscriber<? super T> call(Subscriber<? super R> child) {
+        final ParentSubscriber parent = new ParentSubscriber(child);
+        child.add(parent);
+        alternate.unsafeSubscribe(child);
+        return parent;
+    }
+
+    private final class ParentSubscriber extends Subscriber<T> {
+
+        private final Subscriber<? super R> child;
+
+        ParentSubscriber(Subscriber<? super R> child) {
+            this.child = child;
+        }
+
+        @Override
+        public void setProducer(final Producer producer) {
+            /*Nothing from this producer will ever be requested as we never expect any items to be emitted from the
+            parent.*/
+        }
+
+        @Override
+        public void onCompleted() {
+            // Nothing to do as the child completes with the alternate observable.
+        }
+
+        @Override
+        public void onError(Throwable e) {
+            child.onError(e);
+        }
+
+        @Override
+        public void onNext(T t) {
+            onError(new IllegalStateException("Merge empty with source emitted an item: " + t));
+        }
+    }
+}

src/test/java/rx/internal/operators/OperatorContinueWithTest.java renamed to src/test/java/rx/internal/operators/OperatorConcatEmptyWithTest.java

@@ -21,13 +21,13 @@
 import rx.schedulers.Schedulers;
 import rx.schedulers.TestScheduler;
 
-public class OperatorContinueWithTest {
+public class OperatorConcatEmptyWithTest {
 
     @Test(timeout = 60000)
     public void testWithVoid() {
         final String soleValue = "Hello";
         Observable<String> source = Observable.<Void>empty()
-                                              .continueWith(Observable.just(soleValue));
+                                              .concatEmptyWith(Observable.just(soleValue));
 
         TestSubscriber<String> testSubscriber = new TestSubscriber<String>();
         source.subscribe(testSubscriber);
@@ -40,7 +40,7 @@ public void testWithVoid() {
     @Test(timeout = 60000)
     public void testErrorOnSourceEmitItem() {
         Observable<String> source = Observable.just(1)
-                                              .continueWith(Observable.just("Hello"));
+                                              .concatEmptyWith(Observable.just("Hello"));
 
         TestSubscriber<String> testSubscriber = new TestSubscriber<String>();
         source.subscribe(testSubscriber);
@@ -53,7 +53,7 @@ public void testErrorOnSourceEmitItem() {
     @Test(timeout = 60000)
     public void testSourceError() throws Exception {
         Observable<String> source = Observable.<Void>error(new IllegalStateException())
-                                              .continueWith(Observable.just("Hello"));
+                                              .concatEmptyWith(Observable.just("Hello"));
 
         TestSubscriber<String> testSubscriber = new TestSubscriber<String>();
         source.subscribe(testSubscriber);
@@ -71,7 +71,7 @@ public void testNoSubscribeBeforeSourceCompletion() {
         /*Delaying on complete event so to check that the subscription does not happen before completion*/
         Observable<String> source = Observable.<Void>empty()
                                               .observeOn(testScheduler)
-                                              .continueWith(Observable.just(soleValue));
+                                              .concatEmptyWith(Observable.just(soleValue));
 
         TestSubscriber<String> testSubscriber = new TestSubscriber<String>();
         source.subscribe(testSubscriber);