Skip to content

Commit 53049b2

Browse files
ZacSweersZacSweers
committed
Fix calling dispose() in terminal event
Matches ReactiveX/RxJava#4957, which we were doing wrong. Note the lifecycle disposable is not lazily set because that's internally managed and non-terminating.
1 parent f329e12 commit 53049b2

File tree

5 files changed

+121
-66
lines changed

5 files changed

+121
-66
lines changed

‎autodispose/src/main/java/com/uber/autodispose/AutoDisposingCompletableObserverImpl.java

Lines changed: 24 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -76,25 +76,36 @@ final class AutoDisposingCompletableObserverImpl implements AutoDisposingComplet
7676
@Override public final void dispose() {
7777
synchronized (this) {
7878
AutoDisposableHelper.dispose(lifecycleDisposable);
79+
callMainSubscribeIfNecessary();
80+
AutoDisposableHelper.dispose(mainDisposable);
81+
}
82+
}
7983

80-
// If we've never actually called the downstream onSubscribe (i.e. requested immediately in
81-
// onSubscribe and had a terminal event), we need to still send an empty disposable instance
82-
// to abide by the Observer contract.
83-
if (mainDisposable.get() == null) {
84-
try {
85-
onSubscribe.accept(Disposables.disposed());
86-
} catch (Exception e) {
87-
Exceptions.throwIfFatal(e);
88-
RxJavaPlugins.onError(e);
89-
}
84+
private void lazyDispose() {
85+
synchronized (this) {
86+
AutoDisposableHelper.dispose(lifecycleDisposable);
87+
callMainSubscribeIfNecessary();
88+
mainDisposable.lazySet(AutoDisposableHelper.DISPOSED);
89+
}
90+
}
91+
92+
private void callMainSubscribeIfNecessary() {
93+
// If we've never actually called the downstream onSubscribe (i.e. requested immediately in
94+
// onSubscribe and had a terminal event), we need to still send an empty disposable instance
95+
// to abide by the Observer contract.
96+
if (mainDisposable.get() == null) {
97+
try {
98+
onSubscribe.accept(Disposables.disposed());
99+
} catch (Exception e) {
100+
Exceptions.throwIfFatal(e);
101+
RxJavaPlugins.onError(e);
90102
}
91-
AutoDisposableHelper.dispose(mainDisposable);
92103
}
93104
}
94105

95106
@Override public final void onComplete() {
96107
if (!isDisposed()) {
97-
dispose();
108+
lazyDispose();
98109
try {
99110
onComplete.run();
100111
} catch (Exception e) {
@@ -106,7 +117,7 @@ final class AutoDisposingCompletableObserverImpl implements AutoDisposingComplet
106117

107118
@Override public final void onError(Throwable e) {
108119
if (!isDisposed()) {
109-
dispose();
120+
lazyDispose();
110121
try {
111122
onError.accept(e);
112123
} catch (Exception e1) {

‎autodispose/src/main/java/com/uber/autodispose/AutoDisposingMaybeObserverImpl.java

Lines changed: 25 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -79,25 +79,36 @@ final class AutoDisposingMaybeObserverImpl<T> implements AutoDisposingMaybeObser
7979
@Override public final void dispose() {
8080
synchronized (this) {
8181
AutoDisposableHelper.dispose(lifecycleDisposable);
82+
callMainSubscribeIfNecessary();
83+
AutoDisposableHelper.dispose(mainDisposable);
84+
}
85+
}
8286

83-
// If we've never actually called the downstream onSubscribe (i.e. requested immediately in
84-
// onSubscribe and had a terminal event), we need to still send an empty disposable instance
85-
// to abide by the Observer contract.
86-
if (mainDisposable.get() == null) {
87-
try {
88-
onSubscribe.accept(Disposables.disposed());
89-
} catch (Exception e) {
90-
Exceptions.throwIfFatal(e);
91-
RxJavaPlugins.onError(e);
92-
}
87+
private void lazyDispose() {
88+
synchronized (this) {
89+
AutoDisposableHelper.dispose(lifecycleDisposable);
90+
callMainSubscribeIfNecessary();
91+
mainDisposable.lazySet(AutoDisposableHelper.DISPOSED);
92+
}
93+
}
94+
95+
private void callMainSubscribeIfNecessary() {
96+
// If we've never actually called the downstream onSubscribe (i.e. requested immediately in
97+
// onSubscribe and had a terminal event), we need to still send an empty disposable instance
98+
// to abide by the Observer contract.
99+
if (mainDisposable.get() == null) {
100+
try {
101+
onSubscribe.accept(Disposables.disposed());
102+
} catch (Exception e) {
103+
Exceptions.throwIfFatal(e);
104+
RxJavaPlugins.onError(e);
93105
}
94-
AutoDisposableHelper.dispose(mainDisposable);
95106
}
96107
}
97108

98109
@Override public final void onSuccess(T value) {
99110
if (!isDisposed()) {
100-
dispose();
111+
lazyDispose();
101112
try {
102113
onSuccess.accept(value);
103114
} catch (Exception e) {
@@ -109,7 +120,7 @@ final class AutoDisposingMaybeObserverImpl<T> implements AutoDisposingMaybeObser
109120

110121
@Override public final void onError(Throwable e) {
111122
if (!isDisposed()) {
112-
dispose();
123+
lazyDispose();
113124
try {
114125
onError.accept(e);
115126
} catch (Exception e1) {
@@ -121,7 +132,7 @@ final class AutoDisposingMaybeObserverImpl<T> implements AutoDisposingMaybeObser
121132

122133
@Override public final void onComplete() {
123134
if (!isDisposed()) {
124-
dispose();
135+
lazyDispose();
125136
try {
126137
onComplete.run();
127138
} catch (Exception e) {

‎autodispose/src/main/java/com/uber/autodispose/AutoDisposingObserverImpl.java

Lines changed: 24 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -79,19 +79,30 @@ final class AutoDisposingObserverImpl<T> implements AutoDisposingObserver<T> {
7979
@Override public final void dispose() {
8080
synchronized (this) {
8181
AutoDisposableHelper.dispose(lifecycleDisposable);
82+
callMainSubscribeIfNecessary();
83+
AutoDisposableHelper.dispose(mainDisposable);
84+
}
85+
}
8286

83-
// If we've never actually called the downstream onSubscribe (i.e. requested immediately in
84-
// onSubscribe and had a terminal event), we need to still send an empty disposable instance
85-
// to abide by the Observer contract.
86-
if (mainDisposable.get() == null) {
87-
try {
88-
onSubscribe.accept(Disposables.disposed());
89-
} catch (Exception e) {
90-
Exceptions.throwIfFatal(e);
91-
RxJavaPlugins.onError(e);
92-
}
87+
private void lazyDispose() {
88+
synchronized (this) {
89+
AutoDisposableHelper.dispose(lifecycleDisposable);
90+
callMainSubscribeIfNecessary();
91+
mainDisposable.lazySet(AutoDisposableHelper.DISPOSED);
92+
}
93+
}
94+
95+
private void callMainSubscribeIfNecessary() {
96+
// If we've never actually called the downstream onSubscribe (i.e. requested immediately in
97+
// onSubscribe and had a terminal event), we need to still send an empty disposable instance
98+
// to abide by the Observer contract.
99+
if (mainDisposable.get() == null) {
100+
try {
101+
onSubscribe.accept(Disposables.disposed());
102+
} catch (Exception e) {
103+
Exceptions.throwIfFatal(e);
104+
RxJavaPlugins.onError(e);
93105
}
94-
AutoDisposableHelper.dispose(mainDisposable);
95106
}
96107
}
97108

@@ -108,7 +119,7 @@ final class AutoDisposingObserverImpl<T> implements AutoDisposingObserver<T> {
108119

109120
@Override public final void onError(Throwable e) {
110121
if (!isDisposed()) {
111-
dispose();
122+
lazyDispose();
112123
try {
113124
onError.accept(e);
114125
} catch (Exception e1) {
@@ -120,7 +131,7 @@ final class AutoDisposingObserverImpl<T> implements AutoDisposingObserver<T> {
120131

121132
@Override public final void onComplete() {
122133
if (!isDisposed()) {
123-
dispose();
134+
lazyDispose();
124135
try {
125136
onComplete.run();
126137
} catch (Exception e) {

‎autodispose/src/main/java/com/uber/autodispose/AutoDisposingSingleObserverImpl.java

Lines changed: 24 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -75,25 +75,36 @@ final class AutoDisposingSingleObserverImpl<T> implements AutoDisposingSingleObs
7575
@Override public final void dispose() {
7676
synchronized (this) {
7777
AutoDisposableHelper.dispose(lifecycleDisposable);
78+
callMainSubscribeIfNecessary();
79+
AutoDisposableHelper.dispose(mainDisposable);
80+
}
81+
}
7882

79-
// If we've never actually called the downstream onSubscribe (i.e. requested immediately in
80-
// onSubscribe and had a terminal event), we need to still send an empty disposable instance
81-
// to abide by the Observer contract.
82-
if (mainDisposable.get() == null) {
83-
try {
84-
onSubscribe.accept(Disposables.disposed());
85-
} catch (Exception e) {
86-
Exceptions.throwIfFatal(e);
87-
RxJavaPlugins.onError(e);
88-
}
83+
private void lazyDispose() {
84+
synchronized (this) {
85+
AutoDisposableHelper.dispose(lifecycleDisposable);
86+
callMainSubscribeIfNecessary();
87+
mainDisposable.lazySet(AutoDisposableHelper.DISPOSED);
88+
}
89+
}
90+
91+
private void callMainSubscribeIfNecessary() {
92+
// If we've never actually called the downstream onSubscribe (i.e. requested immediately in
93+
// onSubscribe and had a terminal event), we need to still send an empty disposable instance
94+
// to abide by the Observer contract.
95+
if (mainDisposable.get() == null) {
96+
try {
97+
onSubscribe.accept(Disposables.disposed());
98+
} catch (Exception e) {
99+
Exceptions.throwIfFatal(e);
100+
RxJavaPlugins.onError(e);
89101
}
90-
AutoDisposableHelper.dispose(mainDisposable);
91102
}
92103
}
93104

94105
@Override public final void onSuccess(T value) {
95106
if (!isDisposed()) {
96-
dispose();
107+
lazyDispose();
97108
try {
98109
onSuccess.accept(value);
99110
} catch (Exception e) {
@@ -105,7 +116,7 @@ final class AutoDisposingSingleObserverImpl<T> implements AutoDisposingSingleObs
105116

106117
@Override public final void onError(Throwable e) {
107118
if (!isDisposed()) {
108-
dispose();
119+
lazyDispose();
109120
try {
110121
onError.accept(e);
111122
} catch (Exception e1) {

‎autodispose/src/main/java/com/uber/autodispose/AutoDisposingSubscriberImpl.java

Lines changed: 24 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -95,19 +95,30 @@ final class AutoDisposingSubscriberImpl<T> implements AutoDisposingSubscriber<T>
9595
@Override public final void cancel() {
9696
synchronized (this) {
9797
AutoDisposableHelper.dispose(lifecycleDisposable);
98+
callMainSubscribeIfNecessary();
99+
AutoSubscriptionHelper.cancel(mainSubscription);
100+
}
101+
}
98102

99-
// If we've never actually started the upstream subscription (i.e. requested immediately in
100-
// onSubscribe and had a terminal event), we need to still send an empty subscription instance
101-
// to abide by the Subscriber contract.
102-
if (mainSubscription.get() == null) {
103-
try {
104-
onSubscribe.accept(EmptySubscription.INSTANCE);
105-
} catch (Exception e) {
106-
Exceptions.throwIfFatal(e);
107-
RxJavaPlugins.onError(e);
108-
}
103+
private void lazyCancel() {
104+
synchronized (this) {
105+
AutoDisposableHelper.dispose(lifecycleDisposable);
106+
callMainSubscribeIfNecessary();
107+
mainSubscription.lazySet(AutoSubscriptionHelper.CANCELLED);
108+
}
109+
}
110+
111+
private void callMainSubscribeIfNecessary() {
112+
// If we've never actually started the upstream subscription (i.e. requested immediately in
113+
// onSubscribe and had a terminal event), we need to still send an empty subscription instance
114+
// to abide by the Subscriber contract.
115+
if (mainSubscription.get() == null) {
116+
try {
117+
onSubscribe.accept(EmptySubscription.INSTANCE);
118+
} catch (Exception e) {
119+
Exceptions.throwIfFatal(e);
120+
RxJavaPlugins.onError(e);
109121
}
110-
AutoSubscriptionHelper.cancel(mainSubscription);
111122
}
112123
}
113124

@@ -132,7 +143,7 @@ final class AutoDisposingSubscriberImpl<T> implements AutoDisposingSubscriber<T>
132143

133144
@Override public void onError(Throwable e) {
134145
if (!isDisposed()) {
135-
cancel();
146+
lazyCancel();
136147
try {
137148
onError.accept(e);
138149
} catch (Exception e1) {
@@ -144,7 +155,7 @@ final class AutoDisposingSubscriberImpl<T> implements AutoDisposingSubscriber<T>
144155

145156
@Override public final void onComplete() {
146157
if (!isDisposed()) {
147-
cancel();
158+
lazyCancel();
148159
try {
149160
onComplete.run();
150161
} catch (Exception e) {

0 commit comments

Comments
 (0)