Skip to content
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

2.x: Improve coverage, fix operator logic 03/12 #5910

Merged
merged 1 commit into from
Mar 13, 2018
Merged

2.x: Improve coverage, fix operator logic 03/12 #5910

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/io/reactivex/exceptions/CompositeException.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -278,7 +278,7 @@ public int size() {
* @param e the {@link Throwable} {@code e}.
* @return The root cause of {@code e}. If {@code e.getCause()} returns {@code null} or {@code e}, just return {@code e} itself.
*/
private Throwable getRootCause(Throwable e) {
/*private */Throwable getRootCause(Throwable e) {
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That cause == root case is borderline impossible, unless this method gets called with this.

Throwable root = e.getCause();
if (root == null || cause == root) {
return e;
Expand Down
2 changes: 1 addition & 1 deletion src/main/java/io/reactivex/internal/functions/ObjectHelper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,7 +71,7 @@ public static int compare(int v1, int v2) {
}

/**
* Compares two integer values similar to Long.compare.
* Compares two long values similar to Long.compare.
* @param v1 the first value
* @param v2 the second value
* @return the comparison result
Expand Down
26 changes: 11 additions & 15 deletions src/main/java/io/reactivex/internal/operators/flowable/FlowableBufferBoundarySupplier.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -196,24 +196,20 @@ void next() {

BufferBoundarySubscriber<T, U, B> bs = new BufferBoundarySubscriber<T, U, B>(this);

Disposable o = other.get();

if (!other.compareAndSet(o, bs)) {
return;
}

U b;
synchronized (this) {
b = buffer;
if (b == null) {
return;
if (DisposableHelper.replace(other, bs)) {
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Use dedicated helper instead of a CAS, plus the old code could unset a terminal indicator.

U b;
synchronized (this) {
b = buffer;
if (b == null) {
return;
}
buffer = next;
}
buffer = next;
}

boundary.subscribe(bs);
boundary.subscribe(bs);

fastPathEmitMax(b, false, this);
fastPathEmitMax(b, false, this);
}
}

@Override
Expand Down
15 changes: 6 additions & 9 deletions src/main/java/io/reactivex/internal/operators/flowable/FlowableBufferTimed.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -177,8 +177,8 @@ public void request(long n) {

@Override
public void cancel() {
cancelled = true;
s.cancel();

DisposableHelper.dispose(timer);
}

Expand All @@ -199,14 +199,10 @@ public void run() {

synchronized (this) {
current = buffer;
if (current != null) {
buffer = next;
if (current == null) {
return;
}
}

if (current == null) {
DisposableHelper.dispose(timer);
return;
buffer = next;
}

fastPathEmitMax(current, false, this);
Expand Down Expand Up @@ -324,9 +320,10 @@ public void request(long n) {

@Override
public void cancel() {
clear();
cancelled = true;
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

cancelled is checked at various places but was never actually set.

s.cancel();
w.dispose();
clear();
}

void clear() {
Expand Down
37 changes: 20 additions & 17 deletions src/main/java/io/reactivex/internal/operators/flowable/FlowableDebounceTimed.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,7 +58,7 @@ static final class DebounceTimedSubscriber<T> extends AtomicLong

Subscription s;

final SequentialDisposable timer = new SequentialDisposable();
Disposable timer;
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Timer is only manipulated from onNext, onError or onComplete and doesn't have to be atomically safe. The operator disposes the entire worker to stop any outstanding timer activity anyway.


volatile long index;

Expand Down Expand Up @@ -88,17 +88,15 @@ public void onNext(T t) {
long idx = index + 1;
index = idx;

Disposable d = timer.get();
Disposable d = timer;
if (d != null) {
d.dispose();
}

DebounceEmitter<T> de = new DebounceEmitter<T>(t, idx, this);
if (timer.replace(de)) {
d = worker.schedule(de, timeout, unit);

de.setResource(d);
}
timer = de;
d = worker.schedule(de, timeout, unit);
de.setResource(d);
}

@Override
Expand All @@ -108,6 +106,10 @@ public void onError(Throwable t) {
return;
}
done = true;
Disposable d = timer;
if (d != null) {
d.dispose();
}
actual.onError(t);
worker.dispose();
}
Expand All @@ -119,17 +121,18 @@ public void onComplete() {
}
done = true;

Disposable d = timer.get();
if (!DisposableHelper.isDisposed(d)) {
@SuppressWarnings("unchecked")
DebounceEmitter<T> de = (DebounceEmitter<T>)d;
if (de != null) {
de.emit();
}
DisposableHelper.dispose(timer);
actual.onComplete();
worker.dispose();
Disposable d = timer;
if (d != null) {
d.dispose();
}
@SuppressWarnings("unchecked")
DebounceEmitter<T> de = (DebounceEmitter<T>)d;
if (de != null) {
de.emit();
}

actual.onComplete();
worker.dispose();
}

@Override
Expand Down
10 changes: 6 additions & 4 deletions src/main/java/io/reactivex/internal/operators/observable/ObservableBuffer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,11 +127,13 @@ public void onError(Throwable t) {
@Override
public void onComplete() {
U b = buffer;
buffer = null;
if (b != null && !b.isEmpty()) {
actual.onNext(b);
if (b != null) {
buffer = null;
if (!b.isEmpty()) {
actual.onNext(b);
}
actual.onComplete();
}
actual.onComplete();
}
}

Expand Down
26 changes: 11 additions & 15 deletions ...ain/java/io/reactivex/internal/operators/observable/ObservableBufferBoundarySupplier.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -190,24 +190,20 @@ void next() {

BufferBoundaryObserver<T, U, B> bs = new BufferBoundaryObserver<T, U, B>(this);

Disposable o = other.get();

if (!other.compareAndSet(o, bs)) {
return;
}

U b;
synchronized (this) {
b = buffer;
if (b == null) {
return;
if (DisposableHelper.replace(other, bs)) {
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Use the helper method instead of a CAS that could replace a terminal indicator.

U b;
synchronized (this) {
b = buffer;
if (b == null) {
return;
}
buffer = next;
}
buffer = next;
}

boundary.subscribe(bs);
boundary.subscribe(bs);

fastPathEmit(b, false, this);
fastPathEmit(b, false, this);
}
}

@Override
Expand Down
36 changes: 19 additions & 17 deletions src/main/java/io/reactivex/internal/operators/observable/ObservableDebounceTimed.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,7 @@ static final class DebounceTimedObserver<T>

Disposable s;

final AtomicReference<Disposable> timer = new AtomicReference<Disposable>();
Disposable timer;
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

timer is only manipulated from onNext, onError and onComplete.


volatile long index;

Expand Down Expand Up @@ -80,18 +80,15 @@ public void onNext(T t) {
long idx = index + 1;
index = idx;

Disposable d = timer.get();
Disposable d = timer;
if (d != null) {
d.dispose();
}

DebounceEmitter<T> de = new DebounceEmitter<T>(t, idx, this);
if (timer.compareAndSet(d, de)) {
d = worker.schedule(de, timeout, unit);

de.setResource(d);
}

timer = de;
d = worker.schedule(de, timeout, unit);
de.setResource(d);
}

@Override
Expand All @@ -100,6 +97,10 @@ public void onError(Throwable t) {
RxJavaPlugins.onError(t);
return;
}
Disposable d = timer;
if (d != null) {
d.dispose();
}
done = true;
actual.onError(t);
worker.dispose();
Expand All @@ -112,16 +113,17 @@ public void onComplete() {
}
done = true;

Disposable d = timer.get();
if (d != DisposableHelper.DISPOSED) {
@SuppressWarnings("unchecked")
DebounceEmitter<T> de = (DebounceEmitter<T>)d;
if (de != null) {
de.run();
}
actual.onComplete();
worker.dispose();
Disposable d = timer;
if (d != null) {
d.dispose();
}
@SuppressWarnings("unchecked")
DebounceEmitter<T> de = (DebounceEmitter<T>)d;
if (de != null) {
de.run();
}
actual.onComplete();
worker.dispose();
}

@Override
Expand Down
7 changes: 2 additions & 5 deletions src/main/java/io/reactivex/internal/operators/single/SingleTakeUntil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,12 +85,9 @@ public void onSubscribe(Disposable d) {
public void onSuccess(T value) {
other.dispose();

Disposable a = get();
Disposable a = getAndSet(DisposableHelper.DISPOSED);
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should be called at most once anyway, so biasing towards the a != DisposableHelper.DISPOSED being true.

if (a != DisposableHelper.DISPOSED) {
a = getAndSet(DisposableHelper.DISPOSED);
if (a != DisposableHelper.DISPOSED) {
actual.onSuccess(value);
}
actual.onSuccess(value);
}
}

Expand Down
3 changes: 1 addition & 2 deletions src/main/java/io/reactivex/internal/util/MergerBiFunction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,8 +58,7 @@ public List<T> apply(List<T> a, List<T> b) throws Exception {
while (at.hasNext()) {
both.add(at.next());
}
} else
if (s2 != null) {
} else {
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

At this point either s1 or s2 can be null.

both.add(s2);
while (bt.hasNext()) {
both.add(bt.next());
Expand Down
16 changes: 16 additions & 0 deletions src/test/java/io/reactivex/exceptions/CompositeExceptionTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -349,6 +349,22 @@ public void badException() {
assertSame(e, new CompositeException(e).getCause().getCause());
assertSame(e, new CompositeException(new RuntimeException(e)).getCause().getCause().getCause());
}

@Test
public void rootCauseEval() {
final TestException ex0 = new TestException();
Throwable throwable = new Throwable() {

private static final long serialVersionUID = 3597694032723032281L;

@Override
public synchronized Throwable getCause() {
return ex0;
}
};
CompositeException ex = new CompositeException(throwable);
assertSame(ex, ex.getRootCause(ex));
}
}

final class BadException extends Throwable {
Expand Down
7 changes: 7 additions & 0 deletions src/test/java/io/reactivex/internal/functions/ObjectHelperTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,4 +58,11 @@ public void compare() {
assertEquals(0, ObjectHelper.compare(0, 0));
assertEquals(1, ObjectHelper.compare(2, 0));
}

@Test
public void compareLong() {
assertEquals(-1, ObjectHelper.compare(0L, 2L));
assertEquals(0, ObjectHelper.compare(0L, 0L));
assertEquals(1, ObjectHelper.compare(2L, 0L));
}
}
Loading