Skip to content

Commit 39e159f

Browse files
akarnokdakarnokd
authored
2.x: fix timed exact buffer calling cancel unnecessarily (#5761)
1 parent 9f2beac commit 39e159f

File tree

5 files changed

+132
-5
lines changed

5 files changed

+132
-5
lines changed

‎src/main/java/io/reactivex/internal/operators/flowable/FlowableBufferTimed.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -166,7 +166,7 @@ public void onComplete() {
166166
queue.offer(b);
167167
done = true;
168168
if (enter()) {
169-
QueueDrainHelper.drainMaxLoop(queue, actual, false, this, this);
169+
QueueDrainHelper.drainMaxLoop(queue, actual, false, null, this);
170170
}
171171
}
172172

‎src/main/java/io/reactivex/internal/operators/observable/ObservableBufferTimed.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -161,7 +161,7 @@ public void onComplete() {
161161
queue.offer(b);
162162
done = true;
163163
if (enter()) {
164-
QueueDrainHelper.drainLoop(queue, actual, false, this, this);
164+
QueueDrainHelper.drainLoop(queue, actual, false, null, this);
165165
}
166166
}
167167
DisposableHelper.dispose(timer);

‎src/main/java/io/reactivex/internal/util/QueueDrainHelper.java

Lines changed: 9 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -168,7 +168,9 @@ public static <T, U> boolean checkTerminated(boolean d, boolean empty,
168168
if (d) {
169169
if (delayError) {
170170
if (empty) {
171-
disposable.dispose();
171+
if (disposable != null) {
172+
disposable.dispose();
173+
}
172174
Throwable err = qd.error();
173175
if (err != null) {
174176
s.onError(err);
@@ -181,12 +183,16 @@ public static <T, U> boolean checkTerminated(boolean d, boolean empty,
181183
Throwable err = qd.error();
182184
if (err != null) {
183185
q.clear();
184-
disposable.dispose();
186+
if (disposable != null) {
187+
disposable.dispose();
188+
}
185189
s.onError(err);
186190
return true;
187191
} else
188192
if (empty) {
189-
disposable.dispose();
193+
if (disposable != null) {
194+
disposable.dispose();
195+
}
190196
s.onComplete();
191197
return true;
192198
}

‎src/test/java/io/reactivex/internal/operators/flowable/FlowableBufferTest.java

Lines changed: 60 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2014,4 +2014,64 @@ public void run() {
20142014
assertEquals("Round: " + i, 5, items);
20152015
}
20162016
}
2017+
2018+
@SuppressWarnings("unchecked")
2019+
@Test
2020+
public void noCompletionCancelExact() {
2021+
final AtomicInteger counter = new AtomicInteger();
2022+
2023+
Flowable.<Integer>empty()
2024+
.doOnCancel(new Action() {
2025+
@Override
2026+
public void run() throws Exception {
2027+
counter.getAndIncrement();
2028+
}
2029+
})
2030+
.buffer(5, TimeUnit.SECONDS)
2031+
.test()
2032+
.awaitDone(5, TimeUnit.SECONDS)
2033+
.assertResult(Collections.<Integer>emptyList());
2034+
2035+
assertEquals(0, counter.get());
2036+
}
2037+
2038+
@SuppressWarnings("unchecked")
2039+
@Test
2040+
public void noCompletionCancelSkip() {
2041+
final AtomicInteger counter = new AtomicInteger();
2042+
2043+
Flowable.<Integer>empty()
2044+
.doOnCancel(new Action() {
2045+
@Override
2046+
public void run() throws Exception {
2047+
counter.getAndIncrement();
2048+
}
2049+
})
2050+
.buffer(5, 10, TimeUnit.SECONDS)
2051+
.test()
2052+
.awaitDone(5, TimeUnit.SECONDS)
2053+
.assertResult(Collections.<Integer>emptyList());
2054+
2055+
assertEquals(0, counter.get());
2056+
}
2057+
2058+
@SuppressWarnings("unchecked")
2059+
@Test
2060+
public void noCompletionCancelOverlap() {
2061+
final AtomicInteger counter = new AtomicInteger();
2062+
2063+
Flowable.<Integer>empty()
2064+
.doOnCancel(new Action() {
2065+
@Override
2066+
public void run() throws Exception {
2067+
counter.getAndIncrement();
2068+
}
2069+
})
2070+
.buffer(10, 5, TimeUnit.SECONDS)
2071+
.test()
2072+
.awaitDone(5, TimeUnit.SECONDS)
2073+
.assertResult(Collections.<Integer>emptyList());
2074+
2075+
assertEquals(0, counter.get());
2076+
}
20172077
}

‎src/test/java/io/reactivex/internal/operators/observable/ObservableBufferTest.java

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919

2020
import java.util.*;
2121
import java.util.concurrent.*;
22+
import java.util.concurrent.atomic.AtomicInteger;
2223

2324
import org.junit.*;
2425
import org.mockito.*;
@@ -1439,4 +1440,64 @@ public void run() {
14391440
assertEquals("Round: " + i, 5, items);
14401441
}
14411442
}
1443+
1444+
@SuppressWarnings("unchecked")
1445+
@Test
1446+
public void noCompletionCancelExact() {
1447+
final AtomicInteger counter = new AtomicInteger();
1448+
1449+
Observable.<Integer>empty()
1450+
.doOnDispose(new Action() {
1451+
@Override
1452+
public void run() throws Exception {
1453+
counter.getAndIncrement();
1454+
}
1455+
})
1456+
.buffer(5, TimeUnit.SECONDS)
1457+
.test()
1458+
.awaitDone(5, TimeUnit.SECONDS)
1459+
.assertResult(Collections.<Integer>emptyList());
1460+
1461+
assertEquals(0, counter.get());
1462+
}
1463+
1464+
@SuppressWarnings("unchecked")
1465+
@Test
1466+
public void noCompletionCancelSkip() {
1467+
final AtomicInteger counter = new AtomicInteger();
1468+
1469+
Observable.<Integer>empty()
1470+
.doOnDispose(new Action() {
1471+
@Override
1472+
public void run() throws Exception {
1473+
counter.getAndIncrement();
1474+
}
1475+
})
1476+
.buffer(5, 10, TimeUnit.SECONDS)
1477+
.test()
1478+
.awaitDone(5, TimeUnit.SECONDS)
1479+
.assertResult(Collections.<Integer>emptyList());
1480+
1481+
assertEquals(0, counter.get());
1482+
}
1483+
1484+
@SuppressWarnings("unchecked")
1485+
@Test
1486+
public void noCompletionCancelOverlap() {
1487+
final AtomicInteger counter = new AtomicInteger();
1488+
1489+
Observable.<Integer>empty()
1490+
.doOnDispose(new Action() {
1491+
@Override
1492+
public void run() throws Exception {
1493+
counter.getAndIncrement();
1494+
}
1495+
})
1496+
.buffer(10, 5, TimeUnit.SECONDS)
1497+
.test()
1498+
.awaitDone(5, TimeUnit.SECONDS)
1499+
.assertResult(Collections.<Integer>emptyList());
1500+
1501+
assertEquals(0, counter.get());
1502+
}
14421503
}

0 commit comments

Comments
 (0)