66 lines
1.9 KiB
TypeScript
66 lines
1.9 KiB
TypeScript
import { expect, tap } from '@push.rocks/tapbundle';
|
|
import * as smartfile from '@push.rocks/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 streamWrapper = new smartstream.StreamWrapper([
|
|
smartfile.fsStream.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(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();
|