Skip to content

Commit

Permalink
FlowableWithLatestFrom forgets request (#5494)
Browse files Browse the repository at this point in the history
* FlowableWithLatestFrom forgets request

* Add unit test and correct testBackpressure

* Revert indentation change

* Revert import change
  • Loading branch information
matgabriel authored and akarnokd committed Jul 18, 2017
1 parent cd62d62 commit 340bed9
Show file tree
Hide file tree
Showing 2 changed files with 37 additions and 2 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,8 @@ public void onNext(T t) {
return;
}
actual.onNext(r);
} else{
request(1);
}
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -261,7 +261,7 @@ public void testNoDownstreamUnsubscribe() {

@Test
public void testBackpressure() {
Flowable<Integer> source = Flowable.range(1, 10);
PublishProcessor<Integer> source = PublishProcessor.create();
PublishProcessor<Integer> other = PublishProcessor.create();

Flowable<Integer> result = source.withLatestFrom(other, COMBINER);
Expand All @@ -274,17 +274,24 @@ public void testBackpressure() {

ts.request(1);

source.onNext(1);

assertTrue("Other has no observers!", other.hasSubscribers());

ts.assertNoValues();

other.onNext(1);

ts.request(1);
source.onNext(2);

ts.assertValue((2 << 8) + 1);

ts.request(5);
source.onNext(3);
source.onNext(4);
source.onNext(5);
source.onNext(6);
source.onNext(7);
ts.assertValues(
(2 << 8) + 1, (3 << 8) + 1, (4 << 8) + 1, (5 << 8) + 1,
(6 << 8) + 1, (7 << 8) + 1
Expand Down Expand Up @@ -717,4 +724,30 @@ public void zeroOtherCombinerReturnsNull() {
.test()
.assertFailureAndMessage(NullPointerException.class, "The combiner returned a null value");
}

@Test
public void testSingleRequestNotForgottenWhenNoData() {
PublishProcessor<Integer> source = PublishProcessor.create();
PublishProcessor<Integer> other = PublishProcessor.create();

Flowable<Integer> result = source.withLatestFrom(other, COMBINER);

TestSubscriber<Integer> ts = new TestSubscriber<Integer>(0L);

result.subscribe(ts);

ts.request(1);

source.onNext(1);

ts.assertNoValues();

other.onNext(1);

ts.assertNoValues();

source.onNext(2);

ts.assertValue((2 << 8) + 1);
}
}

0 comments on commit 340bed9

Please sign in to comment.