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(PairObservable): add PairObservable creation method
closes ReactiveX#1804
- Loading branch information
Showing
5 changed files
with
133 additions
and
0 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 |
---|---|---|
@@ -0,0 +1,43 @@ | ||
import {expect} from 'chai'; | ||
import * as Rx from '../../dist/cjs/Rx'; | ||
|
||
declare const {hot, asDiagram, expectObservable, expectSubscriptions}; | ||
declare const rxTestScheduler: Rx.TestScheduler; | ||
const Observable = Rx.Observable; | ||
|
||
describe('Observable.pairs', () => { | ||
asDiagram('pairs({a: 1, b:2})')('should create an observable emits key-value pair', () => { | ||
const e1 = Observable.pairs({a: 1, b: 2}, rxTestScheduler); | ||
const expected = '(ab|)'; | ||
const values = { | ||
a: ['a', 1], | ||
b: ['b', 2] | ||
}; | ||
|
||
expectObservable(e1).toBe(expected, values); | ||
}); | ||
|
||
it('should create an observable without scheduler', (done: MochaDone) => { | ||
let expected = [ | ||
['a', 1], | ||
['b', 2], | ||
['c', 3] | ||
]; | ||
|
||
Observable.pairs({a: 1, b: 2, c: 3}).subscribe(x => { | ||
expect(x).to.deep.equal(expected.shift()); | ||
}, x => { | ||
done(new Error('should not be called')); | ||
}, () => { | ||
expect(expected).to.be.empty; | ||
done(); | ||
}); | ||
}); | ||
|
||
it('should work with empty object', () => { | ||
const e1 = Observable.pairs({}, rxTestScheduler); | ||
const 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 |
---|---|---|
@@ -0,0 +1,10 @@ | ||
import {Observable} from '../../Observable'; | ||
import {pairs as staticPairs} from '../../observable/pairs'; | ||
|
||
Observable.pairs = staticPairs; | ||
|
||
declare module '../../Observable' { | ||
namespace Observable { | ||
export let pairs: typeof staticPairs; | ||
} | ||
} |
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,76 @@ | ||
import {Scheduler} from '../Scheduler'; | ||
import {Observable} from '../Observable'; | ||
import {Subscriber} from '../Subscriber'; | ||
import {TeardownLogic} from '../Subscription'; | ||
|
||
interface PairsContext<T> { | ||
obj: Object; | ||
keys: Array<string>; | ||
length: number; | ||
index: number; | ||
subscriber: Subscriber<Array<string | T>>; | ||
} | ||
|
||
function dispatch<T>(state: PairsContext<T>) { | ||
const {obj, keys, length, index, subscriber} = state; | ||
|
||
if (index === length) { | ||
subscriber.complete(); | ||
return; | ||
} | ||
|
||
const key = keys[index]; | ||
const value = obj[key]; | ||
subscriber.next([key, value]); | ||
|
||
state.index = index + 1; | ||
|
||
(<any> this).schedule(state); | ||
} | ||
|
||
/** | ||
* We need this JSDoc comment for affecting ESDoc. | ||
* @extends {Ignored} | ||
* @hide true | ||
*/ | ||
export class PairsObservable<T> extends Observable<Array<string | T>> { | ||
private keys: Array<string>; | ||
|
||
/** | ||
* Convert an object into an observable sequence of [key, value] pairs | ||
* using an optional Scheduler to enumerate the object. | ||
* | ||
* @param {Object} obj The object to inspect and turn into an | ||
* Observable sequence. | ||
* @param {Scheduler} [scheduler] An optional Scheduler to run the | ||
* enumeration of the input sequence on. | ||
* @returns {(Observable<Array<string | T>>)} An observable sequence of | ||
* [key, value] pairs from the object. | ||
*/ | ||
static create<T>(obj: Object, scheduler?: Scheduler): Observable<Array<string | T>> { | ||
return new PairsObservable<T>(obj, scheduler); | ||
} | ||
|
||
constructor(private obj: Object, private scheduler?: Scheduler) { | ||
super(); | ||
this.keys = Object.keys(obj); | ||
} | ||
|
||
protected _subscribe(subscriber: Subscriber<Array<string | T>>): TeardownLogic { | ||
const {keys, scheduler} = this; | ||
const length = keys.length; | ||
|
||
if (scheduler) { | ||
return scheduler.schedule(dispatch, 0, { | ||
obj: this.obj, keys, length, index: 0, subscriber | ||
}); | ||
} else { | ||
for (let idx = 0; idx < length; idx++) { | ||
const key = keys[idx]; | ||
const value = this.obj[key]; | ||
subscriber.next([key, value]); | ||
} | ||
subscriber.complete(); | ||
} | ||
} | ||
} |
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 @@ | ||
import { PairsObservable } from './PairsObservable'; | ||
|
||
export const pairs = PairsObservable.create; |