Merge pull request #3150 from akarnokd/BufferUntilSubscriberBackpressureV2

Window operators now support backpressure in the inner observable.

Author: akarnokd

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

@@ -154,7 +154,7 @@ void replaceSubject() {
             child.onNext(producer);
         }
         void createNewWindow() {
-            BufferUntilSubscriber<T> bus = BufferUntilSubscriber.create();
+            UnicastSubject<T> bus = UnicastSubject.create();
             consumer = bus;
             producer = bus;
         }

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

@@ -160,7 +160,7 @@ void replaceSubject() {
             child.onNext(producer);
         }
         void createNewWindow() {
-            BufferUntilSubscriber<T> bus = BufferUntilSubscriber.create();
+            UnicastSubject<T> bus = UnicastSubject.create();
             consumer = bus;
             producer = bus;
             Observable<? extends U> other;

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

@@ -60,7 +60,7 @@ public Subscriber<? super T> call(Subscriber<? super Observable<T>> child) {
     final class ExactSubscriber extends Subscriber<T> {
         final Subscriber<? super Observable<T>> child;
         int count;
-        BufferUntilSubscriber<T> window;
+        UnicastSubject<T> window;
         volatile boolean noWindow = true;
         public ExactSubscriber(Subscriber<? super Observable<T>> child) {
             /**
@@ -107,7 +107,7 @@ void requestMore(long n) {
         public void onNext(T t) {
             if (window == null) {
                 noWindow = false;
-                window = BufferUntilSubscriber.create();
+                window = UnicastSubject.create();
                 child.onNext(window);
             }
             window.onNext(t);
@@ -241,7 +241,7 @@ public void onCompleted() {
         }
 
         CountedSubject<T> createCountedSubject() {
-            final BufferUntilSubscriber<T> bus = BufferUntilSubscriber.create();
+            final UnicastSubject<T> bus = UnicastSubject.create();
             return new CountedSubject<T>(bus, bus);
         }
     }

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

@@ -233,7 +233,7 @@ void endWindow(SerializedSubject<T> window) {
             }
         }
         SerializedSubject<T> createSerializedSubject() {
-            BufferUntilSubscriber<T> bus = BufferUntilSubscriber.create();
+            UnicastSubject<T> bus = UnicastSubject.create();
             return new SerializedSubject<T>(bus, bus);
         }
     }

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

@@ -214,7 +214,7 @@ boolean replaceSubject() {
                 unsubscribe();
                 return false;
             }
-            BufferUntilSubscriber<T> bus = BufferUntilSubscriber.create();
+            UnicastSubject<T> bus = UnicastSubject.create();
             state = state.create(bus, bus);
             child.onNext(bus);
             return true;
@@ -492,7 +492,7 @@ void terminateChunk(CountedSerializedSubject<T> chunk) {
             }
         }
         CountedSerializedSubject<T> createCountedSerializedSubject() {
-            BufferUntilSubscriber<T> bus = BufferUntilSubscriber.create();
+            UnicastSubject<T> bus = UnicastSubject.create();
             return new CountedSerializedSubject<T>(bus, bus);
         }
     }