From a4e79fe36095ec9aef034057f0b9d933ee4f35de Mon Sep 17 00:00:00 2001 From: Phil Pluckthun Date: Sat, 13 Aug 2022 14:07:58 +0100 Subject: [PATCH] Apply eslint/prettier fixes --- gatsby-config.js | 4 +- gatsby/theme.js | 8 +- gatsby/typography.js | 2 +- rollup.config.js | 20 +- scripts/eslint-preset.js | 10 +- scripts/flow-typings-plugin.js | 4 +- src/__tests__/operators.test.ts | 286 ++++++++---------- src/__tests__/sinks.test.ts | 220 +++++++------- src/__tests__/sources.test.ts | 123 ++------ src/callbag.ts | 24 +- src/helpers.ts | 14 +- src/index.ts | 14 +- src/observable.ts | 27 +- src/operators.ts | 510 +++++++++++++++++--------------- src/pipe.ts | 3 +- src/sinks.ts | 20 +- src/sources.ts | 154 +++++----- src/types.ts | 6 +- 18 files changed, 698 insertions(+), 751 deletions(-) diff --git a/gatsby-config.js b/gatsby-config.js index cea9dd1..c888666 100644 --- a/gatsby-config.js +++ b/gatsby-config.js @@ -3,8 +3,8 @@ module.exports = { title: 'Wonka', description: 'A fast push & pull stream library for Reason, OCaml, and TypeScript', siteUrl: 'https://wonka.kitten.sh', - githubUrl: 'https://github.com/kitten/wonka' + githubUrl: 'https://github.com/kitten/wonka', }, plugins: ['gatsby-plugin-netlify'], - __experimentalThemes: ['gatsby-theme-docs-system'] + __experimentalThemes: ['gatsby-theme-docs-system'], }; diff --git a/gatsby/theme.js b/gatsby/theme.js index c1f5b77..1fa8db6 100644 --- a/gatsby/theme.js +++ b/gatsby/theme.js @@ -13,13 +13,13 @@ const systemFonts = [ 'Apple Color Emoji', 'Segoe UI Emoji', 'Segoe UI Symbol', - 'Noto Color Emoji' + 'Noto Color Emoji', ]; export const fonts = { header: ['phantom-sans', ...systemFonts], code: ['space-mono', 'monospace'], - body: systemFonts + body: systemFonts, }; export const colors = { @@ -29,7 +29,7 @@ export const colors = { fg: '#36313d', fgHeading: '#000000', fgPassive: '#78757a', - fgActive: '#f5735f' + fgActive: '#f5735f', }; export const prismTheme = nightOwlLight; @@ -46,5 +46,5 @@ export const zIndices = [0, 1, 2, 3]; export const borders = [ `${borderWidths[0]} solid ${colors.bgPassive}`, `${borderWidths[1]} solid ${colors.bgPassive}`, - `${borderWidths[2]} solid ${colors.bgPassive}` + `${borderWidths[2]} solid ${colors.bgPassive}`, ]; diff --git a/gatsby/typography.js b/gatsby/typography.js index 92cb4e2..3b4da25 100644 --- a/gatsby/typography.js +++ b/gatsby/typography.js @@ -9,7 +9,7 @@ const typography = new Typography({ headerFontFamily: theme.fonts.header, headerWeight: '500', bodyFontFamily: theme.fonts.body, - bodyWeight: '400' + bodyWeight: '400', }); export default typography; diff --git a/rollup.config.js b/rollup.config.js index a4af434..c1a66ac 100644 --- a/rollup.config.js +++ b/rollup.config.js @@ -22,10 +22,7 @@ const plugins = [ typescript({ typescript: require('typescript'), - exclude: [ - 'src/**/*.test.ts', - '**/__tests__/*', - ], + exclude: ['src/**/*.test.ts', '**/__tests__/*'], compilerOptions: { sourceMap: true, noEmit: false, @@ -49,7 +46,7 @@ const plugins = [ templateString: false, objectRestSpread: false, }, - exclude: 'node_modules/**' + exclude: 'node_modules/**', }), terser({ @@ -68,7 +65,7 @@ const plugins = [ sequences: false, loops: false, conditionals: false, - join_vars: false + join_vars: false, }, mangle: { module: true, @@ -77,12 +74,12 @@ const plugins = [ output: { beautify: true, braces: true, - indent_level: 2 - } + indent_level: 2, + }, }), ]; -const output = (format) => { +const output = format => { const extension = format === 'esm' ? '.mjs' : '.js'; return { chunkFileNames: '[hash]' + extension, @@ -119,10 +116,7 @@ const config = { tryCatchDeoptimization: false, moduleSideEffects: false, }, - output: [ - output('esm'), - output('cjs'), - ], + output: [output('esm'), output('cjs')], }; export default config; diff --git a/scripts/eslint-preset.js b/scripts/eslint-preset.js index 7ebf7ef..bf3dbbd 100644 --- a/scripts/eslint-preset.js +++ b/scripts/eslint-preset.js @@ -13,6 +13,7 @@ module.exports = { 'sort-keys': 'off', 'no-console': ['error', { allow: ['warn', 'error'] }], 'prefer-arrow/prefer-arrow-functions': 'off', + 'prefer-rest-params': 'off', 'prettier/prettier': [ 'error', @@ -55,12 +56,9 @@ module.exports = { '@typescript-eslint/no-misused-new': 'off', '@typescript-eslint/no-explicit-any': 'off', '@typescript-eslint/array-type': 'off', - '@typescript-eslint/no-unused-vars': [ - 'error', - { - argsIgnorePattern: '^_', - }, - ], + '@typescript-eslint/no-empty-function': 'off', + '@typescript-eslint/no-unused-vars': 'off', + 'prefer-rest-params': 'off', }, }, ], diff --git a/scripts/flow-typings-plugin.js b/scripts/flow-typings-plugin.js index da4fd1f..739737f 100644 --- a/scripts/flow-typings-plugin.js +++ b/scripts/flow-typings-plugin.js @@ -5,7 +5,7 @@ import { compiler, beautify } from 'flowgen'; function flowTypings() { return { - name: "flow-typings", + name: 'flow-typings', async writeBundle() { const cwd = process.cwd(); for (const file of glob('dist/types/**/*.d.ts')) { @@ -17,7 +17,7 @@ function flowTypings() { const definition = flowdef.replace(/import/g, 'import type'); writeFileSync(newpath, '// @flow\n\n' + definition); } - } + }, }; } diff --git a/src/__tests__/operators.test.ts b/src/__tests__/operators.test.ts index 54b80f4..1155096 100644 --- a/src/__tests__/operators.test.ts +++ b/src/__tests__/operators.test.ts @@ -9,25 +9,24 @@ import * as operators from '../operators'; A Pull will be sent from the sink upwards and should pass through the operator until the source receives it, which then pushes a value down. */ -const passesPassivePull = ( - operator: Operator, - output: any = 0 -) => { +const passesPassivePull = (operator: Operator, output: any = 0) => { it('responds to Pull talkback signals (spec)', () => { let talkback: TalkbackFn | null = null; let pushes = 0; const values: any[] = []; const source: Source = sink => { - sink(start((signal) => { - if (!pushes && signal === TalkbackKind.Pull) { - pushes++; - sink(push(0)); - } - })); + sink( + start(signal => { + if (!pushes && signal === TalkbackKind.Pull) { + pushes++; + sink(push(0)); + } + }) + ); }; - const sink: Sink = (signal) => { + const sink: Sink = signal => { expect(signal).not.toBe(SignalKind.End); if (signal === SignalKind.End) { /*noop*/ @@ -55,24 +54,22 @@ const passesPassivePull = ( A Push will be sent downwards from the source, through the operator to the sink. Pull events should be let through from the sink after every Push event. */ -const passesActivePush = ( - operator: Operator, - result: any = 0 -) => { +const passesActivePush = (operator: Operator, result: any = 0) => { it('responds to eager Push signals (spec)', () => { const values: any[] = []; let talkback: TalkbackFn | null = null; let sink: Sink | null = null; let pulls = 0; - const source: Source = (_sink) => { - (sink = _sink)(start((signal) => { - if (signal === TalkbackKind.Pull) - pulls++; - })); + const source: Source = _sink => { + (sink = _sink)( + start(signal => { + if (signal === TalkbackKind.Pull) pulls++; + }) + ); }; - operator(source)((signal) => { + operator(source)(signal => { expect(signal).not.toBe(SignalKind.End); if (signal === SignalKind.End) { /*noop*/ @@ -105,13 +102,15 @@ const passesSinkClose = (operator: Operator) => { let closing = 0; const source: Source = sink => { - sink(start((signal) => { - if (signal === TalkbackKind.Pull && !closing) { - sink(push(0)); - } else if (signal === TalkbackKind.Close) { - closing++; - } - })); + sink( + start(signal => { + if (signal === TalkbackKind.Pull && !closing) { + sink(push(0)); + } else if (signal === TalkbackKind.Close) { + closing++; + } + }) + ); }; const sink: Sink = signal => { @@ -138,10 +137,7 @@ const passesSinkClose = (operator: Operator) => { A Push and End signal will be sent after the first Pull talkback signal from the sink, which shouldn't lead to any extra Close or Pull talkback signals. */ -const passesSourceEnd = ( - operator: Operator, - result: any = 0 -) => { +const passesSourceEnd = (operator: Operator, result: any = 0) => { it('passes on immediate Push then End signals from source (spec)', () => { const signals: Signal[] = []; let talkback: TalkbackFn | null = null; @@ -149,16 +145,18 @@ const passesSourceEnd = ( let ending = 0; const source: Source = sink => { - sink(start((signal) => { - expect(signal).not.toBe(TalkbackKind.Close); - if (signal === TalkbackKind.Pull) { - pulls++; - if (pulls === 1) { - sink(push(0)); - sink(SignalKind.End); + sink( + start(signal => { + expect(signal).not.toBe(TalkbackKind.Close); + if (signal === TalkbackKind.Pull) { + pulls++; + if (pulls === 1) { + sink(push(0)); + sink(SignalKind.End); + } } - } - })); + }) + ); }; const sink: Sink = signal => { @@ -188,10 +186,7 @@ const passesSourceEnd = ( after the first pull in response to another. This is similar to passesSourceEnd but more well behaved since mergeMap/switchMap/concatMap are eager operators. */ -const passesSourcePushThenEnd = ( - operator: Operator, - result: any = 0 -) => { +const passesSourcePushThenEnd = (operator: Operator, result: any = 0) => { it('passes on End signals from source (spec)', () => { const signals: Signal[] = []; let talkback: TalkbackFn | null = null; @@ -199,14 +194,19 @@ const passesSourcePushThenEnd = ( let ending = 0; const source: Source = sink => { - sink(start((signal) => { - expect(signal).not.toBe(TalkbackKind.Close); - if (signal === TalkbackKind.Pull) { - pulls++; - if (pulls <= 2) { sink(push(0)); } - else { sink(SignalKind.End); } - } - })); + sink( + start(signal => { + expect(signal).not.toBe(TalkbackKind.Close); + if (signal === TalkbackKind.Pull) { + pulls++; + if (pulls <= 2) { + sink(push(0)); + } else { + sink(SignalKind.End); + } + } + }) + ); }; const sink: Sink = signal => { @@ -228,11 +228,7 @@ const passesSourcePushThenEnd = ( jest.runAllTimers(); expect(ending).toBe(1); expect(pulls).toBe(3); - expect(signals).toEqual([ - push(result), - push(result), - SignalKind.End - ]); + expect(signals).toEqual([push(result), push(result), SignalKind.End]); }); }; @@ -272,13 +268,15 @@ const passesStrictEnd = (operator: Operator) => { const signals: Signal[] = []; const source: Source = sink => { - sink(start((signal) => { - if (signal === TalkbackKind.Pull) { - pulls++; - sink(SignalKind.End); - sink(push(123)); - } - })); + sink( + start(signal => { + if (signal === TalkbackKind.Pull) { + pulls++; + sink(SignalKind.End); + sink(push(123)); + } + }) + ); }; const sink: Sink = signal => { @@ -303,11 +301,13 @@ const passesStrictEnd = (operator: Operator) => { const signals: Signal[] = []; const source: Source = sink => { - sink(start((signal) => { - if (signal === TalkbackKind.Close) { - sink(push(123)); - } - })); + sink( + start(signal => { + if (signal === TalkbackKind.Close) { + sink(push(123)); + } + }) + ); }; const sink: Sink = signal => { @@ -339,14 +339,16 @@ const passesCloseAndEnd = (closingOperator: Operator) => { let ending = 0; const source: Source = sink => { - sink(start((signal) => { - // For some operator tests we do need to send a single value - if (signal === TalkbackKind.Pull) { - sink(push(null)); - } else { - closing++; - } - })); + sink( + start(signal => { + // For some operator tests we do need to send a single value + if (signal === TalkbackKind.Pull) { + sink(push(null)); + } else { + closing++; + } + }) + ); }; const sink: Sink = signal => { @@ -364,22 +366,21 @@ const passesCloseAndEnd = (closingOperator: Operator) => { }); }; -const passesAsyncSequence = ( - operator: Operator, - result: any = 0 -) => { +const passesAsyncSequence = (operator: Operator, result: any = 0) => { it('passes an async push with an async end (spec)', () => { let hasPushed = false; const signals: Signal[] = []; const source: Source = sink => { - sink(start((signal) => { - if (signal === TalkbackKind.Pull && !hasPushed) { - hasPushed = true; - setTimeout(() => sink(push(0)), 10); - setTimeout(() => sink(SignalKind.End), 20); - } - })); + sink( + start(signal => { + if (signal === TalkbackKind.Pull && !hasPushed) { + hasPushed = true; + setTimeout(() => sink(push(0)), 10); + setTimeout(() => sink(SignalKind.End), 20); + } + }) + ); }; const sink: Sink = signal => { @@ -402,10 +403,7 @@ const passesAsyncSequence = ( expect(hasPushed).toBeTruthy(); jest.runAllTimers(); - expect(signals).toEqual([ - push(result), - SignalKind.End - ]); + expect(signals).toEqual([push(result), SignalKind.End]); }); }; @@ -439,10 +437,11 @@ describe('combine', () => { describe('buffer', () => { const valueThenNever: Source = sink => - sink(start((signal) => { - if (signal === TalkbackKind.Pull) - sink(push(null)); - })); + sink( + start(signal => { + if (signal === TalkbackKind.Pull) sink(push(null)); + }) + ); const noop = operators.buffer(valueThenNever); @@ -522,9 +521,7 @@ describe('concatMap', () => { })(sources.fromValue(null))(fn); expect(teardown).toHaveBeenCalled(); - expect(signals).toEqual([ - start(expect.any(Function)), - ]); + expect(signals).toEqual([start(expect.any(Function))]); }); // This asynchronous test for concatMap will behave differently than mergeMap & switchMap @@ -539,18 +536,10 @@ describe('concatMap', () => { ); jest.advanceTimersByTime(14); - expect(fn.mock.calls).toEqual([ - [1], - [2], - ]); + expect(fn.mock.calls).toEqual([[1], [2]]); jest.runAllTimers(); - expect(fn.mock.calls).toEqual([ - [1], - [2], - [10], - [20], - ]); + expect(fn.mock.calls).toEqual([[1], [2], [10], [20]]); }); it('works for fully asynchronous sources', () => { @@ -561,7 +550,7 @@ describe('concatMap', () => { return sources.make(observer => { setTimeout(() => observer.next(1)); return () => {}; - }) + }); })(sources.fromValue(null)) ); @@ -573,13 +562,10 @@ describe('concatMap', () => { const values: any[] = []; sinks.forEach(x => values.push(x))( - operators.concat([ - sources.fromArray([1, 2]), - sources.fromArray([3, 4]) - ]) + operators.concat([sources.fromArray([1, 2]), sources.fromArray([3, 4])]) ); - expect(values).toEqual([ 1, 2, 3, 4 ]); + expect(values).toEqual([1, 2, 3, 4]); }); }); @@ -738,9 +724,7 @@ describe('mergeMap', () => { })(sources.fromValue(null))(fn); expect(teardown).toHaveBeenCalled(); - expect(values).toEqual([ - start(expect.any(Function)), - ]); + expect(values).toEqual([start(expect.any(Function))]); }); // This asynchronous test for mergeMap will behave differently than concatMap & switchMap @@ -755,25 +739,17 @@ describe('mergeMap', () => { ); jest.runAllTimers(); - expect(fn.mock.calls).toEqual([ - [1], - [10], - [2], - [20], - ]); + expect(fn.mock.calls).toEqual([[1], [10], [2], [20]]); }); it('emits synchronous values in order', () => { const values: any[] = []; sinks.forEach(x => values.push(x))( - operators.merge([ - sources.fromArray([1, 2]), - sources.fromArray([3, 4]) - ]) + operators.merge([sources.fromArray([1, 2]), sources.fromArray([3, 4])]) ); - expect(values).toEqual([ 1, 2, 3, 4 ]); + expect(values).toEqual([1, 2, 3, 4]); }); }); @@ -841,7 +817,9 @@ describe('onStart', () => { let sink: Sink; const fn = jest.fn(); - const source: Source = _sink => { sink = _sink; }; + const source: Source = _sink => { + sink = _sink; + }; sinks.forEach(() => {})(operators.onStart(fn)(source)); @@ -854,10 +832,11 @@ describe('onStart', () => { describe('sample', () => { const valueThenNever: Source = sink => - sink(start((signal) => { - if (signal === TalkbackKind.Pull) - sink(push(null)); - })); + sink( + start(signal => { + if (signal === TalkbackKind.Pull) sink(push(null)); + }) + ); const noop = operators.sample(valueThenNever); @@ -1059,9 +1038,7 @@ describe('switchMap', () => { })(sources.fromValue(null))(fn); expect(teardown).toHaveBeenCalled(); - expect(signals).toEqual([ - start(expect.any(Function)), - ]); + expect(signals).toEqual([start(expect.any(Function))]); }); // This asynchronous test for switchMap will behave differently than concatMap & mergeMap @@ -1070,17 +1047,13 @@ describe('switchMap', () => { const fn = jest.fn(); sinks.forEach(fn)( - operators.switchMap((x: number) => ( + operators.switchMap((x: number) => operators.take(2)(operators.map((y: number) => x * (y + 1))(sources.interval(5))) - ))(source) + )(source) ); jest.runAllTimers(); - expect(fn.mock.calls).toEqual([ - [1], - [10], - [20], - ]); + expect(fn.mock.calls).toEqual([[1], [10], [20]]); }); }); @@ -1104,11 +1077,7 @@ describe('take', () => { next(1); expect(fn).toHaveBeenCalledTimes(3); - expect(fn.mock.calls).toEqual([ - [start(expect.any(Function))], - [push(1)], - [SignalKind.End], - ]); + expect(fn.mock.calls).toEqual([[start(expect.any(Function))], [push(1)], [SignalKind.End]]); }); }); @@ -1134,10 +1103,7 @@ describe('takeUntil', () => { next(1); expect(fn).toHaveBeenCalledTimes(2); - expect(fn.mock.calls).toEqual([ - [start(expect.any(Function))], - [push(1)], - ]); + expect(fn.mock.calls).toEqual([[start(expect.any(Function))], [push(1)]]); notify(null); expect(fn).toHaveBeenCalledTimes(3); @@ -1165,11 +1131,7 @@ describe('takeWhile', () => { next(1); next(2); - expect(fn.mock.calls).toEqual([ - [start(expect.any(Function))], - [push(1)], - [SignalKind.End], - ]); + expect(fn.mock.calls).toEqual([[start(expect.any(Function))], [push(1)], [SignalKind.End]]); }); }); @@ -1199,11 +1161,7 @@ describe('takeLast', () => { expect(signals.length).toBe(0); complete(); - expect(signals).toEqual([ - start(expect.any(Function)), - push(2), - SignalKind.End, - ]); + expect(signals).toEqual([start(expect.any(Function)), push(2), SignalKind.End]); }); }); diff --git a/src/__tests__/sinks.test.ts b/src/__tests__/sinks.test.ts index 5ce29f7..f7d9cf6 100644 --- a/src/__tests__/sinks.test.ts +++ b/src/__tests__/sinks.test.ts @@ -15,18 +15,20 @@ describe('subscribe', () => { let pulls = 0; const fn = jest.fn(); - const source: Source = (sink) => { - sink(start((signal) => { - if (signal === TalkbackKind.Pull) { - if (pulls < 3) { - pulls++; - sink(push(0)); - } else { - sink(SignalKind.End); - expect(pulls).toBe(3); + const source: Source = sink => { + sink( + start(signal => { + if (signal === TalkbackKind.Pull) { + if (pulls < 3) { + pulls++; + sink(push(0)); + } else { + sink(SignalKind.End); + expect(pulls).toBe(3); + } } - } - })); + }) + ); }; sinks.subscribe(fn)(source); @@ -38,17 +40,19 @@ describe('subscribe', () => { let pulls = 0; let closing = 0; - const source: Source = (sink) => { - sink(start((signal) => { - if (signal === TalkbackKind.Pull) { - if (!pulls) { - pulls++; - sink(push(0)); + const source: Source = sink => { + sink( + start(signal => { + if (signal === TalkbackKind.Pull) { + if (!pulls) { + pulls++; + sink(push(0)); + } + } else { + closing++; } - } else { - closing++; - } - })); + }) + ); }; const sub = sinks.subscribe(() => {})(source); @@ -62,15 +66,17 @@ describe('subscribe', () => { let pulls = 0; let closing = 0; - const source: Source = (sink) => { - sink(start((signal) => { - if (signal === TalkbackKind.Pull) { - pulls++; - sink(SignalKind.End); - } else { - closing++; - } - })); + const source: Source = sink => { + sink( + start(signal => { + if (signal === TalkbackKind.Pull) { + pulls++; + sink(SignalKind.End); + } else { + closing++; + } + }) + ); }; const sub = sinks.subscribe(() => {})(source); @@ -81,13 +87,15 @@ describe('subscribe', () => { it('ignores Push signals after the source has ended', () => { const fn = jest.fn(); - const source: Source = (sink) => { - sink(start((signal) => { - if (signal === TalkbackKind.Pull) { - sink(SignalKind.End); - sink(push(0)); - } - })); + const source: Source = sink => { + sink( + start(signal => { + if (signal === TalkbackKind.Pull) { + sink(SignalKind.End); + sink(push(0)); + } + }) + ); }; sinks.subscribe(fn)(source); @@ -96,12 +104,14 @@ describe('subscribe', () => { it('ignores Push signals after cancellation', () => { const fn = jest.fn(); - const source: Source = (sink) => { - sink(start((signal) => { - if (signal === TalkbackKind.Close) { - sink(push(0)); - } - })); + const source: Source = sink => { + sink( + start(signal => { + if (signal === TalkbackKind.Close) { + sink(push(0)); + } + }) + ); }; sinks.subscribe(fn)(source).unsubscribe(); @@ -112,18 +122,20 @@ describe('subscribe', () => { describe('publish', () => { it('sends Pull talkback signals every Push signal', () => { let pulls = 0; - const source: Source = (sink) => { - sink(start((signal) => { - if (signal === TalkbackKind.Pull) { - if (pulls < 3) { - pulls++; - sink(push(0)); - } else { - sink(SignalKind.End); - expect(pulls).toBe(3); + const source: Source = sink => { + sink( + start(signal => { + if (signal === TalkbackKind.Pull) { + if (pulls < 3) { + pulls++; + sink(push(0)); + } else { + sink(SignalKind.End); + expect(pulls).toBe(3); + } } - } - })); + }) + ); }; sinks.publish(source); @@ -134,18 +146,20 @@ describe('publish', () => { describe('toArray', () => { it('sends Pull talkback signals every Push signal', () => { let pulls = 0; - const source: Source = (sink) => { - sink(start((signal) => { - if (signal === TalkbackKind.Pull) { - if (pulls < 3) { - pulls++; - sink(push(0)); - } else { - sink(SignalKind.End); - expect(pulls).toBe(3); + const source: Source = sink => { + sink( + start(signal => { + if (signal === TalkbackKind.Pull) { + if (pulls < 3) { + pulls++; + sink(push(0)); + } else { + sink(SignalKind.End); + expect(pulls).toBe(3); + } } - } - })); + }) + ); }; const array = sinks.toArray(source); @@ -157,17 +171,19 @@ describe('toArray', () => { let pulls = 0; let ending = 0; - const source: Source = (sink) => { - sink(start((signal) => { - if (signal === TalkbackKind.Pull) { - if (!pulls) { - pulls++; - sink(push(0)); + const source: Source = sink => { + sink( + start(signal => { + if (signal === TalkbackKind.Pull) { + if (!pulls) { + pulls++; + sink(push(0)); + } + } else { + ending++; } - } else { - ending++; - } - })); + }) + ); }; const array = sinks.toArray(source); @@ -181,12 +197,13 @@ describe('toPromise', () => { let pulls = 0; let sink: Sink | null = null; - const source: Source = (_sink) => { + const source: Source = _sink => { sink = _sink; - sink(start((signal) => { - if (signal === TalkbackKind.Pull) - pulls++; - })); + sink( + start(signal => { + if (signal === TalkbackKind.Pull) pulls++; + }) + ); }; const fn = jest.fn(); @@ -219,10 +236,11 @@ describe('toObservable', () => { const source: Source = _sink => { sink = _sink; - sink(start((signal) => { - if (signal === TalkbackKind.Pull) - pulls++; - })); + sink( + start(signal => { + if (signal === TalkbackKind.Pull) pulls++; + }) + ); }; Observable.from(observable.toObservable(source) as any).subscribe({ @@ -242,10 +260,11 @@ describe('toObservable', () => { it('forwards cancellations from the Observable as a talkback', () => { let ending = 0; const source: Source = sink => - sink(start((signal) => { - if (signal === TalkbackKind.Close) - ending++; - })); + sink( + start(signal => { + if (signal === TalkbackKind.Close) ending++; + }) + ); const sub = Observable.from(observable.toObservable(source) as any).subscribe({}); @@ -263,10 +282,11 @@ describe('toCallbag', () => { const source: Source = _sink => { sink = _sink; - sink(start((signal) => { - if (signal === TalkbackKind.Pull) - pulls++; - })); + sink( + start(signal => { + if (signal === TalkbackKind.Pull) pulls++; + }) + ); }; callbagIterate(fn)(callbag.toCallbag(source)); @@ -284,13 +304,15 @@ describe('toCallbag', () => { const fn = jest.fn(); const source: Source = sink => - sink(start((signal) => { - if (signal === TalkbackKind.Pull) { - sink(push(0)); - } else { - ending++; - } - })); + sink( + start(signal => { + if (signal === TalkbackKind.Pull) { + sink(push(0)); + } else { + ending++; + } + }) + ); callbagIterate(fn)(callbagTake(1)(callbag.toCallbag(source) as any)); diff --git a/src/__tests__/sources.test.ts b/src/__tests__/sources.test.ts index 4bf125b..a7e3b49 100644 --- a/src/__tests__/sources.test.ts +++ b/src/__tests__/sources.test.ts @@ -9,10 +9,7 @@ import * as observable from '../observable'; import callbagFromArray from 'callbag-from-iter'; import Observable from 'zen-observable'; -const collectSignals = ( - source: Source, - onStart?: (talkbackCb: TalkbackFn) => void -) => { +const collectSignals = (source: Source, onStart?: (talkbackCb: TalkbackFn) => void) => { let talkback = talkbackPlaceholder; const signals: Signal[] = []; source(signal => { @@ -26,7 +23,7 @@ const collectSignals = ( } else { talkback(TalkbackKind.Pull); } - }) + }); return signals; }; @@ -103,11 +100,7 @@ const passesTrampoline = (source: Source) => { }; source(sink); - expect(signals).toEqual([ - push(1), - push(2), - SignalKind.End, - ]); + expect(signals).toEqual([push(1), push(2), SignalKind.End]); }); }; @@ -135,25 +128,15 @@ describe('fromValue', () => { }); describe('merge', () => { - const source = operators.merge([ - sources.fromValue(0), - sources.empty - ]); + const source = operators.merge([sources.fromValue(0), sources.empty]); passesColdPull(source); passesActiveClose(source); it('correctly merges two sources where the second is empty', () => { - const source = operators.merge([ - sources.fromValue(0), - sources.empty - ]); + const source = operators.merge([sources.fromValue(0), sources.empty]); - expect(collectSignals(source)).toEqual([ - start(expect.any(Function)), - push(0), - SignalKind.End, - ]); + expect(collectSignals(source)).toEqual([start(expect.any(Function)), push(0), SignalKind.End]); }); it('correctly merges hot sources', () => { @@ -166,11 +149,7 @@ describe('merge', () => { const signals = collectSignals(source); expect(onStart).toHaveBeenCalledTimes(2); - expect(signals).toEqual([ - start(expect.any(Function)), - push(1), - push(2), - ]); + expect(signals).toEqual([start(expect.any(Function)), push(1), push(2)]); }); it('correctly merges asynchronous sources', () => { @@ -179,9 +158,7 @@ describe('merge', () => { const onStart = jest.fn(); const source = operators.merge([ operators.onStart(onStart)(sources.fromValue(-1)), - operators.onStart(onStart)( - operators.take(2)(sources.interval(50)) - ), + operators.onStart(onStart)(operators.take(2)(sources.interval(50))), ]); const signals = collectSignals(source); @@ -199,25 +176,15 @@ describe('merge', () => { }); describe('concat', () => { - const source = operators.concat([ - sources.fromValue(0), - sources.empty - ]); + const source = operators.concat([sources.fromValue(0), sources.empty]); passesColdPull(source); passesActiveClose(source); it('correctly concats two sources where the second is empty', () => { - const source = operators.concat([ - sources.fromValue(0), - sources.empty - ]); + const source = operators.concat([sources.fromValue(0), sources.empty]); - expect(collectSignals(source)).toEqual([ - start(expect.any(Function)), - push(0), - SignalKind.End, - ]); + expect(collectSignals(source)).toEqual([start(expect.any(Function)), push(0), SignalKind.End]); }); }); @@ -234,11 +201,7 @@ describe('make', () => { expect(signals).toEqual([start(expect.any(Function))]); jest.runAllTimers(); - expect(signals).toEqual([ - start(expect.any(Function)), - push(1), - SignalKind.End, - ]); + expect(signals).toEqual([start(expect.any(Function)), push(1), SignalKind.End]); }); it('supports active cancellation', () => { @@ -263,24 +226,15 @@ describe('makeSubject', () => { const { source, next, complete } = sources.makeSubject(); const signals = collectSignals(source); - expect(signals).toEqual([ - start(expect.any(Function)), - ]); + expect(signals).toEqual([start(expect.any(Function))]); next(1); - expect(signals).toEqual([ - start(expect.any(Function)), - push(1), - ]); + expect(signals).toEqual([start(expect.any(Function)), push(1)]); complete(); - expect(signals).toEqual([ - start(expect.any(Function)), - push(1), - SignalKind.End, - ]); + expect(signals).toEqual([start(expect.any(Function)), push(1), SignalKind.End]); }); it('ignores signals after complete has been called', () => { @@ -288,10 +242,7 @@ describe('makeSubject', () => { const signals = collectSignals(source); complete(); - expect(signals).toEqual([ - start(expect.any(Function)), - SignalKind.End, - ]); + expect(signals).toEqual([start(expect.any(Function)), SignalKind.End]); next(1); complete(); @@ -302,7 +253,7 @@ describe('makeSubject', () => { describe('never', () => { it('emits nothing and ends immediately', () => { const signals = collectSignals(sources.never); - expect(signals).toEqual([start(expect.any(Function)) ]); + expect(signals).toEqual([start(expect.any(Function))]); }); }); @@ -310,10 +261,7 @@ describe('empty', () => { it('emits nothing and ends immediately', () => { const signals = collectSignals(sources.empty); - expect(signals).toEqual([ - start(expect.any(Function)), - SignalKind.End, - ]); + expect(signals).toEqual([start(expect.any(Function)), SignalKind.End]); }); }); @@ -324,17 +272,11 @@ describe('fromPromise', () => { const promise = Promise.resolve(1); const signals = collectSignals(sources.fromPromise(promise)); - expect(signals).toEqual([ - start(expect.any(Function)), - ]); + expect(signals).toEqual([start(expect.any(Function))]); await promise; - expect(signals).toEqual([ - start(expect.any(Function)), - push(1), - SignalKind.End, - ]); + expect(signals).toEqual([start(expect.any(Function)), push(1), SignalKind.End]); }); }); @@ -349,12 +291,7 @@ describe('fromObservable', () => { await new Promise(resolve => setTimeout(resolve)); - expect(signals).toEqual([ - start(expect.any(Function)), - push(1), - push(2), - SignalKind.End, - ]); + expect(signals).toEqual([start(expect.any(Function)), push(1), push(2), SignalKind.End]); }); it('supports cancellation on converted Observables', async () => { @@ -365,9 +302,7 @@ describe('fromObservable', () => { await new Promise(resolve => setTimeout(resolve)); - expect(signals).toEqual([ - start(expect.any(Function)), - ]); + expect(signals).toEqual([start(expect.any(Function))]); }); }); @@ -376,12 +311,7 @@ describe('fromCallbag', () => { const source = callbag.fromCallbag(callbagFromArray([1, 2]) as any); const signals = collectSignals(source); - expect(signals).toEqual([ - start(expect.any(Function)), - push(1), - push(2), - SignalKind.End, - ]); + expect(signals).toEqual([start(expect.any(Function)), push(1), push(2), SignalKind.End]); }); it('supports cancellation on converted Observables', () => { @@ -390,9 +320,7 @@ describe('fromCallbag', () => { talkback(TalkbackKind.Close); }); - expect(signals).toEqual([ - start(expect.any(Function)), - ]); + expect(signals).toEqual([start(expect.any(Function))]); }); }); @@ -437,8 +365,7 @@ describe('fromDomEvent', () => { const sink: Sink = signal => { expect(signal).not.toBe(SignalKind.End); - if ((signal as any).tag === SignalKind.Start) - talkback = signal[0]; + if ((signal as any).tag === SignalKind.Start) talkback = signal[0]; }; sources.fromDomEvent(element as any, 'click')(sink); diff --git a/src/callbag.ts b/src/callbag.ts index 169eefa..f5d63b5 100644 --- a/src/callbag.ts +++ b/src/callbag.ts @@ -1,5 +1,5 @@ -import { Source, SignalKind, TalkbackKind } from './types' -import { push, start } from './helpers' +import { Source, SignalKind, TalkbackKind } from './types'; +import { push, start } from './helpers'; interface Callbag { (t: 0, d: Callbag): void; @@ -8,16 +8,18 @@ interface Callbag { } export function fromCallbag(callbag: Callbag): Source { - return (sink) => { + return sink => { callbag(0, (signal: number, data: any) => { if (signal === 0) { - sink(start((signal) => { - if (signal === TalkbackKind.Pull) { - data(1); - } else { - data(2); - } - })); + sink( + start(signal => { + if (signal === TalkbackKind.Pull) { + data(1); + } else { + data(2); + } + }) + ); } else if (signal === 1) { sink(push(data)); } else if (signal === 2) { @@ -30,7 +32,7 @@ export function fromCallbag(callbag: Callbag): Source { export function toCallbag(source: Source): Callbag { return (signal: number, sink: any) => { if (signal === 0) { - source((signal) => { + source(signal => { if (signal === SignalKind.End) { sink(2); } else if (signal.tag === SignalKind.Start) { diff --git a/src/helpers.ts b/src/helpers.ts index 6bf521a..d166eaa 100644 --- a/src/helpers.ts +++ b/src/helpers.ts @@ -1,16 +1,20 @@ -import { TalkbackFn, TeardownFn, Start, Push, SignalKind } from './types' +import { TalkbackFn, TeardownFn, Start, Push, SignalKind } from './types'; -export const talkbackPlaceholder: TalkbackFn = (_signal) => {/*noop*/}; -export const teardownPlaceholder: TeardownFn = () => {/*noop*/}; +export const talkbackPlaceholder: TalkbackFn = _signal => { + /*noop*/ +}; +export const teardownPlaceholder: TeardownFn = () => { + /*noop*/ +}; export function start(talkback: TalkbackFn): Start { const box: any = [talkback]; box.tag = SignalKind.Start; - return box + return box; } export function push(value: T): Push { const box: any = [value]; box.tag = SignalKind.Push; - return box + return box; } diff --git a/src/index.ts b/src/index.ts index 00f4a02..a376594 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,7 +1,7 @@ -export * from './types' -export * from './sources' -export * from './operators' -export * from './sinks' -export * from './observable' -export * from './callbag' -export * from './pipe' +export * from './types'; +export * from './sources'; +export * from './operators'; +export * from './sinks'; +export * from './observable'; +export * from './callbag'; +export * from './pipe'; diff --git a/src/observable.ts b/src/observable.ts index 7aa7e0f..e09a921 100644 --- a/src/observable.ts +++ b/src/observable.ts @@ -1,5 +1,5 @@ -import { Source, SignalKind, TalkbackKind } from './types' -import { push, start, talkbackPlaceholder } from './helpers' +import { Source, SignalKind, TalkbackKind } from './types'; +import { push, start, talkbackPlaceholder } from './helpers'; interface ObservableSubscription { closed?: boolean; @@ -16,15 +16,16 @@ interface Observable { subscribe(observer: ObservableObserver): ObservableSubscription; } -const observableSymbol: unique symbol = typeof Symbol === 'function' - ? (Symbol as any).observable || ((Symbol as any).observable = Symbol('observable')) - : '@@observable' +const observableSymbol: unique symbol = + typeof Symbol === 'function' + ? (Symbol as any).observable || ((Symbol as any).observable = Symbol('observable')) + : '@@observable'; export function fromObservable(input: Observable): Source { const observable: Observable = input[observableSymbol] ? (input as any)[observableSymbol]() : input; - return (sink) => { + return sink => { const subscription = observable.subscribe({ next(value: T) { sink(push(value)); @@ -32,11 +33,15 @@ export function fromObservable(input: Observable): Source { complete() { sink(SignalKind.End); }, - error() {/*noop*/}, + error() { + /*noop*/ + }, }); - sink(start((signal) => { - if (signal === TalkbackKind.Close) subscription.unsubscribe(); - })); + sink( + start(signal => { + if (signal === TalkbackKind.Close) subscription.unsubscribe(); + }) + ); }; } @@ -45,7 +50,7 @@ export function toObservable(source: Source): Observable { subscribe(observer: ObservableObserver) { let talkback = talkbackPlaceholder; let ended = false; - source((signal) => { + source(signal => { if (ended) { /*noop*/ } else if (signal === SignalKind.End) { diff --git a/src/operators.ts b/src/operators.ts index 41b94f1..ba377a9 100644 --- a/src/operators.ts +++ b/src/operators.ts @@ -1,17 +1,17 @@ -import { Source, Sink, Operator, SignalKind, TalkbackKind, TalkbackFn } from './types' -import { push, start, talkbackPlaceholder } from './helpers' -import { fromArray } from './sources' +import { Source, Sink, Operator, SignalKind, TalkbackKind, TalkbackFn } from './types'; +import { push, start, talkbackPlaceholder } from './helpers'; +import { fromArray } from './sources'; const identity = (x: T): T => x; export function buffer(notifier: Source): Operator { - return (source) => (sink) => { + return source => sink => { let buffer: T[] = []; let sourceTalkback = talkbackPlaceholder; let notifierTalkback = talkbackPlaceholder; let pulled = false; let ended = false; - source((signal) => { + source(signal => { if (ended) { /*noop*/ } else if (signal === SignalKind.End) { @@ -21,7 +21,7 @@ export function buffer(notifier: Source): Operator { sink(SignalKind.End); } else if (signal.tag === SignalKind.Start) { sourceTalkback = signal[0]; - notifier((signal) => { + notifier(signal => { if (ended) { /*noop*/ } else if (signal === SignalKind.End) { @@ -48,22 +48,24 @@ export function buffer(notifier: Source): Operator { } } }); - sink(start((signal) => { - if (signal === TalkbackKind.Close && !ended) { - ended = true; - sourceTalkback(TalkbackKind.Close); - notifierTalkback(TalkbackKind.Close); - } else if (!ended && !pulled) { - pulled = true; - sourceTalkback(TalkbackKind.Pull); - notifierTalkback(TalkbackKind.Pull); - } - })); + sink( + start(signal => { + if (signal === TalkbackKind.Close && !ended) { + ended = true; + sourceTalkback(TalkbackKind.Close); + notifierTalkback(TalkbackKind.Close); + } else if (!ended && !pulled) { + pulled = true; + sourceTalkback(TalkbackKind.Pull); + notifierTalkback(TalkbackKind.Pull); + } + }) + ); }; } export function combine(sourceA: Source, sourceB: Source): Source<[A, B]> { - return (sink) => { + return sink => { let lastValA: A | void; let lastValB: B | void; let talkbackA = talkbackPlaceholder; @@ -71,7 +73,7 @@ export function combine(sourceA: Source, sourceB: Source): Source<[A let gotSignal = false; let gotEnd = false; let ended = false; - sourceA((signal) => { + sourceA(signal => { if (signal === SignalKind.End) { if (!gotEnd) { gotEnd = true; @@ -94,7 +96,7 @@ export function combine(sourceA: Source, sourceB: Source): Source<[A sink(push([lastValA, lastValB] as [A, B])); } }); - sourceB((signal) => { + sourceB(signal => { if (signal === SignalKind.End) { if (!gotEnd) { gotEnd = true; @@ -117,24 +119,26 @@ export function combine(sourceA: Source, sourceB: Source): Source<[A sink(push([lastValA, lastValB] as [A, B])); } }); - sink(start((signal) => { - if (ended) { - /*noop*/ - } else if (signal === TalkbackKind.Close) { - ended = true; - talkbackA(TalkbackKind.Close); - talkbackB(TalkbackKind.Close); - } else if (!gotSignal) { - gotSignal = true; - talkbackA(TalkbackKind.Pull); - talkbackB(TalkbackKind.Pull); - } - })); + sink( + start(signal => { + if (ended) { + /*noop*/ + } else if (signal === TalkbackKind.Close) { + ended = true; + talkbackA(TalkbackKind.Close); + talkbackB(TalkbackKind.Close); + } else if (!gotSignal) { + gotSignal = true; + talkbackA(TalkbackKind.Pull); + talkbackB(TalkbackKind.Pull); + } + }) + ); }; } export function concatMap(map: (value: In) => Source): Operator { - return (source) => (sink) => { + return source => sink => { const inputQueue: In[] = []; let outerTalkback = talkbackPlaceholder; let innerTalkback = talkbackPlaceholder; @@ -144,12 +148,12 @@ export function concatMap(map: (value: In) => Source): Operator): void { innerActive = true; - innerSource((signal) => { + innerSource(signal => { if (signal === SignalKind.End) { if (innerActive) { innerActive = false; if (inputQueue.length) { - applyInnerSource(map(inputQueue.shift()!)) + applyInnerSource(map(inputQueue.shift()!)); } else if (ended) { sink(SignalKind.End); } else if (!outerPulled) { @@ -170,13 +174,12 @@ export function concatMap(map: (value: In) => Source): Operator { + source(signal => { if (ended) { /*noop*/ } else if (signal === SignalKind.End) { ended = true; - if (!innerActive && !inputQueue.length) - sink(SignalKind.End); + if (!innerActive && !inputQueue.length) sink(SignalKind.End); } else if (signal.tag === SignalKind.Start) { outerTalkback = signal[0]; } else { @@ -188,27 +191,29 @@ export function concatMap(map: (value: In) => Source): Operator { - if (signal === TalkbackKind.Close) { - if (!ended) { - ended = true; - outerTalkback(TalkbackKind.Close); - } - if (innerActive) { - innerActive = false; - innerTalkback(TalkbackKind.Close); - } - } else { - if (!ended && !outerPulled) { - outerPulled = true; - outerTalkback(TalkbackKind.Pull); - } - if (innerActive && !innerPulled) { - innerPulled = true; - innerTalkback(TalkbackKind.Pull); + sink( + start(signal => { + if (signal === TalkbackKind.Close) { + if (!ended) { + ended = true; + outerTalkback(TalkbackKind.Close); + } + if (innerActive) { + innerActive = false; + innerTalkback(TalkbackKind.Close); + } + } else { + if (!ended && !outerPulled) { + outerPulled = true; + outerTalkback(TalkbackKind.Pull); + } + if (innerActive && !innerPulled) { + innerPulled = true; + innerTalkback(TalkbackKind.Pull); + } } - } - })); + }) + ); }; } @@ -221,9 +226,9 @@ export function concat(sources: Source[]): Source { } export function filter(predicate: (value: T) => boolean): Operator { - return (source) => (sink) => { + return source => sink => { let talkback = talkbackPlaceholder; - source((signal) => { + source(signal => { if (signal === SignalKind.End) { sink(signal); } else if (signal.tag === SignalKind.Start) { @@ -239,24 +244,25 @@ export function filter(predicate: (value: T) => boolean): Operator { } export function map(map: (value: In) => Out): Operator { - return (source) => (sink) => source((signal) => { - if (signal === SignalKind.End || signal.tag === SignalKind.Start) { - sink(signal); - } else { - sink(push(map(signal[0]))); - } - }); + return source => sink => + source(signal => { + if (signal === SignalKind.End || signal.tag === SignalKind.Start) { + sink(signal); + } else { + sink(push(map(signal[0]))); + } + }); } export function mergeMap(map: (value: In) => Source): Operator { - return (source) => (sink) => { + return source => sink => { const innerTalkbacks: TalkbackFn[] = []; let outerTalkback = talkbackPlaceholder; let outerPulled = false; let ended = false; function applyInnerSource(innerSource: Source): void { let talkback = talkbackPlaceholder; - innerSource((signal) => { + innerSource(signal => { if (signal === SignalKind.End) { if (innerTalkbacks.length) { const index = innerTalkbacks.indexOf(talkback); @@ -271,7 +277,7 @@ export function mergeMap(map: (value: In) => Source): Operator(map: (value: In) => Source): Operator { + source(signal => { if (ended) { /*noop*/ } else if (signal === SignalKind.End) { ended = true; - if (!innerTalkbacks.length) - sink(SignalKind.End); + if (!innerTalkbacks.length) sink(SignalKind.End); } else if (signal.tag === SignalKind.Start) { outerTalkback = signal[0]; } else { @@ -297,25 +302,25 @@ export function mergeMap(map: (value: In) => Source): Operator { - if (signal === TalkbackKind.Close) { - if (!ended) { - ended = true; - outerTalkback(TalkbackKind.Close); - } - while (innerTalkbacks.length) - innerTalkbacks.pop()!(TalkbackKind.Close); - } else { - if (!ended && !outerPulled) { - outerPulled = true; - outerTalkback(TalkbackKind.Pull); + sink( + start(signal => { + if (signal === TalkbackKind.Close) { + if (!ended) { + ended = true; + outerTalkback(TalkbackKind.Close); + } + while (innerTalkbacks.length) innerTalkbacks.pop()!(TalkbackKind.Close); } else { - outerPulled = false; + if (!ended && !outerPulled) { + outerPulled = true; + outerTalkback(TalkbackKind.Pull); + } else { + outerPulled = false; + } + for (let i = 0; i < innerTalkbacks.length; i++) innerTalkbacks[i](TalkbackKind.Pull); } - for (let i = 0; i < innerTalkbacks.length; i++) - innerTalkbacks[i](TalkbackKind.Pull); - } - })); + }) + ); }; } @@ -328,9 +333,9 @@ export function merge(sources: Source[]): Source { } export function onEnd(callback: () => void): Operator { - return (source) => (sink) => { + return source => sink => { let ended = false; - source((signal) => { + source(signal => { if (ended) { /*noop*/ } else if (signal === SignalKind.End) { @@ -339,15 +344,17 @@ export function onEnd(callback: () => void): Operator { callback(); } else if (signal.tag === SignalKind.Start) { const talkback = signal[0]; - sink(start((signal) => { - if (signal === TalkbackKind.Close) { - ended = true; - talkback(TalkbackKind.Close); - callback(); - } else { - talkback(signal); - } - })); + sink( + start(signal => { + if (signal === TalkbackKind.Close) { + ended = true; + talkback(TalkbackKind.Close); + callback(); + } else { + talkback(signal); + } + }) + ); } else { sink(signal); } @@ -356,9 +363,9 @@ export function onEnd(callback: () => void): Operator { } export function onPush(callback: (value: T) => void): Operator { - return (source) => (sink) => { + return source => sink => { let ended = false; - source((signal) => { + source(signal => { if (ended) { /*noop*/ } else if (signal === SignalKind.End) { @@ -366,10 +373,12 @@ export function onPush(callback: (value: T) => void): Operator { sink(SignalKind.End); } else if (signal.tag === SignalKind.Start) { const talkback = signal[0]; - sink(start((signal) => { - if (signal === TalkbackKind.Close) ended = true; - talkback(signal); - })); + sink( + start(signal => { + if (signal === TalkbackKind.Close) ended = true; + talkback(signal); + }) + ); } else { callback(signal[0]); sink(signal); @@ -379,26 +388,27 @@ export function onPush(callback: (value: T) => void): Operator { } export function onStart(callback: () => void): Operator { - return (source) => (sink) => source((signal) => { - if (signal === SignalKind.End) { - sink(SignalKind.End); - } else if (signal.tag === SignalKind.Start) { - sink(signal); - callback(); - } else { - sink(signal); - } - }); + return source => sink => + source(signal => { + if (signal === SignalKind.End) { + sink(SignalKind.End); + } else if (signal.tag === SignalKind.Start) { + sink(signal); + callback(); + } else { + sink(signal); + } + }); } export function sample(notifier: Source): Operator { - return (source) => (sink) => { + return source => sink => { let sourceTalkback = talkbackPlaceholder; let notifierTalkback = talkbackPlaceholder; let value: T | void; let pulled = false; let ended = false; - source((signal) => { + source(signal => { if (ended) { /*noop*/ } else if (signal === SignalKind.End) { @@ -418,7 +428,7 @@ export function sample(notifier: Source): Operator { } } }); - notifier((signal) => { + notifier(signal => { if (ended) { /*noop*/ } else if (signal === SignalKind.End) { @@ -433,30 +443,32 @@ export function sample(notifier: Source): Operator { sink(signal); } }); - sink(start((signal) => { - if (signal === TalkbackKind.Close && !ended) { - ended = true; - sourceTalkback(TalkbackKind.Close); - notifierTalkback(TalkbackKind.Close); - } else if (!ended && !pulled) { - pulled = true; - sourceTalkback(TalkbackKind.Pull); - notifierTalkback(TalkbackKind.Pull); - } - })); + sink( + start(signal => { + if (signal === TalkbackKind.Close && !ended) { + ended = true; + sourceTalkback(TalkbackKind.Close); + notifierTalkback(TalkbackKind.Close); + } else if (!ended && !pulled) { + pulled = true; + sourceTalkback(TalkbackKind.Pull); + notifierTalkback(TalkbackKind.Pull); + } + }) + ); }; } export function scan(reducer: (acc: Out, value: In) => Out, seed: Out): Operator { - return (source) => (sink) => { + return source => sink => { let acc = seed; - source((signal) => { + source(signal => { if (signal === SignalKind.End) { sink(SignalKind.End); } else if (signal.tag === SignalKind.Start) { sink(signal); } else { - sink(push(acc = reducer(acc, signal[0]))); + sink(push((acc = reducer(acc, signal[0])))); } }); }; @@ -466,10 +478,10 @@ export function share(source: Source): Source { const sinks: Sink[] = []; let talkback = talkbackPlaceholder; let gotSignal = false; - return (sink) => { + return sink => { sinks.push(sink); if (sinks.length === 1) { - source((signal) => { + source(signal => { if (signal === SignalKind.End) { while (sinks.length) sinks.pop()!(SignalKind.End); } else if (signal.tag === SignalKind.Start) { @@ -480,24 +492,26 @@ export function share(source: Source): Source { } }); } - sink(start((signal) => { - if (signal === TalkbackKind.Close) { - const index = sinks.indexOf(sink); - if (index > -1) sinks.splice(index, 1); - if (!sinks.length) talkback(TalkbackKind.Close); - } else if (signal === TalkbackKind.Pull && !gotSignal) { - gotSignal = true; - talkback(TalkbackKind.Pull); - } - })); + sink( + start(signal => { + if (signal === TalkbackKind.Close) { + const index = sinks.indexOf(sink); + if (index > -1) sinks.splice(index, 1); + if (!sinks.length) talkback(TalkbackKind.Close); + } else if (signal === TalkbackKind.Pull && !gotSignal) { + gotSignal = true; + talkback(TalkbackKind.Pull); + } + }) + ); }; } export function skip(wait: number): Operator { - return (source) => (sink) => { + return source => sink => { let talkback = talkbackPlaceholder; let rest = wait; - source((signal) => { + source(signal => { if (signal === SignalKind.End) { sink(SignalKind.End); } else if (signal.tag === SignalKind.Start) { @@ -513,13 +527,13 @@ export function skip(wait: number): Operator { } export function skipUntil(notifier: Source): Operator { - return (source) => (sink) => { + return source => sink => { let sourceTalkback = talkbackPlaceholder; let notifierTalkback = talkbackPlaceholder; let skip = true; let pulled = false; let ended = false; - source((signal) => { + source(signal => { if (ended) { /*noop*/ } else if (signal === SignalKind.End) { @@ -528,7 +542,7 @@ export function skipUntil(notifier: Source): Operator { sink(SignalKind.End); } else if (signal.tag === SignalKind.Start) { sourceTalkback = signal[0]; - notifier((signal) => { + notifier(signal => { if (signal === SignalKind.End) { if (skip) { ended = true; @@ -552,25 +566,27 @@ export function skipUntil(notifier: Source): Operator { pulled = false; } }); - sink(start((signal) => { - if (signal === TalkbackKind.Close && !ended) { - ended = true; - sourceTalkback(TalkbackKind.Close); - if (skip) notifierTalkback(TalkbackKind.Close); - } else if (!ended && !pulled) { - pulled = true; - if (skip) notifierTalkback(TalkbackKind.Pull); - sourceTalkback(TalkbackKind.Pull); - } - })); + sink( + start(signal => { + if (signal === TalkbackKind.Close && !ended) { + ended = true; + sourceTalkback(TalkbackKind.Close); + if (skip) notifierTalkback(TalkbackKind.Close); + } else if (!ended && !pulled) { + pulled = true; + if (skip) notifierTalkback(TalkbackKind.Pull); + sourceTalkback(TalkbackKind.Pull); + } + }) + ); }; } export function skipWhile(predicate: (value: T) => boolean): Operator { - return (source) => (sink) => { + return source => sink => { let talkback = talkbackPlaceholder; let skip = true; - source((signal) => { + source(signal => { if (signal === SignalKind.End) { sink(SignalKind.End); } else if (signal.tag === SignalKind.Start) { @@ -591,7 +607,7 @@ export function skipWhile(predicate: (value: T) => boolean): Operator { } export function switchMap(map: (value: In) => Source): Operator { - return (source) => (sink) => { + return source => sink => { let outerTalkback = talkbackPlaceholder; let innerTalkback = talkbackPlaceholder; let outerPulled = false; @@ -600,7 +616,7 @@ export function switchMap(map: (value: In) => Source): Operator): void { innerActive = true; - innerSource((signal) => { + innerSource(signal => { if (!innerActive) { /*noop*/ } else if (signal === SignalKind.End) { @@ -624,13 +640,12 @@ export function switchMap(map: (value: In) => Source): Operator { + source(signal => { if (ended) { /*noop*/ } else if (signal === SignalKind.End) { ended = true; - if (!innerActive) - sink(SignalKind.End); + if (!innerActive) sink(SignalKind.End); } else if (signal.tag === SignalKind.Start) { outerTalkback = signal[0]; } else { @@ -647,27 +662,29 @@ export function switchMap(map: (value: In) => Source): Operator { - if (signal === TalkbackKind.Close) { - if (!ended) { - ended = true; - outerTalkback(TalkbackKind.Close); - } - if (innerActive) { - innerActive = false; - innerTalkback(TalkbackKind.Close); - } - } else { - if (!ended && !outerPulled) { - outerPulled = true; - outerTalkback(TalkbackKind.Pull); - } - if (innerActive && !innerPulled) { - innerPulled = true; - innerTalkback(TalkbackKind.Pull); + sink( + start(signal => { + if (signal === TalkbackKind.Close) { + if (!ended) { + ended = true; + outerTalkback(TalkbackKind.Close); + } + if (innerActive) { + innerActive = false; + innerTalkback(TalkbackKind.Close); + } + } else { + if (!ended && !outerPulled) { + outerPulled = true; + outerTalkback(TalkbackKind.Pull); + } + if (innerActive && !innerPulled) { + innerPulled = true; + innerTalkback(TalkbackKind.Pull); + } } - } - })); + }) + ); }; } @@ -676,11 +693,11 @@ export function switchAll(source: Source>): Source { } export function take(max: number): Operator { - return (source) => (sink) => { + return source => sink => { let talkback = talkbackPlaceholder; let ended = false; let taken = 0; - source((signal) => { + source(signal => { if (ended) { /*noop*/ } else if (signal === SignalKind.End) { @@ -705,22 +722,24 @@ export function take(max: number): Operator { sink(signal); } }); - sink(start((signal) => { - if (signal === TalkbackKind.Close && !ended) { - ended = true; - talkback(TalkbackKind.Close); - } else if (signal === TalkbackKind.Pull && !ended && taken < max) { - talkback(TalkbackKind.Pull); - } - })); + sink( + start(signal => { + if (signal === TalkbackKind.Close && !ended) { + ended = true; + talkback(TalkbackKind.Close); + } else if (signal === TalkbackKind.Pull && !ended && taken < max) { + talkback(TalkbackKind.Pull); + } + }) + ); }; } export function takeLast(max: number): Operator { - return (source) => (sink) => { + return source => sink => { const queue: T[] = []; let talkback = talkbackPlaceholder; - source((signal) => { + source(signal => { if (signal === SignalKind.End) { fromArray(queue)(sink); } else if (signal.tag === SignalKind.Start) { @@ -740,11 +759,11 @@ export function takeLast(max: number): Operator { } export function takeUntil(notifier: Source): Operator { - return (source) => (sink) => { + return source => sink => { let sourceTalkback = talkbackPlaceholder; let notifierTalkback = talkbackPlaceholder; let ended = false; - source((signal) => { + source(signal => { if (ended) { /*noop*/ } else if (signal === SignalKind.End) { @@ -753,7 +772,7 @@ export function takeUntil(notifier: Source): Operator { sink(SignalKind.End); } else if (signal.tag === SignalKind.Start) { sourceTalkback = signal[0]; - notifier((signal) => { + notifier(signal => { if (signal === SignalKind.End) { /*noop*/ } else if (signal.tag === SignalKind.Start) { @@ -768,23 +787,25 @@ export function takeUntil(notifier: Source): Operator { sink(signal); } }); - sink(start((signal) => { - if (signal === TalkbackKind.Close && !ended) { - ended = true; - sourceTalkback(TalkbackKind.Close); - notifierTalkback(TalkbackKind.Close); - } else if (!ended) { - sourceTalkback(TalkbackKind.Pull); - } - })); + sink( + start(signal => { + if (signal === TalkbackKind.Close && !ended) { + ended = true; + sourceTalkback(TalkbackKind.Close); + notifierTalkback(TalkbackKind.Close); + } else if (!ended) { + sourceTalkback(TalkbackKind.Pull); + } + }) + ); }; } export function takeWhile(predicate: (value: T) => boolean): Operator { - return (source) => (sink) => { + return source => sink => { let talkback = talkbackPlaceholder; let ended = false; - source((signal) => { + source(signal => { if (ended) { /*noop*/ } else if (signal === SignalKind.End) { @@ -805,11 +826,11 @@ export function takeWhile(predicate: (value: T) => boolean): Operator { } export function debounce(timing: (value: T) => number): Operator { - return (source) => (sink) => { + return source => sink => { let id: any | void; let deferredEnded = false; let ended = false; - source((signal) => { + source(signal => { if (ended) { /*noop*/ } else if (signal === SignalKind.End) { @@ -821,16 +842,18 @@ export function debounce(timing: (value: T) => number): Operator { } } else if (signal.tag === SignalKind.Start) { const talkback = signal[0]; - sink(start((signal) => { - if (signal === TalkbackKind.Close && !ended) { - ended = true; - deferredEnded = false; - if (id) clearTimeout(id); - talkback(TalkbackKind.Close); - } else if (!ended) { - talkback(TalkbackKind.Pull); - } - })); + sink( + start(signal => { + if (signal === TalkbackKind.Close && !ended) { + ended = true; + deferredEnded = false; + if (id) clearTimeout(id); + talkback(TalkbackKind.Close); + } else if (!ended) { + talkback(TalkbackKind.Pull); + } + }) + ); } else { if (id) clearTimeout(id); id = setTimeout(() => { @@ -844,9 +867,9 @@ export function debounce(timing: (value: T) => number): Operator { } export function delay(wait: number): Operator { - return (source) => (sink) => { + return source => sink => { let active = 0; - source((signal) => { + source(signal => { if (typeof signal !== 'number' && signal.tag === SignalKind.Start) { sink(signal); } else { @@ -863,23 +886,25 @@ export function delay(wait: number): Operator { } export function throttle(timing: (value: T) => number): Operator { - return (source) => (sink) => { + return source => sink => { let skip = false; let id: any | void; - source((signal) => { + source(signal => { if (signal === SignalKind.End) { if (id) clearTimeout(id); sink(SignalKind.End); } else if (signal.tag === SignalKind.Start) { const talkback = signal[0]; - sink(start((signal) => { - if (signal === TalkbackKind.Close) { - if (id) clearTimeout(id); - talkback(TalkbackKind.Close); - } else { - talkback(TalkbackKind.Pull); - } - })); + sink( + start(signal => { + if (signal === TalkbackKind.Close) { + if (id) clearTimeout(id); + talkback(TalkbackKind.Close); + } else { + talkback(TalkbackKind.Pull); + } + }) + ); } else if (!skip) { skip = true; if (id) clearTimeout(id); @@ -893,7 +918,4 @@ export function throttle(timing: (value: T) => number): Operator { }; } -export { - mergeAll as flatten, - onPush as tap, -} +export { mergeAll as flatten, onPush as tap }; diff --git a/src/pipe.ts b/src/pipe.ts index e8374dc..1fc0391 100644 --- a/src/pipe.ts +++ b/src/pipe.ts @@ -153,8 +153,7 @@ function pipe( function pipe() { let x = arguments[0]; - for (let i = 1, l = arguments.length; i < l; i++) - x = arguments[i](x); + for (let i = 1, l = arguments.length; i < l; i++) x = arguments[i](x); return x; } diff --git a/src/sinks.ts b/src/sinks.ts index ebb28f2..8c43829 100644 --- a/src/sinks.ts +++ b/src/sinks.ts @@ -1,11 +1,11 @@ -import { Source, Subscription, TalkbackKind, SignalKind } from './types' -import { talkbackPlaceholder } from './helpers' +import { Source, Subscription, TalkbackKind, SignalKind } from './types'; +import { talkbackPlaceholder } from './helpers'; export function subscribe(subscriber: (value: T) => void) { return (source: Source): Subscription => { let talkback = talkbackPlaceholder; let ended = false; - source((signal) => { + source(signal => { if (signal === SignalKind.End) { ended = true; } else if (signal.tag === SignalKind.Start) { @@ -22,8 +22,8 @@ export function subscribe(subscriber: (value: T) => void) { talkback(TalkbackKind.Close); } }, - } - } + }; + }; } export function forEach(subscriber: (value: T) => void) { @@ -33,14 +33,16 @@ export function forEach(subscriber: (value: T) => void) { } export function publish(source: Source): void { - subscribe((_value) => {/*noop*/})(source); + subscribe(_value => { + /*noop*/ + })(source); } export function toArray(source: Source): T[] { const values: T[] = []; let talkback = talkbackPlaceholder; let ended = false; - source((signal) => { + source(signal => { if (signal === SignalKind.End) { ended = true; } else if (signal.tag === SignalKind.Start) { @@ -58,7 +60,7 @@ export function toPromise(source: Source): Promise { return new Promise(resolve => { let talkback = talkbackPlaceholder; let value: T | void; - source((signal) => { + source(signal => { if (signal === SignalKind.End) { resolve(value!); } else if (signal.tag === SignalKind.Start) { @@ -68,5 +70,5 @@ export function toPromise(source: Source): Promise { talkback(TalkbackKind.Pull); } }); - }) + }); } diff --git a/src/sources.ts b/src/sources.ts index 4e7336d..9e314c3 100644 --- a/src/sources.ts +++ b/src/sources.ts @@ -1,52 +1,56 @@ -import { Source, Sink, SignalKind, TalkbackKind, Observer, Subject, TeardownFn } from './types' -import { push, start, talkbackPlaceholder } from './helpers' +import { Source, Sink, SignalKind, TalkbackKind, Observer, Subject, TeardownFn } from './types'; +import { push, start, talkbackPlaceholder } from './helpers'; export function fromArray(array: T[]): Source { - return (sink) => { + return sink => { let ended = false; let looping = false; let pulled = false; let current = 0; - sink(start((signal) => { - if (signal === TalkbackKind.Close) { - ended = true; - } else if (looping) { - pulled = true; - } else { - pulled = true; - looping = true; - for (pulled = looping = true; pulled && !ended; current++) { - if (current < array.length) { - pulled = false; - sink(push(array[current])); - } else { - ended = true; - sink(SignalKind.End); + sink( + start(signal => { + if (signal === TalkbackKind.Close) { + ended = true; + } else if (looping) { + pulled = true; + } else { + pulled = true; + looping = true; + for (pulled = looping = true; pulled && !ended; current++) { + if (current < array.length) { + pulled = false; + sink(push(array[current])); + } else { + ended = true; + sink(SignalKind.End); + } } + looping = false; } - looping = false; - } - })) - } + }) + ); + }; } export function fromValue(value: T): Source { - return (sink) => { + return sink => { let ended = false; - sink(start((signal) => { - if (signal === TalkbackKind.Close) { - ended = true; - } else if (!ended) { - ended = true; - sink(push(value)); - sink(SignalKind.End); - } - })) - } + sink( + start(signal => { + if (signal === TalkbackKind.Close) { + ended = true; + } else if (!ended) { + ended = true; + sink(push(value)); + sink(SignalKind.End); + } + }) + ); + }; } export function make(produce: (observer: Observer) => TeardownFn): Source { - return (sink) => { + return sink => { let ended = false; const teardown = produce({ next(value: T) { @@ -59,13 +63,15 @@ export function make(produce: (observer: Observer) => TeardownFn): Source< } }, }); - sink(start((signal) => { - if (signal === TalkbackKind.Close && !ended) { - ended = true; - teardown(); - } - })); - } + sink( + start(signal => { + if (signal === TalkbackKind.Close && !ended) { + ended = true; + teardown(); + } + }) + ); + }; } export function makeSubject(): Subject { @@ -74,12 +80,14 @@ export function makeSubject(): Subject { return { source(sink: Sink) { sinks.push(sink); - sink(start((signal) => { - if (signal === TalkbackKind.Close) { - const index = sinks.indexOf(sink); - if (index > -1) sinks.splice(index, 1); - } - })); + sink( + start(signal => { + if (signal === TalkbackKind.Close) { + const index = sinks.indexOf(sink); + if (index > -1) sinks.splice(index, 1); + } + }) + ); }, next(value: T) { if (!ended) { @@ -98,14 +106,16 @@ export function makeSubject(): Subject { export const empty: Source = (sink: Sink): void => { let ended = false; - sink(start((signal) => { - if (signal === TalkbackKind.Close) { - ended = true; - } else if (!ended) { - ended = true; - sink(SignalKind.End); - } - })); + sink( + start(signal => { + if (signal === TalkbackKind.Close) { + ended = true; + } else if (!ended) { + ended = true; + sink(SignalKind.End); + } + }) + ); }; export const never: Source = (sink: Sink): void => { @@ -113,42 +123,46 @@ export const never: Source = (sink: Sink): void => { }; export function interval(ms: number): Source { - return (sink) => { + return sink => { let i = 0; const id = setInterval(() => { sink(push(i++)); }, ms); - sink(start((signal) => { - if (signal === TalkbackKind.Close) - clearInterval(id); - })); + sink( + start(signal => { + if (signal === TalkbackKind.Close) clearInterval(id); + }) + ); }; } export function fromDomEvent(element: HTMLElement, event: string): Source { - return (sink) => { + return sink => { const handler = (payload: Event) => { sink(push(payload)); }; - sink(start((signal) => { - if (signal === TalkbackKind.Close) - element.removeEventListener(event, handler); - })); + sink( + start(signal => { + if (signal === TalkbackKind.Close) element.removeEventListener(event, handler); + }) + ); element.addEventListener(event, handler); }; } export function fromPromise(promise: Promise): Source { - return (sink) => { + return sink => { let ended = false; - promise.then((value) => { + promise.then(value => { if (!ended) { sink(push(value)); sink(SignalKind.End); } }); - sink(start((signal) => { - if (signal === TalkbackKind.Close) ended = true; - })); + sink( + start(signal => { + if (signal === TalkbackKind.Close) ended = true; + }) + ); }; } diff --git a/src/types.ts b/src/types.ts index 54c9cec..b5c1a70 100644 --- a/src/types.ts +++ b/src/types.ts @@ -13,11 +13,11 @@ export const enum SignalKind { } export interface Tag { - tag: T + tag: T; } -export type Start<_T> = (Tag & [TalkbackFn]) -export type Push = (Tag & [T]) +export type Start<_T> = Tag & [TalkbackFn]; +export type Push = Tag & [T]; export type Signal = Start | Push | SignalKind.End; export type Sink = (signal: Signal) => void; -- knotmirror 0.0.1