Merge pull request #3039 from akarnokd/WindowFuncFix

Window with Observable: fixed unsubscription and behavior

Author: benjchristensen

src/main/java/rx/Observable.java

@@ -9059,7 +9059,7 @@ public final <U, R> Observable<R> withLatestFrom(Observable<? extends U> other,
      * @see <a href="http://reactivex.io/documentation/operators/window.html">ReactiveX operators documentation: Window</a>
      */
     public final <TClosing> Observable<Observable<T>> window(Func0<? extends Observable<? extends TClosing>> closingSelector) {
-        return lift(new OperatorWindowWithObservable<T, TClosing>(closingSelector));
+        return lift(new OperatorWindowWithObservableFactory<T, TClosing>(closingSelector));
     }
 
     /**

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

@@ -15,16 +15,13 @@
  */
 package rx.internal.operators;
 
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-import rx.Observable;
+import java.util.*;
+
+import rx.*;
 import rx.Observable.Operator;
+import rx.Observable;
 import rx.Observer;
-import rx.Subscriber;
-import rx.functions.Func0;
 import rx.observers.SerializedSubscriber;
-import rx.observers.Subscribers;
 
 /**
  * Creates non-overlapping windows of items where each window is terminated by
@@ -34,36 +31,21 @@
  * @param <U> the boundary value type
  */
 public final class OperatorWindowWithObservable<T, U> implements Operator<Observable<T>, T> {
-    final Func0<? extends Observable<? extends U>> otherFactory;
+    final Observable<U> other;
 
-    public OperatorWindowWithObservable(Func0<? extends Observable<? extends U>> otherFactory) {
-        this.otherFactory = otherFactory;
-    }
     public OperatorWindowWithObservable(final Observable<U> other) {
-        this.otherFactory = new Func0<Observable<U>>() {
-
-            @Override
-            public Observable<U> call() {
-                return other;
-            }
-            
-        };
+        this.other = other;
     }
 
     @Override
     public Subscriber<? super T> call(Subscriber<? super Observable<T>> child) {
 
-        Observable<? extends U> other;
-        try {
-            other = otherFactory.call();
-        } catch (Throwable e) {
-            child.onError(e);
-            return Subscribers.empty();
-        }
-        
         SourceSubscriber<T> sub = new SourceSubscriber<T>(child);
         BoundarySubscriber<T, U> bs = new BoundarySubscriber<T, U>(child, sub);
 
+        child.add(sub);
+        child.add(bs);
+        
         sub.replaceWindow();
 
         other.unsafeSubscribe(bs);
@@ -88,7 +70,6 @@ static final class SourceSubscriber<T> extends Subscriber<T> {
         List<Object> queue;
 
         public SourceSubscriber(Subscriber<? super Observable<T>> child) {
-            super(child);
             this.child = new SerializedSubscriber<Observable<T>>(child);
             this.guard = new Object();
         }
@@ -288,7 +269,6 @@ void error(Throwable e) {
     static final class BoundarySubscriber<T, U> extends Subscriber<U> {
         final SourceSubscriber<T> sub;
         public BoundarySubscriber(Subscriber<?> child, SourceSubscriber<T> sub) {
-            super(child);
             this.sub = sub;
         }