Removing SynchronizedObserver usage from Subject implementations.

- We don't need to add synchronization as the subjects can trust their source Observables to comply with the Rx contract.
- This optimization follows Rx Design Guidelines 6.8. Avoid serializing operators

This was discussed at ReactiveX#256

Author: benjchristensen

rxjava-core/src/main/java/rx/subjects/AsyncSubject.java

@@ -15,11 +15,8 @@
  */
 package rx.subjects;
 
-import static org.mockito.Matchers.any;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Matchers.anyString;
+import static org.mockito.Matchers.*;
+import static org.mockito.Mockito.*;
 
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicReference;
@@ -30,7 +27,6 @@
 import rx.Observer;
 import rx.Subscription;
 import rx.util.AtomicObservableSubscription;
-import rx.util.SynchronizedObserver;
 import rx.util.functions.Action1;
 import rx.util.functions.Func0;
 import rx.util.functions.Func1;
@@ -80,7 +76,7 @@ public void unsubscribe() {
                 });
 
                 // on subscribe add it to the map of outbound observers to notify
-                observers.put(subscription, new SynchronizedObserver<T>(observer, subscription));
+                observers.put(subscription, observer);
                 return subscription;
             }
         };

rxjava-core/src/main/java/rx/subjects/BehaviorSubject.java

@@ -27,7 +27,6 @@
 import rx.Observer;
 import rx.Subscription;
 import rx.util.AtomicObservableSubscription;
-import rx.util.SynchronizedObserver;
 import rx.util.functions.Action1;
 import rx.util.functions.Func0;
 import rx.util.functions.Func1;
@@ -86,11 +85,10 @@ public void unsubscribe() {
                     }
                 });
 
-                SynchronizedObserver<T> synchronizedObserver = new SynchronizedObserver<T>(observer, subscription);
-                synchronizedObserver.onNext(currentValue.get());
+                observer.onNext(currentValue.get());
 
                 // on subscribe add it to the map of outbound observers to notify
-                observers.put(subscription, synchronizedObserver);
+                observers.put(subscription, observer);
                 return subscription;
             }
         };

rxjava-core/src/main/java/rx/subjects/PublishSubject.java

@@ -34,7 +34,6 @@
 import rx.Observer;
 import rx.Subscription;
 import rx.util.AtomicObservableSubscription;
-import rx.util.SynchronizedObserver;
 import rx.util.functions.Action1;
 import rx.util.functions.Func0;
 import rx.util.functions.Func1;
@@ -78,7 +77,7 @@ public void unsubscribe() {
                 });
 
                 // on subscribe add it to the map of outbound observers to notify
-                observers.put(subscription, new SynchronizedObserver<T>(observer, subscription));
+                observers.put(subscription, observer);
                 return subscription;
             }
         };