ReactiveX
diff --git a/‎src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindow.java
Copy file name to clipboard
+5-5Lines changed: 5 additions & 5 deletions b/‎src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindow.java
Copy file name to clipboard
+5-5Lines changed: 5 additions & 5 deletions
diff --git a/‎src/main/java/io/reactivex/rxjava3/internal/operators/flowable/WindowSubscribeIntercept.java renamed to ‎src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindowSubscribeIntercept.java
Copy file name to clipboard
+2-2Lines changed: 2 additions & 2 deletions b/‎src/main/java/io/reactivex/rxjava3/internal/operators/flowable/WindowSubscribeIntercept.java renamed to ‎src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindowSubscribeIntercept.java
Copy file name to clipboard
+2-2Lines changed: 2 additions & 2 deletions
diff --git a/‎src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindowTimed.java
Copy file name to clipboard
+12-15Lines changed: 12 additions & 15 deletions b/‎src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindowTimed.java
Copy file name to clipboard
+12-15Lines changed: 12 additions & 15 deletions
diff --git a/‎src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableWindow.java
Copy file name to clipboard
+18-2Lines changed: 18 additions & 2 deletions b/‎src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableWindow.java
Copy file name to clipboard
+18-2Lines changed: 18 additions & 2 deletions
diff --git a/‎src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableWindowSubscribeIntercept.java
Copy file name to clipboard
+46Lines changed: 46 additions & 0 deletions b/‎src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableWindowSubscribeIntercept.java
Copy file name to clipboard
+46Lines changed: 46 additions & 0 deletions
@@ -92,14 +92,14 @@ public void onNext(T t) {
             long i = index;
 
             UnicastProcessor<T> w = window;
-            WindowSubscribeIntercept<T> intercept = null;
+            FlowableWindowSubscribeIntercept<T> intercept = null;
             if (i == 0) {
                 getAndIncrement();
 
                 w = UnicastProcessor.<T>create(bufferSize, this);
                 window = w;
 
-                intercept = new WindowSubscribeIntercept<T>(w);
+                intercept = new FlowableWindowSubscribeIntercept<T>(w);
                 downstream.onNext(intercept);
             }
 
@@ -211,15 +211,15 @@ public void onSubscribe(Subscription s) {
         public void onNext(T t) {
             long i = index;
 
-            WindowSubscribeIntercept<T> intercept = null;
+            FlowableWindowSubscribeIntercept<T> intercept = null;
             UnicastProcessor<T> w = window;
             if (i == 0) {
                 getAndIncrement();
 
                 w = UnicastProcessor.<T>create(bufferSize, this);
                 window = w;
 
-                intercept = new WindowSubscribeIntercept<T>(w);
+                intercept = new FlowableWindowSubscribeIntercept<T>(w);
                 downstream.onNext(intercept);
             }
 
@@ -477,7 +477,7 @@ void drain() {
                             break;
                         }
 
-                        WindowSubscribeIntercept<T> intercept = new WindowSubscribeIntercept<T>(t);
+                        FlowableWindowSubscribeIntercept<T> intercept = new FlowableWindowSubscribeIntercept<T>(t);
                         a.onNext(intercept);
 
                         if (intercept.tryAbandon()) {
 
@@ -25,13 +25,13 @@
  * @param <T> the element type of the flow.
  * @since 3.0.0
  */
-final class WindowSubscribeIntercept<T> extends Flowable<T> {
+final class FlowableWindowSubscribeIntercept<T> extends Flowable<T> {
 
     final FlowableProcessor<T> window;
 
     final AtomicBoolean once;
 
-    WindowSubscribeIntercept(FlowableProcessor<T> source) {
+    FlowableWindowSubscribeIntercept(FlowableProcessor<T> source) {
         this.window = source;
         this.once = new AtomicBoolean();
     }
 
@@ -28,7 +28,6 @@
 import io.reactivex.rxjava3.internal.subscriptions.SubscriptionHelper;
 import io.reactivex.rxjava3.internal.util.BackpressureHelper;
 import io.reactivex.rxjava3.processors.UnicastProcessor;
-import io.reactivex.rxjava3.subscribers.SerializedSubscriber;
 
 public final class FlowableWindowTimed<T> extends AbstractFlowableWithUpstream<T, Flowable<T>> {
     final long timespan;
@@ -53,23 +52,21 @@ public FlowableWindowTimed(Flowable<T> source,
     }
 
     @Override
-    protected void subscribeActual(Subscriber<? super Flowable<T>> s) {
-        SerializedSubscriber<Flowable<T>> actual = new SerializedSubscriber<Flowable<T>>(s);
-
+    protected void subscribeActual(Subscriber<? super Flowable<T>> downstream) {
         if (timespan == timeskip) {
             if (maxSize == Long.MAX_VALUE) {
                 source.subscribe(new WindowExactUnboundedSubscriber<T>(
-                        actual,
+                        downstream,
                         timespan, unit, scheduler, bufferSize));
                 return;
             }
             source.subscribe(new WindowExactBoundedSubscriber<T>(
-                        actual,
+                        downstream,
                         timespan, unit, scheduler,
                         bufferSize, maxSize, restartTimerOnMaxSize));
             return;
         }
-        source.subscribe(new WindowSkipSubscriber<T>(actual,
+        source.subscribe(new WindowSkipSubscriber<T>(downstream,
                 timespan, timeskip, unit, scheduler.createWorker(), bufferSize));
     }
 
@@ -100,8 +97,8 @@ abstract static class AbstractWindowSubscriber<T>
 
         final AtomicInteger windowCount;
 
-        AbstractWindowSubscriber(Subscriber<? super Flowable<T>> actual, long timespan, TimeUnit unit, int bufferSize) {
-            this.downstream = actual;
+        AbstractWindowSubscriber(Subscriber<? super Flowable<T>> downstream, long timespan, TimeUnit unit, int bufferSize) {
+            this.downstream = downstream;
             this.queue = new MpscLinkedQueue<Object>();
             this.timespan = timespan;
             this.unit = unit;
@@ -204,7 +201,7 @@ void createFirstWindow() {
 
                     emitted = 1;
 
-                    WindowSubscribeIntercept<T> intercept = new WindowSubscribeIntercept<T>(window);
+                    FlowableWindowSubscribeIntercept<T> intercept = new FlowableWindowSubscribeIntercept<T>(window);
                     downstream.onNext(intercept);
 
                     timer.replace(scheduler.schedulePeriodicallyDirect(this, timespan, timespan, unit));
@@ -293,7 +290,7 @@ else if (!isEmpty) {
                                     window = UnicastProcessor.create(bufferSize, windowRunnable);
                                     this.window = window;
 
-                                    WindowSubscribeIntercept<T> intercept = new WindowSubscribeIntercept<T>(window);
+                                    FlowableWindowSubscribeIntercept<T> intercept = new FlowableWindowSubscribeIntercept<T>(window);
                                     downstream.onNext(intercept);
 
                                     if (intercept.tryAbandon()) {
@@ -372,7 +369,7 @@ void createFirstWindow() {
                     windowCount.getAndIncrement();
                     window = UnicastProcessor.create(bufferSize, this);
 
-                    WindowSubscribeIntercept<T> intercept = new WindowSubscribeIntercept<T>(window);
+                    FlowableWindowSubscribeIntercept<T> intercept = new FlowableWindowSubscribeIntercept<T>(window);
                     downstream.onNext(intercept);
 
                     Runnable boundaryTask = new WindowBoundaryRunnable(this, 1L);
@@ -510,7 +507,7 @@ UnicastProcessor<T> createNewWindow(UnicastProcessor<T> window) {
                     window = UnicastProcessor.create(bufferSize, this);
                     this.window = window;
 
-                    WindowSubscribeIntercept<T> intercept = new WindowSubscribeIntercept<T>(window);
+                    FlowableWindowSubscribeIntercept<T> intercept = new FlowableWindowSubscribeIntercept<T>(window);
                     downstream.onNext(intercept);
 
                     if (restartTimerOnMaxSize) {
@@ -573,7 +570,7 @@ void createFirstWindow() {
                     UnicastProcessor<T> window = UnicastProcessor.create(bufferSize, this);
                     windows.add(window);
 
-                    WindowSubscribeIntercept<T> intercept = new WindowSubscribeIntercept<T>(window);
+                    FlowableWindowSubscribeIntercept<T> intercept = new FlowableWindowSubscribeIntercept<T>(window);
                     downstream.onNext(intercept);
 
                     worker.schedule(new WindowBoundaryRunnable(this, false), timespan, unit);
@@ -647,7 +644,7 @@ void drain() {
                                     UnicastProcessor<T> window = UnicastProcessor.create(bufferSize, this);
                                     windows.add(window);
 
-                                    WindowSubscribeIntercept<T> intercept = new WindowSubscribeIntercept<T>(window);
+                                    FlowableWindowSubscribeIntercept<T> intercept = new FlowableWindowSubscribeIntercept<T>(window);
                                     downstream.onNext(intercept);
 
                                     worker.schedule(new WindowBoundaryRunnable(this, false), timespan, unit);
 
@@ -77,14 +77,17 @@ public void onSubscribe(Disposable d) {
         @Override
         public void onNext(T t) {
             UnicastSubject<T> w = window;
+            ObservableWindowSubscribeIntercept<T> intercept = null;
             if (w == null && !cancelled) {
                 w = UnicastSubject.create(capacityHint, this);
                 window = w;
-                downstream.onNext(w);
+                intercept = new ObservableWindowSubscribeIntercept<T>(w);
+                downstream.onNext(intercept);
             }
 
             if (w != null) {
                 w.onNext(t);
+
                 if (++size >= count) {
                     size = 0;
                     window = null;
@@ -93,6 +96,12 @@ public void onNext(T t) {
                         upstream.dispose();
                     }
                 }
+
+                if (intercept != null && intercept.tryAbandon()) {
+                    w.onComplete();
+                    w = null;
+                    window = null;
+                }
             }
         }
 
@@ -180,11 +189,14 @@ public void onNext(T t) {
 
             long s = skip;
 
+            ObservableWindowSubscribeIntercept<T> intercept = null;
+
             if (i % s == 0 && !cancelled) {
                 wip.getAndIncrement();
                 UnicastSubject<T> w = UnicastSubject.create(capacityHint, this);
+                intercept = new ObservableWindowSubscribeIntercept<T>(w);
                 ws.offer(w);
-                downstream.onNext(w);
+                downstream.onNext(intercept);
             }
 
             long c = firstEmission + 1;
@@ -205,6 +217,10 @@ public void onNext(T t) {
             }
 
             index = i + 1;
+
+            if (intercept != null && intercept.tryAbandon()) {
+                intercept.window.onComplete();
+            }
         }
 
         @Override
 
@@ -0,0 +1,46 @@
+/**
+ * Copyright (c) 2016-present, RxJava Contributors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
+ */
+
+package io.reactivex.rxjava3.internal.operators.observable;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import io.reactivex.rxjava3.core.*;
+import io.reactivex.rxjava3.subjects.Subject;
+
+/**
+ * Wrapper for a Subject that detects an incoming subscriber.
+ * @param <T> the element type of the flow.
+ * @since 3.0.0
+ */
+final class ObservableWindowSubscribeIntercept<T> extends Observable<T> {
+
+    final Subject<T> window;
+
+    final AtomicBoolean once;
+
+    ObservableWindowSubscribeIntercept(Subject<T> source) {
+        this.window = source;
+        this.once = new AtomicBoolean();
+    }
+
+    @Override
+    protected void subscribeActual(Observer<? super T> s) {
+        window.subscribe(s);
+        once.set(true);
+    }
+
+    boolean tryAbandon() {
+        return !once.get() && once.compareAndSet(false, true);
+    }
+}