-
Notifications
You must be signed in to change notification settings - Fork 7.6k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #785 from benjchristensen/operator-zip
Reimplement Zip Operator Using Lift [Preview]
- Loading branch information
Showing
10 changed files
with
1,080 additions
and
294 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
151 changes: 151 additions & 0 deletions
151
rxjava-core/src/main/java/rx/observers/Subscribers.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,151 @@ | ||
| package rx.observers; | ||
|
|
||
| import rx.Observer; | ||
| import rx.Subscriber; | ||
| import rx.util.OnErrorNotImplementedException; | ||
| import rx.util.functions.Action0; | ||
| import rx.util.functions.Action1; | ||
|
|
||
| public class Subscribers { | ||
|
|
||
| public static <T> Subscriber<T> from(final Observer<? super T> o) { | ||
| return new Subscriber<T>() { | ||
|
|
||
| @Override | ||
| public void onCompleted() { | ||
| o.onCompleted(); | ||
| } | ||
|
|
||
| @Override | ||
| public void onError(Throwable e) { | ||
| o.onError(e); | ||
| } | ||
|
|
||
| @Override | ||
| public void onNext(T t) { | ||
| o.onNext(t); | ||
| } | ||
|
|
||
| }; | ||
| } | ||
|
|
||
| /** | ||
| * Create an empty Subscriber that ignores all events. | ||
| */ | ||
| public static final <T> Subscriber<T> create() { | ||
| return new Subscriber<T>() { | ||
|
|
||
| @Override | ||
| public final void onCompleted() { | ||
| // do nothing | ||
| } | ||
|
|
||
| @Override | ||
| public final void onError(Throwable e) { | ||
| throw new OnErrorNotImplementedException(e); | ||
| } | ||
|
|
||
| @Override | ||
| public final void onNext(T args) { | ||
| // do nothing | ||
| } | ||
|
|
||
| }; | ||
| } | ||
|
|
||
| /** | ||
| * Create an Subscriber that receives `onNext` and ignores `onError` and `onCompleted`. | ||
| */ | ||
| public static final <T> Subscriber<T> create(final Action1<? super T> onNext) { | ||
| if (onNext == null) { | ||
| throw new IllegalArgumentException("onNext can not be null"); | ||
| } | ||
|
|
||
| return new Subscriber<T>() { | ||
|
|
||
| @Override | ||
| public final void onCompleted() { | ||
| // do nothing | ||
| } | ||
|
|
||
| @Override | ||
| public final void onError(Throwable e) { | ||
| throw new OnErrorNotImplementedException(e); | ||
| } | ||
|
|
||
| @Override | ||
| public final void onNext(T args) { | ||
| onNext.call(args); | ||
| } | ||
|
|
||
| }; | ||
| } | ||
|
|
||
| /** | ||
| * Create an Subscriber that receives `onNext` and `onError` and ignores `onCompleted`. | ||
| * | ||
| */ | ||
| public static final <T> Subscriber<T> create(final Action1<? super T> onNext, final Action1<Throwable> onError) { | ||
| if (onNext == null) { | ||
| throw new IllegalArgumentException("onNext can not be null"); | ||
| } | ||
| if (onError == null) { | ||
| throw new IllegalArgumentException("onError can not be null"); | ||
| } | ||
|
|
||
| return new Subscriber<T>() { | ||
|
|
||
| @Override | ||
| public final void onCompleted() { | ||
| // do nothing | ||
| } | ||
|
|
||
| @Override | ||
| public final void onError(Throwable e) { | ||
| onError.call(e); | ||
| } | ||
|
|
||
| @Override | ||
| public final void onNext(T args) { | ||
| onNext.call(args); | ||
| } | ||
|
|
||
| }; | ||
| } | ||
|
|
||
| /** | ||
| * Create an Subscriber that receives `onNext`, `onError` and `onCompleted`. | ||
| * | ||
| */ | ||
| public static final <T> Subscriber<T> create(final Action1<? super T> onNext, final Action1<Throwable> onError, final Action0 onComplete) { | ||
| if (onNext == null) { | ||
| throw new IllegalArgumentException("onNext can not be null"); | ||
| } | ||
| if (onError == null) { | ||
| throw new IllegalArgumentException("onError can not be null"); | ||
| } | ||
| if (onComplete == null) { | ||
| throw new IllegalArgumentException("onComplete can not be null"); | ||
| } | ||
|
|
||
| return new Subscriber<T>() { | ||
|
|
||
| @Override | ||
| public final void onCompleted() { | ||
| onComplete.call(); | ||
| } | ||
|
|
||
| @Override | ||
| public final void onError(Throwable e) { | ||
| onError.call(e); | ||
| } | ||
|
|
||
| @Override | ||
| public final void onNext(T args) { | ||
| onNext.call(args); | ||
| } | ||
|
|
||
| }; | ||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.