forked from ReactiveX/rxjs
-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(forkJoin): accept promise, resultselector as parameter of forkJoin
closes ReactiveX#507
- Loading branch information
Showing
5 changed files
with
220 additions
and
32 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
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,14 +1,155 @@ | ||
/* globals describe, it, expect */ | ||
/* globals describe, it, expect, lowerCaseO, hot, expectObservable */ | ||
var Rx = require('../../dist/cjs/Rx'); | ||
var Observable = Rx.Observable; | ||
|
||
describe('Observable.forkJoin', function () { | ||
it('should join the last values of the provided observables into an array', function (done) { | ||
Observable.forkJoin(Observable.of(1, 2, 3, 'a'), | ||
Observable.of('b'), | ||
Observable.of(1, 2, 3, 4, 'c')) | ||
.subscribe(function (x) { | ||
expect(x).toEqual(['a', 'b', 'c']); | ||
}, null, done); | ||
}); | ||
}); | ||
it('should join the last values of the provided observables into an array', function () { | ||
var e1 = Observable.forkJoin( | ||
hot('--a--b--c--d--|'), | ||
hot('(b|)'), | ||
hot('--1--2--3--|') | ||
); | ||
var expected = '--------------(x|)'; | ||
|
||
expectObservable(e1).toBe(expected, {x: ['d', 'b', '3']}); | ||
}); | ||
|
||
it('should accept lowercase-o observables', function () { | ||
var e1 = Observable.forkJoin( | ||
hot('--a--b--c--d--|'), | ||
hot('(b|)'), | ||
lowerCaseO('1', '2', '3') | ||
); | ||
var expected = '--------------(x|)'; | ||
|
||
expectObservable(e1).toBe(expected, {x: ['d', 'b', '3']}); | ||
}); | ||
|
||
it('should accept promise', function (done) { | ||
var e1 = Observable.forkJoin( | ||
Observable.of(1), | ||
Promise.resolve(2) | ||
); | ||
|
||
e1.subscribe(function (x) { | ||
expect(x).toEqual([1,2]); | ||
}, | ||
function (err) { | ||
done.fail('should not be called'); | ||
}, | ||
done); | ||
}); | ||
|
||
it('forkJoin n-ary parameters empty', function () { | ||
var e1 = Observable.forkJoin( | ||
hot('--a--b--c--d--|'), | ||
hot('(b|)'), | ||
hot('------------------|') | ||
); | ||
var expected = '------------------|'; | ||
|
||
expectObservable(e1).toBe(expected); | ||
}); | ||
|
||
it('forkJoin n-ary parameters empty before end', function () { | ||
var e1 = Observable.forkJoin( | ||
hot('--a--b--c--d--|'), | ||
hot('(b|)'), | ||
hot('---------|') | ||
); | ||
var expected = '---------|'; | ||
|
||
expectObservable(e1).toBe(expected); | ||
}); | ||
|
||
it('forkJoin empty empty', function () { | ||
var e1 = Observable.forkJoin( | ||
hot('--------------|'), | ||
hot('---------|') | ||
); | ||
var expected = '---------|'; | ||
|
||
expectObservable(e1).toBe(expected); | ||
}); | ||
|
||
it('forkJoin none', function () { | ||
var e1 = Observable.forkJoin(); | ||
var expected = '|'; | ||
|
||
expectObservable(e1).toBe(expected); | ||
}); | ||
|
||
it('forkJoin empty return', function () { | ||
function selector(x, y) { | ||
return x + y; | ||
} | ||
|
||
var e1 = Observable.forkJoin( | ||
hot('--a--b--c--d--|'), | ||
hot('---------|'), | ||
selector); | ||
var expected = '---------|'; | ||
|
||
expectObservable(e1).toBe(expected); | ||
}); | ||
|
||
it('forkJoin return return', function () { | ||
function selector(x, y) { | ||
return x + y; | ||
} | ||
|
||
var e1 = Observable.forkJoin( | ||
hot('--a--b--c--d--|'), | ||
hot('---2-----|'), | ||
selector); | ||
var expected = '--------------(x|)'; | ||
|
||
expectObservable(e1).toBe(expected, {x: 'd2'}); | ||
}); | ||
|
||
it('forkJoin empty throw', function () { | ||
var e1 = Observable.forkJoin( | ||
hot('------#'), | ||
hot('---------|')); | ||
var expected = '------#'; | ||
|
||
expectObservable(e1).toBe(expected); | ||
}); | ||
|
||
it('forkJoin empty throw', function () { | ||
function selector(x, y) { | ||
return x + y; | ||
} | ||
|
||
var e1 = Observable.forkJoin( | ||
hot('------#'), | ||
hot('---------|'), | ||
selector); | ||
var expected = '------#'; | ||
|
||
expectObservable(e1).toBe(expected); | ||
}); | ||
|
||
it('forkJoin return throw', function () { | ||
var e1 = Observable.forkJoin( | ||
hot('------#'), | ||
hot('---a-----|')); | ||
var expected = '------#'; | ||
|
||
expectObservable(e1).toBe(expected); | ||
}); | ||
|
||
it('forkJoin return throw', function () { | ||
function selector(x, y) { | ||
return x + y; | ||
} | ||
|
||
var e1 = Observable.forkJoin( | ||
hot('------#'), | ||
hot('-------b-|'), | ||
selector); | ||
var expected = '------#'; | ||
|
||
expectObservable(e1).toBe(expected); | ||
}); | ||
}); |
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
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,50 +1,94 @@ | ||
import {Observable} from '../Observable'; | ||
import {Subscriber} from '../Subscriber'; | ||
import {PromiseObservable} from './PromiseObservable'; | ||
import {EmptyObservable} from './EmptyObservable'; | ||
import {isPromise} from '../util/isPromise'; | ||
|
||
export class ForkJoinObservable<T> extends Observable<T> { | ||
constructor(private observables: Observable<any>[]) { | ||
super(); | ||
constructor(private sources: Array<Observable<any> | | ||
Promise<any> | | ||
((...values: Array<any>) => any)>) { | ||
super(); | ||
} | ||
|
||
static create(...sources: Array<Observable<any> | | ||
Promise<any> | | ||
((...values: Array<any>) => any)>) | ||
: Observable<any> { | ||
if (sources === null || sources.length === 0) { | ||
return new EmptyObservable(); | ||
} | ||
return new ForkJoinObservable(sources); | ||
} | ||
|
||
static create<R>(...observables: Observable<any>[]): Observable<R> { | ||
return new ForkJoinObservable(observables); | ||
private getResultSelector(): (...values: Array<any>) => any { | ||
const sources = this.sources; | ||
|
||
let resultSelector = sources[sources.length - 1]; | ||
if (typeof resultSelector !== 'function') { | ||
return null; | ||
} | ||
this.sources.pop(); | ||
return <(...values: Array<any>) => any>resultSelector; | ||
} | ||
|
||
_subscribe(subscriber: Subscriber<any>) { | ||
const observables = this.observables; | ||
const len = observables.length; | ||
let context = { complete: 0, total: len, values: emptyArray(len) }; | ||
let resultSelector = this.getResultSelector(); | ||
const sources = this.sources; | ||
const len = sources.length; | ||
|
||
let context = { completed: 0, total: len, values: emptyArray(len), selector: resultSelector }; | ||
for (let i = 0; i < len; i++) { | ||
observables[i].subscribe(new AllSubscriber(subscriber, this, i, context)); | ||
let source = sources[i]; | ||
if (isPromise(source)) { | ||
source = PromiseObservable.create(<Promise<any>>source); | ||
This comment has been minimized.
Sorry, something went wrong.
This comment has been minimized.
Sorry, something went wrong. |
||
} | ||
(<Observable<any>>source).subscribe(new AllSubscriber(subscriber, i, context)); | ||
} | ||
} | ||
} | ||
|
||
class AllSubscriber<T> extends Subscriber<T> { | ||
private _value: T; | ||
private _value: any = null; | ||
|
||
constructor(destination: Subscriber<T>, | ||
private parent: ForkJoinObservable<T>, | ||
constructor(destination: Subscriber<any>, | ||
private index: number, | ||
private context: { complete: number, total: number, values: any[] }) { | ||
private context: { completed: number, | ||
total: number, | ||
values: any[], | ||
selector: (...values: Array<any>) => any }) { | ||
super(destination); | ||
} | ||
|
||
_next(value: T) { | ||
_next(value: any): void { | ||
this._value = value; | ||
} | ||
|
||
_complete() { | ||
_complete(): void { | ||
if (this._value == null) { | ||
this.destination.complete(); | ||
} | ||
|
||
const destination = this.destination; | ||
This comment has been minimized.
Sorry, something went wrong. |
||
const context = this.context; | ||
context.completed++; | ||
context.values[this.index] = this._value; | ||
This comment has been minimized.
Sorry, something went wrong.
benlesh
|
||
|
||
if (context.completed !== context.values.length) { | ||
return; | ||
} | ||
|
||
if (context.values.every(hasValue)) { | ||
this.destination.next(context.values); | ||
this.destination.complete(); | ||
let value = context.selector ? context.selector.apply(this, context.values) : | ||
context.values; | ||
destination.next(value); | ||
} | ||
|
||
destination.complete(); | ||
} | ||
} | ||
|
||
function hasValue(x) { | ||
function hasValue(x: any): boolean { | ||
return x !== null; | ||
} | ||
|
||
|
@@ -54,4 +98,4 @@ function emptyArray(len: number): any[] { | |
arr.push(null); | ||
} | ||
return arr; | ||
} | ||
} |
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
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,3 @@ | ||
export function isPromise(value: any): boolean { | ||
return value && typeof value.subscribe !== 'function' && typeof value.then === 'function'; | ||
} |
1 comment
on commit 1900a6f
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Great work, @kwonoj! I've added some feedback.
This should be
new PromiseObservable
, it will skip the extra function call, and it may even minify a little better. Same for below.