smartstream/test/test.streamfunction.ts
2022-06-07 16:16:14 +02:00

69 lines
2.0 KiB
TypeScript

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();