Mirror: 🎩 A tiny but capable push & pull stream library for TypeScript and Flow
feat: Completely fulfill the Observable spec in toObservable (#132)
authored by
Phil Pluckthun
and committed by
GitHub
5517a763
6c1413a9
+32
-8
+5
.changeset/seven-meals-film.md
+5
.changeset/seven-meals-film.md
+27
-8
src/observable.ts
+27
-8
src/observable.ts
···
8
8
9
9
interface ObservableObserver<T> {
10
10
next(value: T): void;
11
-
error(error: any): void;
12
-
complete(): void;
11
+
error?(error: any): void;
12
+
complete?(): void;
13
+
}
14
+
15
+
interface ObservableLike<T> {
16
+
subscribe(observer: ObservableObserver<T>): ObservableSubscription;
17
+
[Symbol.observable]?(): Observable<T>;
13
18
}
14
19
15
20
interface Observable<T> {
16
21
subscribe(observer: ObservableObserver<T>): ObservableSubscription;
22
+
23
+
subscribe(
24
+
onNext: (value: T) => any,
25
+
onError?: (error: any) => any,
26
+
onComplete?: () => any
27
+
): ObservableSubscription;
28
+
29
+
[Symbol.observable](): Observable<T>;
17
30
}
18
31
19
-
const observableSymbol = (): symbol | string => Symbol.observable || '@@observable';
32
+
const observableSymbol = (): typeof Symbol.observable => Symbol.observable || '@@observable';
20
33
21
-
export function fromObservable<T>(input: Observable<T>): Source<T> {
34
+
export function fromObservable<T>(input: ObservableLike<T>): Source<T> {
22
35
input = input[observableSymbol()] ? (input as any)[observableSymbol()]() : input;
23
36
return sink => {
24
37
const subscription = input.subscribe({
···
28
41
complete() {
29
42
sink(SignalKind.End);
30
43
},
31
-
error() {
32
-
/*noop*/
44
+
error(error) {
45
+
throw error;
33
46
},
34
47
});
35
48
sink(
···
42
55
43
56
export function toObservable<T>(source: Source<T>): Observable<T> {
44
57
return {
45
-
subscribe(observer: ObservableObserver<T>) {
58
+
subscribe(
59
+
next: ObservableObserver<T> | ((value: T) => any),
60
+
error?: (error: any) => any | undefined,
61
+
complete?: () => any | undefined
62
+
) {
63
+
const observer: ObservableObserver<T> =
64
+
typeof next == 'object' ? next : { next, error, complete };
46
65
let talkback = talkbackPlaceholder;
47
66
let ended = false;
48
67
source(signal => {
···
50
69
/*noop*/
51
70
} else if (signal === SignalKind.End) {
52
71
ended = true;
53
-
observer.complete();
72
+
if (observer.complete) observer.complete();
54
73
} else if (signal.tag === SignalKind.Start) {
55
74
(talkback = signal[0])(TalkbackKind.Pull);
56
75
} else {