Remove Unnecessary Subscription

- be explicit for error case in JoinObserver

Author: benjchristensen

rxjava-core/src/main/java/rx/joins/JoinObserver1.java

@@ -19,9 +19,11 @@
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Queue;
+import java.util.concurrent.atomic.AtomicBoolean;
+
 import rx.Notification;
 import rx.Observable;
-import rx.subscriptions.SingleAssignmentSubscription;
+import rx.operators.SafeObservableSubscription;
 import rx.util.functions.Action1;
 
 /**
@@ -33,14 +35,15 @@ public final class JoinObserver1<T> extends ObserverBase<Notification<T>> implem
     private final Action1<Throwable> onError;
     private final List<ActivePlan0> activePlans;
     private final Queue<Notification<T>> queue;
-    private final SingleAssignmentSubscription subscription;
+    private final SafeObservableSubscription subscription;
     private volatile boolean done;
+    private final AtomicBoolean subscribed = new AtomicBoolean(false);
 
     public JoinObserver1(Observable<T> source, Action1<Throwable> onError) {
         this.source = source;
         this.onError = onError;
         queue = new LinkedList<Notification<T>>();
-        subscription = new SingleAssignmentSubscription();
+        subscription = new SafeObservableSubscription();
         activePlans = new ArrayList<ActivePlan0>();
     }
     public Queue<Notification<T>> queue() {
@@ -51,8 +54,12 @@ public void addActivePlan(ActivePlan0 activePlan) {
     }
     @Override
     public void subscribe(Object gate) {
-        this.gate = gate;
-        subscription.set(source.materialize().subscribe(this));
+        if (subscribed.compareAndSet(false, true)) {
+            this.gate = gate;
+            subscription.wrap(source.materialize().subscribe(this));
+        } else {
+            throw new IllegalStateException("Can only be subscribed to once.");
+        }
     }
 
     @Override