Incorporate feedback from #130

android/autodispose-android-archcomponents/src/main/java/com/uber/autodispose/android/lifecycle/LifecycleEventsObservable.java

@@ -75,15 +75,18 @@ void backfillEvents() {
   }
 
   @Override protected void subscribeActual(Observer<? super Event> observer) {
+    ArchLifecycleObserver archObserver =
+        new ArchLifecycleObserver(lifecycle, observer, eventsObservable);
+    observer.onSubscribe(archObserver);
     if (!isMainThread()) {
       observer.onError(
           new IllegalStateException("Lifecycles can only be bound to on the main thread!"));
       return;
     }
-    ArchLifecycleObserver archObserver =
-        new ArchLifecycleObserver(lifecycle, observer, eventsObservable);
-    observer.onSubscribe(archObserver);
     lifecycle.addObserver(archObserver);
+    if (archObserver.isDisposed()) {
+      lifecycle.removeObserver(archObserver);
+    }
   }
 
   static final class ArchLifecycleObserver extends MainThreadDisposable

android/autodispose-android/src/main/java/com/uber/autodispose/android/ViewAttachEventsObservable.java

@@ -37,6 +37,8 @@ final class ViewAttachEventsObservable extends Observable<ViewLifecycleEvent> {
   }
 
   @Override protected void subscribeActual(Observer<? super ViewLifecycleEvent> observer) {
+    Listener listener = new Listener(view, observer);
+    observer.onSubscribe(listener);
     if (!isMainThread()) {
       observer.onError(new IllegalStateException("Views can only be bound to on the main thread!"));
       return;
@@ -46,9 +48,10 @@ final class ViewAttachEventsObservable extends Observable<ViewLifecycleEvent> {
       // Emit the last event, like a behavior subject
       observer.onNext(ViewLifecycleEvent.ATTACH);
     }
-    Listener listener = new Listener(view, observer);
-    observer.onSubscribe(listener);
     view.addOnAttachStateChangeListener(listener);
+    if (listener.isDisposed()) {
+      view.removeOnAttachStateChangeListener(listener);
+    }
   }
 
   static final class Listener extends MainThreadDisposable

autodispose/src/main/java/com/uber/autodispose/AtomicThrowable.java

@@ -0,0 +1,58 @@
+/*
+ * Copyright (c) 2016-present, RxJava Contributors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package com.uber.autodispose;
+
+import java.util.concurrent.atomic.AtomicReference;
+import javax.annotation.Nullable;
+
+/**
+ * Atomic container for Throwables including combining and having a
+ * terminal state via ExceptionHelper.
+ * <p>
+ * Watch out for the leaked AtomicReference methods!
+ */
+final class AtomicThrowable extends AtomicReference<Throwable> {
+
+  private static final long serialVersionUID = 3949248817947090603L;
+
+  /**
+   * Atomically adds a Throwable to this container (combining with a previous Throwable is
+   * necessary).
+   *
+   * @param t the throwable to add
+   * @return true if successful, false if the container has been terminated
+   */
+  public boolean addThrowable(Throwable t) {
+    return ExceptionHelper.addThrowable(this, t);
+  }
+
+  /**
+   * Atomically terminate the container and return the contents of the last
+   * non-terminal Throwable of it.
+   *
+   * @return the last Throwable
+   */
+  @Nullable
+  public Throwable terminate() {
+    return ExceptionHelper.terminate(this);
+  }
+
+  public boolean isTerminated() {
+    return get() == ExceptionHelper.TERMINATED;
+  }
+}

autodispose/src/main/java/com/uber/autodispose/AutoDispose.java

@@ -166,9 +166,9 @@ public interface ScopeHandler {
     return new LifecycleScopeProviderHandlerImpl(scope);
   }
 
-  private static class MaybeScopeHandlerImpl implements ScopeHandler {
+  private static final class MaybeScopeHandlerImpl implements ScopeHandler {
 
-    private final Maybe<?> scope;
+    final Maybe<?> scope;
 
     MaybeScopeHandlerImpl(Maybe<?> scope) {
       this.scope = scope;
@@ -196,9 +196,9 @@ public <T> Function<Observable<? extends T>, ObservableSubscribeProxy<T>> forObs
     }
   }
 
-  private static class ScopeProviderHandlerImpl implements ScopeHandler {
+  private static final class ScopeProviderHandlerImpl implements ScopeHandler {
 
-    private final ScopeProvider scope;
+    final ScopeProvider scope;
 
     ScopeProviderHandlerImpl(ScopeProvider scope) {
       this.scope = scope;
@@ -226,9 +226,9 @@ public <T> Function<Observable<? extends T>, ObservableSubscribeProxy<T>> forObs
     }
   }
 
-  private static class LifecycleScopeProviderHandlerImpl implements ScopeHandler {
+  private static final class LifecycleScopeProviderHandlerImpl implements ScopeHandler {
 
-    private final LifecycleScopeProvider<?> scope;
+    final LifecycleScopeProvider<?> scope;
 
     LifecycleScopeProviderHandlerImpl(LifecycleScopeProvider<?> scope) {
       this.scope = scope;

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

@@ -20,7 +20,6 @@
 import io.reactivex.CompletableObserver;
 import io.reactivex.Maybe;
 import io.reactivex.disposables.Disposable;
-import io.reactivex.disposables.Disposables;
 import io.reactivex.observers.DisposableMaybeObserver;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -41,27 +40,27 @@ final class AutoDisposingCompletableObserverImpl implements AutoDisposingComplet
   }
 
   @Override public void onSubscribe(final Disposable d) {
-    if (AutoDisposeEndConsumerHelper.setOnce(lifecycleDisposable,
-        lifecycle.subscribeWith(new DisposableMaybeObserver<Object>() {
-          @Override public void onSuccess(Object o) {
-            callMainSubscribeIfNecessary(d);
-            AutoDisposingCompletableObserverImpl.this.dispose();
-          }
+    DisposableMaybeObserver<Object> o = new DisposableMaybeObserver<Object>() {
+      @Override public void onSuccess(Object o) {
+        lifecycleDisposable.lazySet(AutoDisposableHelper.DISPOSED);
+        AutoDisposableHelper.dispose(mainDisposable);
+      }
 
-          @Override public void onError(Throwable e) {
-            callMainSubscribeIfNecessary(d);
-            AutoDisposingCompletableObserverImpl.this.onError(e);
-          }
+      @Override public void onError(Throwable e) {
+        lifecycleDisposable.lazySet(AutoDisposableHelper.DISPOSED);
+        AutoDisposingCompletableObserverImpl.this.onError(e);
+      }
 
-          @Override public void onComplete() {
-            callMainSubscribeIfNecessary(d);
-            // Noop - we're unbound now
-          }
-        }),
-        getClass())) {
-      if (AutoDisposeEndConsumerHelper.setOnce(mainDisposable, d, getClass())) {
-        delegate.onSubscribe(this);
+      @Override public void onComplete() {
+        mainDisposable.lazySet(AutoDisposableHelper.DISPOSED);
+        lifecycleDisposable.lazySet(AutoDisposableHelper.DISPOSED);
+        // Noop - we're unbound now
       }
+    };
+    if (AutoDisposeEndConsumerHelper.setOnce(lifecycleDisposable, o, getClass())) {
+      delegate.onSubscribe(this);
+      lifecycle.subscribe(o);
+      AutoDisposeEndConsumerHelper.setOnce(mainDisposable, d, getClass());
     }
   }
 
@@ -70,39 +69,22 @@ final class AutoDisposingCompletableObserverImpl implements AutoDisposingComplet
   }
 
   @Override public void dispose() {
-    synchronized (this) {
-      AutoDisposableHelper.dispose(lifecycleDisposable);
-      AutoDisposableHelper.dispose(mainDisposable);
-    }
-  }
-
-  private void lazyDispose() {
-    synchronized (this) {
-      AutoDisposableHelper.dispose(lifecycleDisposable);
-      mainDisposable.lazySet(AutoDisposableHelper.DISPOSED);
-    }
-  }
-
-  @SuppressWarnings("WeakerAccess") // Avoiding synthetic accessors
-  void callMainSubscribeIfNecessary(Disposable d) {
-    // If we've never actually called the downstream onSubscribe (i.e. requested immediately in
-    // onSubscribe and had a terminal event), we need to still send an empty disposable instance
-    // to abide by the Observer contract.
-    if (AutoDisposableHelper.setIfNotSet(mainDisposable, d)) {
-      delegate.onSubscribe(Disposables.disposed());
-    }
+    AutoDisposableHelper.dispose(lifecycleDisposable);
+    AutoDisposableHelper.dispose(mainDisposable);
   }
 
   @Override public void onComplete() {
     if (!isDisposed()) {
-      lazyDispose();
+      mainDisposable.lazySet(AutoDisposableHelper.DISPOSED);
+      AutoDisposableHelper.dispose(lifecycleDisposable);
       delegate.onComplete();
     }
   }
 
   @Override public void onError(Throwable e) {
     if (!isDisposed()) {
-      lazyDispose();
+      mainDisposable.lazySet(AutoDisposableHelper.DISPOSED);
+      AutoDisposableHelper.dispose(lifecycleDisposable);
       delegate.onError(e);
     }
   }

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

@@ -20,7 +20,6 @@
 import io.reactivex.Maybe;
 import io.reactivex.MaybeObserver;
 import io.reactivex.disposables.Disposable;
-import io.reactivex.disposables.Disposables;
 import io.reactivex.observers.DisposableMaybeObserver;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -41,27 +40,27 @@ final class AutoDisposingMaybeObserverImpl<T> implements AutoDisposingMaybeObser
   }
 
   @Override public void onSubscribe(final Disposable d) {
-    if (AutoDisposeEndConsumerHelper.setOnce(lifecycleDisposable,
-        lifecycle.subscribeWith(new DisposableMaybeObserver<Object>() {
-          @Override public void onSuccess(Object o) {
-            callMainSubscribeIfNecessary(d);
-            AutoDisposingMaybeObserverImpl.this.dispose();
-          }
+    DisposableMaybeObserver<Object> o = new DisposableMaybeObserver<Object>() {
+      @Override public void onSuccess(Object o) {
+        lifecycleDisposable.lazySet(AutoDisposableHelper.DISPOSED);
+        AutoDisposableHelper.dispose(mainDisposable);
+      }
 
-          @Override public void onError(Throwable e) {
-            callMainSubscribeIfNecessary(d);
-            AutoDisposingMaybeObserverImpl.this.onError(e);
-          }
+      @Override public void onError(Throwable e) {
+        lifecycleDisposable.lazySet(AutoDisposableHelper.DISPOSED);
+        AutoDisposingMaybeObserverImpl.this.onError(e);
+      }
 
-          @Override public void onComplete() {
-            callMainSubscribeIfNecessary(d);
-            // Noop - we're unbound now
-          }
-        }),
-        getClass())) {
-      if (AutoDisposeEndConsumerHelper.setOnce(mainDisposable, d, getClass())) {
-        delegate.onSubscribe(this);
+      @Override public void onComplete() {
+        mainDisposable.lazySet(AutoDisposableHelper.DISPOSED);
+        lifecycleDisposable.lazySet(AutoDisposableHelper.DISPOSED);
+        // Noop - we're unbound now
       }
+    };
+    if (AutoDisposeEndConsumerHelper.setOnce(lifecycleDisposable, o, getClass())) {
+      delegate.onSubscribe(this);
+      lifecycle.subscribe(o);
+      AutoDisposeEndConsumerHelper.setOnce(mainDisposable, d, getClass());
     }
   }
 
@@ -70,46 +69,30 @@ final class AutoDisposingMaybeObserverImpl<T> implements AutoDisposingMaybeObser
   }
 
   @Override public void dispose() {
-    synchronized (this) {
-      AutoDisposableHelper.dispose(lifecycleDisposable);
-      AutoDisposableHelper.dispose(mainDisposable);
-    }
-  }
-
-  private void lazyDispose() {
-    synchronized (this) {
-      AutoDisposableHelper.dispose(lifecycleDisposable);
-      mainDisposable.lazySet(AutoDisposableHelper.DISPOSED);
-    }
-  }
-
-  @SuppressWarnings("WeakerAccess") // Avoiding synthetic accessors
-  void callMainSubscribeIfNecessary(Disposable d) {
-    // If we've never actually called the downstream onSubscribe (i.e. requested immediately in
-    // onSubscribe and had a terminal event), we need to still send an empty disposable instance
-    // to abide by the Observer contract.
-    if (AutoDisposableHelper.setIfNotSet(mainDisposable, d)) {
-      delegate.onSubscribe(Disposables.disposed());
-    }
+    AutoDisposableHelper.dispose(lifecycleDisposable);
+    AutoDisposableHelper.dispose(mainDisposable);
   }
 
   @Override public void onSuccess(T value) {
     if (!isDisposed()) {
-      lazyDispose();
+      mainDisposable.lazySet(AutoDisposableHelper.DISPOSED);
+      AutoDisposableHelper.dispose(lifecycleDisposable);
       delegate.onSuccess(value);
     }
   }
 
   @Override public void onError(Throwable e) {
     if (!isDisposed()) {
-      lazyDispose();
+      mainDisposable.lazySet(AutoDisposableHelper.DISPOSED);
+      AutoDisposableHelper.dispose(lifecycleDisposable);
       delegate.onError(e);
     }
   }
 
   @Override public void onComplete() {
     if (!isDisposed()) {
-      lazyDispose();
+      mainDisposable.lazySet(AutoDisposableHelper.DISPOSED);
+      AutoDisposableHelper.dispose(lifecycleDisposable);
       delegate.onComplete();
     }
   }