Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

OperatorMulticastAndReplay #1108

Merged
merged 1 commit into from
Apr 30, 2014
Merged

OperatorMulticastAndReplay #1108

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
65 changes: 33 additions & 32 deletions rxjava-core/src/main/java/rx/Observable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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));
}

/**
Expand All @@ -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);
}

/**
Expand Down Expand Up @@ -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());
}

/**
Expand Down Expand Up @@ -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));
}

/**
Expand All @@ -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());
}

/**
Expand Down Expand Up @@ -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());
}

/**
Expand All @@ -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));
}

/**
Expand All @@ -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));
}

/**
Expand Down Expand Up @@ -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));
}

/**
Expand All @@ -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));
}

/**
Expand Down Expand Up @@ -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));
}

/**
Expand All @@ -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));
}

/**
Expand All @@ -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));
}

/**
Expand Down Expand Up @@ -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));
}

/**
Expand All @@ -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));
}

/**
Expand Down Expand Up @@ -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));
}

/**
Expand All @@ -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));
}

/**
Expand Down
148 changes: 0 additions & 148 deletions rxjava-core/src/main/java/rx/operators/OperationMulticast.java
View file
Open in desktop

This file was deleted.

Loading