Skip to content

Commit f57e1fc

Browse files
benleshbenlesh
authored
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`.
1 parent 4dd0be0 commit f57e1fc

27 files changed

Lines changed: 448 additions & 269 deletions

‎spec/helpers/observableMatcher.ts‎

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,48 @@
1+
import * as _ from 'lodash';
2+
3+
function stringify(x: any): string {
4+
return JSON.stringify(x, function (key: string, value: any) {
5+
if (Array.isArray(value)) {
6+
return '[' + value
7+
.map(function (i) {
8+
return '\n\t' + stringify(i);
9+
}) + '\n]';
10+
}
11+
return value;
12+
})
13+
.replace(/\\"/g, '"')
14+
.replace(/\\t/g, '\t')
15+
.replace(/\\n/g, '\n');
16+
}
17+
18+
function deleteErrorNotificationStack(marble: any) {
19+
const { notification } = marble;
20+
if (notification) {
21+
const { kind, error } = notification;
22+
if (kind === 'E' && error instanceof Error) {
23+
notification.error = { name: error.name, message: error.message };
24+
}
25+
}
26+
return marble;
27+
}
28+
29+
export function observableMatcher(actual: any, expected: any) {
30+
if (Array.isArray(actual) && Array.isArray(expected)) {
31+
actual = actual.map(deleteErrorNotificationStack);
32+
expected = expected.map(deleteErrorNotificationStack);
33+
const passed = _.isEqual(actual, expected);
34+
if (passed) {
35+
return;
36+
}
37+
38+
let message = '\nExpected \n';
39+
actual.forEach((x: any) => message += `\t${stringify(x)}\n`);
40+
41+
message += '\t\nto deep equal \n';
42+
expected.forEach((x: any) => message += `\t${stringify(x)}\n`);
43+
44+
chai.assert(passed, message);
45+
} else {
46+
chai.assert.deepEqual(actual, expected);
47+
}
48+
}

‎spec/observables/ScalarObservable-spec.ts‎

Lines changed: 0 additions & 17 deletions
This file was deleted.

‎spec/observables/of-spec.ts‎

Lines changed: 2 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
import { expect } from 'chai';
2-
import { empty, of, Observable } from 'rxjs';
2+
import { of, Observable } from 'rxjs';
33
import { expectObservable } from '../helpers/marble-testing';
44
import { TestScheduler } from 'rxjs/testing';
55
import { concatMap, delay, concatAll } from 'rxjs/operators';
@@ -33,18 +33,13 @@ describe('of', () => {
3333
});
3434
});
3535

36-
it('should return an empty observable if passed no values', () => {
37-
const obs = of();
38-
expect(obs).to.equal(empty());
39-
});
40-
4136
it('should emit one value', (done: MochaDone) => {
4237
let calls = 0;
4338

4439
of(42).subscribe((x: number) => {
4540
expect(++calls).to.equal(1);
4641
expect(x).to.equal(42);
47-
}, (err: any) => {
42+
}, (err: any) => {
4843
done(new Error('should not be called'));
4944
}, () => {
5045
done();

‎spec/scheduled/scheduled-spec.ts‎

Lines changed: 63 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,63 @@
1+
import { scheduled, of } from 'rxjs';
2+
import { TestScheduler } from 'rxjs/testing';
3+
import { lowerCaseO } from '../helpers/test-helper';
4+
import { observableMatcher } from '../helpers/observableMatcher';
5+
import { expect } from 'chai';
6+
7+
describe('scheduled', () => {
8+
let testScheduler: TestScheduler;
9+
10+
beforeEach(() => {
11+
testScheduler = new TestScheduler(observableMatcher);
12+
});
13+
14+
it('should schedule a sync observable', () => {
15+
const input = of('a', 'b', 'c');
16+
testScheduler.run(({ expectObservable }) => {
17+
expectObservable(scheduled(input, testScheduler)).toBe('(abc|)');
18+
});
19+
});
20+
21+
it('should schedule an array', () => {
22+
const input = ['a', 'b', 'c'];
23+
testScheduler.run(({ expectObservable }) => {
24+
expectObservable(scheduled(input, testScheduler)).toBe('(abc|)');
25+
});
26+
});
27+
28+
it('should schedule an iterable', () => {
29+
const input = 'abc'; // strings are iterables
30+
testScheduler.run(({ expectObservable }) => {
31+
expectObservable(scheduled(input, testScheduler)).toBe('(abc|)');
32+
});
33+
});
34+
35+
it('should schedule an observable-like', () => {
36+
const input = lowerCaseO('a', 'b', 'c'); // strings are iterables
37+
testScheduler.run(({ expectObservable }) => {
38+
expectObservable(scheduled(input, testScheduler)).toBe('(abc|)');
39+
});
40+
});
41+
42+
it('should schedule a promise', done => {
43+
const results: any[] = [];
44+
const input = Promise.resolve('x'); // strings are iterables
45+
scheduled(input, testScheduler).subscribe({
46+
next(value) { results.push(value); },
47+
complete() { results.push('done'); },
48+
});
49+
50+
expect(results).to.deep.equal([]);
51+
52+
// Promises force async, so we can't schedule synchronously, no matter what.
53+
testScheduler.flush();
54+
expect(results).to.deep.equal([]);
55+
56+
Promise.resolve().then(() => {
57+
// NOW it should work, as the other promise should have resolved.
58+
testScheduler.flush();
59+
expect(results).to.deep.equal(['x', 'done']);
60+
done();
61+
});
62+
});
63+
});

‎spec/util/subscribeToResult-spec.ts‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -6,15 +6,15 @@ import $$symbolObservable from 'symbol-observable';
66
import { of, range, throwError } from 'rxjs';
77

88
describe('subscribeToResult', () => {
9-
it('should synchronously complete when subscribe to scalarObservable', () => {
9+
it('should synchronously complete when subscribed to scalarObservable', () => {
1010
const result = of(42);
1111
let expected: number;
1212
const subscriber = new OuterSubscriber<number, number>((x) => expected = x);
1313

1414
const subscription = subscribeToResult(subscriber, result);
1515

1616
expect(expected).to.be.equal(42);
17-
expect(subscription).to.not.exist;
17+
expect(subscription.closed).to.be.true;
1818
});
1919

2020
it('should subscribe to observables that are an instanceof Observable', (done) => {

‎src/index.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,7 @@ export { throwError } from './internal/observable/throwError';
6464
export { timer } from './internal/observable/timer';
6565
export { using } from './internal/observable/using';
6666
export { zip } from './internal/observable/zip';
67+
export { scheduled } from './internal/scheduled/scheduled';
6768

6869
/* Constants */
6970
export { EMPTY } from './internal/observable/empty';

‎src/internal/Observable.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@ import { canReportError } from './util/canReportError';
66
import { toSubscriber } from './util/toSubscriber';
77
import { iif } from './observable/iif';
88
import { throwError } from './observable/throwError';
9-
import { observable as Symbol_observable } from '../internal/symbol/observable';
9+
import { observable as Symbol_observable } from './symbol/observable';
1010
import { pipeFromArray } from './util/pipe';
1111
import { config } from './config';
1212

‎src/internal/observable/concat.ts‎

Lines changed: 23 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -6,13 +6,30 @@ import { from } from './from';
66
import { concatAll } from '../operators/concatAll';
77

88
/* tslint:disable:max-line-length */
9-
export function concat<O1 extends ObservableInput<any>>(v1: O1, scheduler?: SchedulerLike): Observable<ObservedValueOf<O1>>;
10-
export function concat<O1 extends ObservableInput<any>, O2 extends ObservableInput<any>>(v1: O1, v2: O2, scheduler?: SchedulerLike): Observable<ObservedValueOf<O1> | ObservedValueOf<O2>>;
11-
export function concat<O1 extends ObservableInput<any>, O2 extends ObservableInput<any>, O3 extends ObservableInput<any>>(v1: O1, v2: O2, v3: O3, scheduler?: SchedulerLike): Observable<ObservedValueOf<O1> | ObservedValueOf<O2> | ObservedValueOf<O3>>;
12-
export function concat<O1 extends ObservableInput<any>, O2 extends ObservableInput<any>, O3 extends ObservableInput<any>, O4 extends ObservableInput<any>>(v1: O1, v2: O2, v3: O3, v4: O4, scheduler?: SchedulerLike): Observable<ObservedValueOf<O1> | ObservedValueOf<O2> | ObservedValueOf<O3> | ObservedValueOf<O4>>;
13-
export function concat<O1 extends ObservableInput<any>, O2 extends ObservableInput<any>, O3 extends ObservableInput<any>, O4 extends ObservableInput<any>, O5 extends ObservableInput<any>>(v1: O1, v2: O2, v3: O3, v4: O4, v5: O5, scheduler?: SchedulerLike): Observable<ObservedValueOf<O1> | ObservedValueOf<O2> | ObservedValueOf<O3> | ObservedValueOf<O4> | ObservedValueOf<O5>>;
14-
export function concat<O1 extends ObservableInput<any>, O2 extends ObservableInput<any>, O3 extends ObservableInput<any>, O4 extends ObservableInput<any>, O5 extends ObservableInput<any>, O6 extends ObservableInput<any>>(v1: O1, v2: O2, v3: O3, v4: O4, v5: O5, v6: O6, scheduler?: SchedulerLike): Observable<ObservedValueOf<O1> | ObservedValueOf<O2> | ObservedValueOf<O3> | ObservedValueOf<O4> | ObservedValueOf<O5> | ObservedValueOf<O6>>;
9+
/** @deprecated Use {@link scheduled} and {@link concatAll} (e.g. `scheduled([o1, o2, o3], scheduler).pipe(concatAll())`) */
10+
export function concat<O1 extends ObservableInput<any>>(v1: O1, scheduler: SchedulerLike): Observable<ObservedValueOf<O1>>;
11+
/** @deprecated Use {@link scheduled} and {@link concatAll} (e.g. `scheduled([o1, o2, o3], scheduler).pipe(concatAll())`) */
12+
export function concat<O1 extends ObservableInput<any>, O2 extends ObservableInput<any>>(v1: O1, v2: O2, scheduler: SchedulerLike): Observable<ObservedValueOf<O1> | ObservedValueOf<O2>>;
13+
/** @deprecated Use {@link scheduled} and {@link concatAll} (e.g. `scheduled([o1, o2, o3], scheduler).pipe(concatAll())`) */
14+
export function concat<O1 extends ObservableInput<any>, O2 extends ObservableInput<any>, O3 extends ObservableInput<any>>(v1: O1, v2: O2, v3: O3, scheduler: SchedulerLike): Observable<ObservedValueOf<O1> | ObservedValueOf<O2> | ObservedValueOf<O3>>;
15+
/** @deprecated Use {@link scheduled} and {@link concatAll} (e.g. `scheduled([o1, o2, o3], scheduler).pipe(concatAll())`) */
16+
export function concat<O1 extends ObservableInput<any>, O2 extends ObservableInput<any>, O3 extends ObservableInput<any>, O4 extends ObservableInput<any>>(v1: O1, v2: O2, v3: O3, v4: O4, scheduler: SchedulerLike): Observable<ObservedValueOf<O1> | ObservedValueOf<O2> | ObservedValueOf<O3> | ObservedValueOf<O4>>;
17+
/** @deprecated Use {@link scheduled} and {@link concatAll} (e.g. `scheduled([o1, o2, o3], scheduler).pipe(concatAll())`) */
18+
export function concat<O1 extends ObservableInput<any>, O2 extends ObservableInput<any>, O3 extends ObservableInput<any>, O4 extends ObservableInput<any>, O5 extends ObservableInput<any>>(v1: O1, v2: O2, v3: O3, v4: O4, v5: O5, scheduler: SchedulerLike): Observable<ObservedValueOf<O1> | ObservedValueOf<O2> | ObservedValueOf<O3> | ObservedValueOf<O4> | ObservedValueOf<O5>>;
19+
/** @deprecated Use {@link scheduled} and {@link concatAll} (e.g. `scheduled([o1, o2, o3], scheduler).pipe(concatAll())`) */
20+
export function concat<O1 extends ObservableInput<any>, O2 extends ObservableInput<any>, O3 extends ObservableInput<any>, O4 extends ObservableInput<any>, O5 extends ObservableInput<any>, O6 extends ObservableInput<any>>(v1: O1, v2: O2, v3: O3, v4: O4, v5: O5, v6: O6, scheduler: SchedulerLike): Observable<ObservedValueOf<O1> | ObservedValueOf<O2> | ObservedValueOf<O3> | ObservedValueOf<O4> | ObservedValueOf<O5> | ObservedValueOf<O6>>;
21+
22+
export function concat<O1 extends ObservableInput<any>>(v1: O1): Observable<ObservedValueOf<O1>>;
23+
export function concat<O1 extends ObservableInput<any>, O2 extends ObservableInput<any>>(v1: O1, v2: O2): Observable<ObservedValueOf<O1> | ObservedValueOf<O2>>;
24+
export function concat<O1 extends ObservableInput<any>, O2 extends ObservableInput<any>, O3 extends ObservableInput<any>>(v1: O1, v2: O2, v3: O3): Observable<ObservedValueOf<O1> | ObservedValueOf<O2> | ObservedValueOf<O3>>;
25+
export function concat<O1 extends ObservableInput<any>, O2 extends ObservableInput<any>, O3 extends ObservableInput<any>, O4 extends ObservableInput<any>>(v1: O1, v2: O2, v3: O3, v4: O4): Observable<ObservedValueOf<O1> | ObservedValueOf<O2> | ObservedValueOf<O3> | ObservedValueOf<O4>>;
26+
export function concat<O1 extends ObservableInput<any>, O2 extends ObservableInput<any>, O3 extends ObservableInput<any>, O4 extends ObservableInput<any>, O5 extends ObservableInput<any>>(v1: O1, v2: O2, v3: O3, v4: O4, v5: O5): Observable<ObservedValueOf<O1> | ObservedValueOf<O2> | ObservedValueOf<O3> | ObservedValueOf<O4> | ObservedValueOf<O5>>;
27+
export function concat<O1 extends ObservableInput<any>, O2 extends ObservableInput<any>, O3 extends ObservableInput<any>, O4 extends ObservableInput<any>, O5 extends ObservableInput<any>, O6 extends ObservableInput<any>>(v1: O1, v2: O2, v3: O3, v4: O4, v5: O5, v6: O6): Observable<ObservedValueOf<O1> | ObservedValueOf<O2> | ObservedValueOf<O3> | ObservedValueOf<O4> | ObservedValueOf<O5> | ObservedValueOf<O6>>;
28+
export function concat<O extends ObservableInput<any>>(...observables: O[]): Observable<ObservedValueOf<O>>;
29+
/** @deprecated Use {@link scheduled} and {@link concatAll} (e.g. `scheduled([o1, o2, o3], scheduler).pipe(concatAll())`) */
1530
export function concat<O extends ObservableInput<any>>(...observables: (O | SchedulerLike)[]): Observable<ObservedValueOf<O>>;
31+
export function concat<R>(...observables: ObservableInput<any>[]): Observable<R>;
32+
/** @deprecated Use {@link scheduled} and {@link concatAll} (e.g. `scheduled([o1, o2, o3], scheduler).pipe(concatAll())`) */
1633
export function concat<R>(...observables: (ObservableInput<any> | SchedulerLike)[]): Observable<R>;
1734
/* tslint:enable:max-line-length */
1835
/**

‎src/internal/observable/empty.ts‎

Lines changed: 4 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -53,19 +53,16 @@ export const EMPTY = new Observable<never>(subscriber => subscriber.complete());
5353
* @see {@link of}
5454
* @see {@link throwError}
5555
*
56-
* @param {SchedulerLike} [scheduler] A {@link SchedulerLike} to use for scheduling
56+
* @param scheduler A {@link SchedulerLike} to use for scheduling
5757
* the emission of the complete notification.
58-
* @return {Observable} An "empty" Observable: emits only the complete
58+
* @return An "empty" Observable: emits only the complete
5959
* notification.
60-
* @static true
61-
* @name empty
62-
* @owner Observable
63-
* @deprecated Deprecated in favor of using {@link index/EMPTY} constant.
60+
* @deprecated Deprecated in favor of using {@link EMPTY} constant, or {@link scheduled} (e.g. `scheduled([], scheduler)`)
6461
*/
6562
export function empty(scheduler?: SchedulerLike) {
6663
return scheduler ? emptyScheduled(scheduler) : EMPTY;
6764
}
6865

69-
export function emptyScheduled(scheduler: SchedulerLike) {
66+
function emptyScheduled(scheduler: SchedulerLike) {
7067
return new Observable<never>(subscriber => scheduler.schedule(() => subscriber.complete()));
7168
}

‎src/internal/observable/from.ts‎

Lines changed: 6 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -1,16 +1,11 @@
11
import { Observable } from '../Observable';
2-
import { isPromise } from '../util/isPromise';
3-
import { isArrayLike } from '../util/isArrayLike';
4-
import { isInteropObservable } from '../util/isInteropObservable';
5-
import { isIterable } from '../util/isIterable';
6-
import { fromArray } from './fromArray';
7-
import { fromPromise } from './fromPromise';
8-
import { fromIterable } from './fromIterable';
9-
import { fromObservable } from './fromObservable';
102
import { subscribeTo } from '../util/subscribeTo';
113
import { ObservableInput, SchedulerLike, ObservedValueOf } from '../types';
4+
import { scheduled } from '../scheduled/scheduled';
125

13-
export function from<O extends ObservableInput<any>>(input: O, scheduler?: SchedulerLike): Observable<ObservedValueOf<O>>;
6+
export function from<O extends ObservableInput<any>>(input: O): Observable<ObservedValueOf<O>>;
7+
/** @deprecated use {@link scheduled} instead. */
8+
export function from<O extends ObservableInput<any>>(input: O, scheduler: SchedulerLike): Observable<ObservedValueOf<O>>;
149

1510
/**
1611
* Creates an Observable from an Array, an array-like object, a Promise, an iterable object, or an Observable-like object.
@@ -111,26 +106,13 @@ export function from<O extends ObservableInput<any>>(input: O, scheduler?: Sched
111106
* @name from
112107
* @owner Observable
113108
*/
114-
115109
export function from<T>(input: ObservableInput<T>, scheduler?: SchedulerLike): Observable<T> {
116110
if (!scheduler) {
117111
if (input instanceof Observable) {
118112
return input;
119113
}
120114
return new Observable<T>(subscribeTo(input));
115+
} else {
116+
return scheduled(input, scheduler);
121117
}
122-
123-
if (input != null) {
124-
if (isInteropObservable(input)) {
125-
return fromObservable(input, scheduler);
126-
} else if (isPromise(input)) {
127-
return fromPromise(input, scheduler);
128-
} else if (isArrayLike(input)) {
129-
return fromArray(input, scheduler);
130-
} else if (isIterable(input) || typeof input === 'string') {
131-
return fromIterable(input, scheduler);
132-
}
133-
}
134-
135-
throw new TypeError((input !== null && typeof input || input) + ' is not observable');
136118
}

0 commit comments

Comments
 (0)