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;
 
 /**
@@ -35,36 +35,30 @@ public OperatorSwitchIfEmpty(Observable<? extends T> alternate) {
     @Override
     public Subscriber<? super T> call(Subscriber<? super T> child) {
         final SerialSubscription ssub = new SerialSubscription();
-        final SwitchIfEmptySubscriber parent = new SwitchIfEmptySubscriber(child, ssub);
+        ProducerArbiter arbiter = new ProducerArbiter();
+        final SwitchIfEmptySubscriber parent = new SwitchIfEmptySubscriber(child, ssub, arbiter);
         ssub.set(parent);
         child.add(ssub);
+        child.setProducer(arbiter);
         return parent;
     }
 
     private class SwitchIfEmptySubscriber extends Subscriber<T> {
 
-        boolean empty = true;
-        final AtomicLong consumerCapacity = new AtomicLong(0l);
-
+        private boolean empty = true;
         private final Subscriber<? super T> child;
-        final SerialSubscription ssub;
+        private final SerialSubscription ssub;
+        private final ProducerArbiter arbiter;
 
-        public SwitchIfEmptySubscriber(Subscriber<? super T> child, final SerialSubscription ssub) {
+        SwitchIfEmptySubscriber(Subscriber<? super T> child, final SerialSubscription ssub, ProducerArbiter arbiter) {
             this.child = child;
             this.ssub = ssub;
+            this.arbiter = arbiter;
         }
 
         @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
@@ -77,41 +71,7 @@ public void onCompleted() {
         }
 
         private void subscribeToAlternate() {
-            ssub.set(alternate.unsafeSubscribe(new Subscriber<T>() {
-
-                @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);
-                    }
-                }
-
-                @Override
-                public void onCompleted() {
-                    child.onCompleted();
-                }
-
-                @Override
-                public void onError(Throwable e) {
-                    child.onError(e);
-                }
-
-                @Override
-                public void onNext(T t) {
-                    child.onNext(t);
-                }
-            }));
+            ssub.set(alternate.unsafeSubscribe(new AlternateSubscriber<T>(child, arbiter)));
         }
 
         @Override
@@ -125,4 +85,36 @@ public void onNext(T t) {
             child.onNext(t);
         }
     }
+    
+    private static final class AlternateSubscriber<T> extends Subscriber<T> {
+        
+        private final ProducerArbiter arbiter;
+        private final Subscriber<? super T> child;
+
+        AlternateSubscriber(Subscriber<? super T> child, ProducerArbiter arbiter) {
+            this.child = child;
+            this.arbiter = arbiter;
+        }
+        
+        @Override
+        public void setProducer(final Producer producer) {
+            arbiter.setProducer(producer);
+        }
+
+        @Override
+        public void onCompleted() {
+            child.onCompleted();
+        }
+
+        @Override
+        public void onError(Throwable e) {
+            child.onError(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,10 @@ public void onStart() {
 
         assertEquals(Arrays.asList(1), ts.getOnNextEvents());
         ts.assertNoErrors();
+        ts.requestMore(1);
+        ts.assertValueCount(2);
+        ts.requestMore(1);
+        ts.assertValueCount(3);
     }
     @Test
     public void testBackpressureNoRequest() {
@@ -153,8 +161,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();
+    }
 }