innerFrom.js
3.84 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
import { __asyncValues, __awaiter } from "tslib";
import { isArrayLike } from '../util/isArrayLike';
import { isPromise } from '../util/isPromise';
import { Observable } from '../Observable';
import { isInteropObservable } from '../util/isInteropObservable';
import { isAsyncIterable } from '../util/isAsyncIterable';
import { createInvalidObservableTypeError } from '../util/throwUnobservableError';
import { isIterable } from '../util/isIterable';
import { isReadableStreamLike, readableStreamLikeToAsyncGenerator } from '../util/isReadableStreamLike';
import { isFunction } from '../util/isFunction';
import { reportUnhandledError } from '../util/reportUnhandledError';
import { observable as Symbol_observable } from '../symbol/observable';
export function innerFrom(input) {
if (input instanceof Observable) {
return input;
}
if (input != null) {
if (isInteropObservable(input)) {
return fromInteropObservable(input);
}
if (isArrayLike(input)) {
return fromArrayLike(input);
}
if (isPromise(input)) {
return fromPromise(input);
}
if (isAsyncIterable(input)) {
return fromAsyncIterable(input);
}
if (isIterable(input)) {
return fromIterable(input);
}
if (isReadableStreamLike(input)) {
return fromReadableStreamLike(input);
}
}
throw createInvalidObservableTypeError(input);
}
export function fromInteropObservable(obj) {
return new Observable((subscriber) => {
const obs = obj[Symbol_observable]();
if (isFunction(obs.subscribe)) {
return obs.subscribe(subscriber);
}
throw new TypeError('Provided object does not correctly implement Symbol.observable');
});
}
export function fromArrayLike(array) {
return new Observable((subscriber) => {
for (let i = 0; i < array.length && !subscriber.closed; i++) {
subscriber.next(array[i]);
}
subscriber.complete();
});
}
export function fromPromise(promise) {
return new Observable((subscriber) => {
promise
.then((value) => {
if (!subscriber.closed) {
subscriber.next(value);
subscriber.complete();
}
}, (err) => subscriber.error(err))
.then(null, reportUnhandledError);
});
}
export function fromIterable(iterable) {
return new Observable((subscriber) => {
for (const value of iterable) {
subscriber.next(value);
if (subscriber.closed) {
return;
}
}
subscriber.complete();
});
}
export function fromAsyncIterable(asyncIterable) {
return new Observable((subscriber) => {
process(asyncIterable, subscriber).catch((err) => subscriber.error(err));
});
}
export function fromReadableStreamLike(readableStream) {
return fromAsyncIterable(readableStreamLikeToAsyncGenerator(readableStream));
}
function process(asyncIterable, subscriber) {
var asyncIterable_1, asyncIterable_1_1;
var e_1, _a;
return __awaiter(this, void 0, void 0, function* () {
try {
for (asyncIterable_1 = __asyncValues(asyncIterable); asyncIterable_1_1 = yield asyncIterable_1.next(), !asyncIterable_1_1.done;) {
const value = asyncIterable_1_1.value;
subscriber.next(value);
if (subscriber.closed) {
return;
}
}
}
catch (e_1_1) { e_1 = { error: e_1_1 }; }
finally {
try {
if (asyncIterable_1_1 && !asyncIterable_1_1.done && (_a = asyncIterable_1.return)) yield _a.call(asyncIterable_1);
}
finally { if (e_1) throw e_1.error; }
}
subscriber.complete();
});
}
//# sourceMappingURL=innerFrom.js.map