smartstream/test/test.ts.streamfunction.ts

import { expect, tap } from '@push.rocks/tapbundle';
import * as fs from 'fs';

import * as smartstream from '../ts/index.js';

let testIntake: smartstream.StreamIntake<string>;

tap.test('should handle a read stream', async (tools) => {
  const counter = 0;
  const streamWrapper = new smartstream.StreamWrapper([
    fs.createReadStream('./test/assets/readabletext.txt'),
    new smartstream.SmartDuplex({
      writeFunction: async (chunkStringArg: Buffer, streamTools) => {
        // do something with the stream here
        const result = chunkStringArg.toString().substr(0, 100);
        streamTools.push('wow =========== \n');
        return Buffer.from(result);
      },
      finalFunction: async (tools) => {
        return Buffer.from('this is the end');
      },
    }),
    new smartstream.SmartDuplex({
      writeFunction: async (chunkStringArg) => {
        console.log(chunkStringArg.toString());
      },
      finalFunction: async (tools) => {
        tools.push(null);
      },
    })
  ]);
  await streamWrapper.run();
});

tap.test('should create a valid Intake', async (tools) => {
  testIntake = new smartstream.StreamIntake<string>();
  testIntake.pipe(
      new smartstream.SmartDuplex({
        objectMode: true,
        writeFunction: async (chunkStringArg: string, streamTools) => {
          await tools.delayFor(100);
          console.log(chunkStringArg);
          return chunkStringArg;
        }
      })
    )
    .pipe(fs.createWriteStream('./test/assets/writabletext.txt'));
  const testFinished = tools.defer();
  let counter = 0;
  testIntake.pushNextObservable.subscribe(() => {
    if (counter < 50) {
      counter++;
      testIntake.pushData('hi');
      testIntake.pushData('+wow');
      testIntake.pushData('\n');
    } else {
      testIntake.signalEnd();
      testFinished.resolve();
    }
  });
  await testFinished.promise;
  testIntake.signalEnd();
});

export default tap.start();