Skip to content

Fixed window(time) to work properly with unsubscription, added #2972

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Jun 17, 2015
Merged

Fixed window(time) to work properly with unsubscription, added #2972

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion src/main/java/rx/Observable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9017,7 +9017,7 @@ public final Observable<Observable<T>> window(int count) {
* <img width="640" height="365" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/window4.png" alt="">
* <dl>
* <dt><b>Backpressure Support:</b></dt>
* <dd>The operator has limited backpressure support. If {@code count} == {@code skip}, the operator honors backpressure on its outer subscriber, ignores backpressure in its inner Observables
* <dd>The operator honors backpressure on its outer subscriber, ignores backpressure in its inner Observables
* but each of them will emit at most {@code count} elements.</dd>
* <dt><b>Scheduler:</b></dt>
* <dd>This version of {@code window} does not operate by default on a particular {@link Scheduler}.</dd>
Expand Down
60 changes: 43 additions & 17 deletions src/main/java/rx/internal/operators/OperatorWindowWithSize.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,17 +48,20 @@ public OperatorWindowWithSize(int size, int skip) {
@Override
public Subscriber<? super T> call(Subscriber<? super Observable<T>> child) {
if (skip == size) {
return new ExactSubscriber(child);
ExactSubscriber e = new ExactSubscriber(child);
e.init();
return e;
}
return new InexactSubscriber(child);
InexactSubscriber ie = new InexactSubscriber(child);
ie.init();
return ie;
}
/** Subscriber with exact, non-overlapping window bounds. */
final class ExactSubscriber extends Subscriber<T> {
final Subscriber<? super Observable<T>> child;
int count;
BufferUntilSubscriber<T> window;
volatile boolean noWindow = true;
final Subscription parentSubscription = this;
public ExactSubscriber(Subscriber<? super Observable<T>> child) {
/**
* See https://github.com/ReactiveX/RxJava/issues/1546
Expand All @@ -69,13 +72,15 @@ public ExactSubscriber(Subscriber<? super Observable<T>> child) {
/*
* Add unsubscribe hook to child to get unsubscribe on outer (unsubscribing on next window, not on the inner window itself)
*/
}
void init() {
child.add(Subscriptions.create(new Action0() {

@Override
public void call() {
// if no window we unsubscribe up otherwise wait until window ends
if (noWindow) {
parentSubscription.unsubscribe();
unsubscribe();
}
}

Expand Down Expand Up @@ -111,7 +116,7 @@ public void onNext(T t) {
window = null;
noWindow = true;
if (child.isUnsubscribed()) {
parentSubscription.unsubscribe();
unsubscribe();
return;
}
}
Expand Down Expand Up @@ -139,7 +144,7 @@ final class InexactSubscriber extends Subscriber<T> {
final Subscriber<? super Observable<T>> child;
int count;
final List<CountedSubject<T>> chunks = new LinkedList<CountedSubject<T>>();
final Subscription parentSubscription = this;
volatile boolean noWindow = true;

public InexactSubscriber(Subscriber<? super Observable<T>> child) {
/**
Expand All @@ -148,6 +153,9 @@ public InexactSubscriber(Subscriber<? super Observable<T>> child) {
* applies to the outer, not the inner.
*/
this.child = child;
}

void init() {
/*
* Add unsubscribe hook to child to get unsubscribe on outer (unsubscribing on next window, not on the inner window itself)
*/
Expand All @@ -156,24 +164,38 @@ public InexactSubscriber(Subscriber<? super Observable<T>> child) {
@Override
public void call() {
// if no window we unsubscribe up otherwise wait until window ends
if (chunks == null || chunks.size() == 0) {
parentSubscription.unsubscribe();
if (noWindow) {
unsubscribe();
}
}

}));
}

@Override
public void onStart() {
// no backpressure as we are controlling data flow by window size
request(Long.MAX_VALUE);

child.setProducer(new Producer() {
@Override
public void request(long n) {
if (n > 0) {
long u = n * size;
if (((u >>> 31) != 0) && (u / n != size)) {
u = Long.MAX_VALUE;
}
requestMore(u);
}
}
});
}

void requestMore(long n) {
request(n);
}

@Override
public void onNext(T t) {
if (count++ % skip == 0) {
if (!child.isUnsubscribed()) {
if (chunks.isEmpty()) {
noWindow = false;
}
CountedSubject<T> cs = createCountedSubject();
chunks.add(cs);
child.onNext(cs.producer);
Expand All @@ -189,16 +211,19 @@ public void onNext(T t) {
cs.consumer.onCompleted();
}
}
if (chunks.size() == 0 && child.isUnsubscribed()) {
parentSubscription.unsubscribe();
return;
if (chunks.isEmpty()) {
noWindow = true;
if (child.isUnsubscribed()) {
unsubscribe();
}
}
}

@Override
public void onError(Throwable e) {
List<CountedSubject<T>> list = new ArrayList<CountedSubject<T>>(chunks);
chunks.clear();
noWindow = true;
for (CountedSubject<T> cs : list) {
cs.consumer.onError(e);
}
Expand All @@ -209,6 +234,7 @@ public void onError(Throwable e) {
public void onCompleted() {
List<CountedSubject<T>> list = new ArrayList<CountedSubject<T>>(chunks);
chunks.clear();
noWindow = true;
for (CountedSubject<T> cs : list) {
cs.consumer.onCompleted();
}
Expand Down
106 changes: 62 additions & 44 deletions src/main/java/rx/internal/operators/OperatorWindowWithTime.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,21 +15,17 @@
*/
package rx.internal.operators;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.*;
import java.util.concurrent.TimeUnit;
import rx.Observable;

import rx.*;
import rx.Observable.Operator;
import rx.Observer;
import rx.Scheduler;
import rx.Scheduler.Worker;
import rx.Subscriber;
import rx.Observable;
import rx.Observer;
import rx.functions.Action0;
import rx.observers.SerializedObserver;
import rx.observers.SerializedSubscriber;
import rx.observers.*;
import rx.subscriptions.Subscriptions;

/**
* Creates windows of values into the source sequence with timed window creation, length and size bounds.
Expand Down Expand Up @@ -62,15 +58,16 @@ public OperatorWindowWithTime(long timespan, long timeshift, TimeUnit unit, int
@Override
public Subscriber<? super T> call(Subscriber<? super Observable<T>> child) {
Worker worker = scheduler.createWorker();
child.add(worker);

if (timespan == timeshift) {
ExactSubscriber s = new ExactSubscriber(child, worker);
s.add(worker);
s.scheduleExact();
return s;
}

InexactSubscriber s = new InexactSubscriber(child, worker);
s.add(worker);
s.startNewChunk();
s.scheduleChunk();
return s;
Expand Down Expand Up @@ -118,11 +115,19 @@ final class ExactSubscriber extends Subscriber<T> {
volatile State<T> state;

public ExactSubscriber(Subscriber<? super Observable<T>> child, Worker worker) {
super(child);
this.child = new SerializedSubscriber<Observable<T>>(child);
this.worker = worker;
this.guard = new Object();
this.state = State.empty();
child.add(Subscriptions.create(new Action0() {
@Override
public void call() {
// if there is no active window, unsubscribe the upstream
if (state.consumer == null) {
unsubscribe();
}
}
}));
}

@Override
Expand All @@ -132,7 +137,6 @@ public void onStart() {

@Override
public void onNext(T t) {
List<Object> localQueue;
synchronized (guard) {
if (emitting) {
if (queue == null) {
Expand All @@ -141,29 +145,29 @@ public void onNext(T t) {
queue.add(t);
return;
}
localQueue = queue;
queue = null;
emitting = true;
}
boolean once = true;
boolean skipFinal = false;
try {
do {
drain(localQueue);
if (once) {
once = false;
emitValue(t);
}
if (!emitValue(t)) {
return;
}

for (;;) {
List<Object> localQueue;
synchronized (guard) {
localQueue = queue;
queue = null;
if (localQueue == null) {
emitting = false;
skipFinal = true;
return;
}
queue = null;
}
if (!drain(localQueue)) {
return;
}
} while (!child.isUnsubscribed());
}
} finally {
if (!skipFinal) {
synchronized (guard) {
Expand All @@ -172,13 +176,15 @@ public void onNext(T t) {
}
}
}
void drain(List<Object> queue) {
boolean drain(List<Object> queue) {
if (queue == null) {
return;
return true;
}
for (Object o : queue) {
if (o == NEXT_SUBJECT) {
replaceSubject();
if (!replaceSubject()) {
return false;
}
} else
if (nl.isError(o)) {
error(nl.getError(o));
Expand All @@ -190,23 +196,35 @@ void drain(List<Object> queue) {
} else {
@SuppressWarnings("unchecked")
T t = (T)o;
emitValue(t);
if (!emitValue(t)) {
return false;
}
}
}
return true;
}
void replaceSubject() {
boolean replaceSubject() {
Observer<T> s = state.consumer;
if (s != null) {
s.onCompleted();
}
// if child has unsubscribed, unsubscribe upstream instead of opening a new window
if (child.isUnsubscribed()) {
state = state.clear();
unsubscribe();
return false;
}
BufferUntilSubscriber<T> bus = BufferUntilSubscriber.create();
state = state.create(bus, bus);
child.onNext(bus);
return true;
}
void emitValue(T t) {
boolean emitValue(T t) {
State<T> s = state;
if (s.consumer == null) {
replaceSubject();
if (!replaceSubject()) {
return false;
}
s = state;
}
s.consumer.onNext(t);
Expand All @@ -217,6 +235,7 @@ void emitValue(T t) {
s = s.next();
}
state = s;
return true;
}

@Override
Expand Down Expand Up @@ -285,7 +304,6 @@ public void call() {
}, 0, timespan, unit);
}
void nextWindow() {
List<Object> localQueue;
synchronized (guard) {
if (emitting) {
if (queue == null) {
Expand All @@ -294,29 +312,29 @@ void nextWindow() {
queue.add(NEXT_SUBJECT);
return;
}
localQueue = queue;
queue = null;
emitting = true;
}
boolean once = true;
boolean skipFinal = false;
try {
do {
drain(localQueue);
if (once) {
once = false;
replaceSubject();
}
if (!replaceSubject()) {
return;
}
for (;;) {
List<Object> localQueue;
synchronized (guard) {
localQueue = queue;
queue = null;
if (localQueue == null) {
emitting = false;
skipFinal = true;
return;
}
queue = null;
}
} while (!child.isUnsubscribed());

if (!drain(localQueue)) {
return;
}
}
} finally {
if (!skipFinal) {
synchronized (guard) {
Expand Down
Loading