Synchronized SubscriptionList

As per discussion in ReactiveX#1383 ... despite the performance hit.

Author: benjchristensen

rxjava-core/src/main/java/rx/internal/util/SubscriptionList.java

@@ -54,7 +54,7 @@ public synchronized boolean isUnsubscribed() {
      * @param s
      *            the {@link Subscription} to add
      */
-    public void add(final Subscription s) {
+    public synchronized void add(final Subscription s) {
         if (unsubscribed) {
             s.unsubscribe();
         } else {
@@ -71,13 +71,19 @@ public void add(final Subscription s) {
      */
     @Override
     public void unsubscribe() {
-        if (unsubscribed) {
-            return;
+        Collection<Subscription> toUnsubscribe = null;
+        synchronized (this) {
+            if (unsubscribed) {
+                return;
+            }
+            unsubscribed = true;
+            toUnsubscribe = subscriptions;
+            subscriptions = null;
+        }
+
+        if (toUnsubscribe != null) {
+            unsubscribeFromAll(toUnsubscribe);
         }
-        unsubscribed = true;
-        // we will only get here once
-        unsubscribeFromAll(subscriptions);
-        subscriptions = null;
     }
 
     private static void unsubscribeFromAll(Collection<Subscription> subscriptions) {

rxjava-core/src/perf/java/rx/PerfBaseline.java

@@ -39,6 +39,16 @@ public void observableConsumption(Input input) throws InterruptedException {
     public void observableViaRange(Input input) throws InterruptedException {
         input.observable.subscribe(input.observer);
     }
+    
+    @Benchmark
+    public void observableConsumptionUnsafe(Input input) throws InterruptedException {
+        input.firehose.unsafeSubscribe(input.newSubscriber());
+    }
+
+    @Benchmark
+    public void observableViaRangeUnsafe(Input input) throws InterruptedException {
+        input.observable.unsafeSubscribe(input.newSubscriber());
+    }
 
     @Benchmark
     public void iterableViaForLoopConsumption(Input input) throws InterruptedException {

rxjava-core/src/perf/java/rx/jmh/InputWithIncrementingInteger.java

@@ -36,7 +36,7 @@ public abstract class InputWithIncrementingInteger {
     public Observer<Integer> observer;
 
     public abstract int getSize();
-    
+
     @Setup
     public void setup(final Blackhole bh) {
         this.bh = bh;
@@ -106,4 +106,25 @@ public LatchedObserver<Integer> newLatchedObserver() {
         return new LatchedObserver<Integer>(bh);
     }
 
+    public Subscriber<Integer> newSubscriber() {
+        return new Subscriber<Integer>() {
+
+            @Override
+            public void onCompleted() {
+
+            }
+
+            @Override
+            public void onError(Throwable e) {
+
+            }
+
+            @Override
+            public void onNext(Integer t) {
+                bh.consume(t);
+            }
+
+        };
+    }
+
 }