Modified to conform Rx.NET

Author: akarnokd

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

@@ -2170,10 +2170,7 @@ public static Observable<Long> timer(long initialDelay, long period, TimeUnit un
     /**
      * Create an Observable which delays the events via another Observable on a per item-basis.
      * <p>
-     * Note: onError or onCompleted events are immediately propagated.
-     * <p>
-     * Note: if the Observable returned by the {@code itemDelay} just completes, that
-     * particular source item is not emitted.
+     * Note: onError event is immediately propagated.
      * 
      * @param <U> the item delay value type (ignored)
      * @param itemDelay function that returns an Observable for each source item which is
@@ -2187,13 +2184,7 @@ public <U> Observable<T> delay(Func1<? super T, ? extends Observable<U>> itemDel
     /**
      * Create an Observable which delays the subscription and events via another Observables on a per item-basis.
      * <p>
-     * Note: onError or onCompleted events are immediately propagated.
-     * <p>
-     * Note: if the Observable returned by the {@code itemDelay} just completes, that
-     * particular source item is not emitted.
-     * <p>
-     * Note: if the {@code subscriptionDelay}'s Observable just completes, the created
-     * observable will just complete as well.
+     * Note: onError event is immediately propagated.
      * 
      * @param <U> the subscription delay value type (ignored)
      * @param <V> the item delay value type (ignored)

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

@@ -120,8 +120,11 @@ public DelayViaObservable(Observable<? extends T> source,
         public Subscription onSubscribe(Observer<? super T> t1) {
             CompositeSubscription csub = new CompositeSubscription();
 
+            SerialSubscription sosub = new SerialSubscription();
+            csub.add(sosub);
+            SourceObserver<T, V> so = new SourceObserver<T, V>(t1, itemDelay, csub, sosub);
             if (subscriptionDelay == null) {
-                csub.add(source.subscribe(new SourceObserver<T, V>(t1, itemDelay, csub)));
+                sosub.set(source.subscribe(so));
             } else {
                 Observable<U> subscriptionSource;
                 try {
@@ -132,66 +135,70 @@ public Subscription onSubscribe(Observer<? super T> t1) {
                 }
                 SerialSubscription ssub = new SerialSubscription();
                 csub.add(ssub);
-                ssub.set(subscriptionSource.subscribe(new SubscribeDelay<T, U, V>(source, t1, itemDelay, csub, ssub)));
+                ssub.set(subscriptionSource.subscribe(new SubscribeDelay<T, U, V>(source, so, csub, ssub)));
             }
 
             return csub;
         }
+        /** Subscribe delay observer. */
         private static final class SubscribeDelay<T, U, V> implements Observer<U> {
             final Observable<? extends T> source;
-            final Observer<? super T> observer;
-            final Func1<? super T, ? extends Observable<V>> itemDelay;
+            final SourceObserver<T, V> so;
             final CompositeSubscription csub;
             final Subscription self;
-            /** Prevent any onError and onCompleted once the first item was delivered. */
+            /** Prevent any onError once the first item was delivered. */
             boolean subscribed;
 
-            public SubscribeDelay(Observable<? extends T> source, Observer<? super T> observer, Func1<? super T, ? extends Observable<V>> itemDelay, 
+            public SubscribeDelay(
+                    Observable<? extends T> source,
+                    SourceObserver<T, V> so, 
                     CompositeSubscription csub, Subscription self) {
                 this.source = source;
-                this.observer = observer;
-                this.itemDelay = itemDelay;
+                this.so = so;
                 this.csub = csub;
                 this.self = self;
             }
 
             @Override
             public void onNext(U args) {
-                subscribed = true;
-                csub.remove(self);
-                csub.add(source.subscribe(new SourceObserver<T, V>(observer, itemDelay, csub)));
+                onCompleted();
             }
 
             @Override
             public void onError(Throwable e) {
                 if (!subscribed) {
-                    observer.onError(e);
+                    so.observer.onError(e);
                     csub.unsubscribe();
                 }
             }
 
             @Override
             public void onCompleted() {
-                if (!subscribed) {
-                    observer.onCompleted();
-                    csub.unsubscribe();
-                }
+                subscribed = true;
+                csub.remove(self);
+                so.self.set(source.subscribe(so));
             }
         }
         /** The source observer. */
         private static final class SourceObserver<T, U> implements Observer<T> {
             final Observer<? super T> observer;
             final Func1<? super T, ? extends Observable<U>> itemDelay;
             final CompositeSubscription csub;
+            final SerialSubscription self;
             /** Guard to avoid overlapping events from the various sources. */
             final Object guard;
             boolean done;
+            int wip;
 
-            public SourceObserver(Observer<? super T> observer, Func1<? super T, ? extends Observable<U>> itemDelay, CompositeSubscription csub) {
+            public SourceObserver(Observer<? super T> observer, 
+                    Func1<? super T, ? extends Observable<U>> itemDelay, 
+                    CompositeSubscription csub,
+                    SerialSubscription self) {
                 this.observer = observer;
                 this.itemDelay = itemDelay;
                 this.csub = csub;
                 this.guard = new Object();
+                this.self = self;
             }
 
             @Override
@@ -203,47 +210,57 @@ public void onNext(T args) {
                     onError(t);
                     return;
                 }
+                
+                synchronized (guard) {
+                    wip++;
+                }
+                
                 SerialSubscription ssub = new SerialSubscription();
                 csub.add(ssub);
-                
                 ssub.set(delayer.subscribe(new DelayObserver<T, U>(args, this, ssub)));
             }
 
             @Override
             public void onError(Throwable e) {
                 synchronized (guard) {
-                    if (done) {
-                        return;
-                    }
-                    done = true;
                     observer.onError(e);
                 }
                 csub.unsubscribe();
             }
 
             @Override
             public void onCompleted() {
+                boolean b;
                 synchronized (guard) {
-                    if (done) {
-                        return;
-                    }
                     done = true;
-                    observer.onCompleted();
+                    b = checkDone();
+                }
+                if (b) {
+                    csub.unsubscribe();
+                } else {
+                    self.unsubscribe();
                 }
-                csub.unsubscribe();
             }
 
-            public void emit(T value, Subscription token) {
+            void emit(T value, Subscription token) {
+                boolean b;
                 synchronized (guard) {
-                    if (done) {
-                        return;
-                    }
                     observer.onNext(value);
+                    wip--;
+                    b = checkDone();
+                }
+                if (b) {
+                    csub.unsubscribe();
+                } else {
+                    csub.remove(token);
                 }
-                remove(token);
             }
-            public void remove(Subscription token) {
-                csub.remove(token);
+            boolean checkDone() {
+                if (done && wip == 0) {
+                    observer.onCompleted();
+                    return true;
+                }
+                return false;
             }
         }
         /**
@@ -272,7 +289,7 @@ public void onError(Throwable e) {
 
             @Override
             public void onCompleted() {
-                parent.remove(token);
+                parent.emit(value, token);
             }
 
         }

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

@@ -276,47 +276,7 @@ public Observable<Integer> call(Integer t1) {
 
         verify(o, never()).onError(any(Throwable.class));
     }
-    @Test
-    public void testDelayWithObservableSkipper1() {
-        PublishSubject<Integer> source = PublishSubject.create();
-        final List<PublishSubject<Integer>> delays = new ArrayList<PublishSubject<Integer>>();
-        final int n = 10;
-        for (int i = 0; i < n; i++) {
-            PublishSubject<Integer> delay = PublishSubject.create();
-            delays.add(delay);
-        }
-
-        Func1<Integer, Observable<Integer>> delayFunc = new Func1<Integer, Observable<Integer>>() {
-            @Override
-            public Observable<Integer> call(Integer t1) {
-                return delays.get(t1);
-            }
-        };
-        
-        @SuppressWarnings("unchecked")
-        Observer<Object> o = mock(Observer.class);
-        InOrder inOrder = inOrder(o);
-        
-        source.delay(delayFunc).subscribe(o);
-        
-        
-        for (int i = 0; i < n; i++) {
-            source.onNext(i);
-            if (i % 2 == 0) {
-                delays.get(i).onNext(i);
-                inOrder.verify(o).onNext(i);
-            } else {
-                delays.get(i).onCompleted();
-                inOrder.verify(o, never()).onNext(i);
-            }
-        }
-        source.onCompleted();
-        
-        inOrder.verify(o).onCompleted();
-        inOrder.verifyNoMoreInteractions();
-        
-        verify(o, never()).onError(any(Throwable.class));
-    }
+    
     @Test
     public void testDelayWithObservableSingleSend1() {
         PublishSubject<Integer> source = PublishSubject.create();
@@ -521,4 +481,66 @@ public Observable<Integer> call(Integer t1) {
         verify(o, never()).onNext(any());
         verify(o, never()).onCompleted();
     }
+    @Test
+    public void testDelayWithObservableEmptyDelayer() {
+        PublishSubject<Integer> source = PublishSubject.create();
+
+        Func1<Integer, Observable<Integer>> delayFunc = new Func1<Integer, Observable<Integer>>() {
+
+            @Override
+            public Observable<Integer> call(Integer t1) {
+                return Observable.empty();
+            }
+        };
+        @SuppressWarnings("unchecked")
+        Observer<Object> o = mock(Observer.class);
+        InOrder inOrder = inOrder(o);
+
+        source.delay(delayFunc).subscribe(o);
+        
+        source.onNext(1);
+        source.onCompleted();
+        
+        inOrder.verify(o).onNext(1);
+        inOrder.verify(o).onCompleted();
+        inOrder.verifyNoMoreInteractions();
+        verify(o, never()).onError(any(Throwable.class));
+    }
+    
+    @Test
+    public void testDelayWithObservableSubscriptionRunCompletion() {
+        PublishSubject<Integer> source = PublishSubject.create();
+        final PublishSubject<Integer> sdelay = PublishSubject.create();
+        final PublishSubject<Integer> delay = PublishSubject.create();
+        Func0<Observable<Integer>> subFunc = new Func0<Observable<Integer>>() {
+            @Override
+            public Observable<Integer> call() {
+                return sdelay;
+            }
+        };
+        Func1<Integer, Observable<Integer>> delayFunc = new Func1<Integer, Observable<Integer>>() {
+
+            @Override
+            public Observable<Integer> call(Integer t1) {
+                return delay;
+            }
+        };
+        
+        @SuppressWarnings("unchecked")
+        Observer<Object> o = mock(Observer.class);
+        InOrder inOrder = inOrder(o);
+
+        source.delay(subFunc, delayFunc).subscribe(o);
+        
+        source.onNext(1);
+        sdelay.onCompleted();
+
+        source.onNext(2);
+        delay.onNext(2);
+        
+        inOrder.verify(o).onNext(2);
+        inOrder.verifyNoMoreInteractions();
+        verify(o, never()).onError(any(Throwable.class));
+        verify(o, never()).onCompleted();
+    }
 }