Update to EndConsumerHelper (#77)

* Opportunistic formatting fixes

* Update to EndConsumerHelper

Resolves #76

From the original PR:

This PR changes the "Disposable already set!" and "Subscription already set!" messages on the standard consumer classes (DisposableSubscriber, DisposableObserver, etc.) to something more meaningful:

"It is not allowed to subscribe with a(n) <class name> multiple times. Please create a fresh instance of <class name> and subscribe that to the target source instead."

Where <class name> is a placeholder for the getClass().getName() of the subclass of those consumer types. It should clearly state to avoid subscribing with them multiple times as well as printing the full class name to indicate the problem is with the use of the implementor class, and not with the abstract RxJava class.

Inspired by [this Stackoverflow](http://stackoverflow.com/questions/43482263/rxjava2-protocolviolationexception-disposable-already-set) question, one of many such questions.

For the internal operators, the original error message stays because when they appear, that is still likely due to an implementation bug (or a misbehaving user-created custom implementation).

* Fix checkstyle

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

@@ -1,4 +1,4 @@
-/**
+/*
  * Copyright 2016-present, RxJava Contributors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
@@ -17,9 +17,7 @@
 package com.uber.autodispose;
 
 import io.reactivex.disposables.Disposable;
-import io.reactivex.plugins.RxJavaPlugins;
 import java.util.concurrent.atomic.AtomicReference;
-import javax.annotation.Nullable;
 
 /**
  * Utility methods for working with Disposables atomically. Copied from the RxJava implementation.
@@ -30,51 +28,6 @@ enum AutoDisposableHelper implements Disposable {
    */
   DISPOSED;
 
-  static boolean isDisposed(Disposable d) {
-    return d == DISPOSED;
-  }
-
-  static boolean set(AtomicReference<Disposable> field, @Nullable Disposable d) {
-    for (; ; ) {
-      Disposable current = field.get();
-      if (current == DISPOSED) {
-        if (d != null) {
-          d.dispose();
-        }
-        return false;
-      }
-      if (field.compareAndSet(current, d)) {
-        if (current != null) {
-          current.dispose();
-        }
-        return true;
-      }
-    }
-  }
-
-  /**
-   * Atomically sets the field to the given non-null Disposable and returns true
-   * or returns false if the field is non-null.
-   * If the target field contains the common DISPOSED instance, the supplied disposable
-   * is disposed. If the field contains other non-null Disposable, an IllegalStateException
-   * is signalled to the RxJavaPlugins.onError hook.
-   *
-   * @param field the target field
-   * @param d the disposable to set, not null
-   * @return true if the operation succeeded, false
-   */
-  static boolean setOnce(AtomicReference<Disposable> field, Disposable d) {
-    AutoDisposeUtil.checkNotNull(d, "d is null");
-    if (!field.compareAndSet(null, d)) {
-      d.dispose();
-      if (field.get() != DISPOSED) {
-        reportDisposableSet();
-      }
-      return false;
-    }
-    return true;
-  }
-
   /**
    * Atomically sets the field to the given non-null Disposable and returns true
    * or returns false if the field is non-null.
@@ -88,30 +41,6 @@ static boolean setIfNotSet(AtomicReference<Disposable> field, Disposable d) {
     return field.compareAndSet(null, d);
   }
 
-  /**
-   * Atomically replaces the Disposable in the field with the given new Disposable
-   * but does not dispose the old one.
-   *
-   * @param field the target field to change
-   * @param d the new disposable, null allowed
-   * @return true if the operation succeeded, false if the target field contained
-   * the common DISPOSED instance and the given disposable (if not null) is disposed.
-   */
-  static boolean replace(AtomicReference<Disposable> field, @Nullable Disposable d) {
-    for (; ; ) {
-      Disposable current = field.get();
-      if (current == DISPOSED) {
-        if (d != null) {
-          d.dispose();
-        }
-        return false;
-      }
-      if (field.compareAndSet(current, d)) {
-        return true;
-      }
-    }
-  }
-
   /**
    * Atomically disposes the Disposable in the field if not already disposed.
    *
@@ -133,35 +62,6 @@ static boolean dispose(AtomicReference<Disposable> field) {
     return false;
   }
 
-  /**
-   * Verifies that current is null, next is not null, otherwise signals errors
-   * to the RxJavaPlugins and returns false.
-   *
-   * @param current the current Disposable, expected to be null
-   * @param next the next Disposable, expected to be non-null
-   * @return true if the validation succeeded
-   */
-  static boolean validate(@Nullable Disposable current, Disposable next) {
-    //noinspection ConstantConditions leftover from original RxJava implementation
-    if (next == null) {
-      RxJavaPlugins.onError(new NullPointerException("next is null"));
-      return false;
-    }
-    if (current != null) {
-      next.dispose();
-      reportDisposableSet();
-      return false;
-    }
-    return true;
-  }
-
-  /**
-   * Reports that the disposable is already set to the RxJavaPlugins error handler.
-   */
-  static void reportDisposableSet() {
-    RxJavaPlugins.onError(new IllegalStateException("Disposable already set!"));
-  }
-
   @Override public void dispose() {
     // deliberately no-op
   }

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

@@ -0,0 +1,112 @@
+/*
+ * 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 io.reactivex.disposables.Disposable;
+import io.reactivex.exceptions.ProtocolViolationException;
+import io.reactivex.plugins.RxJavaPlugins;
+import java.util.concurrent.atomic.AtomicReference;
+import org.reactivestreams.Subscription;
+
+/**
+ * Utility class to help report multiple subscriptions with the same
+ * consumer type instead of the internal "Disposable already set!" message
+ * that is practically reserved for internal operators and indicate bugs in them.
+ *
+ * Copied from the RxJava implementation.
+ */
+final class AutoDisposeEndConsumerHelper {
+
+  /**
+   * Utility class.
+   */
+  private AutoDisposeEndConsumerHelper() {
+    throw new IllegalStateException("No instances!");
+  }
+
+  /**
+   * Atomically updates the target upstream AtomicReference from null to the non-null
+   * next Disposable, otherwise disposes next and reports a ProtocolViolationException
+   * if the AtomicReference doesn't contain the shared disposed indicator.
+   *
+   * @param upstream the target AtomicReference to update
+   * @param next the Disposable to set on it atomically
+   * @param observer the class of the consumer to have a personalized
+   * error message if the upstream already contains a non-cancelled Disposable.
+   * @return true if successful, false if the content of the AtomicReference was non null
+   */
+  public static boolean setOnce(AtomicReference<Disposable> upstream, Disposable next,
+      Class<?> observer) {
+    AutoDisposeUtil.checkNotNull(next, "next is null");
+    if (!upstream.compareAndSet(null, next)) {
+      next.dispose();
+      if (upstream.get() != AutoDisposableHelper.DISPOSED) {
+        reportDoubleSubscription(observer);
+      }
+      return false;
+    }
+    return true;
+  }
+
+  /**
+   * Atomically updates the target upstream AtomicReference from null to the non-null
+   * next Subscription, otherwise cancels next and reports a ProtocolViolationException
+   * if the AtomicReference doesn't contain the shared cancelled indicator.
+   *
+   * @param upstream the target AtomicReference to update
+   * @param next the Subscription to set on it atomically
+   * @param subscriber the class of the consumer to have a personalized
+   * error message if the upstream already contains a non-cancelled Subscription.
+   * @return true if successful, false if the content of the AtomicReference was non null
+   */
+  public static boolean setOnce(AtomicReference<Subscription> upstream, Subscription next,
+      Class<?> subscriber) {
+    AutoDisposeUtil.checkNotNull(next, "next is null");
+    if (!upstream.compareAndSet(null, next)) {
+      next.cancel();
+      if (upstream.get() != AutoSubscriptionHelper.CANCELLED) {
+        reportDoubleSubscription(subscriber);
+      }
+      return false;
+    }
+    return true;
+  }
+
+  /**
+   * Builds the error message with the consumer class.
+   *
+   * @param consumer the class of the consumer
+   * @return the error message string
+   */
+  public static String composeMessage(String consumer) {
+    return "It is not allowed to subscribe with a(n) "
+        + consumer
+        + " multiple times. "
+        + "Please create a fresh instance of "
+        + consumer
+        + " and subscribe that to the target source instead.";
+  }
+
+  /**
+   * Report a ProtocolViolationException with a personalized message referencing
+   * the simple type name of the consumer class and report it via
+   * RxJavaPlugins.onError.
+   *
+   * @param consumer the class of the consumer
+   */
+  public static void reportDoubleSubscription(Class<?> consumer) {
+    RxJavaPlugins.onError(new ProtocolViolationException(composeMessage(consumer.getName())));
+  }
+}

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

@@ -38,22 +38,22 @@ final class AutoDisposingCompletableObserverImpl implements AutoDisposingComplet
   }
 
   @Override public void onSubscribe(final Disposable d) {
-    if (AutoDisposableHelper.setOnce(lifecycleDisposable,
+    if (AutoDisposeEndConsumerHelper.setOnce(lifecycleDisposable,
         lifecycle.doOnEvent(new BiConsumer<Object, Throwable>() {
-          @Override
-          public void accept(Object o, Throwable throwable) throws Exception {
+          @Override public void accept(Object o, Throwable throwable) throws Exception {
             callMainSubscribeIfNecessary(d);
           }
-        }).subscribe(new Consumer<Object>() {
-          @Override public void accept(Object o) throws Exception {
-            dispose();
-          }
-        }, new Consumer<Throwable>() {
-          @Override public void accept(Throwable e1) throws Exception {
-            onError(e1);
-          }
-        }))) {
-      if (AutoDisposableHelper.setOnce(mainDisposable, d)) {
+        })
+            .subscribe(new Consumer<Object>() {
+              @Override public void accept(Object o) throws Exception {
+                dispose();
+              }
+            }, new Consumer<Throwable>() {
+              @Override public void accept(Throwable e1) throws Exception {
+                onError(e1);
+              }
+            }), getClass())) {
+      if (AutoDisposeEndConsumerHelper.setOnce(mainDisposable, d, getClass())) {
         delegate.onSubscribe(this);
       }
     }

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

@@ -38,24 +38,22 @@ final class AutoDisposingMaybeObserverImpl<T> implements AutoDisposingMaybeObser
   }
 
   @Override public void onSubscribe(final Disposable d) {
-    if (AutoDisposableHelper.setOnce(lifecycleDisposable,
-            lifecycle.doOnEvent(new BiConsumer<Object, Throwable>() {
-              @Override
-              public void accept(Object o, Throwable throwable) throws Exception {
-                callMainSubscribeIfNecessary(d);
-              }
-            }).subscribe(new Consumer<Object>() {
-              @Override
-              public void accept(Object o) throws Exception {
+    if (AutoDisposeEndConsumerHelper.setOnce(lifecycleDisposable,
+        lifecycle.doOnEvent(new BiConsumer<Object, Throwable>() {
+          @Override public void accept(Object o, Throwable throwable) throws Exception {
+            callMainSubscribeIfNecessary(d);
+          }
+        })
+            .subscribe(new Consumer<Object>() {
+              @Override public void accept(Object o) throws Exception {
                 dispose();
               }
             }, new Consumer<Throwable>() {
-              @Override
-              public void accept(Throwable e1) throws Exception {
+              @Override public void accept(Throwable e1) throws Exception {
                 onError(e1);
               }
-            }))) {
-      if (AutoDisposableHelper.setOnce(mainDisposable, d)) {
+            }), getClass())) {
+      if (AutoDisposeEndConsumerHelper.setOnce(mainDisposable, d, getClass())) {
         delegate.onSubscribe(this);
       }
     }

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

@@ -38,22 +38,22 @@ final class AutoDisposingObserverImpl<T> implements AutoDisposingObserver<T> {
   }
 
   @Override public void onSubscribe(final Disposable d) {
-    if (AutoDisposableHelper.setOnce(lifecycleDisposable,
+    if (AutoDisposeEndConsumerHelper.setOnce(lifecycleDisposable,
         lifecycle.doOnEvent(new BiConsumer<Object, Throwable>() {
-          @Override
-          public void accept(Object o, Throwable throwable) throws Exception {
+          @Override public void accept(Object o, Throwable throwable) throws Exception {
             callMainSubscribeIfNecessary(d);
           }
-        }).subscribe(new Consumer<Object>() {
-          @Override public void accept(Object o) throws Exception {
-            dispose();
-          }
-        }, new Consumer<Throwable>() {
-          @Override public void accept(Throwable e) throws Exception {
-            AutoDisposingObserverImpl.this.onError(e);
-          }
-        }))) {
-      if (AutoDisposableHelper.setOnce(mainDisposable, d)) {
+        })
+            .subscribe(new Consumer<Object>() {
+              @Override public void accept(Object o) throws Exception {
+                dispose();
+              }
+            }, new Consumer<Throwable>() {
+              @Override public void accept(Throwable e) throws Exception {
+                AutoDisposingObserverImpl.this.onError(e);
+              }
+            }), getClass())) {
+      if (AutoDisposeEndConsumerHelper.setOnce(mainDisposable, d, getClass())) {
         delegate.onSubscribe(this);
       }
     }

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

@@ -38,22 +38,22 @@ final class AutoDisposingSingleObserverImpl<T> implements AutoDisposingSingleObs
   }
 
   @Override public void onSubscribe(final Disposable d) {
-    if (AutoDisposableHelper.setOnce(lifecycleDisposable,
+    if (AutoDisposeEndConsumerHelper.setOnce(lifecycleDisposable,
         lifecycle.doOnEvent(new BiConsumer<Object, Throwable>() {
-          @Override
-          public void accept(Object o, Throwable throwable) throws Exception {
+          @Override public void accept(Object o, Throwable throwable) throws Exception {
             callMainSubscribeIfNecessary(d);
           }
-        }).subscribe(new Consumer<Object>() {
-          @Override public void accept(Object o) throws Exception {
-            dispose();
-          }
-        }, new Consumer<Throwable>() {
-          @Override public void accept(Throwable e) throws Exception {
-            AutoDisposingSingleObserverImpl.this.onError(e);
-          }
-        }))) {
-      if (AutoDisposableHelper.setOnce(mainDisposable, d)) {
+        })
+            .subscribe(new Consumer<Object>() {
+              @Override public void accept(Object o) throws Exception {
+                dispose();
+              }
+            }, new Consumer<Throwable>() {
+              @Override public void accept(Throwable e) throws Exception {
+                AutoDisposingSingleObserverImpl.this.onError(e);
+              }
+            }), getClass())) {
+      if (AutoDisposeEndConsumerHelper.setOnce(mainDisposable, d, getClass())) {
         delegate.onSubscribe(this);
       }
     }