Merge pull request ReactiveX#1515 from benjchristensen/multicast-factory

Support Subject Factory with Multicast

Author: benjchristensen

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

@@ -5414,8 +5414,43 @@ public final <TIntermediate, TResult> Observable<TResult> multicast(
      * @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>
      */
-    public final <R> ConnectableObservable<R> multicast(Subject<? super T, ? extends R> subject) {
-        return new OperatorMulticast<T, R>(this, subject);
+    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>>() {
+
+            @Override
+            public Subject<? super T, ? extends R> call() {
+                // same one every time, no factory behavior
+                return subject;
+            }
+            
+        });
+    }
+    
+    /**
+     * Returns a {@link ConnectableObservable} that upon connection causes the source Observable to push results
+     * into the specified subject. A Connectable Observable resembles an ordinary Observable, except that it
+     * does not begin emitting items when it is subscribed to, but only when its {@code connect} method
+     * is called.
+     * <dl>
+     *  <dt><b>Backpressure Support:</b></dt>
+     *  <dd>This operator does not support backpressure because multicasting means the stream is "hot" with
+     *      multiple subscribers. Each child will need to manage backpressure independently using operators such
+     *      as {@link #onBackpressureDrop} and {@link #onBackpressureBuffer}.</dd>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code multicast} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     * 
+     * @param subjectFactory
+     *            Func that creates a new {@link Subject} for the {@link ConnectableObservable} to push source items into
+     * @param <R>
+     *            the type of items emitted by the resulting {@code ConnectableObservable}
+     * @return a {@link ConnectableObservable} that upon connection causes the source Observable to push results
+     *         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>
+     */
+    public final <R> ConnectableObservable<R> multicast(Func0<? extends Subject<? super T, ? extends R>> subjectFactory) {
+        return new OperatorMulticast<T, R>(this, subjectFactory);
     }
 
     /**
@@ -5724,7 +5759,14 @@ public final <R> Observable<R> parallel(final Func1<Observable<T>, Observable<R>
      * @see <a href="http://msdn.microsoft.com/en-us/library/system.reactive.linq.observable.publish.aspx">MSDN: Observable.Publish</a>
      */
     public final ConnectableObservable<T> publish() {
-        return new OperatorMulticast<T, T>(this, PublishSubject.<T> create());
+        return new OperatorMulticast<T, T>(this, new Func0<Subject<? super T, ? extends T>>() {
+
+            @Override
+            public Subject<? super T, ? extends T> call() {
+                return PublishSubject.<T> create();
+            }
+            
+        });
     }
 
     /**
@@ -5819,8 +5861,15 @@ public final Subject<T, T> call() {
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Connectable-Observable-Operators#observablepublish-and-observablemulticast">RxJava wiki: publish</a>
      * @see <a href="http://msdn.microsoft.com/en-us/library/system.reactive.linq.observable.publish.aspx">MSDN: Observable.Publish</a>
      */
