kitten.sh / wonka
Mirror: 🎩 A tiny but capable push & pull stream library for TypeScript and Flow
0
fork

Configure Feed

Select the types of activity you want to include in your feed.

Apply eslint/prettier fixes

Phil Pluckthun a4e79fe3 0c08fc08

+698 -751
+2 -2
gatsby-config.js
··· 3 3 title: 'Wonka', 4 4 description: 'A fast push & pull stream library for Reason, OCaml, and TypeScript', 5 5 siteUrl: 'https://wonka.kitten.sh', 6 - githubUrl: 'https://github.com/kitten/wonka' 6 + githubUrl: 'https://github.com/kitten/wonka', 7 7 }, 8 8 plugins: ['gatsby-plugin-netlify'], 9 - __experimentalThemes: ['gatsby-theme-docs-system'] 9 + __experimentalThemes: ['gatsby-theme-docs-system'], 10 10 };
+4 -4
gatsby/theme.js
··· 13 13 'Apple Color Emoji', 14 14 'Segoe UI Emoji', 15 15 'Segoe UI Symbol', 16 - 'Noto Color Emoji' 16 + 'Noto Color Emoji', 17 17 ]; 18 18 19 19 export const fonts = { 20 20 header: ['phantom-sans', ...systemFonts], 21 21 code: ['space-mono', 'monospace'], 22 - body: systemFonts 22 + body: systemFonts, 23 23 }; 24 24 25 25 export const colors = { ··· 29 29 fg: '#36313d', 30 30 fgHeading: '#000000', 31 31 fgPassive: '#78757a', 32 - fgActive: '#f5735f' 32 + fgActive: '#f5735f', 33 33 }; 34 34 35 35 export const prismTheme = nightOwlLight; ··· 46 46 export const borders = [ 47 47 `${borderWidths[0]} solid ${colors.bgPassive}`, 48 48 `${borderWidths[1]} solid ${colors.bgPassive}`, 49 - `${borderWidths[2]} solid ${colors.bgPassive}` 49 + `${borderWidths[2]} solid ${colors.bgPassive}`, 50 50 ];
+1 -1
gatsby/typography.js
··· 9 9 headerFontFamily: theme.fonts.header, 10 10 headerWeight: '500', 11 11 bodyFontFamily: theme.fonts.body, 12 - bodyWeight: '400' 12 + bodyWeight: '400', 13 13 }); 14 14 15 15 export default typography;
+7 -13
rollup.config.js
··· 22 22 23 23 typescript({ 24 24 typescript: require('typescript'), 25 - exclude: [ 26 - 'src/**/*.test.ts', 27 - '**/__tests__/*', 28 - ], 25 + exclude: ['src/**/*.test.ts', '**/__tests__/*'], 29 26 compilerOptions: { 30 27 sourceMap: true, 31 28 noEmit: false, ··· 49 46 templateString: false, 50 47 objectRestSpread: false, 51 48 }, 52 - exclude: 'node_modules/**' 49 + exclude: 'node_modules/**', 53 50 }), 54 51 55 52 terser({ ··· 68 65 sequences: false, 69 66 loops: false, 70 67 conditionals: false, 71 - join_vars: false 68 + join_vars: false, 72 69 }, 73 70 mangle: { 74 71 module: true, ··· 77 74 output: { 78 75 beautify: true, 79 76 braces: true, 80 - indent_level: 2 81 - } 77 + indent_level: 2, 78 + }, 82 79 }), 83 80 ]; 84 81 85 - const output = (format) => { 82 + const output = format => { 86 83 const extension = format === 'esm' ? '.mjs' : '.js'; 87 84 return { 88 85 chunkFileNames: '[hash]' + extension, ··· 119 116 tryCatchDeoptimization: false, 120 117 moduleSideEffects: false, 121 118 }, 122 - output: [ 123 - output('esm'), 124 - output('cjs'), 125 - ], 119 + output: [output('esm'), output('cjs')], 126 120 }; 127 121 128 122 export default config;
+4 -6
scripts/eslint-preset.js
··· 13 13 'sort-keys': 'off', 14 14 'no-console': ['error', { allow: ['warn', 'error'] }], 15 15 'prefer-arrow/prefer-arrow-functions': 'off', 16 + 'prefer-rest-params': 'off', 16 17 17 18 'prettier/prettier': [ 18 19 'error', ··· 55 56 '@typescript-eslint/no-misused-new': 'off', 56 57 '@typescript-eslint/no-explicit-any': 'off', 57 58 '@typescript-eslint/array-type': 'off', 58 - '@typescript-eslint/no-unused-vars': [ 59 - 'error', 60 - { 61 - argsIgnorePattern: '^_', 62 - }, 63 - ], 59 + '@typescript-eslint/no-empty-function': 'off', 60 + '@typescript-eslint/no-unused-vars': 'off', 61 + 'prefer-rest-params': 'off', 64 62 }, 65 63 }, 66 64 ],
+2 -2
scripts/flow-typings-plugin.js
··· 5 5 6 6 function flowTypings() { 7 7 return { 8 - name: "flow-typings", 8 + name: 'flow-typings', 9 9 async writeBundle() { 10 10 const cwd = process.cwd(); 11 11 for (const file of glob('dist/types/**/*.d.ts')) { ··· 17 17 const definition = flowdef.replace(/import/g, 'import type'); 18 18 writeFileSync(newpath, '// @flow\n\n' + definition); 19 19 } 20 - } 20 + }, 21 21 }; 22 22 } 23 23
+122 -164
src/__tests__/operators.test.ts
··· 9 9 A Pull will be sent from the sink upwards and should pass through 10 10 the operator until the source receives it, which then pushes a 11 11 value down. */ 12 - const passesPassivePull = ( 13 - operator: Operator<any, any>, 14 - output: any = 0 15 - ) => { 12 + const passesPassivePull = (operator: Operator<any, any>, output: any = 0) => { 16 13 it('responds to Pull talkback signals (spec)', () => { 17 14 let talkback: TalkbackFn | null = null; 18 15 let pushes = 0; 19 16 const values: any[] = []; 20 17 21 18 const source: Source<any> = sink => { 22 - sink(start((signal) => { 23 - if (!pushes && signal === TalkbackKind.Pull) { 24 - pushes++; 25 - sink(push(0)); 26 - } 27 - })); 19 + sink( 20 + start(signal => { 21 + if (!pushes && signal === TalkbackKind.Pull) { 22 + pushes++; 23 + sink(push(0)); 24 + } 25 + }) 26 + ); 28 27 }; 29 28 30 - const sink: Sink<any> = (signal) => { 29 + const sink: Sink<any> = signal => { 31 30 expect(signal).not.toBe(SignalKind.End); 32 31 if (signal === SignalKind.End) { 33 32 /*noop*/ ··· 55 54 A Push will be sent downwards from the source, through the 56 55 operator to the sink. Pull events should be let through from 57 56 the sink after every Push event. */ 58 - const passesActivePush = ( 59 - operator: Operator<any, any>, 60 - result: any = 0 61 - ) => { 57 + const passesActivePush = (operator: Operator<any, any>, result: any = 0) => { 62 58 it('responds to eager Push signals (spec)', () => { 63 59 const values: any[] = []; 64 60 let talkback: TalkbackFn | null = null; 65 61 let sink: Sink<any> | null = null; 66 62 let pulls = 0; 67 63 68 - const source: Source<any> = (_sink) => { 69 - (sink = _sink)(start((signal) => { 70 - if (signal === TalkbackKind.Pull) 71 - pulls++; 72 - })); 64 + const source: Source<any> = _sink => { 65 + (sink = _sink)( 66 + start(signal => { 67 + if (signal === TalkbackKind.Pull) pulls++; 68 + }) 69 + ); 73 70 }; 74 71 75 - operator(source)((signal) => { 72 + operator(source)(signal => { 76 73 expect(signal).not.toBe(SignalKind.End); 77 74 if (signal === SignalKind.End) { 78 75 /*noop*/ ··· 105 102 let closing = 0; 106 103 107 104 const source: Source<any> = sink => { 108 - sink(start((signal) => { 109 - if (signal === TalkbackKind.Pull && !closing) { 110 - sink(push(0)); 111 - } else if (signal === TalkbackKind.Close) { 112 - closing++; 113 - } 114 - })); 105 + sink( 106 + start(signal => { 107 + if (signal === TalkbackKind.Pull && !closing) { 108 + sink(push(0)); 109 + } else if (signal === TalkbackKind.Close) { 110 + closing++; 111 + } 112 + }) 113 + ); 115 114 }; 116 115 117 116 const sink: Sink<any> = signal => { ··· 138 137 A Push and End signal will be sent after the first Pull talkback 139 138 signal from the sink, which shouldn't lead to any extra Close or Pull 140 139 talkback signals. */ 141 - const passesSourceEnd = ( 142 - operator: Operator<any, any>, 143 - result: any = 0 144 - ) => { 140 + const passesSourceEnd = (operator: Operator<any, any>, result: any = 0) => { 145 141 it('passes on immediate Push then End signals from source (spec)', () => { 146 142 const signals: Signal<any>[] = []; 147 143 let talkback: TalkbackFn | null = null; ··· 149 145 let ending = 0; 150 146 151 147 const source: Source<any> = sink => { 152 - sink(start((signal) => { 153 - expect(signal).not.toBe(TalkbackKind.Close); 154 - if (signal === TalkbackKind.Pull) { 155 - pulls++; 156 - if (pulls === 1) { 157 - sink(push(0)); 158 - sink(SignalKind.End); 148 + sink( 149 + start(signal => { 150 + expect(signal).not.toBe(TalkbackKind.Close); 151 + if (signal === TalkbackKind.Pull) { 152 + pulls++; 153 + if (pulls === 1) { 154 + sink(push(0)); 155 + sink(SignalKind.End); 156 + } 159 157 } 160 - } 161 - })); 158 + }) 159 + ); 162 160 }; 163 161 164 162 const sink: Sink<any> = signal => { ··· 188 186 after the first pull in response to another. 189 187 This is similar to passesSourceEnd but more well behaved since 190 188 mergeMap/switchMap/concatMap are eager operators. */ 191 - const passesSourcePushThenEnd = ( 192 - operator: Operator<any, any>, 193 - result: any = 0 194 - ) => { 189 + const passesSourcePushThenEnd = (operator: Operator<any, any>, result: any = 0) => { 195 190 it('passes on End signals from source (spec)', () => { 196 191 const signals: Signal<any>[] = []; 197 192 let talkback: TalkbackFn | null = null; ··· 199 194 let ending = 0; 200 195 201 196 const source: Source<any> = sink => { 202 - sink(start((signal) => { 203 - expect(signal).not.toBe(TalkbackKind.Close); 204 - if (signal === TalkbackKind.Pull) { 205 - pulls++; 206 - if (pulls <= 2) { sink(push(0)); } 207 - else { sink(SignalKind.End); } 208 - } 209 - })); 197 + sink( 198 + start(signal => { 199 + expect(signal).not.toBe(TalkbackKind.Close); 200 + if (signal === TalkbackKind.Pull) { 201 + pulls++; 202 + if (pulls <= 2) { 203 + sink(push(0)); 204 + } else { 205 + sink(SignalKind.End); 206 + } 207 + } 208 + }) 209 + ); 210 210 }; 211 211 212 212 const sink: Sink<any> = signal => { ··· 228 228 jest.runAllTimers(); 229 229 expect(ending).toBe(1); 230 230 expect(pulls).toBe(3); 231 - expect(signals).toEqual([ 232 - push(result), 233 - push(result), 234 - SignalKind.End 235 - ]); 231 + expect(signals).toEqual([push(result), push(result), SignalKind.End]); 236 232 }); 237 233 }; 238 234 ··· 272 268 const signals: Signal<any>[] = []; 273 269 274 270 const source: Source<any> = sink => { 275 - sink(start((signal) => { 276 - if (signal === TalkbackKind.Pull) { 277 - pulls++; 278 - sink(SignalKind.End); 279 - sink(push(123)); 280 - } 281 - })); 271 + sink( 272 + start(signal => { 273 + if (signal === TalkbackKind.Pull) { 274 + pulls++; 275 + sink(SignalKind.End); 276 + sink(push(123)); 277 + } 278 + }) 279 + ); 282 280 }; 283 281 284 282 const sink: Sink<any> = signal => { ··· 303 301 const signals: Signal<any>[] = []; 304 302 305 303 const source: Source<any> = sink => { 306 - sink(start((signal) => { 307 - if (signal === TalkbackKind.Close) { 308 - sink(push(123)); 309 - } 310 - })); 304 + sink( 305 + start(signal => { 306 + if (signal === TalkbackKind.Close) { 307 + sink(push(123)); 308 + } 309 + }) 310 + ); 311 311 }; 312 312 313 313 const sink: Sink<any> = signal => { ··· 339 339 let ending = 0; 340 340 341 341 const source: Source<any> = sink => { 342 - sink(start((signal) => { 343 - // For some operator tests we do need to send a single value 344 - if (signal === TalkbackKind.Pull) { 345 - sink(push(null)); 346 - } else { 347 - closing++; 348 - } 349 - })); 342 + sink( 343 + start(signal => { 344 + // For some operator tests we do need to send a single value 345 + if (signal === TalkbackKind.Pull) { 346 + sink(push(null)); 347 + } else { 348 + closing++; 349 + } 350 + }) 351 + ); 350 352 }; 351 353 352 354 const sink: Sink<any> = signal => { ··· 364 366 }); 365 367 }; 366 368 367 - const passesAsyncSequence = ( 368 - operator: Operator<any, any>, 369 - result: any = 0 370 - ) => { 369 + const passesAsyncSequence = (operator: Operator<any, any>, result: any = 0) => { 371 370 it('passes an async push with an async end (spec)', () => { 372 371 let hasPushed = false; 373 372 const signals: Signal<any>[] = []; 374 373 375 374 const source: Source<any> = sink => { 376 - sink(start((signal) => { 377 - if (signal === TalkbackKind.Pull && !hasPushed) { 378 - hasPushed = true; 379 - setTimeout(() => sink(push(0)), 10); 380 - setTimeout(() => sink(SignalKind.End), 20); 381 - } 382 - })); 375 + sink( 376 + start(signal => { 377 + if (signal === TalkbackKind.Pull && !hasPushed) { 378 + hasPushed = true; 379 + setTimeout(() => sink(push(0)), 10); 380 + setTimeout(() => sink(SignalKind.End), 20); 381 + } 382 + }) 383 + ); 383 384 }; 384 385 385 386 const sink: Sink<any> = signal => { ··· 402 403 expect(hasPushed).toBeTruthy(); 403 404 jest.runAllTimers(); 404 405 405 - expect(signals).toEqual([ 406 - push(result), 407 - SignalKind.End 408 - ]); 406 + expect(signals).toEqual([push(result), SignalKind.End]); 409 407 }); 410 408 }; 411 409 ··· 439 437 440 438 describe('buffer', () => { 441 439 const valueThenNever: Source<any> = sink => 442 - sink(start((signal) => { 443 - if (signal === TalkbackKind.Pull) 444 - sink(push(null)); 445 - })); 440 + sink( 441 + start(signal => { 442 + if (signal === TalkbackKind.Pull) sink(push(null)); 443 + }) 444 + ); 446 445 447 446 const noop = operators.buffer(valueThenNever); 448 447 ··· 522 521 })(sources.fromValue(null))(fn); 523 522 524 523 expect(teardown).toHaveBeenCalled(); 525 - expect(signals).toEqual([ 526 - start(expect.any(Function)), 527 - ]); 524 + expect(signals).toEqual([start(expect.any(Function))]); 528 525 }); 529 526 530 527 // This asynchronous test for concatMap will behave differently than mergeMap & switchMap ··· 539 536 ); 540 537 541 538 jest.advanceTimersByTime(14); 542 - expect(fn.mock.calls).toEqual([ 543 - [1], 544 - [2], 545 - ]); 539 + expect(fn.mock.calls).toEqual([[1], [2]]); 546 540 547 541 jest.runAllTimers(); 548 - expect(fn.mock.calls).toEqual([ 549 - [1], 550 - [2], 551 - [10], 552 - [20], 553 - ]); 542 + expect(fn.mock.calls).toEqual([[1], [2], [10], [20]]); 554 543 }); 555 544 556 545 it('works for fully asynchronous sources', () => { ··· 561 550 return sources.make(observer => { 562 551 setTimeout(() => observer.next(1)); 563 552 return () => {}; 564 - }) 553 + }); 565 554 })(sources.fromValue(null)) 566 555 ); 567 556 ··· 573 562 const values: any[] = []; 574 563 575 564 sinks.forEach(x => values.push(x))( 576 - operators.concat([ 577 - sources.fromArray([1, 2]), 578 - sources.fromArray([3, 4]) 579 - ]) 565 + operators.concat([sources.fromArray([1, 2]), sources.fromArray([3, 4])]) 580 566 ); 581 567 582 - expect(values).toEqual([ 1, 2, 3, 4 ]); 568 + expect(values).toEqual([1, 2, 3, 4]); 583 569 }); 584 570 }); 585 571 ··· 738 724 })(sources.fromValue(null))(fn); 739 725 740 726 expect(teardown).toHaveBeenCalled(); 741 - expect(values).toEqual([ 742 - start(expect.any(Function)), 743 - ]); 727 + expect(values).toEqual([start(expect.any(Function))]); 744 728 }); 745 729 746 730 // This asynchronous test for mergeMap will behave differently than concatMap & switchMap ··· 755 739 ); 756 740 757 741 jest.runAllTimers(); 758 - expect(fn.mock.calls).toEqual([ 759 - [1], 760 - [10], 761 - [2], 762 - [20], 763 - ]); 742 + expect(fn.mock.calls).toEqual([[1], [10], [2], [20]]); 764 743 }); 765 744 766 745 it('emits synchronous values in order', () => { 767 746 const values: any[] = []; 768 747 769 748 sinks.forEach(x => values.push(x))( 770 - operators.merge([ 771 - sources.fromArray([1, 2]), 772 - sources.fromArray([3, 4]) 773 - ]) 749 + operators.merge([sources.fromArray([1, 2]), sources.fromArray([3, 4])]) 774 750 ); 775 751 776 - expect(values).toEqual([ 1, 2, 3, 4 ]); 752 + expect(values).toEqual([1, 2, 3, 4]); 777 753 }); 778 754 }); 779 755 ··· 841 817 let sink: Sink<any>; 842 818 843 819 const fn = jest.fn(); 844 - const source: Source<any> = _sink => { sink = _sink; }; 820 + const source: Source<any> = _sink => { 821 + sink = _sink; 822 + }; 845 823 846 824 sinks.forEach(() => {})(operators.onStart(fn)(source)); 847 825 ··· 854 832 855 833 describe('sample', () => { 856 834 const valueThenNever: Source<any> = sink => 857 - sink(start((signal) => { 858 - if (signal === TalkbackKind.Pull) 859 - sink(push(null)); 860 - })); 835 + sink( 836 + start(signal => { 837 + if (signal === TalkbackKind.Pull) sink(push(null)); 838 + }) 839 + ); 861 840 862 841 const noop = operators.sample(valueThenNever); 863 842 ··· 1059 1038 })(sources.fromValue(null))(fn); 1060 1039 1061 1040 expect(teardown).toHaveBeenCalled(); 1062 - expect(signals).toEqual([ 1063 - start(expect.any(Function)), 1064 - ]); 1041 + expect(signals).toEqual([start(expect.any(Function))]); 1065 1042 }); 1066 1043 1067 1044 // This asynchronous test for switchMap will behave differently than concatMap & mergeMap ··· 1070 1047 const fn = jest.fn(); 1071 1048 1072 1049 sinks.forEach(fn)( 1073 - operators.switchMap((x: number) => ( 1050 + operators.switchMap((x: number) => 1074 1051 operators.take(2)(operators.map((y: number) => x * (y + 1))(sources.interval(5))) 1075 - ))(source) 1052 + )(source) 1076 1053 ); 1077 1054 1078 1055 jest.runAllTimers(); 1079 - expect(fn.mock.calls).toEqual([ 1080 - [1], 1081 - [10], 1082 - [20], 1083 - ]); 1056 + expect(fn.mock.calls).toEqual([[1], [10], [20]]); 1084 1057 }); 1085 1058 }); 1086 1059 ··· 1104 1077 next(1); 1105 1078 1106 1079 expect(fn).toHaveBeenCalledTimes(3); 1107 - expect(fn.mock.calls).toEqual([ 1108 - [start(expect.any(Function))], 1109 - [push(1)], 1110 - [SignalKind.End], 1111 - ]); 1080 + expect(fn.mock.calls).toEqual([[start(expect.any(Function))], [push(1)], [SignalKind.End]]); 1112 1081 }); 1113 1082 }); 1114 1083 ··· 1134 1103 next(1); 1135 1104 1136 1105 expect(fn).toHaveBeenCalledTimes(2); 1137 - expect(fn.mock.calls).toEqual([ 1138 - [start(expect.any(Function))], 1139 - [push(1)], 1140 - ]); 1106 + expect(fn.mock.calls).toEqual([[start(expect.any(Function))], [push(1)]]); 1141 1107 1142 1108 notify(null); 1143 1109 expect(fn).toHaveBeenCalledTimes(3); ··· 1165 1131 next(1); 1166 1132 next(2); 1167 1133 1168 - expect(fn.mock.calls).toEqual([ 1169 - [start(expect.any(Function))], 1170 - [push(1)], 1171 - [SignalKind.End], 1172 - ]); 1134 + expect(fn.mock.calls).toEqual([[start(expect.any(Function))], [push(1)], [SignalKind.End]]); 1173 1135 }); 1174 1136 }); 1175 1137 ··· 1199 1161 expect(signals.length).toBe(0); 1200 1162 complete(); 1201 1163 1202 - expect(signals).toEqual([ 1203 - start(expect.any(Function)), 1204 - push(2), 1205 - SignalKind.End, 1206 - ]); 1164 + expect(signals).toEqual([start(expect.any(Function)), push(2), SignalKind.End]); 1207 1165 }); 1208 1166 }); 1209 1167
+121 -99
src/__tests__/sinks.test.ts
··· 15 15 let pulls = 0; 16 16 const fn = jest.fn(); 17 17 18 - const source: Source<any> = (sink) => { 19 - sink(start((signal) => { 20 - if (signal === TalkbackKind.Pull) { 21 - if (pulls < 3) { 22 - pulls++; 23 - sink(push(0)); 24 - } else { 25 - sink(SignalKind.End); 26 - expect(pulls).toBe(3); 18 + const source: Source<any> = sink => { 19 + sink( 20 + start(signal => { 21 + if (signal === TalkbackKind.Pull) { 22 + if (pulls < 3) { 23 + pulls++; 24 + sink(push(0)); 25 + } else { 26 + sink(SignalKind.End); 27 + expect(pulls).toBe(3); 28 + } 27 29 } 28 - } 29 - })); 30 + }) 31 + ); 30 32 }; 31 33 32 34 sinks.subscribe(fn)(source); ··· 38 40 let pulls = 0; 39 41 let closing = 0; 40 42 41 - const source: Source<any> = (sink) => { 42 - sink(start((signal) => { 43 - if (signal === TalkbackKind.Pull) { 44 - if (!pulls) { 45 - pulls++; 46 - sink(push(0)); 43 + const source: Source<any> = sink => { 44 + sink( 45 + start(signal => { 46 + if (signal === TalkbackKind.Pull) { 47 + if (!pulls) { 48 + pulls++; 49 + sink(push(0)); 50 + } 51 + } else { 52 + closing++; 47 53 } 48 - } else { 49 - closing++; 50 - } 51 - })); 54 + }) 55 + ); 52 56 }; 53 57 54 58 const sub = sinks.subscribe(() => {})(source); ··· 62 66 let pulls = 0; 63 67 let closing = 0; 64 68 65 - const source: Source<any> = (sink) => { 66 - sink(start((signal) => { 67 - if (signal === TalkbackKind.Pull) { 68 - pulls++; 69 - sink(SignalKind.End); 70 - } else { 71 - closing++; 72 - } 73 - })); 69 + const source: Source<any> = sink => { 70 + sink( 71 + start(signal => { 72 + if (signal === TalkbackKind.Pull) { 73 + pulls++; 74 + sink(SignalKind.End); 75 + } else { 76 + closing++; 77 + } 78 + }) 79 + ); 74 80 }; 75 81 76 82 const sub = sinks.subscribe(() => {})(source); ··· 81 87 82 88 it('ignores Push signals after the source has ended', () => { 83 89 const fn = jest.fn(); 84 - const source: Source<any> = (sink) => { 85 - sink(start((signal) => { 86 - if (signal === TalkbackKind.Pull) { 87 - sink(SignalKind.End); 88 - sink(push(0)); 89 - } 90 - })); 90 + const source: Source<any> = sink => { 91 + sink( 92 + start(signal => { 93 + if (signal === TalkbackKind.Pull) { 94 + sink(SignalKind.End); 95 + sink(push(0)); 96 + } 97 + }) 98 + ); 91 99 }; 92 100 93 101 sinks.subscribe(fn)(source); ··· 96 104 97 105 it('ignores Push signals after cancellation', () => { 98 106 const fn = jest.fn(); 99 - const source: Source<any> = (sink) => { 100 - sink(start((signal) => { 101 - if (signal === TalkbackKind.Close) { 102 - sink(push(0)); 103 - } 104 - })); 107 + const source: Source<any> = sink => { 108 + sink( 109 + start(signal => { 110 + if (signal === TalkbackKind.Close) { 111 + sink(push(0)); 112 + } 113 + }) 114 + ); 105 115 }; 106 116 107 117 sinks.subscribe(fn)(source).unsubscribe(); ··· 112 122 describe('publish', () => { 113 123 it('sends Pull talkback signals every Push signal', () => { 114 124 let pulls = 0; 115 - const source: Source<any> = (sink) => { 116 - sink(start((signal) => { 117 - if (signal === TalkbackKind.Pull) { 118 - if (pulls < 3) { 119 - pulls++; 120 - sink(push(0)); 121 - } else { 122 - sink(SignalKind.End); 123 - expect(pulls).toBe(3); 125 + const source: Source<any> = sink => { 126 + sink( 127 + start(signal => { 128 + if (signal === TalkbackKind.Pull) { 129 + if (pulls < 3) { 130 + pulls++; 131 + sink(push(0)); 132 + } else { 133 + sink(SignalKind.End); 134 + expect(pulls).toBe(3); 135 + } 124 136 } 125 - } 126 - })); 137 + }) 138 + ); 127 139 }; 128 140 129 141 sinks.publish(source); ··· 134 146 describe('toArray', () => { 135 147 it('sends Pull talkback signals every Push signal', () => { 136 148 let pulls = 0; 137 - const source: Source<any> = (sink) => { 138 - sink(start((signal) => { 139 - if (signal === TalkbackKind.Pull) { 140 - if (pulls < 3) { 141 - pulls++; 142 - sink(push(0)); 143 - } else { 144 - sink(SignalKind.End); 145 - expect(pulls).toBe(3); 149 + const source: Source<any> = sink => { 150 + sink( 151 + start(signal => { 152 + if (signal === TalkbackKind.Pull) { 153 + if (pulls < 3) { 154 + pulls++; 155 + sink(push(0)); 156 + } else { 157 + sink(SignalKind.End); 158 + expect(pulls).toBe(3); 159 + } 146 160 } 147 - } 148 - })); 161 + }) 162 + ); 149 163 }; 150 164 151 165 const array = sinks.toArray(source); ··· 157 171 let pulls = 0; 158 172 let ending = 0; 159 173 160 - const source: Source<any> = (sink) => { 161 - sink(start((signal) => { 162 - if (signal === TalkbackKind.Pull) { 163 - if (!pulls) { 164 - pulls++; 165 - sink(push(0)); 174 + const source: Source<any> = sink => { 175 + sink( 176 + start(signal => { 177 + if (signal === TalkbackKind.Pull) { 178 + if (!pulls) { 179 + pulls++; 180 + sink(push(0)); 181 + } 182 + } else { 183 + ending++; 166 184 } 167 - } else { 168 - ending++; 169 - } 170 - })); 185 + }) 186 + ); 171 187 }; 172 188 173 189 const array = sinks.toArray(source); ··· 181 197 let pulls = 0; 182 198 let sink: Sink<any> | null = null; 183 199 184 - const source: Source<any> = (_sink) => { 200 + const source: Source<any> = _sink => { 185 201 sink = _sink; 186 - sink(start((signal) => { 187 - if (signal === TalkbackKind.Pull) 188 - pulls++; 189 - })); 202 + sink( 203 + start(signal => { 204 + if (signal === TalkbackKind.Pull) pulls++; 205 + }) 206 + ); 190 207 }; 191 208 192 209 const fn = jest.fn(); ··· 219 236 220 237 const source: Source<any> = _sink => { 221 238 sink = _sink; 222 - sink(start((signal) => { 223 - if (signal === TalkbackKind.Pull) 224 - pulls++; 225 - })); 239 + sink( 240 + start(signal => { 241 + if (signal === TalkbackKind.Pull) pulls++; 242 + }) 243 + ); 226 244 }; 227 245 228 246 Observable.from(observable.toObservable(source) as any).subscribe({ ··· 242 260 it('forwards cancellations from the Observable as a talkback', () => { 243 261 let ending = 0; 244 262 const source: Source<T> = sink => 245 - sink(start((signal) => { 246 - if (signal === TalkbackKind.Close) 247 - ending++; 248 - })); 263 + sink( 264 + start(signal => { 265 + if (signal === TalkbackKind.Close) ending++; 266 + }) 267 + ); 249 268 250 269 const sub = Observable.from(observable.toObservable(source) as any).subscribe({}); 251 270 ··· 263 282 264 283 const source: Source<any> = _sink => { 265 284 sink = _sink; 266 - sink(start((signal) => { 267 - if (signal === TalkbackKind.Pull) 268 - pulls++; 269 - })); 285 + sink( 286 + start(signal => { 287 + if (signal === TalkbackKind.Pull) pulls++; 288 + }) 289 + ); 270 290 }; 271 291 272 292 callbagIterate(fn)(callbag.toCallbag(source)); ··· 284 304 const fn = jest.fn(); 285 305 286 306 const source: Source<any> = sink => 287 - sink(start((signal) => { 288 - if (signal === TalkbackKind.Pull) { 289 - sink(push(0)); 290 - } else { 291 - ending++; 292 - } 293 - })); 307 + sink( 308 + start(signal => { 309 + if (signal === TalkbackKind.Pull) { 310 + sink(push(0)); 311 + } else { 312 + ending++; 313 + } 314 + }) 315 + ); 294 316 295 317 callbagIterate(fn)(callbagTake(1)(callbag.toCallbag(source) as any)); 296 318
+25 -98
src/__tests__/sources.test.ts
··· 9 9 import callbagFromArray from 'callbag-from-iter'; 10 10 import Observable from 'zen-observable'; 11 11 12 - const collectSignals = ( 13 - source: Source<any>, 14 - onStart?: (talkbackCb: TalkbackFn) => void 15 - ) => { 12 + const collectSignals = (source: Source<any>, onStart?: (talkbackCb: TalkbackFn) => void) => { 16 13 let talkback = talkbackPlaceholder; 17 14 const signals: Signal<any>[] = []; 18 15 source(signal => { ··· 26 23 } else { 27 24 talkback(TalkbackKind.Pull); 28 25 } 29 - }) 26 + }); 30 27 31 28 return signals; 32 29 }; ··· 103 100 }; 104 101 105 102 source(sink); 106 - expect(signals).toEqual([ 107 - push(1), 108 - push(2), 109 - SignalKind.End, 110 - ]); 103 + expect(signals).toEqual([push(1), push(2), SignalKind.End]); 111 104 }); 112 105 }; 113 106 ··· 135 128 }); 136 129 137 130 describe('merge', () => { 138 - const source = operators.merge<any>([ 139 - sources.fromValue(0), 140 - sources.empty 141 - ]); 131 + const source = operators.merge<any>([sources.fromValue(0), sources.empty]); 142 132 143 133 passesColdPull(source); 144 134 passesActiveClose(source); 145 135 146 136 it('correctly merges two sources where the second is empty', () => { 147 - const source = operators.merge<any>([ 148 - sources.fromValue(0), 149 - sources.empty 150 - ]); 137 + const source = operators.merge<any>([sources.fromValue(0), sources.empty]); 151 138 152 - expect(collectSignals(source)).toEqual([ 153 - start(expect.any(Function)), 154 - push(0), 155 - SignalKind.End, 156 - ]); 139 + expect(collectSignals(source)).toEqual([start(expect.any(Function)), push(0), SignalKind.End]); 157 140 }); 158 141 159 142 it('correctly merges hot sources', () => { ··· 166 149 const signals = collectSignals(source); 167 150 expect(onStart).toHaveBeenCalledTimes(2); 168 151 169 - expect(signals).toEqual([ 170 - start(expect.any(Function)), 171 - push(1), 172 - push(2), 173 - ]); 152 + expect(signals).toEqual([start(expect.any(Function)), push(1), push(2)]); 174 153 }); 175 154 176 155 it('correctly merges asynchronous sources', () => { ··· 179 158 const onStart = jest.fn(); 180 159 const source = operators.merge<any>([ 181 160 operators.onStart(onStart)(sources.fromValue(-1)), 182 - operators.onStart(onStart)( 183 - operators.take(2)(sources.interval(50)) 184 - ), 161 + operators.onStart(onStart)(operators.take(2)(sources.interval(50))), 185 162 ]); 186 163 187 164 const signals = collectSignals(source); ··· 199 176 }); 200 177 201 178 describe('concat', () => { 202 - const source = operators.concat<any>([ 203 - sources.fromValue(0), 204 - sources.empty 205 - ]); 179 + const source = operators.concat<any>([sources.fromValue(0), sources.empty]); 206 180 207 181 passesColdPull(source); 208 182 passesActiveClose(source); 209 183 210 184 it('correctly concats two sources where the second is empty', () => { 211 - const source = operators.concat<any>([ 212 - sources.fromValue(0), 213 - sources.empty 214 - ]); 185 + const source = operators.concat<any>([sources.fromValue(0), sources.empty]); 215 186 216 - expect(collectSignals(source)).toEqual([ 217 - start(expect.any(Function)), 218 - push(0), 219 - SignalKind.End, 220 - ]); 187 + expect(collectSignals(source)).toEqual([start(expect.any(Function)), push(0), SignalKind.End]); 221 188 }); 222 189 }); 223 190 ··· 234 201 expect(signals).toEqual([start(expect.any(Function))]); 235 202 jest.runAllTimers(); 236 203 237 - expect(signals).toEqual([ 238 - start(expect.any(Function)), 239 - push(1), 240 - SignalKind.End, 241 - ]); 204 + expect(signals).toEqual([start(expect.any(Function)), push(1), SignalKind.End]); 242 205 }); 243 206 244 207 it('supports active cancellation', () => { ··· 263 226 const { source, next, complete } = sources.makeSubject(); 264 227 const signals = collectSignals(source); 265 228 266 - expect(signals).toEqual([ 267 - start(expect.any(Function)), 268 - ]); 229 + expect(signals).toEqual([start(expect.any(Function))]); 269 230 270 231 next(1); 271 232 272 - expect(signals).toEqual([ 273 - start(expect.any(Function)), 274 - push(1), 275 - ]); 233 + expect(signals).toEqual([start(expect.any(Function)), push(1)]); 276 234 277 235 complete(); 278 236 279 - expect(signals).toEqual([ 280 - start(expect.any(Function)), 281 - push(1), 282 - SignalKind.End, 283 - ]); 237 + expect(signals).toEqual([start(expect.any(Function)), push(1), SignalKind.End]); 284 238 }); 285 239 286 240 it('ignores signals after complete has been called', () => { ··· 288 242 const signals = collectSignals(source); 289 243 complete(); 290 244 291 - expect(signals).toEqual([ 292 - start(expect.any(Function)), 293 - SignalKind.End, 294 - ]); 245 + expect(signals).toEqual([start(expect.any(Function)), SignalKind.End]); 295 246 296 247 next(1); 297 248 complete(); ··· 302 253 describe('never', () => { 303 254 it('emits nothing and ends immediately', () => { 304 255 const signals = collectSignals(sources.never); 305 - expect(signals).toEqual([start(expect.any(Function)) ]); 256 + expect(signals).toEqual([start(expect.any(Function))]); 306 257 }); 307 258 }); 308 259 ··· 310 261 it('emits nothing and ends immediately', () => { 311 262 const signals = collectSignals(sources.empty); 312 263 313 - expect(signals).toEqual([ 314 - start(expect.any(Function)), 315 - SignalKind.End, 316 - ]); 264 + expect(signals).toEqual([start(expect.any(Function)), SignalKind.End]); 317 265 }); 318 266 }); 319 267 ··· 324 272 const promise = Promise.resolve(1); 325 273 const signals = collectSignals(sources.fromPromise(promise)); 326 274 327 - expect(signals).toEqual([ 328 - start(expect.any(Function)), 329 - ]); 275 + expect(signals).toEqual([start(expect.any(Function))]); 330 276 331 277 await promise; 332 278 333 - expect(signals).toEqual([ 334 - start(expect.any(Function)), 335 - push(1), 336 - SignalKind.End, 337 - ]); 279 + expect(signals).toEqual([start(expect.any(Function)), push(1), SignalKind.End]); 338 280 }); 339 281 }); 340 282 ··· 349 291 350 292 await new Promise(resolve => setTimeout(resolve)); 351 293 352 - expect(signals).toEqual([ 353 - start(expect.any(Function)), 354 - push(1), 355 - push(2), 356 - SignalKind.End, 357 - ]); 294 + expect(signals).toEqual([start(expect.any(Function)), push(1), push(2), SignalKind.End]); 358 295 }); 359 296 360 297 it('supports cancellation on converted Observables', async () => { ··· 365 302 366 303 await new Promise(resolve => setTimeout(resolve)); 367 304 368 - expect(signals).toEqual([ 369 - start(expect.any(Function)), 370 - ]); 305 + expect(signals).toEqual([start(expect.any(Function))]); 371 306 }); 372 307 }); 373 308 ··· 376 311 const source = callbag.fromCallbag(callbagFromArray([1, 2]) as any); 377 312 const signals = collectSignals(source); 378 313 379 - expect(signals).toEqual([ 380 - start(expect.any(Function)), 381 - push(1), 382 - push(2), 383 - SignalKind.End, 384 - ]); 314 + expect(signals).toEqual([start(expect.any(Function)), push(1), push(2), SignalKind.End]); 385 315 }); 386 316 387 317 it('supports cancellation on converted Observables', () => { ··· 390 320 talkback(TalkbackKind.Close); 391 321 }); 392 322 393 - expect(signals).toEqual([ 394 - start(expect.any(Function)), 395 - ]); 323 + expect(signals).toEqual([start(expect.any(Function))]); 396 324 }); 397 325 }); 398 326 ··· 437 365 438 366 const sink: Sink<any> = signal => { 439 367 expect(signal).not.toBe(SignalKind.End); 440 - if ((signal as any).tag === SignalKind.Start) 441 - talkback = signal[0]; 368 + if ((signal as any).tag === SignalKind.Start) talkback = signal[0]; 442 369 }; 443 370 444 371 sources.fromDomEvent(element as any, 'click')(sink);
+13 -11
src/callbag.ts
··· 1 - import { Source, SignalKind, TalkbackKind } from './types' 2 - import { push, start } from './helpers' 1 + import { Source, SignalKind, TalkbackKind } from './types'; 2 + import { push, start } from './helpers'; 3 3 4 4 interface Callbag<I, O> { 5 5 (t: 0, d: Callbag<O, I>): void; ··· 8 8 } 9 9 10 10 export function fromCallbag<T>(callbag: Callbag<any, T>): Source<T> { 11 - return (sink) => { 11 + return sink => { 12 12 callbag(0, (signal: number, data: any) => { 13 13 if (signal === 0) { 14 - sink(start((signal) => { 15 - if (signal === TalkbackKind.Pull) { 16 - data(1); 17 - } else { 18 - data(2); 19 - } 20 - })); 14 + sink( 15 + start(signal => { 16 + if (signal === TalkbackKind.Pull) { 17 + data(1); 18 + } else { 19 + data(2); 20 + } 21 + }) 22 + ); 21 23 } else if (signal === 1) { 22 24 sink(push(data)); 23 25 } else if (signal === 2) { ··· 30 32 export function toCallbag<T>(source: Source<T>): Callbag<any, T> { 31 33 return (signal: number, sink: any) => { 32 34 if (signal === 0) { 33 - source((signal) => { 35 + source(signal => { 34 36 if (signal === SignalKind.End) { 35 37 sink(2); 36 38 } else if (signal.tag === SignalKind.Start) {
+9 -5
src/helpers.ts
··· 1 - import { TalkbackFn, TeardownFn, Start, Push, SignalKind } from './types' 1 + import { TalkbackFn, TeardownFn, Start, Push, SignalKind } from './types'; 2 2 3 - export const talkbackPlaceholder: TalkbackFn = (_signal) => {/*noop*/}; 4 - export const teardownPlaceholder: TeardownFn = () => {/*noop*/}; 3 + export const talkbackPlaceholder: TalkbackFn = _signal => { 4 + /*noop*/ 5 + }; 6 + export const teardownPlaceholder: TeardownFn = () => { 7 + /*noop*/ 8 + }; 5 9 6 10 export function start<T>(talkback: TalkbackFn): Start<T> { 7 11 const box: any = [talkback]; 8 12 box.tag = SignalKind.Start; 9 - return box 13 + return box; 10 14 } 11 15 12 16 export function push<T>(value: T): Push<T> { 13 17 const box: any = [value]; 14 18 box.tag = SignalKind.Push; 15 - return box 19 + return box; 16 20 }
+7 -7
src/index.ts
··· 1 - export * from './types' 2 - export * from './sources' 3 - export * from './operators' 4 - export * from './sinks' 5 - export * from './observable' 6 - export * from './callbag' 7 - export * from './pipe' 1 + export * from './types'; 2 + export * from './sources'; 3 + export * from './operators'; 4 + export * from './sinks'; 5 + export * from './observable'; 6 + export * from './callbag'; 7 + export * from './pipe';
+16 -11
src/observable.ts
··· 1 - import { Source, SignalKind, TalkbackKind } from './types' 2 - import { push, start, talkbackPlaceholder } from './helpers' 1 + import { Source, SignalKind, TalkbackKind } from './types'; 2 + import { push, start, talkbackPlaceholder } from './helpers'; 3 3 4 4 interface ObservableSubscription { 5 5 closed?: boolean; ··· 16 16 subscribe(observer: ObservableObserver<T>): ObservableSubscription; 17 17 } 18 18 19 - const observableSymbol: unique symbol = typeof Symbol === 'function' 20 - ? (Symbol as any).observable || ((Symbol as any).observable = Symbol('observable')) 21 - : '@@observable' 19 + const observableSymbol: unique symbol = 20 + typeof Symbol === 'function' 21 + ? (Symbol as any).observable || ((Symbol as any).observable = Symbol('observable')) 22 + : '@@observable'; 22 23 23 24 export function fromObservable<T>(input: Observable<T>): Source<T> { 24 25 const observable: Observable<T> = input[observableSymbol] 25 26 ? (input as any)[observableSymbol]() 26 27 : input; 27 - return (sink) => { 28 + return sink => { 28 29 const subscription = observable.subscribe({ 29 30 next(value: T) { 30 31 sink(push(value)); ··· 32 33 complete() { 33 34 sink(SignalKind.End); 34 35 }, 35 - error() {/*noop*/}, 36 + error() { 37 + /*noop*/ 38 + }, 36 39 }); 37 - sink(start((signal) => { 38 - if (signal === TalkbackKind.Close) subscription.unsubscribe(); 39 - })); 40 + sink( 41 + start(signal => { 42 + if (signal === TalkbackKind.Close) subscription.unsubscribe(); 43 + }) 44 + ); 40 45 }; 41 46 } 42 47 ··· 45 50 subscribe(observer: ObservableObserver<T>) { 46 51 let talkback = talkbackPlaceholder; 47 52 let ended = false; 48 - source((signal) => { 53 + source(signal => { 49 54 if (ended) { 50 55 /*noop*/ 51 56 } else if (signal === SignalKind.End) {
+266 -244
src/operators.ts
··· 1 - import { Source, Sink, Operator, SignalKind, TalkbackKind, TalkbackFn } from './types' 2 - import { push, start, talkbackPlaceholder } from './helpers' 3 - import { fromArray } from './sources' 1 + import { Source, Sink, Operator, SignalKind, TalkbackKind, TalkbackFn } from './types'; 2 + import { push, start, talkbackPlaceholder } from './helpers'; 3 + import { fromArray } from './sources'; 4 4 5 5 const identity = <T>(x: T): T => x; 6 6 7 7 export function buffer<S, T>(notifier: Source<S>): Operator<T, T[]> { 8 - return (source) => (sink) => { 8 + return source => sink => { 9 9 let buffer: T[] = []; 10 10 let sourceTalkback = talkbackPlaceholder; 11 11 let notifierTalkback = talkbackPlaceholder; 12 12 let pulled = false; 13 13 let ended = false; 14 - source((signal) => { 14 + source(signal => { 15 15 if (ended) { 16 16 /*noop*/ 17 17 } else if (signal === SignalKind.End) { ··· 21 21 sink(SignalKind.End); 22 22 } else if (signal.tag === SignalKind.Start) { 23 23 sourceTalkback = signal[0]; 24 - notifier((signal) => { 24 + notifier(signal => { 25 25 if (ended) { 26 26 /*noop*/ 27 27 } else if (signal === SignalKind.End) { ··· 48 48 } 49 49 } 50 50 }); 51 - sink(start((signal) => { 52 - if (signal === TalkbackKind.Close && !ended) { 53 - ended = true; 54 - sourceTalkback(TalkbackKind.Close); 55 - notifierTalkback(TalkbackKind.Close); 56 - } else if (!ended && !pulled) { 57 - pulled = true; 58 - sourceTalkback(TalkbackKind.Pull); 59 - notifierTalkback(TalkbackKind.Pull); 60 - } 61 - })); 51 + sink( 52 + start(signal => { 53 + if (signal === TalkbackKind.Close && !ended) { 54 + ended = true; 55 + sourceTalkback(TalkbackKind.Close); 56 + notifierTalkback(TalkbackKind.Close); 57 + } else if (!ended && !pulled) { 58 + pulled = true; 59 + sourceTalkback(TalkbackKind.Pull); 60 + notifierTalkback(TalkbackKind.Pull); 61 + } 62 + }) 63 + ); 62 64 }; 63 65 } 64 66 65 67 export function combine<A, B>(sourceA: Source<A>, sourceB: Source<B>): Source<[A, B]> { 66 - return (sink) => { 68 + return sink => { 67 69 let lastValA: A | void; 68 70 let lastValB: B | void; 69 71 let talkbackA = talkbackPlaceholder; ··· 71 73 let gotSignal = false; 72 74 let gotEnd = false; 73 75 let ended = false; 74 - sourceA((signal) => { 76 + sourceA(signal => { 75 77 if (signal === SignalKind.End) { 76 78 if (!gotEnd) { 77 79 gotEnd = true; ··· 94 96 sink(push([lastValA, lastValB] as [A, B])); 95 97 } 96 98 }); 97 - sourceB((signal) => { 99 + sourceB(signal => { 98 100 if (signal === SignalKind.End) { 99 101 if (!gotEnd) { 100 102 gotEnd = true; ··· 117 119 sink(push([lastValA, lastValB] as [A, B])); 118 120 } 119 121 }); 120 - sink(start((signal) => { 121 - if (ended) { 122 - /*noop*/ 123 - } else if (signal === TalkbackKind.Close) { 124 - ended = true; 125 - talkbackA(TalkbackKind.Close); 126 - talkbackB(TalkbackKind.Close); 127 - } else if (!gotSignal) { 128 - gotSignal = true; 129 - talkbackA(TalkbackKind.Pull); 130 - talkbackB(TalkbackKind.Pull); 131 - } 132 - })); 122 + sink( 123 + start(signal => { 124 + if (ended) { 125 + /*noop*/ 126 + } else if (signal === TalkbackKind.Close) { 127 + ended = true; 128 + talkbackA(TalkbackKind.Close); 129 + talkbackB(TalkbackKind.Close); 130 + } else if (!gotSignal) { 131 + gotSignal = true; 132 + talkbackA(TalkbackKind.Pull); 133 + talkbackB(TalkbackKind.Pull); 134 + } 135 + }) 136 + ); 133 137 }; 134 138 } 135 139 136 140 export function concatMap<In, Out>(map: (value: In) => Source<Out>): Operator<In, Out> { 137 - return (source) => (sink) => { 141 + return source => sink => { 138 142 const inputQueue: In[] = []; 139 143 let outerTalkback = talkbackPlaceholder; 140 144 let innerTalkback = talkbackPlaceholder; ··· 144 148 let ended = false; 145 149 function applyInnerSource(innerSource: Source<Out>): void { 146 150 innerActive = true; 147 - innerSource((signal) => { 151 + innerSource(signal => { 148 152 if (signal === SignalKind.End) { 149 153 if (innerActive) { 150 154 innerActive = false; 151 155 if (inputQueue.length) { 152 - applyInnerSource(map(inputQueue.shift()!)) 156 + applyInnerSource(map(inputQueue.shift()!)); 153 157 } else if (ended) { 154 158 sink(SignalKind.End); 155 159 } else if (!outerPulled) { ··· 170 174 } 171 175 }); 172 176 } 173 - source((signal) => { 177 + source(signal => { 174 178 if (ended) { 175 179 /*noop*/ 176 180 } else if (signal === SignalKind.End) { 177 181 ended = true; 178 - if (!innerActive && !inputQueue.length) 179 - sink(SignalKind.End); 182 + if (!innerActive && !inputQueue.length) sink(SignalKind.End); 180 183 } else if (signal.tag === SignalKind.Start) { 181 184 outerTalkback = signal[0]; 182 185 } else { ··· 188 191 } 189 192 } 190 193 }); 191 - sink(start((signal) => { 192 - if (signal === TalkbackKind.Close) { 193 - if (!ended) { 194 - ended = true; 195 - outerTalkback(TalkbackKind.Close); 196 - } 197 - if (innerActive) { 198 - innerActive = false; 199 - innerTalkback(TalkbackKind.Close); 200 - } 201 - } else { 202 - if (!ended && !outerPulled) { 203 - outerPulled = true; 204 - outerTalkback(TalkbackKind.Pull); 205 - } 206 - if (innerActive && !innerPulled) { 207 - innerPulled = true; 208 - innerTalkback(TalkbackKind.Pull); 194 + sink( 195 + start(signal => { 196 + if (signal === TalkbackKind.Close) { 197 + if (!ended) { 198 + ended = true; 199 + outerTalkback(TalkbackKind.Close); 200 + } 201 + if (innerActive) { 202 + innerActive = false; 203 + innerTalkback(TalkbackKind.Close); 204 + } 205 + } else { 206 + if (!ended && !outerPulled) { 207 + outerPulled = true; 208 + outerTalkback(TalkbackKind.Pull); 209 + } 210 + if (innerActive && !innerPulled) { 211 + innerPulled = true; 212 + innerTalkback(TalkbackKind.Pull); 213 + } 209 214 } 210 - } 211 - })); 215 + }) 216 + ); 212 217 }; 213 218 } 214 219 ··· 221 226 } 222 227 223 228 export function filter<T>(predicate: (value: T) => boolean): Operator<T, T> { 224 - return (source) => (sink) => { 229 + return source => sink => { 225 230 let talkback = talkbackPlaceholder; 226 - source((signal) => { 231 + source(signal => { 227 232 if (signal === SignalKind.End) { 228 233 sink(signal); 229 234 } else if (signal.tag === SignalKind.Start) { ··· 239 244 } 240 245 241 246 export function map<In, Out>(map: (value: In) => Out): Operator<In, Out> { 242 - return (source) => (sink) => source((signal) => { 243 - if (signal === SignalKind.End || signal.tag === SignalKind.Start) { 244 - sink(signal); 245 - } else { 246 - sink(push(map(signal[0]))); 247 - } 248 - }); 247 + return source => sink => 248 + source(signal => { 249 + if (signal === SignalKind.End || signal.tag === SignalKind.Start) { 250 + sink(signal); 251 + } else { 252 + sink(push(map(signal[0]))); 253 + } 254 + }); 249 255 } 250 256 251 257 export function mergeMap<In, Out>(map: (value: In) => Source<Out>): Operator<In, Out> { 252 - return (source) => (sink) => { 258 + return source => sink => { 253 259 const innerTalkbacks: TalkbackFn[] = []; 254 260 let outerTalkback = talkbackPlaceholder; 255 261 let outerPulled = false; 256 262 let ended = false; 257 263 function applyInnerSource(innerSource: Source<Out>): void { 258 264 let talkback = talkbackPlaceholder; 259 - innerSource((signal) => { 265 + innerSource(signal => { 260 266 if (signal === SignalKind.End) { 261 267 if (innerTalkbacks.length) { 262 268 const index = innerTalkbacks.indexOf(talkback); ··· 271 277 } 272 278 } 273 279 } else if (signal.tag === SignalKind.Start) { 274 - innerTalkbacks.push(talkback = signal[0]); 280 + innerTalkbacks.push((talkback = signal[0])); 275 281 talkback(TalkbackKind.Pull); 276 282 } else if (innerTalkbacks.length) { 277 283 sink(signal); ··· 279 285 } 280 286 }); 281 287 } 282 - source((signal) => { 288 + source(signal => { 283 289 if (ended) { 284 290 /*noop*/ 285 291 } else if (signal === SignalKind.End) { 286 292 ended = true; 287 - if (!innerTalkbacks.length) 288 - sink(SignalKind.End); 293 + if (!innerTalkbacks.length) sink(SignalKind.End); 289 294 } else if (signal.tag === SignalKind.Start) { 290 295 outerTalkback = signal[0]; 291 296 } else { ··· 297 302 } 298 303 } 299 304 }); 300 - sink(start((signal) => { 301 - if (signal === TalkbackKind.Close) { 302 - if (!ended) { 303 - ended = true; 304 - outerTalkback(TalkbackKind.Close); 305 - } 306 - while (innerTalkbacks.length) 307 - innerTalkbacks.pop()!(TalkbackKind.Close); 308 - } else { 309 - if (!ended && !outerPulled) { 310 - outerPulled = true; 311 - outerTalkback(TalkbackKind.Pull); 305 + sink( 306 + start(signal => { 307 + if (signal === TalkbackKind.Close) { 308 + if (!ended) { 309 + ended = true; 310 + outerTalkback(TalkbackKind.Close); 311 + } 312 + while (innerTalkbacks.length) innerTalkbacks.pop()!(TalkbackKind.Close); 312 313 } else { 313 - outerPulled = false; 314 + if (!ended && !outerPulled) { 315 + outerPulled = true; 316 + outerTalkback(TalkbackKind.Pull); 317 + } else { 318 + outerPulled = false; 319 + } 320 + for (let i = 0; i < innerTalkbacks.length; i++) innerTalkbacks[i](TalkbackKind.Pull); 314 321 } 315 - for (let i = 0; i < innerTalkbacks.length; i++) 316 - innerTalkbacks[i](TalkbackKind.Pull); 317 - } 318 - })); 322 + }) 323 + ); 319 324 }; 320 325 } 321 326 ··· 328 333 } 329 334 330 335 export function onEnd<T>(callback: () => void): Operator<T, T> { 331 - return (source) => (sink) => { 336 + return source => sink => { 332 337 let ended = false; 333 - source((signal) => { 338 + source(signal => { 334 339 if (ended) { 335 340 /*noop*/ 336 341 } else if (signal === SignalKind.End) { ··· 339 344 callback(); 340 345 } else if (signal.tag === SignalKind.Start) { 341 346 const talkback = signal[0]; 342 - sink(start((signal) => { 343 - if (signal === TalkbackKind.Close) { 344 - ended = true; 345 - talkback(TalkbackKind.Close); 346 - callback(); 347 - } else { 348 - talkback(signal); 349 - } 350 - })); 347 + sink( 348 + start(signal => { 349 + if (signal === TalkbackKind.Close) { 350 + ended = true; 351 + talkback(TalkbackKind.Close); 352 + callback(); 353 + } else { 354 + talkback(signal); 355 + } 356 + }) 357 + ); 351 358 } else { 352 359 sink(signal); 353 360 } ··· 356 363 } 357 364 358 365 export function onPush<T>(callback: (value: T) => void): Operator<T, T> { 359 - return (source) => (sink) => { 366 + return source => sink => { 360 367 let ended = false; 361 - source((signal) => { 368 + source(signal => { 362 369 if (ended) { 363 370 /*noop*/ 364 371 } else if (signal === SignalKind.End) { ··· 366 373 sink(SignalKind.End); 367 374 } else if (signal.tag === SignalKind.Start) { 368 375 const talkback = signal[0]; 369 - sink(start((signal) => { 370 - if (signal === TalkbackKind.Close) ended = true; 371 - talkback(signal); 372 - })); 376 + sink( 377 + start(signal => { 378 + if (signal === TalkbackKind.Close) ended = true; 379 + talkback(signal); 380 + }) 381 + ); 373 382 } else { 374 383 callback(signal[0]); 375 384 sink(signal); ··· 379 388 } 380 389 381 390 export function onStart<T>(callback: () => void): Operator<T, T> { 382 - return (source) => (sink) => source((signal) => { 383 - if (signal === SignalKind.End) { 384 - sink(SignalKind.End); 385 - } else if (signal.tag === SignalKind.Start) { 386 - sink(signal); 387 - callback(); 388 - } else { 389 - sink(signal); 390 - } 391 - }); 391 + return source => sink => 392 + source(signal => { 393 + if (signal === SignalKind.End) { 394 + sink(SignalKind.End); 395 + } else if (signal.tag === SignalKind.Start) { 396 + sink(signal); 397 + callback(); 398 + } else { 399 + sink(signal); 400 + } 401 + }); 392 402 } 393 403 394 404 export function sample<S, T>(notifier: Source<S>): Operator<T, T> { 395 - return (source) => (sink) => { 405 + return source => sink => { 396 406 let sourceTalkback = talkbackPlaceholder; 397 407 let notifierTalkback = talkbackPlaceholder; 398 408 let value: T | void; 399 409 let pulled = false; 400 410 let ended = false; 401 - source((signal) => { 411 + source(signal => { 402 412 if (ended) { 403 413 /*noop*/ 404 414 } else if (signal === SignalKind.End) { ··· 418 428 } 419 429 } 420 430 }); 421 - notifier((signal) => { 431 + notifier(signal => { 422 432 if (ended) { 423 433 /*noop*/ 424 434 } else if (signal === SignalKind.End) { ··· 433 443 sink(signal); 434 444 } 435 445 }); 436 - sink(start((signal) => { 437 - if (signal === TalkbackKind.Close && !ended) { 438 - ended = true; 439 - sourceTalkback(TalkbackKind.Close); 440 - notifierTalkback(TalkbackKind.Close); 441 - } else if (!ended && !pulled) { 442 - pulled = true; 443 - sourceTalkback(TalkbackKind.Pull); 444 - notifierTalkback(TalkbackKind.Pull); 445 - } 446 - })); 446 + sink( 447 + start(signal => { 448 + if (signal === TalkbackKind.Close && !ended) { 449 + ended = true; 450 + sourceTalkback(TalkbackKind.Close); 451 + notifierTalkback(TalkbackKind.Close); 452 + } else if (!ended && !pulled) { 453 + pulled = true; 454 + sourceTalkback(TalkbackKind.Pull); 455 + notifierTalkback(TalkbackKind.Pull); 456 + } 457 + }) 458 + ); 447 459 }; 448 460 } 449 461 450 462 export function scan<In, Out>(reducer: (acc: Out, value: In) => Out, seed: Out): Operator<In, Out> { 451 - return (source) => (sink) => { 463 + return source => sink => { 452 464 let acc = seed; 453 - source((signal) => { 465 + source(signal => { 454 466 if (signal === SignalKind.End) { 455 467 sink(SignalKind.End); 456 468 } else if (signal.tag === SignalKind.Start) { 457 469 sink(signal); 458 470 } else { 459 - sink(push(acc = reducer(acc, signal[0]))); 471 + sink(push((acc = reducer(acc, signal[0])))); 460 472 } 461 473 }); 462 474 }; ··· 466 478 const sinks: Sink<T>[] = []; 467 479 let talkback = talkbackPlaceholder; 468 480 let gotSignal = false; 469 - return (sink) => { 481 + return sink => { 470 482 sinks.push(sink); 471 483 if (sinks.length === 1) { 472 - source((signal) => { 484 + source(signal => { 473 485 if (signal === SignalKind.End) { 474 486 while (sinks.length) sinks.pop()!(SignalKind.End); 475 487 } else if (signal.tag === SignalKind.Start) { ··· 480 492 } 481 493 }); 482 494 } 483 - sink(start((signal) => { 484 - if (signal === TalkbackKind.Close) { 485 - const index = sinks.indexOf(sink); 486 - if (index > -1) sinks.splice(index, 1); 487 - if (!sinks.length) talkback(TalkbackKind.Close); 488 - } else if (signal === TalkbackKind.Pull && !gotSignal) { 489 - gotSignal = true; 490 - talkback(TalkbackKind.Pull); 491 - } 492 - })); 495 + sink( 496 + start(signal => { 497 + if (signal === TalkbackKind.Close) { 498 + const index = sinks.indexOf(sink); 499 + if (index > -1) sinks.splice(index, 1); 500 + if (!sinks.length) talkback(TalkbackKind.Close); 501 + } else if (signal === TalkbackKind.Pull && !gotSignal) { 502 + gotSignal = true; 503 + talkback(TalkbackKind.Pull); 504 + } 505 + }) 506 + ); 493 507 }; 494 508 } 495 509 496 510 export function skip<T>(wait: number): Operator<T, T> { 497 - return (source) => (sink) => { 511 + return source => sink => { 498 512 let talkback = talkbackPlaceholder; 499 513 let rest = wait; 500 - source((signal) => { 514 + source(signal => { 501 515 if (signal === SignalKind.End) { 502 516 sink(SignalKind.End); 503 517 } else if (signal.tag === SignalKind.Start) { ··· 513 527 } 514 528 515 529 export function skipUntil<S, T>(notifier: Source<S>): Operator<T, T> { 516 - return (source) => (sink) => { 530 + return source => sink => { 517 531 let sourceTalkback = talkbackPlaceholder; 518 532 let notifierTalkback = talkbackPlaceholder; 519 533 let skip = true; 520 534 let pulled = false; 521 535 let ended = false; 522 - source((signal) => { 536 + source(signal => { 523 537 if (ended) { 524 538 /*noop*/ 525 539 } else if (signal === SignalKind.End) { ··· 528 542 sink(SignalKind.End); 529 543 } else if (signal.tag === SignalKind.Start) { 530 544 sourceTalkback = signal[0]; 531 - notifier((signal) => { 545 + notifier(signal => { 532 546 if (signal === SignalKind.End) { 533 547 if (skip) { 534 548 ended = true; ··· 552 566 pulled = false; 553 567 } 554 568 }); 555 - sink(start((signal) => { 556 - if (signal === TalkbackKind.Close && !ended) { 557 - ended = true; 558 - sourceTalkback(TalkbackKind.Close); 559 - if (skip) notifierTalkback(TalkbackKind.Close); 560 - } else if (!ended && !pulled) { 561 - pulled = true; 562 - if (skip) notifierTalkback(TalkbackKind.Pull); 563 - sourceTalkback(TalkbackKind.Pull); 564 - } 565 - })); 569 + sink( 570 + start(signal => { 571 + if (signal === TalkbackKind.Close && !ended) { 572 + ended = true; 573 + sourceTalkback(TalkbackKind.Close); 574 + if (skip) notifierTalkback(TalkbackKind.Close); 575 + } else if (!ended && !pulled) { 576 + pulled = true; 577 + if (skip) notifierTalkback(TalkbackKind.Pull); 578 + sourceTalkback(TalkbackKind.Pull); 579 + } 580 + }) 581 + ); 566 582 }; 567 583 } 568 584 569 585 export function skipWhile<T>(predicate: (value: T) => boolean): Operator<T, T> { 570 - return (source) => (sink) => { 586 + return source => sink => { 571 587 let talkback = talkbackPlaceholder; 572 588 let skip = true; 573 - source((signal) => { 589 + source(signal => { 574 590 if (signal === SignalKind.End) { 575 591 sink(SignalKind.End); 576 592 } else if (signal.tag === SignalKind.Start) { ··· 591 607 } 592 608 593 609 export function switchMap<In, Out>(map: (value: In) => Source<Out>): Operator<In, Out> { 594 - return (source) => (sink) => { 610 + return source => sink => { 595 611 let outerTalkback = talkbackPlaceholder; 596 612 let innerTalkback = talkbackPlaceholder; 597 613 let outerPulled = false; ··· 600 616 let ended = false; 601 617 function applyInnerSource(innerSource: Source<Out>): void { 602 618 innerActive = true; 603 - innerSource((signal) => { 619 + innerSource(signal => { 604 620 if (!innerActive) { 605 621 /*noop*/ 606 622 } else if (signal === SignalKind.End) { ··· 624 640 } 625 641 }); 626 642 } 627 - source((signal) => { 643 + source(signal => { 628 644 if (ended) { 629 645 /*noop*/ 630 646 } else if (signal === SignalKind.End) { 631 647 ended = true; 632 - if (!innerActive) 633 - sink(SignalKind.End); 648 + if (!innerActive) sink(SignalKind.End); 634 649 } else if (signal.tag === SignalKind.Start) { 635 650 outerTalkback = signal[0]; 636 651 } else { ··· 647 662 applyInnerSource(map(signal[0])); 648 663 } 649 664 }); 650 - sink(start((signal) => { 651 - if (signal === TalkbackKind.Close) { 652 - if (!ended) { 653 - ended = true; 654 - outerTalkback(TalkbackKind.Close); 665 + sink( 666 + start(signal => { 667 + if (signal === TalkbackKind.Close) { 668 + if (!ended) { 669 + ended = true; 670 + outerTalkback(TalkbackKind.Close); 671 + } 672 + if (innerActive) { 673 + innerActive = false; 674 + innerTalkback(TalkbackKind.Close); 675 + } 676 + } else { 677 + if (!ended && !outerPulled) { 678 + outerPulled = true; 679 + outerTalkback(TalkbackKind.Pull); 680 + } 681 + if (innerActive && !innerPulled) { 682 + innerPulled = true; 683 + innerTalkback(TalkbackKind.Pull); 684 + } 655 685 } 656 - if (innerActive) { 657 - innerActive = false; 658 - innerTalkback(TalkbackKind.Close); 659 - } 660 - } else { 661 - if (!ended && !outerPulled) { 662 - outerPulled = true; 663 - outerTalkback(TalkbackKind.Pull); 664 - } 665 - if (innerActive && !innerPulled) { 666 - innerPulled = true; 667 - innerTalkback(TalkbackKind.Pull); 668 - } 669 - } 670 - })); 686 + }) 687 + ); 671 688 }; 672 689 } 673 690 ··· 676 693 } 677 694 678 695 export function take<T>(max: number): Operator<T, T> { 679 - return (source) => (sink) => { 696 + return source => sink => { 680 697 let talkback = talkbackPlaceholder; 681 698 let ended = false; 682 699 let taken = 0; 683 - source((signal) => { 700 + source(signal => { 684 701 if (ended) { 685 702 /*noop*/ 686 703 } else if (signal === SignalKind.End) { ··· 705 722 sink(signal); 706 723 } 707 724 }); 708 - sink(start((signal) => { 709 - if (signal === TalkbackKind.Close && !ended) { 710 - ended = true; 711 - talkback(TalkbackKind.Close); 712 - } else if (signal === TalkbackKind.Pull && !ended && taken < max) { 713 - talkback(TalkbackKind.Pull); 714 - } 715 - })); 725 + sink( 726 + start(signal => { 727 + if (signal === TalkbackKind.Close && !ended) { 728 + ended = true; 729 + talkback(TalkbackKind.Close); 730 + } else if (signal === TalkbackKind.Pull && !ended && taken < max) { 731 + talkback(TalkbackKind.Pull); 732 + } 733 + }) 734 + ); 716 735 }; 717 736 } 718 737 719 738 export function takeLast<T>(max: number): Operator<T, T> { 720 - return (source) => (sink) => { 739 + return source => sink => { 721 740 const queue: T[] = []; 722 741 let talkback = talkbackPlaceholder; 723 - source((signal) => { 742 + source(signal => { 724 743 if (signal === SignalKind.End) { 725 744 fromArray(queue)(sink); 726 745 } else if (signal.tag === SignalKind.Start) { ··· 740 759 } 741 760 742 761 export function takeUntil<S, T>(notifier: Source<S>): Operator<T, T> { 743 - return (source) => (sink) => { 762 + return source => sink => { 744 763 let sourceTalkback = talkbackPlaceholder; 745 764 let notifierTalkback = talkbackPlaceholder; 746 765 let ended = false; 747 - source((signal) => { 766 + source(signal => { 748 767 if (ended) { 749 768 /*noop*/ 750 769 } else if (signal === SignalKind.End) { ··· 753 772 sink(SignalKind.End); 754 773 } else if (signal.tag === SignalKind.Start) { 755 774 sourceTalkback = signal[0]; 756 - notifier((signal) => { 775 + notifier(signal => { 757 776 if (signal === SignalKind.End) { 758 777 /*noop*/ 759 778 } else if (signal.tag === SignalKind.Start) { ··· 768 787 sink(signal); 769 788 } 770 789 }); 771 - sink(start((signal) => { 772 - if (signal === TalkbackKind.Close && !ended) { 773 - ended = true; 774 - sourceTalkback(TalkbackKind.Close); 775 - notifierTalkback(TalkbackKind.Close); 776 - } else if (!ended) { 777 - sourceTalkback(TalkbackKind.Pull); 778 - } 779 - })); 790 + sink( 791 + start(signal => { 792 + if (signal === TalkbackKind.Close && !ended) { 793 + ended = true; 794 + sourceTalkback(TalkbackKind.Close); 795 + notifierTalkback(TalkbackKind.Close); 796 + } else if (!ended) { 797 + sourceTalkback(TalkbackKind.Pull); 798 + } 799 + }) 800 + ); 780 801 }; 781 802 } 782 803 783 804 export function takeWhile<T>(predicate: (value: T) => boolean): Operator<T, T> { 784 - return (source) => (sink) => { 805 + return source => sink => { 785 806 let talkback = talkbackPlaceholder; 786 807 let ended = false; 787 - source((signal) => { 808 + source(signal => { 788 809 if (ended) { 789 810 /*noop*/ 790 811 } else if (signal === SignalKind.End) { ··· 805 826 } 806 827 807 828 export function debounce<T>(timing: (value: T) => number): Operator<T, T> { 808 - return (source) => (sink) => { 829 + return source => sink => { 809 830 let id: any | void; 810 831 let deferredEnded = false; 811 832 let ended = false; 812 - source((signal) => { 833 + source(signal => { 813 834 if (ended) { 814 835 /*noop*/ 815 836 } else if (signal === SignalKind.End) { ··· 821 842 } 822 843 } else if (signal.tag === SignalKind.Start) { 823 844 const talkback = signal[0]; 824 - sink(start((signal) => { 825 - if (signal === TalkbackKind.Close && !ended) { 826 - ended = true; 827 - deferredEnded = false; 828 - if (id) clearTimeout(id); 829 - talkback(TalkbackKind.Close); 830 - } else if (!ended) { 831 - talkback(TalkbackKind.Pull); 832 - } 833 - })); 845 + sink( 846 + start(signal => { 847 + if (signal === TalkbackKind.Close && !ended) { 848 + ended = true; 849 + deferredEnded = false; 850 + if (id) clearTimeout(id); 851 + talkback(TalkbackKind.Close); 852 + } else if (!ended) { 853 + talkback(TalkbackKind.Pull); 854 + } 855 + }) 856 + ); 834 857 } else { 835 858 if (id) clearTimeout(id); 836 859 id = setTimeout(() => { ··· 844 867 } 845 868 846 869 export function delay<T>(wait: number): Operator<T, T> { 847 - return (source) => (sink) => { 870 + return source => sink => { 848 871 let active = 0; 849 - source((signal) => { 872 + source(signal => { 850 873 if (typeof signal !== 'number' && signal.tag === SignalKind.Start) { 851 874 sink(signal); 852 875 } else { ··· 863 886 } 864 887 865 888 export function throttle<T>(timing: (value: T) => number): Operator<T, T> { 866 - return (source) => (sink) => { 889 + return source => sink => { 867 890 let skip = false; 868 891 let id: any | void; 869 - source((signal) => { 892 + source(signal => { 870 893 if (signal === SignalKind.End) { 871 894 if (id) clearTimeout(id); 872 895 sink(SignalKind.End); 873 896 } else if (signal.tag === SignalKind.Start) { 874 897 const talkback = signal[0]; 875 - sink(start((signal) => { 876 - if (signal === TalkbackKind.Close) { 877 - if (id) clearTimeout(id); 878 - talkback(TalkbackKind.Close); 879 - } else { 880 - talkback(TalkbackKind.Pull); 881 - } 882 - })); 898 + sink( 899 + start(signal => { 900 + if (signal === TalkbackKind.Close) { 901 + if (id) clearTimeout(id); 902 + talkback(TalkbackKind.Close); 903 + } else { 904 + talkback(TalkbackKind.Pull); 905 + } 906 + }) 907 + ); 883 908 } else if (!skip) { 884 909 skip = true; 885 910 if (id) clearTimeout(id); ··· 893 918 }; 894 919 } 895 920 896 - export { 897 - mergeAll as flatten, 898 - onPush as tap, 899 - } 921 + export { mergeAll as flatten, onPush as tap };
+1 -2
src/pipe.ts
··· 153 153 154 154 function pipe() { 155 155 let x = arguments[0]; 156 - for (let i = 1, l = arguments.length; i < l; i++) 157 - x = arguments[i](x); 156 + for (let i = 1, l = arguments.length; i < l; i++) x = arguments[i](x); 158 157 return x; 159 158 } 160 159
+11 -9
src/sinks.ts
··· 1 - import { Source, Subscription, TalkbackKind, SignalKind } from './types' 2 - import { talkbackPlaceholder } from './helpers' 1 + import { Source, Subscription, TalkbackKind, SignalKind } from './types'; 2 + import { talkbackPlaceholder } from './helpers'; 3 3 4 4 export function subscribe<T>(subscriber: (value: T) => void) { 5 5 return (source: Source<T>): Subscription => { 6 6 let talkback = talkbackPlaceholder; 7 7 let ended = false; 8 - source((signal) => { 8 + source(signal => { 9 9 if (signal === SignalKind.End) { 10 10 ended = true; 11 11 } else if (signal.tag === SignalKind.Start) { ··· 22 22 talkback(TalkbackKind.Close); 23 23 } 24 24 }, 25 - } 26 - } 25 + }; 26 + }; 27 27 } 28 28 29 29 export function forEach<T>(subscriber: (value: T) => void) { ··· 33 33 } 34 34 35 35 export function publish<T>(source: Source<T>): void { 36 - subscribe((_value) => {/*noop*/})(source); 36 + subscribe(_value => { 37 + /*noop*/ 38 + })(source); 37 39 } 38 40 39 41 export function toArray<T>(source: Source<T>): T[] { 40 42 const values: T[] = []; 41 43 let talkback = talkbackPlaceholder; 42 44 let ended = false; 43 - source((signal) => { 45 + source(signal => { 44 46 if (signal === SignalKind.End) { 45 47 ended = true; 46 48 } else if (signal.tag === SignalKind.Start) { ··· 58 60 return new Promise(resolve => { 59 61 let talkback = talkbackPlaceholder; 60 62 let value: T | void; 61 - source((signal) => { 63 + source(signal => { 62 64 if (signal === SignalKind.End) { 63 65 resolve(value!); 64 66 } else if (signal.tag === SignalKind.Start) { ··· 68 70 talkback(TalkbackKind.Pull); 69 71 } 70 72 }); 71 - }) 73 + }); 72 74 }
+84 -70
src/sources.ts
··· 1 - import { Source, Sink, SignalKind, TalkbackKind, Observer, Subject, TeardownFn } from './types' 2 - import { push, start, talkbackPlaceholder } from './helpers' 1 + import { Source, Sink, SignalKind, TalkbackKind, Observer, Subject, TeardownFn } from './types'; 2 + import { push, start, talkbackPlaceholder } from './helpers'; 3 3 4 4 export function fromArray<T>(array: T[]): Source<T> { 5 - return (sink) => { 5 + return sink => { 6 6 let ended = false; 7 7 let looping = false; 8 8 let pulled = false; 9 9 let current = 0; 10 - sink(start((signal) => { 11 - if (signal === TalkbackKind.Close) { 12 - ended = true; 13 - } else if (looping) { 14 - pulled = true; 15 - } else { 16 - pulled = true; 17 - looping = true; 18 - for (pulled = looping = true; pulled && !ended; current++) { 19 - if (current < array.length) { 20 - pulled = false; 21 - sink(push(array[current])); 22 - } else { 23 - ended = true; 24 - sink(SignalKind.End); 10 + sink( 11 + start(signal => { 12 + if (signal === TalkbackKind.Close) { 13 + ended = true; 14 + } else if (looping) { 15 + pulled = true; 16 + } else { 17 + pulled = true; 18 + looping = true; 19 + for (pulled = looping = true; pulled && !ended; current++) { 20 + if (current < array.length) { 21 + pulled = false; 22 + sink(push(array[current])); 23 + } else { 24 + ended = true; 25 + sink(SignalKind.End); 26 + } 25 27 } 28 + looping = false; 26 29 } 27 - looping = false; 28 - } 29 - })) 30 - } 30 + }) 31 + ); 32 + }; 31 33 } 32 34 33 35 export function fromValue<T>(value: T): Source<T> { 34 - return (sink) => { 36 + return sink => { 35 37 let ended = false; 36 - sink(start((signal) => { 37 - if (signal === TalkbackKind.Close) { 38 - ended = true; 39 - } else if (!ended) { 40 - ended = true; 41 - sink(push(value)); 42 - sink(SignalKind.End); 43 - } 44 - })) 45 - } 38 + sink( 39 + start(signal => { 40 + if (signal === TalkbackKind.Close) { 41 + ended = true; 42 + } else if (!ended) { 43 + ended = true; 44 + sink(push(value)); 45 + sink(SignalKind.End); 46 + } 47 + }) 48 + ); 49 + }; 46 50 } 47 51 48 52 export function make<T>(produce: (observer: Observer<T>) => TeardownFn): Source<T> { 49 - return (sink) => { 53 + return sink => { 50 54 let ended = false; 51 55 const teardown = produce({ 52 56 next(value: T) { ··· 59 63 } 60 64 }, 61 65 }); 62 - sink(start((signal) => { 63 - if (signal === TalkbackKind.Close && !ended) { 64 - ended = true; 65 - teardown(); 66 - } 67 - })); 68 - } 66 + sink( 67 + start(signal => { 68 + if (signal === TalkbackKind.Close && !ended) { 69 + ended = true; 70 + teardown(); 71 + } 72 + }) 73 + ); 74 + }; 69 75 } 70 76 71 77 export function makeSubject<T>(): Subject<T> { ··· 74 80 return { 75 81 source(sink: Sink<T>) { 76 82 sinks.push(sink); 77 - sink(start((signal) => { 78 - if (signal === TalkbackKind.Close) { 79 - const index = sinks.indexOf(sink); 80 - if (index > -1) sinks.splice(index, 1); 81 - } 82 - })); 83 + sink( 84 + start(signal => { 85 + if (signal === TalkbackKind.Close) { 86 + const index = sinks.indexOf(sink); 87 + if (index > -1) sinks.splice(index, 1); 88 + } 89 + }) 90 + ); 83 91 }, 84 92 next(value: T) { 85 93 if (!ended) { ··· 98 106 99 107 export const empty: Source<any> = (sink: Sink<any>): void => { 100 108 let ended = false; 101 - sink(start((signal) => { 102 - if (signal === TalkbackKind.Close) { 103 - ended = true; 104 - } else if (!ended) { 105 - ended = true; 106 - sink(SignalKind.End); 107 - } 108 - })); 109 + sink( 110 + start(signal => { 111 + if (signal === TalkbackKind.Close) { 112 + ended = true; 113 + } else if (!ended) { 114 + ended = true; 115 + sink(SignalKind.End); 116 + } 117 + }) 118 + ); 109 119 }; 110 120 111 121 export const never: Source<any> = (sink: Sink<any>): void => { ··· 113 123 }; 114 124 115 125 export function interval(ms: number): Source<number> { 116 - return (sink) => { 126 + return sink => { 117 127 let i = 0; 118 128 const id = setInterval(() => { 119 129 sink(push(i++)); 120 130 }, ms); 121 - sink(start((signal) => { 122 - if (signal === TalkbackKind.Close) 123 - clearInterval(id); 124 - })); 131 + sink( 132 + start(signal => { 133 + if (signal === TalkbackKind.Close) clearInterval(id); 134 + }) 135 + ); 125 136 }; 126 137 } 127 138 128 139 export function fromDomEvent(element: HTMLElement, event: string): Source<Event> { 129 - return (sink) => { 140 + return sink => { 130 141 const handler = (payload: Event) => { 131 142 sink(push(payload)); 132 143 }; 133 - sink(start((signal) => { 134 - if (signal === TalkbackKind.Close) 135 - element.removeEventListener(event, handler); 136 - })); 144 + sink( 145 + start(signal => { 146 + if (signal === TalkbackKind.Close) element.removeEventListener(event, handler); 147 + }) 148 + ); 137 149 element.addEventListener(event, handler); 138 150 }; 139 151 } 140 152 141 153 export function fromPromise<T>(promise: Promise<T>): Source<T> { 142 - return (sink) => { 154 + return sink => { 143 155 let ended = false; 144 - promise.then((value) => { 156 + promise.then(value => { 145 157 if (!ended) { 146 158 sink(push(value)); 147 159 sink(SignalKind.End); 148 160 } 149 161 }); 150 - sink(start((signal) => { 151 - if (signal === TalkbackKind.Close) ended = true; 152 - })); 162 + sink( 163 + start(signal => { 164 + if (signal === TalkbackKind.Close) ended = true; 165 + }) 166 + ); 153 167 }; 154 168 }
+3 -3
src/types.ts
··· 13 13 } 14 14 15 15 export interface Tag<T> { 16 - tag: T 16 + tag: T; 17 17 } 18 18 19 - export type Start<_T> = (Tag<SignalKind.Start> & [TalkbackFn]) 20 - export type Push<T> = (Tag<SignalKind.Push> & [T]) 19 + export type Start<_T> = Tag<SignalKind.Start> & [TalkbackFn]; 20 + export type Push<T> = Tag<SignalKind.Push> & [T]; 21 21 export type Signal<T> = Start<T> | Push<T> | SignalKind.End; 22 22 23 23 export type Sink<T> = (signal: Signal<T>) => void;