-
Notifications
You must be signed in to change notification settings - Fork 3k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(scheduled): Add scheduled creation method (#4595)
- Simplifies code around common observable creation and subscription - Removes `scalar` internal impl - Deprecates a number of APIs that accept schedulers where we would rather people use `scheduled`.
- Loading branch information
Showing
27 changed files
with
448 additions
and
269 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,48 @@ | ||
| import * as _ from 'lodash'; | ||
|
|
||
| function stringify(x: any): string { | ||
| return JSON.stringify(x, function (key: string, value: any) { | ||
| if (Array.isArray(value)) { | ||
| return '[' + value | ||
| .map(function (i) { | ||
| return '\n\t' + stringify(i); | ||
| }) + '\n]'; | ||
| } | ||
| return value; | ||
| }) | ||
| .replace(/\\"/g, '"') | ||
| .replace(/\\t/g, '\t') | ||
| .replace(/\\n/g, '\n'); | ||
| } | ||
|
|
||
| function deleteErrorNotificationStack(marble: any) { | ||
| const { notification } = marble; | ||
| if (notification) { | ||
| const { kind, error } = notification; | ||
| if (kind === 'E' && error instanceof Error) { | ||
| notification.error = { name: error.name, message: error.message }; | ||
| } | ||
| } | ||
| return marble; | ||
| } | ||
|
|
||
| export function observableMatcher(actual: any, expected: any) { | ||
| if (Array.isArray(actual) && Array.isArray(expected)) { | ||
| actual = actual.map(deleteErrorNotificationStack); | ||
| expected = expected.map(deleteErrorNotificationStack); | ||
| const passed = _.isEqual(actual, expected); | ||
| if (passed) { | ||
| return; | ||
| } | ||
|
|
||
| let message = '\nExpected \n'; | ||
| actual.forEach((x: any) => message += `\t${stringify(x)}\n`); | ||
|
|
||
| message += '\t\nto deep equal \n'; | ||
| expected.forEach((x: any) => message += `\t${stringify(x)}\n`); | ||
|
|
||
| chai.assert(passed, message); | ||
| } else { | ||
| chai.assert.deepEqual(actual, expected); | ||
| } | ||
| } |
This file was deleted.
Oops, something went wrong.
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,63 @@ | ||
| import { scheduled, of } from 'rxjs'; | ||
| import { TestScheduler } from 'rxjs/testing'; | ||
| import { lowerCaseO } from '../helpers/test-helper'; | ||
| import { observableMatcher } from '../helpers/observableMatcher'; | ||
| import { expect } from 'chai'; | ||
|
|
||
| describe('scheduled', () => { | ||
| let testScheduler: TestScheduler; | ||
|
|
||
| beforeEach(() => { | ||
| testScheduler = new TestScheduler(observableMatcher); | ||
| }); | ||
|
|
||
| it('should schedule a sync observable', () => { | ||
| const input = of('a', 'b', 'c'); | ||
| testScheduler.run(({ expectObservable }) => { | ||
| expectObservable(scheduled(input, testScheduler)).toBe('(abc|)'); | ||
| }); | ||
| }); | ||
|
|
||
| it('should schedule an array', () => { | ||
| const input = ['a', 'b', 'c']; | ||
| testScheduler.run(({ expectObservable }) => { | ||
| expectObservable(scheduled(input, testScheduler)).toBe('(abc|)'); | ||
| }); | ||
| }); | ||
|
|
||
| it('should schedule an iterable', () => { | ||
| const input = 'abc'; // strings are iterables | ||
| testScheduler.run(({ expectObservable }) => { | ||
| expectObservable(scheduled(input, testScheduler)).toBe('(abc|)'); | ||
| }); | ||
| }); | ||
|
|
||
| it('should schedule an observable-like', () => { | ||
| const input = lowerCaseO('a', 'b', 'c'); // strings are iterables | ||
| testScheduler.run(({ expectObservable }) => { | ||
| expectObservable(scheduled(input, testScheduler)).toBe('(abc|)'); | ||
| }); | ||
| }); | ||
|
|
||
| it('should schedule a promise', done => { | ||
| const results: any[] = []; | ||
| const input = Promise.resolve('x'); // strings are iterables | ||
| scheduled(input, testScheduler).subscribe({ | ||
| next(value) { results.push(value); }, | ||
| complete() { results.push('done'); }, | ||
| }); | ||
|
|
||
| expect(results).to.deep.equal([]); | ||
|
|
||
| // Promises force async, so we can't schedule synchronously, no matter what. | ||
| testScheduler.flush(); | ||
| expect(results).to.deep.equal([]); | ||
|
|
||
| Promise.resolve().then(() => { | ||
| // NOW it should work, as the other promise should have resolved. | ||
| testScheduler.flush(); | ||
| expect(results).to.deep.equal(['x', 'done']); | ||
| done(); | ||
| }); | ||
| }); | ||
| }); |
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
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
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
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,26 +1,12 @@ | ||
| import { Observable } from '../Observable'; | ||
| import { SchedulerLike } from '../types'; | ||
| import { Subscription } from '../Subscription'; | ||
| import { subscribeToArray } from '../util/subscribeToArray'; | ||
| import { scheduleArray } from '../scheduled/scheduleArray'; | ||
|
|
||
| export function fromArray<T>(input: ArrayLike<T>, scheduler?: SchedulerLike) { | ||
| if (!scheduler) { | ||
| return new Observable<T>(subscribeToArray(input)); | ||
| } else { | ||
| return new Observable<T>(subscriber => { | ||
| const sub = new Subscription(); | ||
| let i = 0; | ||
| sub.add(scheduler.schedule(function () { | ||
| if (i === input.length) { | ||
| subscriber.complete(); | ||
| return; | ||
| } | ||
| subscriber.next(input[i++]); | ||
| if (!subscriber.closed) { | ||
| sub.add(this.schedule()); | ||
| } | ||
| })); | ||
| return sub; | ||
| }); | ||
| return scheduleArray(input, scheduler); | ||
| } | ||
| } |
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,15 @@ | ||
| import { Observable } from '../Observable'; | ||
| import { SchedulerLike } from '../types'; | ||
| import { Subscription } from '../Subscription'; | ||
| import { iterator as Symbol_iterator } from '../symbol/iterator'; | ||
| import { subscribeToIterable } from '../util/subscribeToIterable'; | ||
| import { scheduleIterable } from '../scheduled/scheduleIterable'; | ||
|
|
||
| export function fromIterable<T>(input: Iterable<T>, scheduler: SchedulerLike) { | ||
| export function fromIterable<T>(input: Iterable<T>, scheduler?: SchedulerLike) { | ||
| if (!input) { | ||
| throw new Error('Iterable cannot be null'); | ||
| } | ||
| if (!scheduler) { | ||
| return new Observable<T>(subscribeToIterable(input)); | ||
| } else { | ||
| return new Observable<T>(subscriber => { | ||
| const sub = new Subscription(); | ||
| let iterator: Iterator<T>; | ||
| sub.add(() => { | ||
| // Finalize generators | ||
| if (iterator && typeof iterator.return === 'function') { | ||
| iterator.return(); | ||
| } | ||
| }); | ||
| sub.add(scheduler.schedule(() => { | ||
| iterator = input[Symbol_iterator](); | ||
| sub.add(scheduler.schedule(function () { | ||
| if (subscriber.closed) { | ||
| return; | ||
| } | ||
| let value: T; | ||
| let done: boolean; | ||
| try { | ||
| const result = iterator.next(); | ||
| value = result.value; | ||
| done = result.done; | ||
| } catch (err) { | ||
| subscriber.error(err); | ||
| return; | ||
| } | ||
| if (done) { | ||
| subscriber.complete(); | ||
| } else { | ||
| subscriber.next(value); | ||
| this.schedule(); | ||
| } | ||
| })); | ||
| })); | ||
| return sub; | ||
| }); | ||
| return scheduleIterable(input, scheduler); | ||
| } | ||
| } |
Oops, something went wrong.