/**
 * Test idiomatic use-cases of ts-stream.
 *
 * Copyright (C) 2018 Martin Poelstra
 * License: MIT
 */ import { expect } from "chai" ; import { Readable , ReadableStream , Stream , Transform , Writable , WritableStream , } from "../lib/index" ; import { delay , noop , settle } from "./util" ; enum MockDatabaseState { Opening , Open , Reading , Writing , Closing , Closed } /**
 * Mock object to simulate a stateful resource that needs
 * proper opening and closing, and which can be only opened
 * by one 'user'.
 *
 * It allows only one operation to be in-progress at any point
 * in time (e.g. reading, closing), and the database needs to
 * be 'open' for all operations (except open itself, of course).
 *
 * All operations take some time, to allow any internal promises
 * in the library to settle.
 */ class MockDatabase < T > { private _state : MockDatabaseState = MockDatabaseState . Closed ; private _values : T [ ] = [ ] ; constructor ( values = [ ] ) { this . setValues ( values ) ; } public async open ( ) { this . _transition ( MockDatabaseState . Closed , MockDatabaseState . Opening ) ; await delay ( 1 ) ; this . _transition ( MockDatabaseState . Opening , MockDatabaseState . Open ) ; } public async close ( ) { this . _transition ( MockDatabaseState . Open , MockDatabaseState . Closing ) ; await delay ( 1 ) ; this . _transition ( MockDatabaseState . Closing , MockDatabaseState . Closed ) ; } public async read ( ) { this . assertState ( MockDatabaseState . Open ) ; const value = this . _values . shift ( ) ; this . _transition ( MockDatabaseState . Open , MockDatabaseState . Reading ) ; await delay ( 1 ) ; this . _transition ( MockDatabaseState . Reading , MockDatabaseState . Open ) ; return value ; } public async write ( value ) { this . assertState ( MockDatabaseState . Open ) ; this . _values . push ( value ) ; this . _transition ( MockDatabaseState . Open , MockDatabaseState . Writing ) ; await delay ( 1 ) ; this . _transition ( MockDatabaseState . Writing , MockDatabaseState . Open ) ; } public abort ( ) { } /**
     * Can be used after tests to verify that e.g. values were
     * correctly written, or which values were not read yet.
     */ public getValues ( ) { return this . _values . slice ( ) ; } /**
     * Can be used before tests to reinitialize values to be read/written.
     */ public setValues ( values ) { this . _values = values . slice ( ) ; } public assertState ( state ) { expect ( MockDatabaseState [ this . _state ] , "invalid database state" ) . to . equal ( MockDatabaseState [ state ] ) ; } private _transition ( expectedState , newState ) { this . assertState ( expectedState ) ; this . _state = newState ; } } /**
 * Simplest way of creating a source, using .writeEach().
 *
 * It correctly handles backpressure, aborting, and cleaning up
 * the underlying resource.
 * This ensures that other elements in the stream also wait until
 * the resource is fully cleaned up before settling their .result().
 */ async function idiomaticSource < T > ( db , destination ) { await db . open ( ) ; await destination . writeEach ( ( ) => db . read ( ) , ( ) => db . close ( ) , ( err ) => db . abort ( ) ) ; } /**
 * Simplest way of creating a sink, using .forEach().
 *
 * It correctly handles backpressure, aborting, and cleaning up
 * the underlying resource.
 * This ensures that other elements in the stream also wait until
 * the resource is fully cleaned up before settling their .result().
 */ async function idiomaticSink < T > ( source , db ) { await db . open ( ) ; await source . forEach ( ( value ) => db . write ( value ) , ( endError ) => db . close ( ) , ( abortError ) => db . abort ( ) ) ; } /**
 * Writing a source using .write() and .end() calls.
 *
 * Useful if a source doesn't provide a way of pulling data
 * out of it, and instead pushes data.
 *
 * It correctly handles backpressure, aborting, and cleaning up
 * the underlying resource.
 * This ensures that other elements in the stream also wait until
 * the resource is fully cleaned up before settling their .result().
 */ async function idiomaticManualSource < T > ( db , destination ) { let endError ; try { try { await db . open ( ) ; destination . aborted ( ) . catch ( ( ) => db . abort ( ) ) ; while ( true ) { const value = await db . read ( ) ; if ( value === undefined ) { break ; } else { await destination . write ( value ) ; } } } finally { await db . close ( ) ; } } catch ( error ) { endError = error ; } await destination . end ( endError ) ; } /**
 * Simple transform.
 *
 * Correctly handles backpressure, and makes sure to pass aborts along to
 * other elements in the chain. Ensures final stream captures result of all
 * previous streams.
 */ function createTransform < T , R > ( transformer ) { return function idiomaticTransform ( readable , writable ) : void { writable . aborted ( ) . catch ( ( err ) => readable . abort ( err ) ) ; readable . forEach ( ( v ) => writable . write ( transformer ( v ) ) , ( error ? ) => writable . end ( error , readable . result ( ) ) , ( abortReason ) : void => { writable . abort ( abortReason ) ; } ) ; } ; } describe ( "idiomatic examples" , ( ) => { it ( "supports trivial example" , async ( ) => { const s = Stream . from ( [ 1 , 2 , 3 , 4 ] ) ; const values = await s . toArray ( ) ; expect ( values ) . to . deep . equal ( [ 1 , 2 , 3 , 4 ] ) ; } ) ; for ( const testSource of [ idiomaticSource , idiomaticManualSource ] ) { for ( const testSink of [ idiomaticSink ] ) { let sourceDb ; let destDb ; beforeEach ( ( ) => { sourceDb = new MockDatabase ( [ 1 , 2 , 3 , 4 ] ) ; destDb = new MockDatabase ( [ ] ) ; } ) ; afterEach ( ( ) => { sourceDb . assertState ( MockDatabaseState . Closed ) ; destDb . assertState ( MockDatabaseState . Closed ) ; } ) ; describe ( `${ testSource . name } -> ${ testSink . name }` , ( ) => { it ( "supports normal flow" , async ( ) => { const stream = new Stream < number > ( ) ; const sourceResult = testSource ( sourceDb , stream ) ; const destResult = testSink ( stream , destDb ) ; await Promise . all ( [ sourceResult , destResult ] ) ; expect ( destDb . getValues ( ) ) . to . deep . equal ( [ 1 , 2 , 3 , 4 ] ) ; } ) ; it ( "waits before starting next" , async ( ) => { const stream1 = new Stream < number > ( ) ; await Promise . all ( [ testSource ( sourceDb , stream1 ) , testSink ( stream1 , destDb ) , ] ) ; sourceDb . setValues ( [ 5 , 6 , 7 , 8 ] ) ; const stream2 = new Stream < number > ( ) ; await Promise . all ( [ testSource ( sourceDb , stream2 ) , testSink ( stream2 , destDb ) , ] ) ; expect ( destDb . getValues ( ) ) . to . deep . equal ( [ 1 , 2 , 3 , 4 , 5 , 6 , 7 , 8 , ] ) ; } ) ; it ( "ends stream even with abort" , async ( ) => { const stream = new Stream < number > ( ) ; stream . aborted ( ) . catch ( noop ) ; const sourceResult = testSource ( sourceDb , stream ) ; const destResult = testSink ( stream , destDb ) ; stream . abort ( ) ; await settle ( [ sourceResult , destResult ] ) ; expect ( destDb . getValues ( ) . length ) . to . be . lessThan ( 4 ) ; } ) ; it ( "handles errors in transforms" , async ( ) => { const stream = new Stream < number > ( ) ; const sourceResult = testSource ( sourceDb , stream ) ; const mappedStream = stream . map ( ( n ) => { if ( n === 3 ) { throw new Error ( "boom" ) ; } else { return n * 2 ; } } ) ; const destResult = testSink ( mappedStream , destDb ) ; await settle ( [ sourceResult , destResult ] ) ; expect ( destDb . getValues ( ) ) . to . deep . equal ( [ 2 , 4 ] ) ; expect ( stream . isEnded ( ) ) . to . equal ( true ) ; } ) ; } ) ; } } describe ( "transform" , ( ) => { it ( "supports idiomatic transform" , async ( ) => { const source = Stream . from ( [ 1 , 2 , 3 , 4 ] ) ; const timesTwo = ( n ) => n * 2 ; const transformed = source . transform ( createTransform ( timesTwo ) ) ; const result = await transformed . toArray ( ) ; expect ( result ) . to . deep . equal ( [ 2 , 4 , 6 , 8 ] ) ; } ) ; } ) ; } ) ;