Added support for sever connect method calls

Author: mairbek

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

@@ -36,6 +36,7 @@
 import org.mockito.Mockito;
 import org.mockito.MockitoAnnotations;
 
+import rx.observables.ConnectableObservable;
 import rx.observables.GroupedObservable;
 import rx.operators.OperationAll;
 import rx.operators.OperationConcat;
@@ -48,6 +49,7 @@
 import rx.operators.OperationMerge;
 import rx.operators.OperationMergeDelayError;
 import rx.operators.OperationMostRecent;
+import rx.operators.OperatorMulticast;
 import rx.operators.OperationNext;
 import rx.operators.OperationObserveOn;
 import rx.operators.OperationOnErrorResumeNextViaFunction;
@@ -72,6 +74,7 @@
 import rx.plugins.RxJavaErrorHandler;
 import rx.plugins.RxJavaObservableExecutionHook;
 import rx.plugins.RxJavaPlugins;
+import rx.subjects.Subject;
 import rx.subscriptions.BooleanSubscription;
 import rx.subscriptions.Subscriptions;
 import rx.util.AtomicObservableSubscription;
@@ -585,6 +588,17 @@ public void call(Object args) {
         });
     }
 
+    /**
+     * Returns a connectable observable sequence that upon connection causes the source sequence to push results into the specified subject.
+     *
+     * @param subject the subject to push source elements into.
+     * @param <R> result type
+     * @return a connectable observable sequence that upon connection causes the source sequence to push results into the specified subject.
+     */
+    public <R> ConnectableObservable<R> multicast(Subject<T, R> subject) {
+        return multicast(this, subject);
+    }
+
     /**
      * Returns the only element of an observable sequence and throws an exception if there is not exactly one element in the observable sequence.
      * 
@@ -2072,9 +2086,22 @@ public static <T> Iterable<T> mostRecent(Observable<T> source, T initialValue) {
         return OperationMostRecent.mostRecent(source, initialValue);
     }
 
+    /**
+     * Returns a connectable observable sequence that upon connection causes the source sequence to push results into the specified subject.
+     *
+     * @param source the source sequence whose elements will be pushed into the specified subject.
+     * @param subject the subject to push source elements into.
+     * @param <T> source type
+     * @param <R> result type
+     * @return a connectable observable sequence that upon connection causes the source sequence to push results into the specified subject.
+     */
+    public static <T, R> ConnectableObservable<R> multicast(Observable<T> source, final Subject<T, R> subject) {
+        return OperatorMulticast.multicast(source, subject);
+    }
+
     /**
      * Returns the only element of an observable sequence and throws an exception if there is not exactly one element in the observable sequence.
-     * 
+     *
      * @param that
      *            the source Observable
      * @return The single element in the observable sequence.

rxjava-core/src/main/java/rx/observables/ConnectableObservable.java

@@ -1,3 +1,18 @@
+/**
+ * Copyright 2013 Netflix, Inc.
+ *
+ * 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 rx.observables;
 
 import rx.Observable;

rxjava-core/src/main/java/rx/operators/OperatorMulticast.java

@@ -1,3 +1,18 @@
+/**
+ * Copyright 2013 Netflix, Inc.
+ *
+ * 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 rx.operators;
 
 import org.junit.Test;
@@ -9,17 +24,24 @@
 import rx.subjects.Subject;
 import rx.util.functions.Func1;
 
+import javax.annotation.concurrent.GuardedBy;
+
 import static org.mockito.Mockito.*;
 
 public class OperatorMulticast {
     public static <T, R> ConnectableObservable<R> multicast(Observable<T> source, final Subject<T, R> subject) {
-        return new MulticastConnectableObservable<T ,R>(source, subject);
+        return new MulticastConnectableObservable<T, R>(source, subject);
     }
 
     private static class MulticastConnectableObservable<T, R> extends ConnectableObservable<R> {
+        private final Object lock = new Object();
+
         private final Observable<T> source;
         private final Subject<T, R> subject;
 
+        @GuardedBy("lock")
+        private Subscription subscription;
+
         public MulticastConnectableObservable(Observable<T> source, final Subject<T, R> subject) {
             super(new Func1<Observer<R>, Subscription>() {
                 @Override
@@ -32,22 +54,39 @@ public Subscription call(Observer<R> observer) {
         }
 
         public Subscription connect() {
-            return source.subscribe(new Observer<T>() {
-                @Override
-                public void onCompleted() {
-                    subject.onCompleted();
+            synchronized (lock) {
+                if (subscription == null) {
+                    subscription = source.subscribe(new Observer<T>() {
+                        @Override
+                        public void onCompleted() {
+                            subject.onCompleted();
+                        }
+
+                        @Override
+                        public void onError(Exception e) {
+                            subject.onError(e);
+                        }
+
+                        @Override
+                        public void onNext(T args) {
+                            subject.onNext(args);
+                        }
+                    });
                 }
+            }
 
-                @Override
-                public void onError(Exception e) {
-                    subject.onError(e);
-                }
 
+            return new Subscription() {
                 @Override
-                public void onNext(T args) {
-                    subject.onNext(args);
+                public void unsubscribe() {
+                    synchronized (lock) {
+                        if (subscription != null) {
+                            subscription.unsubscribe();
+                            subscription = null;
+                        }
+                    }
                 }
-            });
+            };
         }
 
 
@@ -57,8 +96,7 @@ public static class UnitTest {
 
         @Test
         public void testMulticast() {
-            Subscription s = mock(Subscription.class);
-            TestObservable source = new TestObservable(s);
+            TestObservable source = new TestObservable();
 
             ConnectableObservable<String> multicasted = OperatorMulticast.multicast(source,
                     DefaultSubject.<String>create());
@@ -83,6 +121,60 @@ public void testMulticast() {
 
         }
 
+        @Test
+        public void testMulticastConnectTwice() {
+            TestObservable source = new TestObservable();
+
+            ConnectableObservable<String> multicasted = OperatorMulticast.multicast(source,
+                    DefaultSubject.<String>create());
+
+            Observer<String> observer = mock(Observer.class);
+            multicasted.subscribe(observer);
+
+            source.sendOnNext("one");
+
+            multicasted.connect();
+            multicasted.connect();
+
+            source.sendOnNext("two");
+            source.sendOnCompleted();
+
+            verify(observer, never()).onNext("one");
+            verify(observer, times(1)).onNext("two");
+            verify(observer, times(1)).onCompleted();
+
+        }
+
+        @Test
+        public void testMulticastDisconnect() {
+            TestObservable source = new TestObservable();
+
+            ConnectableObservable<String> multicasted = OperatorMulticast.multicast(source,
+                    DefaultSubject.<String>create());
+
+            Observer<String> observer = mock(Observer.class);
+            multicasted.subscribe(observer);
+
+            source.sendOnNext("one");
+
+            Subscription connection = multicasted.connect();
+            source.sendOnNext("two");
+
+            connection.unsubscribe();
+            source.sendOnNext("three");
+
+            multicasted.connect();
+            source.sendOnNext("four");
+            source.sendOnCompleted();
+
+            verify(observer, never()).onNext("one");
+            verify(observer, times(1)).onNext("two");
+            verify(observer, never()).onNext("three");
+            verify(observer, times(1)).onNext("four");
+            verify(observer, times(1)).onCompleted();
+
+        }
+
 
         private static class TestObservable extends Observable<String> {
 
@@ -102,10 +194,29 @@ public void onNext(String args) {
                     // Do nothing
                 }
             };
-            Subscription s;
+            Subscription s = new Subscription() {
+                @Override
+                public void unsubscribe() {
+                    observer = new Observer<String>() {
+                        @Override
+                        public void onCompleted() {
+                            // Do nothing
+                        }
+
+                        @Override
+                        public void onError(Exception e) {
+                            // Do nothing
+                        }
+
+                        @Override
+                        public void onNext(String args) {
+                            // Do nothing
+                        }
+                    };
+                }
+            };
 
-            public TestObservable(Subscription s) {
-                this.s = s;
+            public TestObservable() {
             }
 
             /* used to simulate subscription */

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

@@ -1,3 +1,18 @@
+/**
+ * Copyright 2013 Netflix, Inc.
+ *
+ * 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 rx.subjects;
 
 import java.util.ArrayList;

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

@@ -1,3 +1,18 @@
+/**
+ * Copyright 2013 Netflix, Inc.
+ *
+ * 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 rx.subjects;
 
 import rx.Observable;