Skip to content

Commit

Permalink
Merge pull request #1912 from akarnokd/RetryWithBackpressureFix
Browse files Browse the repository at this point in the history 
Fixed retry without backpressure & test function to support bp.
  • Loading branch information
@benjchristensen
benjchristensen committed Dec 1, 2014
2 parents 43456c5 + 158d9d9 commit f46d7f9
Show file tree
Hide file tree
Showing 2 changed files with 69 additions and 56 deletions.
11 changes: 8 additions & 3 deletions src/main/java/rx/internal/operators/OnSubscribeRedo.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -225,14 +225,19 @@ public void onError(Throwable e) {

@Override
public void onNext(T v) {
consumerCapacity.decrementAndGet();
if (consumerCapacity.get() != Long.MAX_VALUE) {
consumerCapacity.decrementAndGet();
}
child.onNext(v);
}

@Override
public void setProducer(Producer producer) {
currentProducer.set(producer);
producer.request(consumerCapacity.get());
long c = consumerCapacity.get();
if (c > 0) {
producer.request(c);
}
}
};
// new subscription each time so if it unsubscribes itself it does not prevent retries
Expand Down Expand Up @@ -321,7 +326,7 @@ public void request(final long n) {
long c = consumerCapacity.getAndAdd(n);
Producer producer = currentProducer.get();
if (producer != null) {
producer.request(c + n);
producer.request(n);
} else
if (c == 0 && resumeBoundary.compareAndSet(true, false)) {
worker.schedule(subscribeToSource);
Expand Down
114 changes: 61 additions & 53 deletions src/test/java/rx/internal/operators/OperatorRetryTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,40 +15,26 @@
*/
package rx.internal.operators;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;
import static org.junit.Assert.*;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.*;

import java.util.concurrent.*;
import java.util.concurrent.atomic.*;

import org.junit.Test;
import org.mockito.InOrder;
import org.mockito.Mockito;
import org.mockito.*;

import rx.Observable;
import rx.*;
import rx.Observable.OnSubscribe;
import rx.Observer;
import rx.Subscriber;
import rx.Subscription;
import rx.functions.Action0;
import rx.functions.Action1;
import rx.functions.Func1;
import rx.functions.Func2;
import rx.functions.*;
import rx.internal.util.RxRingBuffer;
import rx.observables.GroupedObservable;
import rx.observers.TestSubscriber;
import rx.schedulers.Schedulers;
import rx.subjects.PublishSubject;
import rx.subscriptions.Subscriptions;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

public class OperatorRetryTest {

@Test
Expand Down Expand Up @@ -403,18 +389,38 @@ public static class FuncWithErrors implements Observable.OnSubscribe<String> {
}

@Override
public void call(Subscriber<? super String> o) {
o.onNext("beginningEveryTime");
if (count.getAndIncrement() < numFailures) {
System.out.println("FuncWithErrors @ " + count.get());
o.onError(new RuntimeException("forced failure: " + count.get()));
} else {
System.out.println("FuncWithErrors @ onSuccessOnly");
o.onNext("onSuccessOnly");
System.out.println("FuncWithErrors @ onCompleted");
o.onCompleted();
System.out.println("FuncWithErrors !");
}
public void call(final Subscriber<? super String> o) {
o.setProducer(new Producer() {
final AtomicLong req = new AtomicLong();
@Override
public void request(long n) {
if (n == Long.MAX_VALUE) {
o.onNext("beginningEveryTime");
if (count.getAndIncrement() < numFailures) {
o.onError(new RuntimeException("forced failure: " + count.get()));
} else {
o.onNext("onSuccessOnly");
o.onCompleted();
}
return;
}
if (n > 0 && req.getAndAdd(1) == 0) {
int i = count.getAndIncrement();
if (i < numFailures) {
o.onNext("beginningEveryTime");
o.onError(new RuntimeException("forced failure: " + count.get()));
} else
if (i == numFailures) {
o.onNext("beginningEveryTime");
} else
if (i > numFailures) {
o.onNext("onSuccessOnly");
o.onCompleted();
}
req.decrementAndGet();
}
}
});
}
}

Expand Down Expand Up @@ -668,26 +674,28 @@ public void testTimeoutWithRetry() {
assertEquals("Start 6 threads, retry 5 then fail on 6", 6, so.efforts.get());
}

@Test(timeout = 3000)
@Test(timeout = 10000)
public void testRetryWithBackpressure() {
@SuppressWarnings("unchecked")
Observer<String> observer = mock(Observer.class);
int NUM_RETRIES = RxRingBuffer.SIZE * 2;
Observable<String> origin = Observable.create(new FuncWithErrors(NUM_RETRIES));
TestSubscriber<String> ts = new TestSubscriber<String>(observer);
origin.retry().observeOn(Schedulers.computation()).unsafeSubscribe(ts);
ts.awaitTerminalEvent();

InOrder inOrder = inOrder(observer);
// should show 3 attempts
inOrder.verify(observer, times(NUM_RETRIES + 1)).onNext("beginningEveryTime");
// should have no errors
inOrder.verify(observer, never()).onError(any(Throwable.class));
// should have a single success
inOrder.verify(observer, times(1)).onNext("onSuccessOnly");
// should have a single successful onCompleted
inOrder.verify(observer, times(1)).onCompleted();
inOrder.verifyNoMoreInteractions();
for (int i = 0; i < 200; i++) {
@SuppressWarnings("unchecked")
Observer<String> observer = mock(Observer.class);
int NUM_RETRIES = RxRingBuffer.SIZE * 2;
Observable<String> origin = Observable.create(new FuncWithErrors(NUM_RETRIES));
TestSubscriber<String> ts = new TestSubscriber<String>(observer);
origin.retry().observeOn(Schedulers.computation()).unsafeSubscribe(ts);
ts.awaitTerminalEvent();

InOrder inOrder = inOrder(observer);
// should have no errors
verify(observer, never()).onError(any(Throwable.class));
// should show NUM_RETRIES attempts
inOrder.verify(observer, times(NUM_RETRIES + 1)).onNext("beginningEveryTime");
// should have a single success
inOrder.verify(observer, times(1)).onNext("onSuccessOnly");
// should have a single successful onCompleted
inOrder.verify(observer, times(1)).onCompleted();
inOrder.verifyNoMoreInteractions();
}
}
@Test(timeout = 3000)
public void testIssue1900() throws InterruptedException {
Expand Down

0 comments on commit f46d7f9

Please sign in to comment.