fix request processing in OperatorSwitchIfNext

Author: davidmoten

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

@@ -15,9 +15,9 @@
  */
 package rx.internal.operators;
 
-import java.util.concurrent.atomic.AtomicLong;
 
 import rx.*;
+import rx.internal.producers.ProducerArbiter;
 import rx.subscriptions.SerialSubscription;
 
 /**
@@ -38,33 +38,35 @@ public Subscriber<? super T> call(Subscriber<? super T> child) {
         final SwitchIfEmptySubscriber parent = new SwitchIfEmptySubscriber(child, ssub);
         ssub.set(parent);
         child.add(ssub);
+        child.setProducer(new Producer() {
+            @Override
+            public void request(long n) {
+                parent.requestMore(n);
+            }});
         return parent;
     }
 
     private class SwitchIfEmptySubscriber extends Subscriber<T> {
 
         boolean empty = true;
-        final AtomicLong consumerCapacity = new AtomicLong(0l);
 
-        private final Subscriber<? super T> child;
+        final Subscriber<? super T> child;
         final SerialSubscription ssub;
+        final ProducerArbiter arbiter;
 
         public SwitchIfEmptySubscriber(Subscriber<? super T> child, final SerialSubscription ssub) {
             this.child = child;
             this.ssub = ssub;
+            this.arbiter = new ProducerArbiter();
+        }
+
+        public void requestMore(long n) {
+            arbiter.request(n);
         }
 
         @Override
         public void setProducer(final Producer producer) {
-            super.setProducer(new Producer() {
-                @Override
-                public void request(long n) {
-                    if (empty) {
-                        consumerCapacity.set(n);
-                    }
-                    producer.request(n);
-                }
-            });
+            arbiter.setProducer(producer);
         }
 
         @Override
@@ -81,20 +83,7 @@ private void subscribeToAlternate() {
 
                 @Override
                 public void setProducer(final Producer producer) {
-                    child.setProducer(new Producer() {
-                        @Override
-                        public void request(long n) {
-                            producer.request(n);
-                        }
-                    });
-                }
-
-                @Override
-                public void onStart() {
-                    final long capacity = consumerCapacity.get();
-                    if (capacity > 0) {
-                        request(capacity);
-                    }
+                    arbiter.setProducer(producer);
                 }
 
                 @Override
@@ -110,6 +99,7 @@ public void onError(Throwable e) {
                 @Override
                 public void onNext(T t) {
                     child.onNext(t);
+                    arbiter.produced(1);
                 }
             }));
         }

src/test/java/rx/internal/operators/OperatorSwitchIfEmptyTest.java

@@ -18,14 +18,18 @@
 import static org.junit.Assert.*;
 
 import java.util.*;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.junit.Test;
 
 import rx.*;
 import rx.Observable;
+import rx.Observable.OnSubscribe;
 import rx.functions.Action0;
+import rx.functions.Action1;
 import rx.observers.TestSubscriber;
+import rx.schedulers.Schedulers;
 import rx.subscriptions.Subscriptions;
 
 public class OperatorSwitchIfEmptyTest {
@@ -142,6 +146,11 @@ public void onStart() {
 
         assertEquals(Arrays.asList(1), ts.getOnNextEvents());
         ts.assertNoErrors();
+        ts.requestMore(1);
+        ts.assertValueCount(2);
+        ts.requestMore(1);
+        ts.assertValueCount(3);
+        System.out.println(ts.getOnNextEvents());
     }
     @Test
     public void testBackpressureNoRequest() {
@@ -153,8 +162,51 @@ public void onStart() {
             }
         };
         Observable.<Integer>empty().switchIfEmpty(Observable.just(1, 2, 3)).subscribe(ts);
-        
         assertTrue(ts.getOnNextEvents().isEmpty());
         ts.assertNoErrors();
     }
+    
+    @Test
+    public void testBackpressureOnFirstObservable() {
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>(0);
+        Observable.just(1,2,3).switchIfEmpty(Observable.just(4, 5, 6)).subscribe(ts);
+        ts.assertNotCompleted();
+        ts.assertNoErrors();
+        ts.assertNoValues();
+    }
+    
+    @Test(timeout = 10000)
+    public void testRequestsNotLost() throws InterruptedException {
+        final TestSubscriber<Long> ts = new TestSubscriber<Long>(0);
+        Observable.create(new OnSubscribe<Long>() {
+
+            @Override
+            public void call(final Subscriber<? super Long> subscriber) {
+                subscriber.setProducer(new Producer() {
+                    final AtomicBoolean completed = new AtomicBoolean(false);
+                    @Override
+                    public void request(long n) {
+                        if (n > 0 && completed.compareAndSet(false, true)) {
+                            Schedulers.io().createWorker().schedule(new Action0() {
+                                @Override
+                                public void call() {
+                                    subscriber.onCompleted();
+                                }}, 100, TimeUnit.MILLISECONDS);
+                        }
+                    }});
+            }})
+          .switchIfEmpty(Observable.from(Arrays.asList(1L, 2L, 3L)))
+          .subscribeOn(Schedulers.computation())
+          .subscribe(ts);
+        ts.requestMore(0);
+        Thread.sleep(50);
+        //request while first observable is still finishing (as empty)
+        ts.requestMore(1);
+        ts.requestMore(1);
+        Thread.sleep(500);
+        ts.assertNotCompleted();
+        ts.assertNoErrors();
+        ts.assertValueCount(2);
+        ts.unsubscribe();
+    }
 }