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

Fix issue #1522 #1523

Merged
merged 3 commits into from
Jul 29, 2014
Merged

Fix issue #1522 #1523

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
212af0a
Fix issue #1522
zsxwing Jul 26, 2014
a2a7a29
Optimize the fix for takeLast and handle the race condition
zsxwing Jul 29, 2014
c1ec1f4
Make "single" support backpressure
zsxwing Jul 29, 2014
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
3 changes: 3 additions & 0 deletions rxjava-core/src/main/java/rx/internal/operators/OperatorSingle.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,9 @@ public void onNext(T value) {
} else {
this.value = value;
isNonEmpty = true;
// Issue: https://github.com/Netflix/RxJava/pull/1527
// Because we cache a value and don't emit now, we need to request another one.
Copy link
Member

Choose a reason for hiding this comment

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

Good catch.

request(1);
}
}

Expand Down
47 changes: 32 additions & 15 deletions rxjava-core/src/main/java/rx/internal/operators/OperatorTakeLast.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -108,9 +108,12 @@ void startEmitting() {

@Override
public void request(long n) {
long _c = 0;
if (requested == Long.MAX_VALUE) {
return;
}
long _c;
if (n == Long.MAX_VALUE) {
requested = Long.MAX_VALUE;
_c = REQUESTED_UPDATER.getAndSet(this, Long.MAX_VALUE);
} else {
_c = REQUESTED_UPDATER.getAndAdd(this, n);
}
Expand All @@ -122,16 +125,20 @@ public void request(long n) {
}

void emit(long previousRequested) {
if (requested < 0) {
if (requested == Long.MAX_VALUE) {
Copy link
Member

Choose a reason for hiding this comment

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

Ha! That's a fun bug, totally missed that when we went from -1 to MAX_VALUE.

I bet it isn't even worth having the complexity of a fast-path because the cost of queue/deque is far worse than the counter, and we only hit the counter at the very end anyways.

// fast-path without backpressure
try {
for (Object value : deque) {
notification.accept(subscriber, value);
if (previousRequested == 0) {
try {
for (Object value : deque) {
notification.accept(subscriber, value);
}
} catch (Throwable e) {
subscriber.onError(e);
} finally {
deque.clear();
}
} catch (Throwable e) {
subscriber.onError(e);
} finally {
deque.clear();
} else {
// backpressure path will handle Long.MAX_VALUE and emit the rest events.
}
} else {
// backpressure is requested
Expand All @@ -155,12 +162,22 @@ void emit(long previousRequested) {
emitted++;
}
}

if (REQUESTED_UPDATER.addAndGet(this, -emitted) == 0) {
// we're done emitting the number requested so return
return;
for (;;) {
long oldRequested = requested;
long newRequested = oldRequested - emitted;
if (oldRequested == Long.MAX_VALUE) {
// became unbounded during the loop
// continue the outer loop to emit the rest events.
break;
}
if (REQUESTED_UPDATER.compareAndSet(this, oldRequested, newRequested)) {
if (newRequested == 0) {
// we're done emitting the number requested so return
return;
}
break;
}
}

}
}
}
Expand Down
55 changes: 52 additions & 3 deletions rxjava-core/src/test/java/rx/internal/operators/OperatorSingleTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,10 +15,9 @@
*/
package rx.internal.operators;

import static org.junit.Assert.assertEquals;
import static org.mockito.Matchers.isA;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.*;

import java.util.NoSuchElementException;

Expand All @@ -27,7 +26,9 @@

import rx.Observable;
import rx.Observer;
import rx.Subscriber;
import rx.functions.Func1;
import rx.functions.Func2;

public class OperatorSingleTest {

Expand Down Expand Up @@ -241,4 +242,52 @@ public Boolean call(Integer t1) {
inOrder.verify(observer, times(1)).onCompleted();
inOrder.verifyNoMoreInteractions();
}

@Test
public void testSingleWithBackpressure() {
Observable<Integer> observable = Observable.from(1, 2).single();

Subscriber<Integer> subscriber = spy(new Subscriber<Integer>() {

@Override
public void onStart() {
request(1);
}

@Override
public void onCompleted() {

}

@Override
public void onError(Throwable e) {

}

@Override
public void onNext(Integer integer) {
request(1);
}
});
observable.subscribe(subscriber);

InOrder inOrder = inOrder(subscriber);
inOrder.verify(subscriber, times(1)).onError(isA(IllegalArgumentException.class));
inOrder.verifyNoMoreInteractions();
}

@Test(timeout = 30000)
public void testIssue1527() throws InterruptedException {
//https://github.com/Netflix/RxJava/pull/1527
Observable<Integer> source = Observable.from(1, 2, 3, 4, 5, 6);
Observable<Integer> reduced = source.reduce(new Func2<Integer, Integer, Integer>() {
@Override
public Integer call(Integer i1, Integer i2) {
return i1 + i2;
}
});

Integer r = reduced.toBlocking().first();
assertEquals(21, r.intValue());
}
}
116 changes: 116 additions & 0 deletions rxjava-core/src/test/java/rx/internal/operators/OperatorTakeLastTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,9 @@

import rx.Observable;
import rx.Observer;
import rx.Subscriber;
import rx.functions.Func1;
import rx.functions.Functions;
import rx.internal.util.RxRingBuffer;
import rx.observers.TestSubscriber;
import rx.schedulers.Schedulers;
Expand Down Expand Up @@ -148,4 +150,118 @@ public Integer call(Integer i) {
};
}

@Test
public void testIssue1522() {
// https://github.com/Netflix/RxJava/issues/1522
assertEquals(0, Observable
.empty()
.count()
.filter(Functions.alwaysFalse())
.toList()
.toBlocking().single().size());
}

@Test
public void testIgnoreRequest1() {
// If `takeLast` does not ignore `request` properly, StackOverflowError will be thrown.
Observable.range(0, 100000).takeLast(100000).subscribe(new Subscriber<Integer>() {

@Override
public void onStart() {
request(Long.MAX_VALUE);
}

@Override
public void onCompleted() {

}

@Override
public void onError(Throwable e) {
}

@Override
public void onNext(Integer integer) {
request(Long.MAX_VALUE);
}
});
}

@Test
public void testIgnoreRequest2() {
// If `takeLast` does not ignore `request` properly, StackOverflowError will be thrown.
Observable.range(0, 100000).takeLast(100000).subscribe(new Subscriber<Integer>() {

@Override
public void onStart() {
request(1);
}

@Override
public void onCompleted() {
}

@Override
public void onError(Throwable e) {
}

@Override
public void onNext(Integer integer) {
request(1);
}
});
}

@Test(timeout = 30000)
public void testIgnoreRequest3() {
// If `takeLast` does not ignore `request` properly, it will enter an infinite loop.
Observable.range(0, 100000).takeLast(100000).subscribe(new Subscriber<Integer>() {

@Override
public void onStart() {
request(1);
}

@Override
public void onCompleted() {

}

@Override
public void onError(Throwable e) {
}

@Override
public void onNext(Integer integer) {
request(Long.MAX_VALUE);
}
});
}


@Test
public void testIgnoreRequest4() {
// If `takeLast` does not ignore `request` properly, StackOverflowError will be thrown.
Observable.range(0, 100000).takeLast(100000).subscribe(new Subscriber<Integer>() {

@Override
public void onStart() {
request(Long.MAX_VALUE);
}

@Override
public void onCompleted() {

}

@Override
public void onError(Throwable e) {
}

@Override
public void onNext(Integer integer) {
request(1);
}
});
}
}