Skip to content

Commit 9b037c7

Browse files
akarnokdakarnokd
authored
2.x: fix Single.timeout unnecessary dispose calls (#5586)
* 2.x: fix Single.timeout unnesessary dispose calls * Route to RxJavaPlugins.onError
1 parent c43229b commit 9b037c7

File tree

2 files changed

+223
-60
lines changed

2 files changed

+223
-60
lines changed

‎src/main/java/io/reactivex/internal/operators/single/SingleTimeout.java‎

Lines changed: 81 additions & 58 deletions
Original file line numberDiff line numberDiff line change
@@ -14,10 +14,12 @@
1414
package io.reactivex.internal.operators.single;
1515

1616
import java.util.concurrent.*;
17-
import java.util.concurrent.atomic.AtomicBoolean;
17+
import java.util.concurrent.atomic.AtomicReference;
1818

1919
import io.reactivex.*;
20-
import io.reactivex.disposables.*;
20+
import io.reactivex.disposables.Disposable;
21+
import io.reactivex.internal.disposables.DisposableHelper;
22+
import io.reactivex.plugins.RxJavaPlugins;
2123

2224
public final class SingleTimeout<T> extends Single<T> {
2325

@@ -43,97 +45,118 @@ public SingleTimeout(SingleSource<T> source, long timeout, TimeUnit unit, Schedu
4345
@Override
4446
protected void subscribeActual(final SingleObserver<? super T> s) {
4547

46-
final CompositeDisposable set = new CompositeDisposable();
47-
s.onSubscribe(set);
48+
TimeoutMainObserver<T> parent = new TimeoutMainObserver<T>(s, other);
49+
s.onSubscribe(parent);
4850

49-
final AtomicBoolean once = new AtomicBoolean();
51+
DisposableHelper.replace(parent.task, scheduler.scheduleDirect(parent, timeout, unit));
5052

51-
Disposable timer = scheduler.scheduleDirect(new TimeoutDispose(once, set, s), timeout, unit);
53+
source.subscribe(parent);
54+
}
5255

53-
set.add(timer);
56+
static final class TimeoutMainObserver<T> extends AtomicReference<Disposable>
57+
implements SingleObserver<T>, Runnable, Disposable {
5458

55-
source.subscribe(new TimeoutObserver(once, set, s));
59+
private static final long serialVersionUID = 37497744973048446L;
5660

57-
}
61+
final SingleObserver<? super T> actual;
5862

59-
final class TimeoutDispose implements Runnable {
60-
private final AtomicBoolean once;
61-
final CompositeDisposable set;
62-
final SingleObserver<? super T> s;
63+
final AtomicReference<Disposable> task;
6364

64-
TimeoutDispose(AtomicBoolean once, CompositeDisposable set, SingleObserver<? super T> s) {
65-
this.once = once;
66-
this.set = set;
67-
this.s = s;
68-
}
65+
final TimeoutFallbackObserver<T> fallback;
6966

70-
@Override
71-
public void run() {
72-
if (once.compareAndSet(false, true)) {
73-
if (other != null) {
74-
set.clear();
75-
other.subscribe(new TimeoutObserver());
76-
} else {
77-
set.dispose();
78-
s.onError(new TimeoutException());
79-
}
80-
}
81-
}
67+
SingleSource<? extends T> other;
8268

83-
final class TimeoutObserver implements SingleObserver<T> {
69+
static final class TimeoutFallbackObserver<T> extends AtomicReference<Disposable>
70+
implements SingleObserver<T> {
8471

85-
@Override
86-
public void onError(Throwable e) {
87-
set.dispose();
88-
s.onError(e);
72+
private static final long serialVersionUID = 2071387740092105509L;
73+
final SingleObserver<? super T> actual;
74+
75+
TimeoutFallbackObserver(SingleObserver<? super T> actual) {
76+
this.actual = actual;
8977
}
9078

9179
@Override
9280
public void onSubscribe(Disposable d) {
93-
set.add(d);
81+
DisposableHelper.setOnce(this, d);
9482
}
9583

9684
@Override
97-
public void onSuccess(T value) {
98-
set.dispose();
99-
s.onSuccess(value);
85+
public void onSuccess(T t) {
86+
actual.onSuccess(t);
10087
}
10188

89+
@Override
90+
public void onError(Throwable e) {
91+
actual.onError(e);
92+
}
10293
}
103-
}
10494

105-
final class TimeoutObserver implements SingleObserver<T> {
95+
TimeoutMainObserver(SingleObserver<? super T> actual, SingleSource<? extends T> other) {
96+
this.actual = actual;
97+
this.other = other;
98+
this.task = new AtomicReference<Disposable>();
99+
if (other != null) {
100+
this.fallback = new TimeoutFallbackObserver<T>(actual);
101+
} else {
102+
this.fallback = null;
103+
}
104+
}
106105

107-
private final AtomicBoolean once;
108-
private final CompositeDisposable set;
109-
private final SingleObserver<? super T> s;
106+
@Override
107+
public void run() {
108+
Disposable d = get();
109+
if (d != DisposableHelper.DISPOSED && compareAndSet(d, DisposableHelper.DISPOSED)) {
110+
if (d != null) {
111+
d.dispose();
112+
}
113+
SingleSource<? extends T> other = this.other;
114+
if (other == null) {
115+
actual.onError(new TimeoutException());
116+
} else {
117+
this.other = null;
118+
other.subscribe(fallback);
119+
}
120+
}
121+
}
110122

111-
TimeoutObserver(AtomicBoolean once, CompositeDisposable set, SingleObserver<? super T> s) {
112-
this.once = once;
113-
this.set = set;
114-
this.s = s;
123+
@Override
124+
public void onSubscribe(Disposable d) {
125+
DisposableHelper.setOnce(this, d);
115126
}
116127

117128
@Override
118-
public void onError(Throwable e) {
119-
if (once.compareAndSet(false, true)) {
120-
set.dispose();
121-
s.onError(e);
129+
public void onSuccess(T t) {
130+
Disposable d = get();
131+
if (d != DisposableHelper.DISPOSED && compareAndSet(d, DisposableHelper.DISPOSED)) {
132+
DisposableHelper.dispose(task);
133+
actual.onSuccess(t);
122134
}
123135
}
124136

125137
@Override
126-
public void onSubscribe(Disposable d) {
127-
set.add(d);
138+
public void onError(Throwable e) {
139+
Disposable d = get();
140+
if (d != DisposableHelper.DISPOSED && compareAndSet(d, DisposableHelper.DISPOSED)) {
141+
DisposableHelper.dispose(task);
142+
actual.onError(e);
143+
} else {
144+
RxJavaPlugins.onError(e);
145+
}
128146
}
129147

130148
@Override
131-
public void onSuccess(T value) {
132-
if (once.compareAndSet(false, true)) {
133-
set.dispose();
134-
s.onSuccess(value);
149+
public void dispose() {
150+
DisposableHelper.dispose(this);
151+
DisposableHelper.dispose(task);
152+
if (fallback != null) {
153+
DisposableHelper.dispose(fallback);
135154
}
136155
}
137156

157+
@Override
158+
public boolean isDisposed() {
159+
return DisposableHelper.isDisposed(get());
160+
}
138161
}
139162
}

‎src/test/java/io/reactivex/internal/operators/single/SingleTimeoutTest.java‎

Lines changed: 142 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -15,15 +15,18 @@
1515

1616
import static org.junit.Assert.*;
1717

18+
import java.util.List;
1819
import java.util.concurrent.TimeUnit;
1920

2021
import org.junit.Test;
2122

22-
import io.reactivex.Single;
23+
import io.reactivex.*;
2324
import io.reactivex.exceptions.TestException;
25+
import io.reactivex.functions.Action;
2426
import io.reactivex.observers.TestObserver;
27+
import io.reactivex.plugins.RxJavaPlugins;
2528
import io.reactivex.schedulers.TestScheduler;
26-
import io.reactivex.subjects.PublishSubject;
29+
import io.reactivex.subjects.*;
2730

2831
public class SingleTimeoutTest {
2932

@@ -69,4 +72,141 @@ public void mainError() {
6972
.awaitDone(5, TimeUnit.SECONDS)
7073
.assertFailure(TestException.class);
7174
}
75+
76+
@Test
77+
public void disposeWhenFallback() {
78+
TestScheduler sch = new TestScheduler();
79+
80+
SingleSubject<Integer> subj = SingleSubject.create();
81+
82+
subj.timeout(1, TimeUnit.SECONDS, sch, Single.just(1))
83+
.test(true)
84+
.assertEmpty();
85+
86+
assertFalse(subj.hasObservers());
87+
}
88+
89+
@Test
90+
public void isDisposed() {
91+
TestHelper.checkDisposed(SingleSubject.create().timeout(1, TimeUnit.DAYS));
92+
}
93+
94+
@Test
95+
public void fallbackDispose() {
96+
TestScheduler sch = new TestScheduler();
97+
98+
SingleSubject<Integer> subj = SingleSubject.create();
99+
100+
SingleSubject<Integer> fallback = SingleSubject.create();
101+
102+
TestObserver<Integer> to = subj.timeout(1, TimeUnit.SECONDS, sch, fallback)
103+
.test();
104+
105+
assertFalse(fallback.hasObservers());
106+
107+
sch.advanceTimeBy(1, TimeUnit.SECONDS);
108+
109+
assertFalse(subj.hasObservers());
110+
assertTrue(fallback.hasObservers());
111+
112+
to.cancel();
113+
114+
assertFalse(fallback.hasObservers());
115+
}
116+
117+
@Test
118+
public void normalSuccessDoesntDisposeMain() {
119+
final int[] calls = { 0 };
120+
121+
Single.just(1)
122+
.doOnDispose(new Action() {
123+
@Override
124+
public void run() throws Exception {
125+
calls[0]++;
126+
}
127+
})
128+
.timeout(1, TimeUnit.DAYS)
129+
.test()
130+
.assertResult(1);
131+
132+
assertEquals(0, calls[0]);
133+
}
134+
135+
@Test
136+
public void successTimeoutRace() {
137+
for (int i = 0; i < 1000; i++) {
138+
final SingleSubject<Integer> subj = SingleSubject.create();
139+
SingleSubject<Integer> fallback = SingleSubject.create();
140+
141+
final TestScheduler sch = new TestScheduler();
142+
143+
TestObserver<Integer> to = subj.timeout(1, TimeUnit.MILLISECONDS, sch, fallback).test();
144+
145+
Runnable r1 = new Runnable() {
146+
@Override
147+
public void run() {
148+
subj.onSuccess(1);
149+
}
150+
};
151+
152+
Runnable r2 = new Runnable() {
153+
@Override
154+
public void run() {
155+
sch.advanceTimeBy(1, TimeUnit.MILLISECONDS);
156+
}
157+
};
158+
159+
TestHelper.race(r1, r2);
160+
161+
if (!fallback.hasObservers()) {
162+
to.assertResult(1);
163+
} else {
164+
to.assertEmpty();
165+
}
166+
}
167+
}
168+
169+
@Test
170+
public void errorTimeoutRace() {
171+
final TestException ex = new TestException();
172+
List<Throwable> errors = TestHelper.trackPluginErrors();
173+
try {
174+
175+
for (int i = 0; i < 1000; i++) {
176+
final SingleSubject<Integer> subj = SingleSubject.create();
177+
SingleSubject<Integer> fallback = SingleSubject.create();
178+
179+
final TestScheduler sch = new TestScheduler();
180+
181+
TestObserver<Integer> to = subj.timeout(1, TimeUnit.MILLISECONDS, sch, fallback).test();
182+
183+
Runnable r1 = new Runnable() {
184+
@Override
185+
public void run() {
186+
subj.onError(ex);
187+
}
188+
};
189+
190+
Runnable r2 = new Runnable() {
191+
@Override
192+
public void run() {
193+
sch.advanceTimeBy(1, TimeUnit.MILLISECONDS);
194+
}
195+
};
196+
197+
TestHelper.race(r1, r2);
198+
199+
if (!fallback.hasObservers()) {
200+
to.assertFailure(TestException.class);
201+
} else {
202+
to.assertEmpty();
203+
}
204+
if (!errors.isEmpty()) {
205+
TestHelper.assertUndeliverable(errors, 0, TestException.class);
206+
}
207+
}
208+
} finally {
209+
RxJavaPlugins.reset();
210+
}
211+
}
72212
}

0 commit comments

Comments
 (0)