3.x: Fix Observable.window (size & time) cancellation and abandonment

src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindow.java

@@ -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()) {

src/main/java/io/reactivex/rxjava3/internal/operators/flowable/WindowSubscribeIntercept.java → src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindowSubscribeIntercept.java

@@ -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();
  }

src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindowTimed.java

@@ -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);

src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableWindow.java

@@ -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

src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableWindowSubscribeIntercept.java

@@ -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
+ *
+ * http://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);
+ }
+}