Merge pull request #1610 from benjchristensen/multicast-without-factory

Deprecate Multicast without Factory

Author: benjchristensen

rxjava-core/src/main/java/rx/Observable.java

@@ -6002,7 +6002,9 @@ public final <TIntermediate, TResult> Observable<TResult> multicast(
      *         into the specified {@link Subject}
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Connectable-Observable-Operators#observablepublish-and-observablemulticast">RxJava wiki: Observable.publish and Observable.multicast</a>
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229708.aspx">MSDN: Observable.Multicast</a>
+     * @deprecated Use multicast(Func0<Subject>) instead. This one caused nuanced bugs as it retains state.
      */
+    @Deprecated
     public final <R> ConnectableObservable<R> multicast(final Subject<? super T, ? extends R> subject) {
         return new OperatorMulticast<T, R>(this, new Func0<Subject<? super T, ? extends R>>() {
 

rxjava-core/src/test/java/rx/internal/operators/OnSubscribeMulticastTest.java

@@ -24,6 +24,7 @@
 
 import rx.Observer;
 import rx.Subscription;
+import rx.functions.Func0;
 import rx.observables.ConnectableObservable;
 import rx.subjects.PublishSubject;
 import rx.subjects.Subject;
@@ -34,8 +35,7 @@ public class OnSubscribeMulticastTest {
     public void testMulticast() {
         Subject<String, String> source = PublishSubject.create();
 
-        ConnectableObservable<String> multicasted = source.multicast(
-                PublishSubject.<String> create());
+        ConnectableObservable<String> multicasted = source.multicast(new PublishSubjectFactory());
 
         @SuppressWarnings("unchecked")
         Observer<String> observer = mock(Observer.class);
@@ -62,8 +62,7 @@ public void testMulticast() {
     public void testMulticastConnectTwice() {
         Subject<String, String> source = PublishSubject.create();
 
-        ConnectableObservable<String> multicasted = source.multicast(
-                PublishSubject.<String> create());
+        ConnectableObservable<String> multicasted = source.multicast(new PublishSubjectFactory());
 
         @SuppressWarnings("unchecked")
         Observer<String> observer = mock(Observer.class);
@@ -87,8 +86,7 @@ public void testMulticastConnectTwice() {
     public void testMulticastDisconnect() {
         Subject<String, String> source = PublishSubject.create();
 
-        ConnectableObservable<String> multicasted = source.multicast(
-                PublishSubject.<String> create());
+        ConnectableObservable<String> multicasted = source.multicast(new PublishSubjectFactory());
 
         @SuppressWarnings("unchecked")
         Observer<String> observer = mock(Observer.class);
@@ -102,6 +100,9 @@ public void testMulticastDisconnect() {
         connection.unsubscribe();
         source.onNext("three");
 
+        // subscribe again
+        multicasted.subscribe(observer);
+        // reconnect
         multicasted.connect();
         source.onNext("four");
         source.onCompleted();
@@ -113,4 +114,13 @@ public void testMulticastDisconnect() {
         verify(observer, times(1)).onCompleted();
 
     }
+    
+    private static final class PublishSubjectFactory implements Func0<Subject<String, String>> {
+
+        @Override
+        public Subject<String, String> call() {
+            return PublishSubject.<String> create();
+        }
+        
+    }
 }