1.x: Merging an observable of singles. #4988
Conversation
Current coverage is 83.54% (diff: 57.57%)@@ 1.x #4988 diff @@
==========================================
Files 288 289 +1
Lines 17806 18235 +429
Methods 0 0
Messages 0 0
Branches 2698 2782 +84
==========================================
+ Hits 15007 15234 +227
- Misses 1950 2096 +146
- Partials 849 905 +56
|
|
Looks like a big copy and paste from an existing merge operator. Can you summarize the differences from the original tried and true operator to help review? I also wonder if we can share a decent chunk of code between operators. |
|
It was a copy for the most part but for changing the class names. It could have be done by mapping the |
| * @see <a href="http://reactivex.io/documentation/operators/merge.html">ReactiveX operators documentation: Merge</a> | ||
| */ | ||
| public static <T> Observable<T> merge(Observable<? extends Single<T>> source) { | ||
| return source.lift(OperatorMergeSingle.<T> instance(false)); |
There was a problem hiding this comment.
This could be
return source.map(new Func1<Single<T>, Observable<T>() {
@Override public Observable<T> call(Single<T> t) {
return t.toObservable();
}
}).lift(OperatorMerge.<T>instance(false));| * @return an Observable that emits all of the items emitted by the source Singles | ||
| * @see <a href="http://reactivex.io/documentation/operators/merge.html">ReactiveX operators documentation: Merge</a> | ||
| */ | ||
| public static <T> Observable<T> merge(Observable<? extends Single<T>> source) { |
| * @return an Observable that emits all of the items emitted by the source Singles | ||
| * @see <a href="http://reactivex.io/documentation/operators/merge.html">ReactiveX operators documentation: Merge</a> | ||
| */ | ||
| public static <T> Observable<T> merge(Observable<? extends Single<T>> source, int maxConcurrent) { |
| * @return an Observable that emits all of the items emitted by the source Singles | ||
| * @see <a href="http://reactivex.io/documentation/operators/merge.html">ReactiveX operators documentation: Merge</a> | ||
| */ | ||
| public static <T> Observable<T> mergeDelayError(Observable<? extends Single<T>> source) { |
| * @return an Observable that emits all of the items emitted by the source Singles | ||
| * @see <a href="http://reactivex.io/documentation/operators/merge.html">ReactiveX operators documentation: Merge</a> | ||
| */ | ||
| public static <T> Observable<T> mergeDelayError(Observable<? extends Single<T>> source, int maxConcurrent) { |
|
@akarnokd actually do you think that the back pressure could be streamlined because of the guarantee that they'll be one onNext for each subscription? Does it even need a custom Producer if we got rid of the maxConcurrent overloads? |
|
The v2 version is optimized. |
You may not want to have a million outstanding |
akarnokd
changed the title
|
Any progress on this? |
|
Closing via #5092. |
The addition of a
Observable<Single<T>> -> Observable<T>to round out the basic API ofrx.Single. I need this for doing a flat scan of sorts.