+1

package.json

reviewed

··· 70 70 ] 71 71 }, 72 72 "jest": { 73 73 + "testRegex": "(src/.*(\\.|/)(test|spec))\\.ts$", 73 74 "transform": { 74 75 "^.+\\.tsx?$": "@sucrase/jest-plugin" 75 76 }

+404

src/__tests__/compliance.ts

reviewed

··· 1 1 + import { Source, Sink, Operator, Signal, SignalKind, TalkbackKind, TalkbackFn } from '../types'; 2 2 + import { push, start } from '../helpers'; 3 3 + 4 4 + /* This tests a noop operator for passive Pull talkback signals. 5 5 + A Pull will be sent from the sink upwards and should pass through 6 6 + the operator until the source receives it, which then pushes a 7 7 + value down. */ 8 8 + export const passesPassivePull = (operator: Operator<any, any>, output: any = 0) => { 9 9 + it('responds to Pull talkback signals (spec)', () => { 10 10 + let talkback: TalkbackFn | null = null; 11 11 + let pushes = 0; 12 12 + const values: any[] = []; 13 13 + 14 14 + const source: Source<any> = sink => { 15 15 + sink( 16 16 + start(signal => { 17 17 + if (!pushes && signal === TalkbackKind.Pull) { 18 18 + pushes++; 19 19 + sink(push(0)); 20 20 + } 21 21 + }) 22 22 + ); 23 23 + }; 24 24 + 25 25 + const sink: Sink<any> = signal => { 26 26 + expect(signal).not.toBe(SignalKind.End); 27 27 + if (signal === SignalKind.End) { 28 28 + /*noop*/ 29 29 + } else if (signal.tag === SignalKind.Push) { 30 30 + values.push(signal[0]); 31 31 + } else { 32 32 + talkback = signal[0]; 33 33 + } 34 34 + }; 35 35 + 36 36 + operator(source)(sink); 37 37 + // The Start signal should always come in immediately 38 38 + expect(talkback).not.toBe(null); 39 39 + // No Push signals should be issued initially 40 40 + expect(values).toEqual([]); 41 41 + 42 42 + // When pulling a value we expect an immediate response 43 43 + talkback!(TalkbackKind.Pull); 44 44 + jest.runAllTimers(); 45 45 + expect(values).toEqual([output]); 46 46 + }); 47 47 + }; 48 48 + 49 49 + /* This tests a noop operator for regular, active Push signals. 50 50 + A Push will be sent downwards from the source, through the 51 51 + operator to the sink. Pull events should be let through from 52 52 + the sink after every Push event. */ 53 53 + export const passesActivePush = (operator: Operator<any, any>, result: any = 0) => { 54 54 + it('responds to eager Push signals (spec)', () => { 55 55 + const values: any[] = []; 56 56 + let talkback: TalkbackFn | null = null; 57 57 + let sink: Sink<any> | null = null; 58 58 + let pulls = 0; 59 59 + 60 60 + const source: Source<any> = _sink => { 61 61 + (sink = _sink)( 62 62 + start(signal => { 63 63 + if (signal === TalkbackKind.Pull) pulls++; 64 64 + }) 65 65 + ); 66 66 + }; 67 67 + 68 68 + operator(source)(signal => { 69 69 + expect(signal).not.toBe(SignalKind.End); 70 70 + if (signal === SignalKind.End) { 71 71 + /*noop*/ 72 72 + } else if (signal.tag === SignalKind.Start) { 73 73 + talkback = signal[0]; 74 74 + } else if (signal.tag === SignalKind.Push) { 75 75 + values.push(signal[0]); 76 76 + talkback!(TalkbackKind.Pull); 77 77 + } 78 78 + }); 79 79 + 80 80 + // No Pull signals should be issued initially 81 81 + expect(pulls).toBe(0); 82 82 + 83 83 + // When pushing a value we expect an immediate response 84 84 + sink!(push(0)); 85 85 + jest.runAllTimers(); 86 86 + expect(values).toEqual([result]); 87 87 + // Subsequently the Pull signal should have travelled upwards 88 88 + expect(pulls).toBe(1); 89 89 + }); 90 90 + }; 91 91 + 92 92 + /* This tests a noop operator for Close talkback signals from the sink. 93 93 + A Close signal will be sent, which should be forwarded to the source, 94 94 + which then ends the communication without sending an End signal. */ 95 95 + export const passesSinkClose = (operator: Operator<any, any>) => { 96 96 + it('responds to Close signals from sink (spec)', () => { 97 97 + let talkback: TalkbackFn | null = null; 98 98 + let closing = 0; 99 99 + 100 100 + const source: Source<any> = sink => { 101 101 + sink( 102 102 + start(signal => { 103 103 + if (signal === TalkbackKind.Pull && !closing) { 104 104 + sink(push(0)); 105 105 + } else if (signal === TalkbackKind.Close) { 106 106 + closing++; 107 107 + } 108 108 + }) 109 109 + ); 110 110 + }; 111 111 + 112 112 + const sink: Sink<any> = signal => { 113 113 + expect(signal).not.toBe(SignalKind.End); 114 114 + if (signal === SignalKind.End) { 115 115 + /*noop*/ 116 116 + } else if (signal.tag === SignalKind.Push) { 117 117 + talkback!(TalkbackKind.Close); 118 118 + } else { 119 119 + talkback = signal[0]; 120 120 + } 121 121 + }; 122 122 + 123 123 + operator(source)(sink); 124 124 + 125 125 + // When pushing a value we expect an immediate close signal 126 126 + talkback!(TalkbackKind.Pull); 127 127 + jest.runAllTimers(); 128 128 + expect(closing).toBe(1); 129 129 + }); 130 130 + }; 131 131 + 132 132 + /* This tests a noop operator for End signals from the source. 133 133 + A Push and End signal will be sent after the first Pull talkback 134 134 + signal from the sink, which shouldn't lead to any extra Close or Pull 135 135 + talkback signals. */ 136 136 + export const passesSourceEnd = (operator: Operator<any, any>, result: any = 0) => { 137 137 + it('passes on immediate Push then End signals from source (spec)', () => { 138 138 + const signals: Signal<any>[] = []; 139 139 + let talkback: TalkbackFn | null = null; 140 140 + let pulls = 0; 141 141 + let ending = 0; 142 142 + 143 143 + const source: Source<any> = sink => { 144 144 + sink( 145 145 + start(signal => { 146 146 + expect(signal).not.toBe(TalkbackKind.Close); 147 147 + if (signal === TalkbackKind.Pull) { 148 148 + pulls++; 149 149 + if (pulls === 1) { 150 150 + sink(push(0)); 151 151 + sink(SignalKind.End); 152 152 + } 153 153 + } 154 154 + }) 155 155 + ); 156 156 + }; 157 157 + 158 158 + const sink: Sink<any> = signal => { 159 159 + if (signal === SignalKind.End) { 160 160 + signals.push(signal); 161 161 + ending++; 162 162 + } else if (signal.tag === SignalKind.Push) { 163 163 + signals.push(signal); 164 164 + } else { 165 165 + talkback = signal[0]; 166 166 + } 167 167 + }; 168 168 + 169 169 + operator(source)(sink); 170 170 + 171 171 + // When pushing a value we expect an immediate Push then End signal 172 172 + talkback!(TalkbackKind.Pull); 173 173 + jest.runAllTimers(); 174 174 + expect(ending).toBe(1); 175 175 + expect(signals).toEqual([push(result), SignalKind.End]); 176 176 + // Also no additional pull event should be created by the operator 177 177 + expect(pulls).toBe(1); 178 178 + }); 179 179 + }; 180 180 + 181 181 + /* This tests a noop operator for End signals from the source 182 182 + after the first pull in response to another. 183 183 + This is similar to passesSourceEnd but more well behaved since 184 184 + mergeMap/switchMap/concatMap are eager operators. */ 185 185 + export const passesSourcePushThenEnd = (operator: Operator<any, any>, result: any = 0) => { 186 186 + it('passes on End signals from source (spec)', () => { 187 187 + const signals: Signal<any>[] = []; 188 188 + let talkback: TalkbackFn | null = null; 189 189 + let pulls = 0; 190 190 + let ending = 0; 191 191 + 192 192 + const source: Source<any> = sink => { 193 193 + sink( 194 194 + start(signal => { 195 195 + expect(signal).not.toBe(TalkbackKind.Close); 196 196 + if (signal === TalkbackKind.Pull) { 197 197 + pulls++; 198 198 + if (pulls <= 2) { 199 199 + sink(push(0)); 200 200 + } else { 201 201 + sink(SignalKind.End); 202 202 + } 203 203 + } 204 204 + }) 205 205 + ); 206 206 + }; 207 207 + 208 208 + const sink: Sink<any> = signal => { 209 209 + if (signal === SignalKind.End) { 210 210 + signals.push(signal); 211 211 + ending++; 212 212 + } else if (signal.tag === SignalKind.Push) { 213 213 + signals.push(signal); 214 214 + talkback!(TalkbackKind.Pull); 215 215 + } else { 216 216 + talkback = signal[0]; 217 217 + } 218 218 + }; 219 219 + 220 220 + operator(source)(sink); 221 221 + 222 222 + // When pushing a value we expect an immediate Push then End signal 223 223 + talkback!(TalkbackKind.Pull); 224 224 + jest.runAllTimers(); 225 225 + expect(ending).toBe(1); 226 226 + expect(pulls).toBe(3); 227 227 + expect(signals).toEqual([push(result), push(result), SignalKind.End]); 228 228 + }); 229 229 + }; 230 230 + 231 231 + /* This tests a noop operator for Start signals from the source. 232 232 + When the operator's sink is started by the source it'll receive 233 233 + a Start event. As a response it should never send more than one 234 234 + Start signals to the sink. */ 235 235 + export const passesSingleStart = (operator: Operator<any, any>) => { 236 236 + it('sends a single Start event to the incoming sink (spec)', () => { 237 237 + let starts = 0; 238 238 + 239 239 + const source: Source<any> = sink => { 240 240 + sink(start(() => {})); 241 241 + }; 242 242 + 243 243 + const sink: Sink<any> = signal => { 244 244 + if (signal !== SignalKind.End && signal.tag === SignalKind.Start) { 245 245 + starts++; 246 246 + } 247 247 + }; 248 248 + 249 249 + // When starting the operator we expect a single start event on the sink 250 250 + operator(source)(sink); 251 251 + expect(starts).toBe(1); 252 252 + }); 253 253 + }; 254 254 + 255 255 + /* This tests a noop operator for silence after End signals from the source. 256 256 + When the operator receives the End signal it shouldn't forward any other 257 257 + signals to the sink anymore. 258 258 + This isn't a strict requirement, but some operators should ensure that 259 259 + all sources are well behaved. This is particularly true for operators 260 260 + that either Close sources themselves or may operate on multiple sources. */ 261 261 + export const passesStrictEnd = (operator: Operator<any, any>) => { 262 262 + it('stops all signals after End has been received (spec: strict end)', () => { 263 263 + let pulls = 0; 264 264 + const signals: Signal<any>[] = []; 265 265 + 266 266 + const source: Source<any> = sink => { 267 267 + sink( 268 268 + start(signal => { 269 269 + if (signal === TalkbackKind.Pull) { 270 270 + pulls++; 271 271 + sink(SignalKind.End); 272 272 + sink(push(123)); 273 273 + } 274 274 + }) 275 275 + ); 276 276 + }; 277 277 + 278 278 + const sink: Sink<any> = signal => { 279 279 + if (signal === SignalKind.End) { 280 280 + signals.push(signal); 281 281 + } else if (signal.tag === SignalKind.Push) { 282 282 + signals.push(signal); 283 283 + } else { 284 284 + signal[0](TalkbackKind.Pull); 285 285 + } 286 286 + }; 287 287 + 288 288 + operator(source)(sink); 289 289 + 290 290 + // The Push signal should've been dropped 291 291 + jest.runAllTimers(); 292 292 + expect(signals).toEqual([SignalKind.End]); 293 293 + expect(pulls).toBe(1); 294 294 + }); 295 295 + 296 296 + it('stops all signals after Close has been received (spec: strict close)', () => { 297 297 + const signals: Signal<any>[] = []; 298 298 + 299 299 + const source: Source<any> = sink => { 300 300 + sink( 301 301 + start(signal => { 302 302 + if (signal === TalkbackKind.Close) { 303 303 + sink(push(123)); 304 304 + } 305 305 + }) 306 306 + ); 307 307 + }; 308 308 + 309 309 + const sink: Sink<any> = signal => { 310 310 + if (signal === SignalKind.End) { 311 311 + signals.push(signal); 312 312 + } else if (signal.tag === SignalKind.Push) { 313 313 + signals.push(signal); 314 314 + } else { 315 315 + signal[0](TalkbackKind.Close); 316 316 + } 317 317 + }; 318 318 + 319 319 + operator(source)(sink); 320 320 + 321 321 + // The Push signal should've been dropped 322 322 + jest.runAllTimers(); 323 323 + expect(signals).toEqual([]); 324 324 + }); 325 325 + }; 326 326 + 327 327 + /* This tests an immediately closing operator for End signals to 328 328 + the sink and Close signals to the source. 329 329 + When an operator closes immediately we expect to see a Close 330 330 + signal at the source and an End signal to the sink, since the 331 331 + closing operator is expected to end the entire chain. */ 332 332 + export const passesCloseAndEnd = (closingOperator: Operator<any, any>) => { 333 333 + it('closes the source and ends the sink correctly (spec: ending operator)', () => { 334 334 + let closing = 0; 335 335 + let ending = 0; 336 336 + 337 337 + const source: Source<any> = sink => { 338 338 + sink( 339 339 + start(signal => { 340 340 + // For some operator tests we do need to send a single value 341 341 + if (signal === TalkbackKind.Pull) { 342 342 + sink(push(null)); 343 343 + } else { 344 344 + closing++; 345 345 + } 346 346 + }) 347 347 + ); 348 348 + }; 349 349 + 350 350 + const sink: Sink<any> = signal => { 351 351 + if (signal === SignalKind.End) { 352 352 + ending++; 353 353 + } else if (signal.tag === SignalKind.Start) { 354 354 + signal[0](TalkbackKind.Pull); 355 355 + } 356 356 + }; 357 357 + 358 358 + // We expect the operator to immediately end and close 359 359 + closingOperator(source)(sink); 360 360 + expect(closing).toBe(1); 361 361 + expect(ending).toBe(1); 362 362 + }); 363 363 + }; 364 364 + 365 365 + export const passesAsyncSequence = (operator: Operator<any, any>, result: any = 0) => { 366 366 + it('passes an async push with an async end (spec)', () => { 367 367 + let hasPushed = false; 368 368 + const signals: Signal<any>[] = []; 369 369 + 370 370 + const source: Source<any> = sink => { 371 371 + sink( 372 372 + start(signal => { 373 373 + if (signal === TalkbackKind.Pull && !hasPushed) { 374 374 + hasPushed = true; 375 375 + setTimeout(() => sink(push(0)), 10); 376 376 + setTimeout(() => sink(SignalKind.End), 20); 377 377 + } 378 378 + }) 379 379 + ); 380 380 + }; 381 381 + 382 382 + const sink: Sink<any> = signal => { 383 383 + if (signal === SignalKind.End) { 384 384 + signals.push(signal); 385 385 + } else if (signal.tag === SignalKind.Push) { 386 386 + signals.push(signal); 387 387 + } else { 388 388 + setTimeout(() => { 389 389 + signal[0](TalkbackKind.Pull); 390 390 + }, 5); 391 391 + } 392 392 + }; 393 393 + 394 394 + // We initially expect to see the push signal 395 395 + // Afterwards after all timers all other signals come in 396 396 + operator(source)(sink); 397 397 + expect(signals.length).toBe(0); 398 398 + jest.advanceTimersByTime(5); 399 399 + expect(hasPushed).toBeTruthy(); 400 400 + jest.runAllTimers(); 401 401 + 402 402 + expect(signals).toEqual([push(result), SignalKind.End]); 403 403 + }); 404 404 + };

+13 -403

src/__tests__/operators.test.ts

reviewed

··· 1 1 - import { Source, Sink, Operator, Signal, SignalKind, TalkbackKind, TalkbackFn } from '../types'; 1 1 + import { Source, Sink, Signal, SignalKind, TalkbackKind, TalkbackFn } from '../types'; 2 2 import { push, start } from '../helpers'; 3 3 4 4 + import { 5 5 + passesPassivePull, 6 6 + passesActivePush, 7 7 + passesSinkClose, 8 8 + passesSourceEnd, 9 9 + passesSingleStart, 10 10 + passesStrictEnd, 11 11 + passesSourcePushThenEnd, 12 12 + passesAsyncSequence, 13 13 + passesCloseAndEnd, 14 14 + } from './compliance'; 15 15 + 4 16 import * as sources from '../sources'; 5 17 import * as sinks from '../sinks'; 6 18 import * as operators from '../operators'; 7 7 - 8 8 - /* This tests a noop operator for passive Pull talkback signals. 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 12 - const passesPassivePull = (operator: Operator<any, any>, output: any = 0) => { 13 13 - it('responds to Pull talkback signals (spec)', () => { 14 14 - let talkback: TalkbackFn | null = null; 15 15 - let pushes = 0; 16 16 - const values: any[] = []; 17 17 - 18 18 - const source: Source<any> = sink => { 19 19 - sink( 20 20 - start(signal => { 21 21 - if (!pushes && signal === TalkbackKind.Pull) { 22 22 - pushes++; 23 23 - sink(push(0)); 24 24 - } 25 25 - }) 26 26 - ); 27 27 - }; 28 28 - 29 29 - const sink: Sink<any> = signal => { 30 30 - expect(signal).not.toBe(SignalKind.End); 31 31 - if (signal === SignalKind.End) { 32 32 - /*noop*/ 33 33 - } else if (signal.tag === SignalKind.Push) { 34 34 - values.push(signal[0]); 35 35 - } else { 36 36 - talkback = signal[0]; 37 37 - } 38 38 - }; 39 39 - 40 40 - operator(source)(sink); 41 41 - // The Start signal should always come in immediately 42 42 - expect(talkback).not.toBe(null); 43 43 - // No Push signals should be issued initially 44 44 - expect(values).toEqual([]); 45 45 - 46 46 - // When pulling a value we expect an immediate response 47 47 - talkback!(TalkbackKind.Pull); 48 48 - jest.runAllTimers(); 49 49 - expect(values).toEqual([output]); 50 50 - }); 51 51 - }; 52 52 - 53 53 - /* This tests a noop operator for regular, active Push signals. 54 54 - A Push will be sent downwards from the source, through the 55 55 - operator to the sink. Pull events should be let through from 56 56 - the sink after every Push event. */ 57 57 - const passesActivePush = (operator: Operator<any, any>, result: any = 0) => { 58 58 - it('responds to eager Push signals (spec)', () => { 59 59 - const values: any[] = []; 60 60 - let talkback: TalkbackFn | null = null; 61 61 - let sink: Sink<any> | null = null; 62 62 - let pulls = 0; 63 63 - 64 64 - const source: Source<any> = _sink => { 65 65 - (sink = _sink)( 66 66 - start(signal => { 67 67 - if (signal === TalkbackKind.Pull) pulls++; 68 68 - }) 69 69 - ); 70 70 - }; 71 71 - 72 72 - operator(source)(signal => { 73 73 - expect(signal).not.toBe(SignalKind.End); 74 74 - if (signal === SignalKind.End) { 75 75 - /*noop*/ 76 76 - } else if (signal.tag === SignalKind.Start) { 77 77 - talkback = signal[0]; 78 78 - } else if (signal.tag === SignalKind.Push) { 79 79 - values.push(signal[0]); 80 80 - talkback!(TalkbackKind.Pull); 81 81 - } 82 82 - }); 83 83 - 84 84 - // No Pull signals should be issued initially 85 85 - expect(pulls).toBe(0); 86 86 - 87 87 - // When pushing a value we expect an immediate response 88 88 - sink!(push(0)); 89 89 - jest.runAllTimers(); 90 90 - expect(values).toEqual([result]); 91 91 - // Subsequently the Pull signal should have travelled upwards 92 92 - expect(pulls).toBe(1); 93 93 - }); 94 94 - }; 95 95 - 96 96 - /* This tests a noop operator for Close talkback signals from the sink. 97 97 - A Close signal will be sent, which should be forwarded to the source, 98 98 - which then ends the communication without sending an End signal. */ 99 99 - const passesSinkClose = (operator: Operator<any, any>) => { 100 100 - it('responds to Close signals from sink (spec)', () => { 101 101 - let talkback: TalkbackFn | null = null; 102 102 - let closing = 0; 103 103 - 104 104 - const source: Source<any> = sink => { 105 105 - sink( 106 106 - start(signal => { 107 107 - if (signal === TalkbackKind.Pull && !closing) { 108 108 - sink(push(0)); 109 109 - } else if (signal === TalkbackKind.Close) { 110 110 - closing++; 111 111 - } 112 112 - }) 113 113 - ); 114 114 - }; 115 115 - 116 116 - const sink: Sink<any> = signal => { 117 117 - expect(signal).not.toBe(SignalKind.End); 118 118 - if (signal === SignalKind.End) { 119 119 - /*noop*/ 120 120 - } else if (signal.tag === SignalKind.Push) { 121 121 - talkback!(TalkbackKind.Close); 122 122 - } else { 123 123 - talkback = signal[0]; 124 124 - } 125 125 - }; 126 126 - 127 127 - operator(source)(sink); 128 128 - 129 129 - // When pushing a value we expect an immediate close signal 130 130 - talkback!(TalkbackKind.Pull); 131 131 - jest.runAllTimers(); 132 132 - expect(closing).toBe(1); 133 133 - }); 134 134 - }; 135 135 - 136 136 - /* This tests a noop operator for End signals from the source. 137 137 - A Push and End signal will be sent after the first Pull talkback 138 138 - signal from the sink, which shouldn't lead to any extra Close or Pull 139 139 - talkback signals. */ 140 140 - const passesSourceEnd = (operator: Operator<any, any>, result: any = 0) => { 141 141 - it('passes on immediate Push then End signals from source (spec)', () => { 142 142 - const signals: Signal<any>[] = []; 143 143 - let talkback: TalkbackFn | null = null; 144 144 - let pulls = 0; 145 145 - let ending = 0; 146 146 - 147 147 - const source: Source<any> = sink => { 148 148 - sink( 149 149 - start(signal => { 150 150 - expect(signal).not.toBe(TalkbackKind.Close); 151 151 - if (signal === TalkbackKind.Pull) { 152 152 - pulls++; 153 153 - if (pulls === 1) { 154 154 - sink(push(0)); 155 155 - sink(SignalKind.End); 156 156 - } 157 157 - } 158 158 - }) 159 159 - ); 160 160 - }; 161 161 - 162 162 - const sink: Sink<any> = signal => { 163 163 - if (signal === SignalKind.End) { 164 164 - signals.push(signal); 165 165 - ending++; 166 166 - } else if (signal.tag === SignalKind.Push) { 167 167 - signals.push(signal); 168 168 - } else { 169 169 - talkback = signal[0]; 170 170 - } 171 171 - }; 172 172 - 173 173 - operator(source)(sink); 174 174 - 175 175 - // When pushing a value we expect an immediate Push then End signal 176 176 - talkback!(TalkbackKind.Pull); 177 177 - jest.runAllTimers(); 178 178 - expect(ending).toBe(1); 179 179 - expect(signals).toEqual([push(result), SignalKind.End]); 180 180 - // Also no additional pull event should be created by the operator 181 181 - expect(pulls).toBe(1); 182 182 - }); 183 183 - }; 184 184 - 185 185 - /* This tests a noop operator for End signals from the source 186 186 - after the first pull in response to another. 187 187 - This is similar to passesSourceEnd but more well behaved since 188 188 - mergeMap/switchMap/concatMap are eager operators. */ 189 189 - const passesSourcePushThenEnd = (operator: Operator<any, any>, result: any = 0) => { 190 190 - it('passes on End signals from source (spec)', () => { 191 191 - const signals: Signal<any>[] = []; 192 192 - let talkback: TalkbackFn | null = null; 193 193 - let pulls = 0; 194 194 - let ending = 0; 195 195 - 196 196 - const source: Source<any> = sink => { 197 197 - sink( 198 198 - start(signal => { 199 199 - expect(signal).not.toBe(TalkbackKind.Close); 200 200 - if (signal === TalkbackKind.Pull) { 201 201 - pulls++; 202 202 - if (pulls <= 2) { 203 203 - sink(push(0)); 204 204 - } else { 205 205 - sink(SignalKind.End); 206 206 - } 207 207 - } 208 208 - }) 209 209 - ); 210 210 - }; 211 211 - 212 212 - const sink: Sink<any> = signal => { 213 213 - if (signal === SignalKind.End) { 214 214 - signals.push(signal); 215 215 - ending++; 216 216 - } else if (signal.tag === SignalKind.Push) { 217 217 - signals.push(signal); 218 218 - talkback!(TalkbackKind.Pull); 219 219 - } else { 220 220 - talkback = signal[0]; 221 221 - } 222 222 - }; 223 223 - 224 224 - operator(source)(sink); 225 225 - 226 226 - // When pushing a value we expect an immediate Push then End signal 227 227 - talkback!(TalkbackKind.Pull); 228 228 - jest.runAllTimers(); 229 229 - expect(ending).toBe(1); 230 230 - expect(pulls).toBe(3); 231 231 - expect(signals).toEqual([push(result), push(result), SignalKind.End]); 232 232 - }); 233 233 - }; 234 234 - 235 235 - /* This tests a noop operator for Start signals from the source. 236 236 - When the operator's sink is started by the source it'll receive 237 237 - a Start event. As a response it should never send more than one 238 238 - Start signals to the sink. */ 239 239 - const passesSingleStart = (operator: Operator<any, any>) => { 240 240 - it('sends a single Start event to the incoming sink (spec)', () => { 241 241 - let starts = 0; 242 242 - 243 243 - const source: Source<any> = sink => { 244 244 - sink(start(() => {})); 245 245 - }; 246 246 - 247 247 - const sink: Sink<any> = signal => { 248 248 - if (signal !== SignalKind.End && signal.tag === SignalKind.Start) { 249 249 - starts++; 250 250 - } 251 251 - }; 252 252 - 253 253 - // When starting the operator we expect a single start event on the sink 254 254 - operator(source)(sink); 255 255 - expect(starts).toBe(1); 256 256 - }); 257 257 - }; 258 258 - 259 259 - /* This tests a noop operator for silence after End signals from the source. 260 260 - When the operator receives the End signal it shouldn't forward any other 261 261 - signals to the sink anymore. 262 262 - This isn't a strict requirement, but some operators should ensure that 263 263 - all sources are well behaved. This is particularly true for operators 264 264 - that either Close sources themselves or may operate on multiple sources. */ 265 265 - const passesStrictEnd = (operator: Operator<any, any>) => { 266 266 - it('stops all signals after End has been received (spec: strict end)', () => { 267 267 - let pulls = 0; 268 268 - const signals: Signal<any>[] = []; 269 269 - 270 270 - const source: Source<any> = sink => { 271 271 - sink( 272 272 - start(signal => { 273 273 - if (signal === TalkbackKind.Pull) { 274 274 - pulls++; 275 275 - sink(SignalKind.End); 276 276 - sink(push(123)); 277 277 - } 278 278 - }) 279 279 - ); 280 280 - }; 281 281 - 282 282 - const sink: Sink<any> = signal => { 283 283 - if (signal === SignalKind.End) { 284 284 - signals.push(signal); 285 285 - } else if (signal.tag === SignalKind.Push) { 286 286 - signals.push(signal); 287 287 - } else { 288 288 - signal[0](TalkbackKind.Pull); 289 289 - } 290 290 - }; 291 291 - 292 292 - operator(source)(sink); 293 293 - 294 294 - // The Push signal should've been dropped 295 295 - jest.runAllTimers(); 296 296 - expect(signals).toEqual([SignalKind.End]); 297 297 - expect(pulls).toBe(1); 298 298 - }); 299 299 - 300 300 - it('stops all signals after Close has been received (spec: strict close)', () => { 301 301 - const signals: Signal<any>[] = []; 302 302 - 303 303 - const source: Source<any> = sink => { 304 304 - sink( 305 305 - start(signal => { 306 306 - if (signal === TalkbackKind.Close) { 307 307 - sink(push(123)); 308 308 - } 309 309 - }) 310 310 - ); 311 311 - }; 312 312 - 313 313 - const sink: Sink<any> = signal => { 314 314 - if (signal === SignalKind.End) { 315 315 - signals.push(signal); 316 316 - } else if (signal.tag === SignalKind.Push) { 317 317 - signals.push(signal); 318 318 - } else { 319 319 - signal[0](TalkbackKind.Close); 320 320 - } 321 321 - }; 322 322 - 323 323 - operator(source)(sink); 324 324 - 325 325 - // The Push signal should've been dropped 326 326 - jest.runAllTimers(); 327 327 - expect(signals).toEqual([]); 328 328 - }); 329 329 - }; 330 330 - 331 331 - /* This tests an immediately closing operator for End signals to 332 332 - the sink and Close signals to the source. 333 333 - When an operator closes immediately we expect to see a Close 334 334 - signal at the source and an End signal to the sink, since the 335 335 - closing operator is expected to end the entire chain. */ 336 336 - const passesCloseAndEnd = (closingOperator: Operator<any, any>) => { 337 337 - it('closes the source and ends the sink correctly (spec: ending operator)', () => { 338 338 - let closing = 0; 339 339 - let ending = 0; 340 340 - 341 341 - const source: Source<any> = sink => { 342 342 - sink( 343 343 - start(signal => { 344 344 - // For some operator tests we do need to send a single value 345 345 - if (signal === TalkbackKind.Pull) { 346 346 - sink(push(null)); 347 347 - } else { 348 348 - closing++; 349 349 - } 350 350 - }) 351 351 - ); 352 352 - }; 353 353 - 354 354 - const sink: Sink<any> = signal => { 355 355 - if (signal === SignalKind.End) { 356 356 - ending++; 357 357 - } else if (signal.tag === SignalKind.Start) { 358 358 - signal[0](TalkbackKind.Pull); 359 359 - } 360 360 - }; 361 361 - 362 362 - // We expect the operator to immediately end and close 363 363 - closingOperator(source)(sink); 364 364 - expect(closing).toBe(1); 365 365 - expect(ending).toBe(1); 366 366 - }); 367 367 - }; 368 368 - 369 369 - const passesAsyncSequence = (operator: Operator<any, any>, result: any = 0) => { 370 370 - it('passes an async push with an async end (spec)', () => { 371 371 - let hasPushed = false; 372 372 - const signals: Signal<any>[] = []; 373 373 - 374 374 - const source: Source<any> = sink => { 375 375 - sink( 376 376 - start(signal => { 377 377 - if (signal === TalkbackKind.Pull && !hasPushed) { 378 378 - hasPushed = true; 379 379 - setTimeout(() => sink(push(0)), 10); 380 380 - setTimeout(() => sink(SignalKind.End), 20); 381 381 - } 382 382 - }) 383 383 - ); 384 384 - }; 385 385 - 386 386 - const sink: Sink<any> = signal => { 387 387 - if (signal === SignalKind.End) { 388 388 - signals.push(signal); 389 389 - } else if (signal.tag === SignalKind.Push) { 390 390 - signals.push(signal); 391 391 - } else { 392 392 - setTimeout(() => { 393 393 - signal[0](TalkbackKind.Pull); 394 394 - }, 5); 395 395 - } 396 396 - }; 397 397 - 398 398 - // We initially expect to see the push signal 399 399 - // Afterwards after all timers all other signals come in 400 400 - operator(source)(sink); 401 401 - expect(signals.length).toBe(0); 402 402 - jest.advanceTimersByTime(5); 403 403 - expect(hasPushed).toBeTruthy(); 404 404 - jest.runAllTimers(); 405 405 - 406 406 - expect(signals).toEqual([push(result), SignalKind.End]); 407 407 - }); 408 408 - }; 409 19 410 20 beforeEach(() => { 411 21 jest.useFakeTimers();