Merge pull request #1833 from benjchristensen/window-tweaks

Fix Thread Safety for Unsubscribe of Window

Author: benjchristensen

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

@@ -61,7 +61,8 @@ final class ExactSubscriber extends Subscriber<T> {
         final Subscriber<? super Observable<T>> child;
         int count;
         BufferUntilSubscriber<T> window;
-        Subscription parentSubscription = this;
+        volatile boolean noWindow = true;
+        final Subscription parentSubscription = this;
         public ExactSubscriber(Subscriber<? super Observable<T>> child) {
             /**
              * See https://github.com/ReactiveX/RxJava/issues/1546
@@ -77,7 +78,7 @@ public ExactSubscriber(Subscriber<? super Observable<T>> child) {
                 @Override
                 public void call() {
                     // if no window we unsubscribe up otherwise wait until window ends
-                    if(window == null) {
+                    if(noWindow) {
                         parentSubscription.unsubscribe();
                     }
                 }
@@ -94,13 +95,15 @@ public void onStart() {
         @Override
         public void onNext(T t) {
             if (window == null) {
+                noWindow = false;
                 window = BufferUntilSubscriber.create();
                 child.onNext(window);                
             }
             window.onNext(t);
             if (++count % size == 0) {
                 window.onCompleted();
                 window = null;
+                noWindow = true;
                 if (child.isUnsubscribed()) {
                     parentSubscription.unsubscribe();
                     return;
@@ -130,7 +133,7 @@ final class InexactSubscriber extends Subscriber<T> {
         final Subscriber<? super Observable<T>> child;
         int count;
         final List<CountedSubject<T>> chunks = new LinkedList<CountedSubject<T>>();
-        Subscription parentSubscription = this;
+        final Subscription parentSubscription = this;
 
         public InexactSubscriber(Subscriber<? super Observable<T>> child) {
             /**