OperatorMulticastAndReplay

Author: akarnokd

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

@@ -56,12 +56,11 @@
 import rx.operators.OperationJoin;
 import rx.operators.OperationMergeDelayError;
 import rx.operators.OperationMergeMaxConcurrent;
-import rx.operators.OperationMulticast;
 import rx.operators.OperationOnErrorResumeNextViaObservable;
 import rx.operators.OperationOnErrorReturn;
 import rx.operators.OperationOnExceptionResumeNextViaObservable;
 import rx.operators.OperationParallelMerge;
-import rx.operators.OperationReplay;
+import rx.operators.OperatorReplay;
 import rx.operators.OperationSample;
 import rx.operators.OperationSequenceEqual;
 import rx.operators.OperationSkip;
@@ -105,6 +104,8 @@
 import rx.operators.OperatorMerge;
 import rx.operators.OperatorMergeMapPair;
 import rx.operators.OperatorMergeMapTransform;
+import rx.operators.OperatorMulticast;
+import rx.operators.OperatorMulticastSelector;
 import rx.operators.OperatorObserveOn;
 import rx.operators.OperatorOnErrorFlatMap;
 import rx.operators.OperatorOnErrorResumeNextViaFunction;
@@ -4419,7 +4420,7 @@ public final <U, R> Observable<R> mergeMapIterable(Func1<? super T, ? extends It
     public final <TIntermediate, TResult> Observable<TResult> multicast(
             final Func0<? extends Subject<? super T, ? extends TIntermediate>> subjectFactory,
             final Func1<? super Observable<TIntermediate>, ? extends Observable<TResult>> selector) {
-        return OperationMulticast.multicast(this, subjectFactory, selector);
+        return create(new OperatorMulticastSelector<T, TIntermediate, TResult>(this, subjectFactory, selector));
     }
 
     /**
@@ -4436,7 +4437,7 @@ public final <TIntermediate, TResult> Observable<TResult> multicast(
      *      Observable.multicast()</a>
      */
     public final <R> ConnectableObservable<R> multicast(Subject<? super T, ? extends R> subject) {
-        return OperationMulticast.multicast(this, subject);
+        return new OperatorMulticast<T, R>(this, subject);
     }
 
     /**
@@ -4641,7 +4642,7 @@ public final <R> Observable<R> parallel(final Func1<Observable<T>, Observable<R>
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Connectable-Observable-Operators#wiki-observablepublish-and-observablemulticast">RxJava Wiki: publish()</a>
      */
     public final ConnectableObservable<T> publish() {
-        return OperationMulticast.multicast(this, PublishSubject.<T> create());
+        return new OperatorMulticast<T, T>(this, PublishSubject.<T> create());
     }
 
     /**
@@ -4704,7 +4705,7 @@ public final Subject<T, T> call() {
      *         and starts with {@code initialValue}
      */
     public final ConnectableObservable<T> publish(T initialValue) {
-        return OperationMulticast.multicast(this, BehaviorSubject.<T> create(initialValue));
+        return new OperatorMulticast<T, T>(this, BehaviorSubject.<T> create(initialValue));
     }
 
     /**
@@ -4716,7 +4717,7 @@ public final ConnectableObservable<T> publish(T initialValue) {
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Connectable-Observable-Operators#wiki-observablepublishlast">RxJava Wiki: publishLast()</a>
      */
     public final ConnectableObservable<T> publishLast() {
-        return OperationMulticast.multicast(this, AsyncSubject.<T> create());
+        return new OperatorMulticast<T, T>(this, AsyncSubject.<T> create());
     }
 
     /**
@@ -4879,7 +4880,7 @@ public final Observable<T> repeat(long count, Scheduler scheduler) {
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Connectable-Observable-Operators#wiki-observablereplay">RxJava Wiki: replay()</a>
      */
     public final ConnectableObservable<T> replay() {
-        return OperationMulticast.multicast(this, ReplaySubject.<T> create());
+        return new OperatorMulticast<T, T>(this, ReplaySubject.<T> create());
     }
 
     /**
@@ -4898,12 +4899,12 @@ public final ConnectableObservable<T> replay() {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229653.aspx">MSDN: Observable.Replay</a>
      */
     public final <R> Observable<R> replay(Func1<? super Observable<T>, ? extends Observable<R>> selector) {
-        return OperationMulticast.multicast(this, new Func0<Subject<T, T>>() {
+        return create(new OperatorMulticastSelector<T, T, R>(this, new Func0<Subject<T, T>>() {
             @Override
             public final Subject<T, T> call() {
                 return ReplaySubject.create();
             }
-        }, selector);
+        }, selector));
     }
 
     /**
@@ -4927,12 +4928,12 @@ public final Subject<T, T> call() {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh211675.aspx">MSDN: Observable.Replay</a>
      */
     public final <R> Observable<R> replay(Func1<? super Observable<T>, ? extends Observable<R>> selector, final int bufferSize) {
-        return OperationMulticast.multicast(this, new Func0<Subject<T, T>>() {
+        return create(new OperatorMulticastSelector<T, T, R>(this, new Func0<Subject<T, T>>() {
             @Override
             public final Subject<T, T> call() {
-                return OperationReplay.replayBuffered(bufferSize);
+                return OperatorReplay.replayBuffered(bufferSize);
             }
-        }, selector);
+        }, selector));
     }
 
     /**
@@ -4995,12 +4996,12 @@ public final <R> Observable<R> replay(Func1<? super Observable<T>, ? extends Obs
         if (bufferSize < 0) {
             throw new IllegalArgumentException("bufferSize < 0");
         }
-        return OperationMulticast.multicast(this, new Func0<Subject<T, T>>() {
+        return create(new OperatorMulticastSelector<T, T, R>(this, new Func0<Subject<T, T>>() {
             @Override
             public final Subject<T, T> call() {
-                return OperationReplay.replayWindowed(time, unit, bufferSize, scheduler);
+                return OperatorReplay.replayWindowed(time, unit, bufferSize, scheduler);
             }
-        }, selector);
+        }, selector));
     }
 
     /**
@@ -5026,12 +5027,12 @@ public final Subject<T, T> call() {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229928.aspx">MSDN: Observable.Replay</a>
      */
     public final <R> Observable<R> replay(Func1<? super Observable<T>, ? extends Observable<R>> selector, final int bufferSize, final Scheduler scheduler) {
-        return OperationMulticast.multicast(this, new Func0<Subject<T, T>>() {
+        return create(new OperatorMulticastSelector<T, T, R>(this, new Func0<Subject<T, T>>() {
             @Override
             public final Subject<T, T> call() {
-                return OperationReplay.<T> createScheduledSubject(OperationReplay.<T> replayBuffered(bufferSize), scheduler);
+                return OperatorReplay.<T> createScheduledSubject(OperatorReplay.<T> replayBuffered(bufferSize), scheduler);
             }
-        }, selector);
+        }, selector));
     }
 
     /**
@@ -5085,12 +5086,12 @@ public final <R> Observable<R> replay(Func1<? super Observable<T>, ? extends Obs
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh244327.aspx">MSDN: Observable.Replay</a>
      */
     public final <R> Observable<R> replay(Func1<? super Observable<T>, ? extends Observable<R>> selector, final long time, final TimeUnit unit, final Scheduler scheduler) {
-        return OperationMulticast.multicast(this, new Func0<Subject<T, T>>() {
+        return create(new OperatorMulticastSelector<T, T, R>(this, new Func0<Subject<T, T>>() {
             @Override
             public final Subject<T, T> call() {
-                return OperationReplay.replayWindowed(time, unit, -1, scheduler);
+                return OperatorReplay.replayWindowed(time, unit, -1, scheduler);
             }
-        }, selector);
+        }, selector));
     }
 
     /**
@@ -5113,12 +5114,12 @@ public final Subject<T, T> call() {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh211644.aspx">MSDN: Observable.Replay</a>
      */
     public final <R> Observable<R> replay(Func1<? super Observable<T>, ? extends Observable<R>> selector, final Scheduler scheduler) {
-        return OperationMulticast.multicast(this, new Func0<Subject<T, T>>() {
+        return create(new OperatorMulticastSelector<T, T, R>(this, new Func0<Subject<T, T>>() {
             @Override
             public final Subject<T, T> call() {
-                return OperationReplay.createScheduledSubject(ReplaySubject.<T> create(), scheduler);
+                return OperatorReplay.createScheduledSubject(ReplaySubject.<T> create(), scheduler);
             }
-        }, selector);
+        }, selector));
     }
 
     /**
@@ -5135,7 +5136,7 @@ public final Subject<T, T> call() {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh211976.aspx">MSDN: Observable.Replay</a>
      */
     public final ConnectableObservable<T> replay(int bufferSize) {
-        return OperationMulticast.multicast(this, OperationReplay.<T> replayBuffered(bufferSize));
+        return new OperatorMulticast<T, T>(this, OperatorReplay.<T> replayBuffered(bufferSize));
     }
 
     /**
@@ -5184,7 +5185,7 @@ public final ConnectableObservable<T> replay(int bufferSize, long time, TimeUnit
         if (bufferSize < 0) {
             throw new IllegalArgumentException("bufferSize < 0");
         }
-        return OperationMulticast.multicast(this, OperationReplay.<T> replayWindowed(time, unit, bufferSize, scheduler));
+        return new OperatorMulticast<T, T>(this, OperatorReplay.<T> replayWindowed(time, unit, bufferSize, scheduler));
     }
 
     /**
@@ -5203,9 +5204,9 @@ public final ConnectableObservable<T> replay(int bufferSize, long time, TimeUnit
      * @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 OperationMulticast.multicast(this,
-                OperationReplay.createScheduledSubject(
-                        OperationReplay.<T> replayBuffered(bufferSize), scheduler));
+        return new OperatorMulticast<T, T>(this,
+                OperatorReplay.createScheduledSubject(
+                        OperatorReplay.<T> replayBuffered(bufferSize), scheduler));
     }
 
     /**
@@ -5245,7 +5246,7 @@ public final ConnectableObservable<T> replay(long time, TimeUnit unit) {
      * @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 OperationMulticast.multicast(this, OperationReplay.<T> replayWindowed(time, unit, -1, scheduler));
+        return new OperatorMulticast<T, T>(this, OperatorReplay.<T> replayWindowed(time, unit, -1, scheduler));
     }
 
     /**
@@ -5262,7 +5263,7 @@ public final ConnectableObservable<T> replay(long time, TimeUnit unit, Scheduler
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh211699.aspx">MSDN: Observable.Replay</a>
      */
     public final ConnectableObservable<T> replay(Scheduler scheduler) {
-        return OperationMulticast.multicast(this, OperationReplay.createScheduledSubject(ReplaySubject.<T> create(), scheduler));
+        return new OperatorMulticast<T, T>(this, OperatorReplay.createScheduledSubject(ReplaySubject.<T> create(), scheduler));
     }
 
     /**