Mirror: 🎩 A tiny but capable push & pull stream library for TypeScript and Flow
chore: Derive common Wonka sources from make source (#114)
* Replace makeSubject with make derivative
* Reuse make for common sources
authored by
Phil Pluckthun
and committed by
GitHub
b8ec16f0
c4373916
+29
-55
+1
src/__tests__/sources.test.ts
+1
src/__tests__/sources.test.ts
+28
-55
src/sources.ts
+28
-55
src/sources.ts
···
1
1
import { Source, Sink, SignalKind, TalkbackKind, Observer, Subject, TeardownFn } from './types';
2
-
import { push, start, talkbackPlaceholder } from './helpers';
2
+
import { push, start, talkbackPlaceholder, teardownPlaceholder } from './helpers';
3
+
import { share } from './operators';
3
4
4
5
export function fromArray<T>(array: T[]): Source<T> {
5
6
return sink => {
···
73
74
}
74
75
75
76
export function makeSubject<T>(): Subject<T> {
76
-
let sinks: Sink<T>[] = [];
77
-
let ended = false;
77
+
let next: Subject<T>['next'] | void;
78
+
let complete: Subject<T>['complete'] | void;
78
79
return {
79
-
source(sink: Sink<T>) {
80
-
sinks.push(sink);
81
-
sink(
82
-
start(signal => {
83
-
if (signal === TalkbackKind.Close) {
84
-
const index = sinks.indexOf(sink);
85
-
if (index > -1) (sinks = sinks.slice()).splice(index, 1);
86
-
}
87
-
})
88
-
);
89
-
},
80
+
source: share(
81
+
make(observer => {
82
+
next = observer.next;
83
+
complete = observer.complete;
84
+
return teardownPlaceholder;
85
+
})
86
+
),
90
87
next(value: T) {
91
-
if (!ended) {
92
-
const signal = push(value);
93
-
for (let i = 0, a = sinks, l = sinks.length; i < l; i++) a[i](signal);
94
-
}
88
+
if (next) next(value);
95
89
},
96
90
complete() {
97
-
if (!ended) {
98
-
ended = true;
99
-
for (let i = 0, a = sinks, l = sinks.length; i < l; i++) a[i](SignalKind.End);
100
-
}
91
+
if (complete) complete();
101
92
},
102
93
};
103
94
}
···
121
112
};
122
113
123
114
export function interval(ms: number): Source<number> {
124
-
return sink => {
115
+
return make(observer => {
125
116
let i = 0;
126
-
const id = setInterval(() => {
127
-
sink(push(i++));
128
-
}, ms);
129
-
sink(
130
-
start(signal => {
131
-
if (signal === TalkbackKind.Close) clearInterval(id);
132
-
})
133
-
);
134
-
};
117
+
const id = setInterval(() => observer.next(i++), ms);
118
+
return () => clearInterval(id);
119
+
});
135
120
}
136
121
137
122
export function fromDomEvent(element: HTMLElement, event: string): Source<Event> {
138
-
return sink => {
139
-
const handler = (payload: Event) => {
140
-
sink(push(payload));
141
-
};
142
-
sink(
143
-
start(signal => {
144
-
if (signal === TalkbackKind.Close) element.removeEventListener(event, handler);
145
-
})
146
-
);
147
-
element.addEventListener(event, handler);
148
-
};
123
+
return make(observer => {
124
+
element.addEventListener(event, observer.next);
125
+
return () => element.removeEventListener(event, observer.next);
126
+
});
149
127
}
150
128
151
129
export function fromPromise<T>(promise: Promise<T>): Source<T> {
152
-
return sink => {
153
-
let ended = false;
130
+
return make(observer => {
154
131
promise.then(value => {
155
-
if (!ended) {
156
-
sink(push(value));
157
-
sink(SignalKind.End);
158
-
}
132
+
Promise.resolve(value).then(() => {
133
+
observer.next(value);
134
+
observer.complete();
135
+
});
159
136
});
160
-
sink(
161
-
start(signal => {
162
-
if (signal === TalkbackKind.Close) ended = true;
163
-
})
164
-
);
165
-
};
137
+
return teardownPlaceholder;
138
+
});
166
139
}