smartstream/test/test.streamfunction.ts

import { expect, tap } from '@pushrocks/tapbundle';
import * as smartfile from '@pushrocks/smartfile';

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 testSmartstream = new smartstream.StreamWrapper([
    smartfile.fsStream.createReadStream('./test/assets/readabletext.txt'),
    smartstream.createDuplexStream<Buffer, Buffer>(
      async (chunkStringArg: Buffer, streamTools) => {
        // do something with the stream here
        const result = chunkStringArg.toString().substr(0, 100);
        streamTools.pipeMore('wow =========== \n');
        return Buffer.from(result);
      },
      async (tools) => {
        // tools.pipeMore('hey, this is the end')
        return Buffer.from('this is the end');
      },
      { objectMode: false }
    ),
    smartstream.createDuplexStream<Buffer, string>(async (chunkStringArg) => {
      console.log(chunkStringArg.toString());
      return null;
    }),
    smartstream.cleanPipe(),
  ]);
  await testSmartstream.run();
});

tap.test('should create a valid Intake', async (tools) => {
  testIntake = new smartstream.StreamIntake<string>();
  testIntake
    .getReadable()
    .pipe(
      smartstream.createDuplexStream<string, string>(
        async (chunkString) => {
          await tools.delayFor(100);
          console.log(chunkString);
          return chunkString;
        },
        async () => {
          return 'noice';
        }
      )
    )
    .pipe(smartfile.fsStream.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();
});

tap.start();