-    public final ConnectableObservable<T> publish(T initialValue) {
-        return new OperatorMulticast<T, T>(this, BehaviorSubject.<T> create(initialValue));
+    public final ConnectableObservable<T> publish(final T initialValue) {
+        return new OperatorMulticast<T, T>(this, new Func0<Subject<? super T, ? extends T>>() {
+
+            @Override
+            public Subject<? super T, ? extends T> call() {
+                return BehaviorSubject.<T> create(initialValue);
+            }
+            
+        });
     }
 
     /**
@@ -5842,7 +5891,14 @@ public final ConnectableObservable<T> publish(T initialValue) {
      * @see <a href="http://msdn.microsoft.com/en-us/library/system.reactive.linq.observable.publishlast.aspx">MSDN: Observable.PublishLast</a>
      */
     public final ConnectableObservable<T> publishLast() {
-        return new OperatorMulticast<T, T>(this, AsyncSubject.<T> create());
+        return new OperatorMulticast<T, T>(this, new Func0<Subject<? super T, ? extends T>>() {
+
+            @Override
+            public Subject<? super T, ? extends T> call() {
+                return AsyncSubject.<T> create();
+            }
+            
+        });
     }
 
     /**
@@ -6112,7 +6168,14 @@ public final Observable<T> repeatWhen(Func1<? super Observable<? extends Notific
      * @see <a href="http://msdn.microsoft.com/en-us/library/system.reactive.linq.observable.replay.aspx">MSDN: Observable.Replay</a>
      */
     public final ConnectableObservable<T> replay() {
-        return new OperatorMulticast<T, T>(this, ReplaySubject.<T> create());
+        return new OperatorMulticast<T, T>(this, new Func0<Subject<? super T, ? extends T>>() {
+
+            @Override
+            public Subject<? super T, ? extends T> call() {
+                return ReplaySubject.<T> create();
+            }
+            
+        });
     }
 
     /**
@@ -6444,8 +6507,15 @@ public final Subject<T, T> call() {
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Connectable-Observable-Operators#observablereplay">RxJava wiki: replay</a>
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh211976.aspx">MSDN: Observable.Replay</a>
      */
-    public final ConnectableObservable<T> replay(int bufferSize) {
-        return new OperatorMulticast<T, T>(this, ReplaySubject.<T>createWithSize(bufferSize));
+    public final ConnectableObservable<T> replay(final int bufferSize) {
+        return new OperatorMulticast<T, T>(this, new Func0<Subject<? super T, ? extends T>>() {
+
+            @Override
+            public Subject<? super T, ? extends T> call() {
+                return ReplaySubject.<T>createWithSize(bufferSize);
+            }
+            
+        });
     }
 
     /**
@@ -6512,11 +6582,18 @@ public final ConnectableObservable<T> replay(int bufferSize, long time, TimeUnit
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Connectable-Observable-Operators#observablereplay">RxJava wiki: replay</a>
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh211759.aspx">MSDN: Observable.Replay</a>
      */
-    public final ConnectableObservable<T> replay(int bufferSize, long time, TimeUnit unit, Scheduler scheduler) {
+    public final ConnectableObservable<T> replay(final int bufferSize, final long time, final TimeUnit unit, final Scheduler scheduler) {
         if (bufferSize < 0) {
             throw new IllegalArgumentException("bufferSize < 0");
         }
-        return new OperatorMulticast<T, T>(this, ReplaySubject.<T>createWithTimeAndSize(time, unit, bufferSize, scheduler));
+        return new OperatorMulticast<T, T>(this, new Func0<Subject<? super T, ? extends T>>() {
+
+            @Override
+            public Subject<? super T, ? extends T> call() {
+                return ReplaySubject.<T>createWithTimeAndSize(time, unit, bufferSize, scheduler);
+            }
+            
+        });
     }
 
     /**
@@ -6544,10 +6621,15 @@ public final ConnectableObservable<T> replay(int bufferSize, long time, TimeUnit
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Connectable-Observable-Operators#observablereplay">RxJava wiki: replay</a>
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229814.aspx">MSDN: Observable.Replay</a>
      */
-    public final ConnectableObservable<T> replay(int bufferSize, Scheduler scheduler) {
-        return new OperatorMulticast<T, T>(this,
-                OperatorReplay.createScheduledSubject(
-                        ReplaySubject.<T>createWithSize(bufferSize), scheduler));
+    public final ConnectableObservable<T> replay(final int bufferSize, final Scheduler scheduler) {
+        return new OperatorMulticast<T, T>(this, new Func0<Subject<? super T, ? extends T>>() {
+
+            @Override
+            public Subject<? super T, ? extends T> call() {
+                return OperatorReplay.createScheduledSubject(ReplaySubject.<T>createWithSize(bufferSize), scheduler);
+            }
+            
+        });
     }
 
     /**
@@ -6606,8 +6688,15 @@ public final ConnectableObservable<T> replay(long time, TimeUnit unit) {
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Connectable-Observable-Operators#observablereplay">RxJava wiki: replay</a>
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh211811.aspx">MSDN: Observable.Replay</a>
      */
-    public final ConnectableObservable<T> replay(long time, TimeUnit unit, Scheduler scheduler) {
-        return new OperatorMulticast<T, T>(this, ReplaySubject.<T>createWithTime(time, unit, scheduler));
+    public final ConnectableObservable<T> replay(final long time, final TimeUnit unit, final Scheduler scheduler) {
+        return new OperatorMulticast<T, T>(this, new Func0<Subject<? super T, ? extends T>>() {
+
+            @Override
+            public Subject<? super T, ? extends T> call() {
+                return ReplaySubject.<T>createWithTime(time, unit, scheduler);
+            }
+            
+        });
     }
 
     /**
@@ -6634,8 +6723,15 @@ public final ConnectableObservable<T> replay(long time, TimeUnit unit, Scheduler
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Connectable-Observable-Operators#observablereplay">RxJava wiki: replay</a>
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh211699.aspx">MSDN: Observable.Replay</a>
      */
-    public final ConnectableObservable<T> replay(Scheduler scheduler) {
-        return new OperatorMulticast<T, T>(this, OperatorReplay.createScheduledSubject(ReplaySubject.<T> create(), scheduler));
+    public final ConnectableObservable<T> replay(final Scheduler scheduler) {
+        return new OperatorMulticast<T, T>(this, new Func0<Subject<? super T, ? extends T>>() {
+
+            @Override
+            public Subject<? super T, ? extends T> call() {
+                return OperatorReplay.createScheduledSubject(ReplaySubject.<T> create(), scheduler);
+            }
+            
+        });
     }
 
     /**

rxjava-core/src/main/java/rx/internal/operators/OnSubscribeMulticastSelector.java

@@ -24,6 +24,7 @@
 import rx.functions.Func1;
 import rx.observables.ConnectableObservable;
 import rx.observers.SafeSubscriber;
+import rx.subjects.ReplaySubject;
 import rx.subjects.Subject;
 
 /**
@@ -54,9 +55,7 @@ public void call(Subscriber<? super TResult> child) {
         Observable<TResult> observable;
         ConnectableObservable<TIntermediate> connectable;
         try {
-            Subject<? super TInput, ? extends TIntermediate> subject = subjectFactory.call();
-            
-            connectable = new OperatorMulticast<TInput, TIntermediate>(source, subject);
+            connectable = new OperatorMulticast<TInput, TIntermediate>(source, subjectFactory);
 
             observable = resultSelector.call(connectable);
         } catch (Throwable t) {

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

@@ -15,58 +15,82 @@
  */
 package rx.internal.operators;
 
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicReference;
+
 import rx.Observable;
 import rx.Subscriber;
 import rx.Subscription;
 import rx.functions.Action0;
 import rx.functions.Action1;
+import rx.functions.Func0;
 import rx.observables.ConnectableObservable;
 import rx.subjects.Subject;
 import rx.subscriptions.Subscriptions;
 
 /**
  * Shares a single subscription to a source through a Subject.
  * 
- * @param <T> the source value type
- * @param <R> the result value type
+ * @param <T>
+ *            the source value type
+ * @param <R>
+ *            the result value type
  */
 public final class OperatorMulticast<T, R> extends ConnectableObservable<R> {
     final Observable<? extends T> source;
-    final Subject<? super T, ? extends R> subject;
-    final Object guard = new Object();
+    final Object guard;
+    final Func0<? extends Subject<? super T, ? extends R>> subjectFactory;
+    private final AtomicReference<Subject<? super T, ? extends R>> connectedSubject;
+    private final List<Subscriber<? super R>> waitingForConnect;
+
     /** Guarded by guard. */
-    Subscription subscription;
+    Subscriber<T> subscription;
+
+    public OperatorMulticast(Observable<? extends T> source, final Func0<? extends Subject<? super T, ? extends R>> subjectFactory) {
+        this(new Object(), new AtomicReference<Subject<? super T, ? extends R>>(), new ArrayList<Subscriber<? super R>>(), source, subjectFactory);
+    }
 
-    public OperatorMulticast(Observable<? extends T> source, final Subject<? super T, ? extends R> subject) {
+    private OperatorMulticast(final Object guard, final AtomicReference<Subject<? super T, ? extends R>> connectedSubject, final List<Subscriber<? super R>> waitingForConnect, Observable<? extends T> source, final Func0<? extends Subject<? super T, ? extends R>> subjectFactory) {
         super(new OnSubscribe<R>() {
-                @Override
-                public void call(Subscriber<? super R> subscriber) {
-                    subject.unsafeSubscribe(subscriber);
+            @Override
+            public void call(Subscriber<? super R> subscriber) {
+                synchronized (guard) {
+                    if (connectedSubject.get() == null) {
+                        // not connected yet, so register
+                        waitingForConnect.add(subscriber);
+                    } else {
+                        // we are already connected so subscribe directly
+                        connectedSubject.get().unsafeSubscribe(subscriber);
+                    }
                 }
-            });
+            }
+        });
+        this.guard = guard;
+        this.connectedSubject = connectedSubject;
+        this.waitingForConnect = waitingForConnect;
         this.source = source;
-        this.subject = subject;
+        this.subjectFactory = subjectFactory;
     }
 
     @Override
     public void connect(Action1<? super Subscription> connection) {
-        connection.call(Subscriptions.create(new Action0() {
-            @Override
-            public void call() {
-                Subscription s;
-                synchronized (guard) {
-                    s = subscription;
-                    subscription = null;
-                }
-                if (s != null) {
-                    s.unsubscribe();
-                }
-            }
-        }));
-        Subscriber<T> s = null;
+        // each time we connect we create a new Subject and Subscription
+
+        boolean shouldSubscribe = false;
+
+        // subscription is the state of whether we are connected or not
         synchronized (guard) {
-            if (subscription == null) {
-                s = new Subscriber<T>() {
+            if (subscription != null) {
+                // already connected, return as there is nothing to do
+                return;
+            } else {
+                shouldSubscribe = true;
+                // we aren't connected, so let's create a new Subject and connect
+                final Subject<? super T, ? extends R> subject = subjectFactory.call();
+                // create new Subscriber that will pass-thru to the subject we just created
+                // we do this since it is also a Subscription whereas the Subject is not
+                subscription = new Subscriber<T>() {
                     @Override
                     public void onCompleted() {
                         subject.onCompleted();
@@ -82,11 +106,38 @@ public void onNext(T args) {
                         subject.onNext(args);
                     }
                 };
-                subscription = s;
+                
+                // register any subscribers that are waiting with this new subject
+                for(Subscriber<? super R> s : waitingForConnect) {
+                    subject.unsafeSubscribe(s);
+                }
+                // clear the waiting list as any new ones that come in after leaving this synchronized block will go direct to the Subject
+                waitingForConnect.clear();
+                // record the Subject so OnSubscribe can see it
+                connectedSubject.set(subject);
             }
         }
-        if (s != null) {
-            source.unsafeSubscribe(s);
+
+        // in the lock above we determined we should subscribe, do it now outside the lock
+        if (shouldSubscribe) {
+            // register a subscription that will shut this down
+            connection.call(Subscriptions.create(new Action0() {
+                @Override
+                public void call() {
+                    Subscription s;
+                    synchronized (guard) {
+                        s = subscription;
+                        subscription = null;
+                        connectedSubject.set(null);
+                    }
+                    if (s != null) {
+                        s.unsubscribe();
+                    }
+                }
+            }));
+
+            // now that everything is hooked up let's subscribe
+            source.unsafeSubscribe(subscription);
         }
     }
 }