Compare commits
42 Commits
Author | SHA1 | Date | |
---|---|---|---|
91392e8bd5 | |||
d161d6613a | |||
7a14e67f4f | |||
465ccfec40 | |||
3adb16d1f8 | |||
a9230ca790 | |||
788f2665c2 | |||
7b678cc856 | |||
12c9d8cc9d | |||
3a2dc1c37e | |||
1f67bc0e1e | |||
b15ddd987c | |||
cc43080513 | |||
49d235411f | |||
d238662bea | |||
8efb2b1093 | |||
4926f57d83 | |||
86552f2b1b | |||
353a8ecde6 | |||
3e03b81a43 | |||
5e4ec5b837 | |||
62796f7151 | |||
2c1d9f05ce | |||
34cbf28972 | |||
1b6e38c040 | |||
b135e6023a | |||
91d01f3689 | |||
e8e067ea77 | |||
2cb490cd2a | |||
98397bb85e | |||
f52b0de21f | |||
1c0e5f264d | |||
8a3c653213 | |||
456ce78917 | |||
5277083097 | |||
8618ac55ef | |||
ea66d1b2fb | |||
c37f62abec | |||
2c904cc1ec | |||
d1561ad1b7 | |||
0ae3fee987 | |||
047c2bd402 |
66
.gitea/workflows/default_nottags.yaml
Normal file
66
.gitea/workflows/default_nottags.yaml
Normal file
@ -0,0 +1,66 @@
|
|||||||
|
name: Default (not tags)
|
||||||
|
|
||||||
|
on:
|
||||||
|
push:
|
||||||
|
tags-ignore:
|
||||||
|
- '**'
|
||||||
|
|
||||||
|
env:
|
||||||
|
IMAGE: registry.gitlab.com/hosttoday/ht-docker-node:npmci
|
||||||
|
NPMCI_COMPUTED_REPOURL: https://${{gitea.repository_owner}}:${{secrets.GITEA_TOKEN}}@gitea.lossless.digital/${{gitea.repository}}.git
|
||||||
|
NPMCI_TOKEN_NPM: ${{secrets.NPMCI_TOKEN_NPM}}
|
||||||
|
NPMCI_TOKEN_NPM2: ${{secrets.NPMCI_TOKEN_NPM2}}
|
||||||
|
NPMCI_GIT_GITHUBTOKEN: ${{secrets.NPMCI_GIT_GITHUBTOKEN}}
|
||||||
|
NPMCI_URL_CLOUDLY: ${{secrets.NPMCI_URL_CLOUDLY}}
|
||||||
|
|
||||||
|
jobs:
|
||||||
|
security:
|
||||||
|
runs-on: ubuntu-latest
|
||||||
|
continue-on-error: true
|
||||||
|
container:
|
||||||
|
image: ${{ env.IMAGE }}
|
||||||
|
|
||||||
|
steps:
|
||||||
|
- uses: actions/checkout@v3
|
||||||
|
|
||||||
|
- name: Install pnpm and npmci
|
||||||
|
run: |
|
||||||
|
pnpm install -g pnpm
|
||||||
|
pnpm install -g @shipzone/npmci
|
||||||
|
|
||||||
|
- name: Run npm prepare
|
||||||
|
run: npmci npm prepare
|
||||||
|
|
||||||
|
- name: Audit production dependencies
|
||||||
|
run: |
|
||||||
|
npmci command npm config set registry https://registry.npmjs.org
|
||||||
|
npmci command pnpm audit --audit-level=high --prod
|
||||||
|
continue-on-error: true
|
||||||
|
|
||||||
|
- name: Audit development dependencies
|
||||||
|
run: |
|
||||||
|
npmci command npm config set registry https://registry.npmjs.org
|
||||||
|
npmci command pnpm audit --audit-level=high --dev
|
||||||
|
continue-on-error: true
|
||||||
|
|
||||||
|
test:
|
||||||
|
if: ${{ always() }}
|
||||||
|
needs: security
|
||||||
|
runs-on: ubuntu-latest
|
||||||
|
container:
|
||||||
|
image: ${{ env.IMAGE }}
|
||||||
|
|
||||||
|
steps:
|
||||||
|
- uses: actions/checkout@v3
|
||||||
|
|
||||||
|
- name: Test stable
|
||||||
|
run: |
|
||||||
|
npmci node install stable
|
||||||
|
npmci npm install
|
||||||
|
npmci npm test
|
||||||
|
|
||||||
|
- name: Test build
|
||||||
|
run: |
|
||||||
|
npmci node install stable
|
||||||
|
npmci npm install
|
||||||
|
npmci npm build
|
124
.gitea/workflows/default_tags.yaml
Normal file
124
.gitea/workflows/default_tags.yaml
Normal file
@ -0,0 +1,124 @@
|
|||||||
|
name: Default (tags)
|
||||||
|
|
||||||
|
on:
|
||||||
|
push:
|
||||||
|
tags:
|
||||||
|
- '*'
|
||||||
|
|
||||||
|
env:
|
||||||
|
IMAGE: registry.gitlab.com/hosttoday/ht-docker-node:npmci
|
||||||
|
NPMCI_COMPUTED_REPOURL: https://${{gitea.repository_owner}}:${{secrets.GITEA_TOKEN}}@gitea.lossless.digital/${{gitea.repository}}.git
|
||||||
|
NPMCI_TOKEN_NPM: ${{secrets.NPMCI_TOKEN_NPM}}
|
||||||
|
NPMCI_TOKEN_NPM2: ${{secrets.NPMCI_TOKEN_NPM2}}
|
||||||
|
NPMCI_GIT_GITHUBTOKEN: ${{secrets.NPMCI_GIT_GITHUBTOKEN}}
|
||||||
|
NPMCI_URL_CLOUDLY: ${{secrets.NPMCI_URL_CLOUDLY}}
|
||||||
|
|
||||||
|
jobs:
|
||||||
|
security:
|
||||||
|
runs-on: ubuntu-latest
|
||||||
|
continue-on-error: true
|
||||||
|
container:
|
||||||
|
image: ${{ env.IMAGE }}
|
||||||
|
|
||||||
|
steps:
|
||||||
|
- uses: actions/checkout@v3
|
||||||
|
|
||||||
|
- name: Prepare
|
||||||
|
run: |
|
||||||
|
pnpm install -g pnpm
|
||||||
|
pnpm install -g @shipzone/npmci
|
||||||
|
npmci npm prepare
|
||||||
|
|
||||||
|
- name: Audit production dependencies
|
||||||
|
run: |
|
||||||
|
npmci command npm config set registry https://registry.npmjs.org
|
||||||
|
npmci command pnpm audit --audit-level=high --prod
|
||||||
|
continue-on-error: true
|
||||||
|
|
||||||
|
- name: Audit development dependencies
|
||||||
|
run: |
|
||||||
|
npmci command npm config set registry https://registry.npmjs.org
|
||||||
|
npmci command pnpm audit --audit-level=high --dev
|
||||||
|
continue-on-error: true
|
||||||
|
|
||||||
|
test:
|
||||||
|
if: ${{ always() }}
|
||||||
|
needs: security
|
||||||
|
runs-on: ubuntu-latest
|
||||||
|
container:
|
||||||
|
image: ${{ env.IMAGE }}
|
||||||
|
|
||||||
|
steps:
|
||||||
|
- uses: actions/checkout@v3
|
||||||
|
|
||||||
|
- name: Prepare
|
||||||
|
run: |
|
||||||
|
pnpm install -g pnpm
|
||||||
|
pnpm install -g @shipzone/npmci
|
||||||
|
npmci npm prepare
|
||||||
|
|
||||||
|
- name: Test stable
|
||||||
|
run: |
|
||||||
|
npmci node install stable
|
||||||
|
npmci npm install
|
||||||
|
npmci npm test
|
||||||
|
|
||||||
|
- name: Test build
|
||||||
|
run: |
|
||||||
|
npmci node install stable
|
||||||
|
npmci npm install
|
||||||
|
npmci npm build
|
||||||
|
|
||||||
|
release:
|
||||||
|
needs: test
|
||||||
|
if: github.event_name == 'push' && startsWith(github.ref, 'refs/tags/')
|
||||||
|
runs-on: ubuntu-latest
|
||||||
|
container:
|
||||||
|
image: ${{ env.IMAGE }}
|
||||||
|
|
||||||
|
steps:
|
||||||
|
- uses: actions/checkout@v3
|
||||||
|
|
||||||
|
- name: Prepare
|
||||||
|
run: |
|
||||||
|
pnpm install -g pnpm
|
||||||
|
pnpm install -g @shipzone/npmci
|
||||||
|
npmci npm prepare
|
||||||
|
|
||||||
|
- name: Release
|
||||||
|
run: |
|
||||||
|
npmci node install stable
|
||||||
|
npmci npm publish
|
||||||
|
|
||||||
|
metadata:
|
||||||
|
needs: test
|
||||||
|
if: github.event_name == 'push' && startsWith(github.ref, 'refs/tags/')
|
||||||
|
runs-on: ubuntu-latest
|
||||||
|
container:
|
||||||
|
image: ${{ env.IMAGE }}
|
||||||
|
continue-on-error: true
|
||||||
|
|
||||||
|
steps:
|
||||||
|
- uses: actions/checkout@v3
|
||||||
|
|
||||||
|
- name: Prepare
|
||||||
|
run: |
|
||||||
|
pnpm install -g pnpm
|
||||||
|
pnpm install -g @shipzone/npmci
|
||||||
|
npmci npm prepare
|
||||||
|
|
||||||
|
- name: Code quality
|
||||||
|
run: |
|
||||||
|
npmci command npm install -g typescript
|
||||||
|
npmci npm install
|
||||||
|
|
||||||
|
- name: Trigger
|
||||||
|
run: npmci trigger
|
||||||
|
|
||||||
|
- name: Build docs and upload artifacts
|
||||||
|
run: |
|
||||||
|
npmci node install stable
|
||||||
|
npmci npm install
|
||||||
|
pnpm install -g @git.zone/tsdoc
|
||||||
|
npmci command tsdoc
|
||||||
|
continue-on-error: true
|
24
package.json
24
package.json
@ -1,6 +1,6 @@
|
|||||||
{
|
{
|
||||||
"name": "@push.rocks/smartstream",
|
"name": "@push.rocks/smartstream",
|
||||||
"version": "2.0.6",
|
"version": "3.0.18",
|
||||||
"private": false,
|
"private": false,
|
||||||
"description": "simplifies access to node streams",
|
"description": "simplifies access to node streams",
|
||||||
"main": "dist_ts/index.js",
|
"main": "dist_ts/index.js",
|
||||||
@ -8,32 +8,30 @@
|
|||||||
"type": "module",
|
"type": "module",
|
||||||
"scripts": {
|
"scripts": {
|
||||||
"test": "(tstest test/)",
|
"test": "(tstest test/)",
|
||||||
"build": "(tsbuild)"
|
"build": "(tsbuild)",
|
||||||
|
"buildDocs": "tsdoc"
|
||||||
},
|
},
|
||||||
"repository": {
|
"repository": {
|
||||||
"type": "git",
|
"type": "git",
|
||||||
"url": "git+ssh://git@gitlab.com/pushrocks/smartstream.git"
|
"url": "git+https://gitlab.com/push.rocks/smartstream.git"
|
||||||
},
|
},
|
||||||
"author": "Lossless GmbH",
|
"author": "Lossless GmbH",
|
||||||
"license": "MIT",
|
"license": "MIT",
|
||||||
"bugs": {
|
"bugs": {
|
||||||
"url": "https://gitlab.com/pushrocks/smartstream/issues"
|
"url": "https://gitlab.com/push.rocks/smartstream/issues"
|
||||||
},
|
},
|
||||||
"homepage": "https://gitlab.com/pushrocks/smartstream#README",
|
"homepage": "https://gitlab.com/push.rocks/smartstream#readme",
|
||||||
"devDependencies": {
|
"devDependencies": {
|
||||||
"@git.zone/tsbuild": "^2.1.66",
|
"@git.zone/tsbuild": "^2.1.66",
|
||||||
"@git.zone/tsrun": "^1.2.44",
|
"@git.zone/tsrun": "^1.2.44",
|
||||||
"@git.zone/tstest": "^1.0.77",
|
"@git.zone/tstest": "^1.0.84",
|
||||||
"@push.rocks/smartfile": "^10.0.33",
|
"@push.rocks/smartfile": "^11.0.0",
|
||||||
"@push.rocks/tapbundle": "^5.0.15"
|
"@push.rocks/tapbundle": "^5.0.15",
|
||||||
|
"@types/node": "^20.9.0"
|
||||||
},
|
},
|
||||||
"dependencies": {
|
"dependencies": {
|
||||||
"@push.rocks/smartpromise": "^4.0.3",
|
"@push.rocks/smartpromise": "^4.0.3",
|
||||||
"@push.rocks/smartrx": "^3.0.7",
|
"@push.rocks/smartrx": "^3.0.7"
|
||||||
"@types/from2": "^2.3.4",
|
|
||||||
"@types/through2": "^2.0.40",
|
|
||||||
"from2": "^2.3.0",
|
|
||||||
"through2": "^4.0.2"
|
|
||||||
},
|
},
|
||||||
"browserslist": [
|
"browserslist": [
|
||||||
"last 1 chrome versions"
|
"last 1 chrome versions"
|
||||||
|
571
pnpm-lock.yaml
generated
571
pnpm-lock.yaml
generated
File diff suppressed because it is too large
Load Diff
0
readme_instructions.md
Normal file
0
readme_instructions.md
Normal file
@ -48,4 +48,3 @@ hi+wow
|
|||||||
hi+wow
|
hi+wow
|
||||||
hi+wow
|
hi+wow
|
||||||
hi+wow
|
hi+wow
|
||||||
noice
|
|
@ -1,11 +1,13 @@
|
|||||||
import { expect, tap } from '@push.rocks/tapbundle';
|
import { expect, tap } from '@push.rocks/tapbundle';
|
||||||
import { SmartStream } from '../ts/smartstream.classes.smartstream.js'; // Adjust the import to your file structure
|
import { SmartDuplex } from '../ts/smartstream.classes.smartduplex.js'; // Adjust the import to your file structure
|
||||||
import * as smartrx from '@push.rocks/smartrx';
|
import * as smartrx from '@push.rocks/smartrx';
|
||||||
import * as fs from 'fs';
|
import * as fs from 'fs';
|
||||||
|
|
||||||
tap.test('should create a SmartStream from a Buffer', async () => {
|
tap.test('should create a SmartStream from a Buffer', async () => {
|
||||||
const bufferData = Buffer.from('This is a test buffer');
|
const bufferData = Buffer.from('This is a test buffer');
|
||||||
const smartStream = SmartStream.fromBuffer(bufferData);
|
const smartStream = SmartDuplex.fromBuffer(bufferData, {
|
||||||
|
handleBackpressure: false,
|
||||||
|
});
|
||||||
|
|
||||||
let receivedData = Buffer.alloc(0);
|
let receivedData = Buffer.alloc(0);
|
||||||
|
|
||||||
@ -25,7 +27,9 @@ tap.test('should create a SmartStream from an Observable', async () => {
|
|||||||
const observableData = 'Observable test data';
|
const observableData = 'Observable test data';
|
||||||
const testObservable = smartrx.rxjs.of(Buffer.from(observableData));
|
const testObservable = smartrx.rxjs.of(Buffer.from(observableData));
|
||||||
|
|
||||||
const smartStream = SmartStream.fromObservable(testObservable);
|
const smartStream = SmartDuplex.fromObservable(testObservable, {
|
||||||
|
handleBackpressure: false,
|
||||||
|
});
|
||||||
|
|
||||||
let receivedData = Buffer.alloc(0);
|
let receivedData = Buffer.alloc(0);
|
||||||
|
|
||||||
|
@ -7,45 +7,42 @@ let testIntake: smartstream.StreamIntake<string>;
|
|||||||
|
|
||||||
tap.test('should handle a read stream', async (tools) => {
|
tap.test('should handle a read stream', async (tools) => {
|
||||||
const counter = 0;
|
const counter = 0;
|
||||||
const testSmartstream = new smartstream.StreamWrapper([
|
const streamWrapper = new smartstream.StreamWrapper([
|
||||||
smartfile.fsStream.createReadStream('./test/assets/readabletext.txt'),
|
smartfile.fsStream.createReadStream('./test/assets/readabletext.txt'),
|
||||||
smartstream.createDuplexStream<Buffer, Buffer>(
|
new smartstream.SmartDuplex({
|
||||||
async (chunkStringArg: Buffer, streamTools) => {
|
writeFunction: async (chunkStringArg: Buffer, streamTools) => {
|
||||||
// do something with the stream here
|
// do something with the stream here
|
||||||
const result = chunkStringArg.toString().substr(0, 100);
|
const result = chunkStringArg.toString().substr(0, 100);
|
||||||
streamTools.pipeMore('wow =========== \n');
|
streamTools.push('wow =========== \n');
|
||||||
return Buffer.from(result);
|
return Buffer.from(result);
|
||||||
},
|
},
|
||||||
async (tools) => {
|
finalFunction: async (tools) => {
|
||||||
// tools.pipeMore('hey, this is the end')
|
|
||||||
return Buffer.from('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(),
|
new smartstream.SmartDuplex({
|
||||||
|
writeFunction: async (chunkStringArg) => {
|
||||||
|
console.log(chunkStringArg.toString());
|
||||||
|
},
|
||||||
|
finalFunction: async (tools) => {
|
||||||
|
tools.push(null);
|
||||||
|
},
|
||||||
|
})
|
||||||
]);
|
]);
|
||||||
await testSmartstream.run();
|
await streamWrapper.run();
|
||||||
});
|
});
|
||||||
|
|
||||||
tap.test('should create a valid Intake', async (tools) => {
|
tap.test('should create a valid Intake', async (tools) => {
|
||||||
testIntake = new smartstream.StreamIntake<string>();
|
testIntake = new smartstream.StreamIntake<string>();
|
||||||
testIntake
|
testIntake.pipe(
|
||||||
.getReadable()
|
new smartstream.SmartDuplex({
|
||||||
.pipe(
|
objectMode: true,
|
||||||
smartstream.createDuplexStream<string, string>(
|
writeFunction: async (chunkStringArg: string, streamTools) => {
|
||||||
async (chunkString) => {
|
|
||||||
await tools.delayFor(100);
|
await tools.delayFor(100);
|
||||||
console.log(chunkString);
|
console.log(chunkStringArg);
|
||||||
return chunkString;
|
return chunkStringArg;
|
||||||
},
|
|
||||||
async () => {
|
|
||||||
return 'noice';
|
|
||||||
}
|
}
|
||||||
)
|
})
|
||||||
)
|
)
|
||||||
.pipe(smartfile.fsStream.createWriteStream('./test/assets/writabletext.txt'));
|
.pipe(smartfile.fsStream.createWriteStream('./test/assets/writabletext.txt'));
|
||||||
const testFinished = tools.defer();
|
const testFinished = tools.defer();
|
||||||
|
@ -3,6 +3,6 @@
|
|||||||
*/
|
*/
|
||||||
export const commitinfo = {
|
export const commitinfo = {
|
||||||
name: '@push.rocks/smartstream',
|
name: '@push.rocks/smartstream',
|
||||||
version: '2.0.6',
|
version: '3.0.18',
|
||||||
description: 'simplifies access to node streams'
|
description: 'simplifies access to node streams'
|
||||||
}
|
}
|
||||||
|
@ -1,4 +1,6 @@
|
|||||||
export * from './smartstream.classes.smartstream.js';
|
export * from './smartstream.classes.passthrough.js';
|
||||||
|
export * from './smartstream.classes.smartduplex.js';
|
||||||
export * from './smartstream.classes.streamwrapper.js';
|
export * from './smartstream.classes.streamwrapper.js';
|
||||||
export * from './smartstream.classes.streamintake.js';
|
export * from './smartstream.classes.streamintake.js';
|
||||||
export * from './smartstream.duplex.js';
|
|
||||||
|
export * from './smartstream.functions.js'
|
||||||
|
21
ts/smartstream.classes.passthrough.ts
Normal file
21
ts/smartstream.classes.passthrough.ts
Normal file
@ -0,0 +1,21 @@
|
|||||||
|
import * as plugins from './smartstream.plugins.js';
|
||||||
|
|
||||||
|
export class PassThrough extends plugins.stream.Duplex {
|
||||||
|
constructor(options?: plugins.stream.DuplexOptions) {
|
||||||
|
super(options);
|
||||||
|
}
|
||||||
|
|
||||||
|
_read(size: number): void {
|
||||||
|
// No-op: Data written will be automatically available for reading.
|
||||||
|
}
|
||||||
|
|
||||||
|
_write(chunk: any, encoding: BufferEncoding, callback: (error?: Error | null) => void): void {
|
||||||
|
if (this.push(chunk, encoding)) {
|
||||||
|
callback();
|
||||||
|
} else {
|
||||||
|
this.once('drain', () => {
|
||||||
|
callback();
|
||||||
|
});
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
205
ts/smartstream.classes.smartduplex.ts
Normal file
205
ts/smartstream.classes.smartduplex.ts
Normal file
@ -0,0 +1,205 @@
|
|||||||
|
import * as plugins from './smartstream.plugins.js';
|
||||||
|
import { Duplex, type DuplexOptions } from 'stream';
|
||||||
|
|
||||||
|
export interface IStreamTools {
|
||||||
|
truncate: () => void;
|
||||||
|
push: (pipeObject: any) => void;
|
||||||
|
}
|
||||||
|
|
||||||
|
export interface IStreamWriteFunction<T, rT> {
|
||||||
|
(chunkArg: T, toolsArg: IStreamTools): Promise<rT>;
|
||||||
|
}
|
||||||
|
|
||||||
|
export interface IStreamFinalFunction<rT> {
|
||||||
|
(toolsArg: IStreamTools): Promise<rT>;
|
||||||
|
}
|
||||||
|
|
||||||
|
export interface ISmartDuplexOptions<TInput, TOutput> extends DuplexOptions {
|
||||||
|
handleBackpressure?: boolean;
|
||||||
|
readFunction?: () => Promise<void>;
|
||||||
|
writeFunction?: IStreamWriteFunction<TInput, TOutput>;
|
||||||
|
finalFunction?: IStreamFinalFunction<TOutput>;
|
||||||
|
// Add other custom options if necessary
|
||||||
|
}
|
||||||
|
|
||||||
|
export class SmartDuplex<TInput = any, TOutput = any> extends Duplex {
|
||||||
|
// STATIC
|
||||||
|
static fromBuffer(buffer: Buffer, options?: ISmartDuplexOptions<any, any>): SmartDuplex {
|
||||||
|
const smartDuplex = new SmartDuplex(options);
|
||||||
|
process.nextTick(() => {
|
||||||
|
smartDuplex.push(buffer);
|
||||||
|
smartDuplex.push(null); // Signal the end of the data
|
||||||
|
});
|
||||||
|
return smartDuplex;
|
||||||
|
}
|
||||||
|
|
||||||
|
static fromObservable(
|
||||||
|
observable: plugins.smartrx.rxjs.Observable<any>,
|
||||||
|
options?: ISmartDuplexOptions<any, any>
|
||||||
|
): SmartDuplex {
|
||||||
|
const smartStream = new SmartDuplex(options);
|
||||||
|
smartStream.observableSubscription = observable.subscribe({
|
||||||
|
next: (data) => {
|
||||||
|
if (!smartStream.push(data) && smartStream.handleBackpressure) {
|
||||||
|
// Pause the observable if the stream buffer is full
|
||||||
|
smartStream.observableSubscription?.unsubscribe();
|
||||||
|
smartStream.once('drain', () => {
|
||||||
|
// Resume the observable when the stream buffer is drained
|
||||||
|
smartStream.observableSubscription?.unsubscribe();
|
||||||
|
smartStream.observableSubscription = observable.subscribe((data) => {
|
||||||
|
smartStream.push(data);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
}
|
||||||
|
},
|
||||||
|
error: (err) => {
|
||||||
|
smartStream.emit('error', err);
|
||||||
|
},
|
||||||
|
complete: () => {
|
||||||
|
smartStream.push(null); // Signal the end of the data
|
||||||
|
},
|
||||||
|
});
|
||||||
|
|
||||||
|
return smartStream;
|
||||||
|
}
|
||||||
|
|
||||||
|
static fromReplaySubject(
|
||||||
|
replaySubject: plugins.smartrx.rxjs.ReplaySubject<any>,
|
||||||
|
options?: DuplexOptions
|
||||||
|
): SmartDuplex {
|
||||||
|
const smartStream = new SmartDuplex(options);
|
||||||
|
let isBackpressured = false;
|
||||||
|
|
||||||
|
// Subscribe to the ReplaySubject
|
||||||
|
const subscription = replaySubject.subscribe({
|
||||||
|
next: (data) => {
|
||||||
|
const canPush = smartStream.push(data);
|
||||||
|
if (!canPush) {
|
||||||
|
// If push returns false, pause the subscription because of backpressure
|
||||||
|
isBackpressured = true;
|
||||||
|
subscription.unsubscribe();
|
||||||
|
}
|
||||||
|
},
|
||||||
|
error: (err) => {
|
||||||
|
smartStream.emit('error', err);
|
||||||
|
},
|
||||||
|
complete: () => {
|
||||||
|
smartStream.push(null); // End the stream when the ReplaySubject completes
|
||||||
|
},
|
||||||
|
});
|
||||||
|
|
||||||
|
// Listen for 'drain' event to resume the subscription if it was paused
|
||||||
|
smartStream.on('drain', () => {
|
||||||
|
if (isBackpressured) {
|
||||||
|
isBackpressured = false;
|
||||||
|
// Resubscribe to the ReplaySubject since we previously paused
|
||||||
|
smartStream.observableSubscription = replaySubject.subscribe({
|
||||||
|
next: (data) => {
|
||||||
|
if (!smartStream.push(data)) {
|
||||||
|
smartStream.observableSubscription?.unsubscribe();
|
||||||
|
isBackpressured = true;
|
||||||
|
}
|
||||||
|
},
|
||||||
|
// No need to repeat error and complete handling here because it's already set up above
|
||||||
|
});
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
return smartStream;
|
||||||
|
}
|
||||||
|
|
||||||
|
// INSTANCE
|
||||||
|
private readFunction?: () => Promise<void>;
|
||||||
|
private handleBackpressure: boolean;
|
||||||
|
private writeFunction?: IStreamWriteFunction<TInput, TOutput>;
|
||||||
|
private finalFunction?: IStreamFinalFunction<TOutput>;
|
||||||
|
private observableSubscription?: plugins.smartrx.rxjs.Subscription;
|
||||||
|
|
||||||
|
constructor(optionsArg?: ISmartDuplexOptions<TInput, TOutput>) {
|
||||||
|
super(optionsArg);
|
||||||
|
this.readFunction = optionsArg?.readFunction;
|
||||||
|
this.writeFunction = optionsArg?.writeFunction;
|
||||||
|
this.finalFunction = optionsArg?.finalFunction;
|
||||||
|
this.handleBackpressure = optionsArg?.handleBackpressure ?? true;
|
||||||
|
}
|
||||||
|
|
||||||
|
public async _read(size: number): Promise<void> {
|
||||||
|
if (this.readFunction) {
|
||||||
|
await this.readFunction();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
public notBackpressured = true;
|
||||||
|
public get backpressured(): boolean {
|
||||||
|
return !this.notBackpressured;
|
||||||
|
}
|
||||||
|
public push(chunkArg?: TOutput | null): boolean {
|
||||||
|
const result = super.push(chunkArg);
|
||||||
|
if (!result && this.handleBackpressure) {
|
||||||
|
this.notBackpressured = false;
|
||||||
|
this.pause();
|
||||||
|
// Listen for 'drain' event to resume
|
||||||
|
this.once('drain', () => {
|
||||||
|
this.notBackpressured = true;
|
||||||
|
this.resume(); // Resume the source of data
|
||||||
|
});
|
||||||
|
}
|
||||||
|
return result;
|
||||||
|
}
|
||||||
|
|
||||||
|
// Ensure the _write method types the chunk as TInput and encodes TOutput
|
||||||
|
public async _write(chunk: TInput, encoding: string, callback: (error?: Error | null) => void) {
|
||||||
|
if (!this.writeFunction) {
|
||||||
|
return callback(new Error('No stream function provided'));
|
||||||
|
}
|
||||||
|
|
||||||
|
const tools: IStreamTools = {
|
||||||
|
truncate: () => {
|
||||||
|
this.push(null);
|
||||||
|
callback();
|
||||||
|
},
|
||||||
|
push: (pushArg: TOutput) => this.push(pushArg),
|
||||||
|
};
|
||||||
|
|
||||||
|
try {
|
||||||
|
const modifiedChunk = await this.writeFunction(chunk, tools);
|
||||||
|
if (modifiedChunk) {
|
||||||
|
this.push(modifiedChunk)
|
||||||
|
if (this.backpressured && this.handleBackpressure) {
|
||||||
|
this.once('drain', () => {
|
||||||
|
callback();
|
||||||
|
});
|
||||||
|
} else {
|
||||||
|
callback();
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
callback();
|
||||||
|
}
|
||||||
|
} catch (err) {
|
||||||
|
callback(err);
|
||||||
|
}
|
||||||
|
return this.notBackpressured;
|
||||||
|
}
|
||||||
|
|
||||||
|
public async _final(callback: (error?: Error | null) => void) {
|
||||||
|
if (this.finalFunction) {
|
||||||
|
const tools: IStreamTools = {
|
||||||
|
truncate: () => callback(),
|
||||||
|
push: (pipeObject) => this.push(pipeObject),
|
||||||
|
};
|
||||||
|
|
||||||
|
try {
|
||||||
|
const finalChunk = await this.finalFunction(tools);
|
||||||
|
if (finalChunk) {
|
||||||
|
this.push(finalChunk);
|
||||||
|
}
|
||||||
|
} catch (err) {
|
||||||
|
callback(err);
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
// nothing here
|
||||||
|
}
|
||||||
|
this.push(null);
|
||||||
|
callback();
|
||||||
|
}
|
||||||
|
}
|
@ -1,55 +0,0 @@
|
|||||||
import * as plugins from './smartstream.plugins.js';
|
|
||||||
import { Duplex, type DuplexOptions } from 'stream';
|
|
||||||
|
|
||||||
export class SmartStream extends Duplex {
|
|
||||||
private observableSubscription?: plugins.smartrx.rxjs.Subscription;
|
|
||||||
|
|
||||||
constructor(options?: DuplexOptions) {
|
|
||||||
super(options);
|
|
||||||
}
|
|
||||||
|
|
||||||
_read(size: number) {
|
|
||||||
// Implement if you need custom behavior, otherwise leave it empty
|
|
||||||
}
|
|
||||||
|
|
||||||
_write(chunk: any, encoding: string, callback: (error?: Error | null) => void) {
|
|
||||||
// Implement if you need custom behavior
|
|
||||||
callback();
|
|
||||||
}
|
|
||||||
|
|
||||||
static fromBuffer(buffer: Buffer, options?: DuplexOptions): SmartStream {
|
|
||||||
const smartStream = new SmartStream(options);
|
|
||||||
process.nextTick(() => {
|
|
||||||
smartStream.push(buffer);
|
|
||||||
smartStream.push(null); // Signal the end of the data
|
|
||||||
});
|
|
||||||
return smartStream;
|
|
||||||
}
|
|
||||||
|
|
||||||
static fromObservable(observable: plugins.smartrx.rxjs.Observable<any>, options?: DuplexOptions): SmartStream {
|
|
||||||
const smartStream = new SmartStream(options);
|
|
||||||
smartStream.observableSubscription = observable.subscribe({
|
|
||||||
next: (data) => {
|
|
||||||
if (!smartStream.push(data)) {
|
|
||||||
// Pause the observable if the stream buffer is full
|
|
||||||
smartStream.observableSubscription?.unsubscribe();
|
|
||||||
smartStream.once('drain', () => {
|
|
||||||
// Resume the observable when the stream buffer is drained
|
|
||||||
smartStream.observableSubscription?.unsubscribe();
|
|
||||||
smartStream.observableSubscription = observable.subscribe(data => {
|
|
||||||
smartStream.push(data);
|
|
||||||
});
|
|
||||||
});
|
|
||||||
}
|
|
||||||
},
|
|
||||||
error: (err) => {
|
|
||||||
smartStream.emit('error', err);
|
|
||||||
},
|
|
||||||
complete: () => {
|
|
||||||
smartStream.push(null); // Signal the end of the data
|
|
||||||
}
|
|
||||||
});
|
|
||||||
|
|
||||||
return smartStream;
|
|
||||||
}
|
|
||||||
}
|
|
@ -1,56 +1,42 @@
|
|||||||
import * as plugins from './smartstream.plugins.js';
|
import * as plugins from './smartstream.plugins.js';
|
||||||
|
|
||||||
export class StreamIntake<T> {
|
export class StreamIntake<T> extends plugins.stream.Readable {
|
||||||
private signalEndBoolean = false;
|
private signalEndBoolean = false;
|
||||||
private chunkStore: T[] = [];
|
private chunkStore: T[] = [];
|
||||||
|
|
||||||
public pushNextObservable = new plugins.smartrx.ObservableIntake<any>();
|
public pushNextObservable = new plugins.smartrx.ObservableIntake<any>();
|
||||||
|
|
||||||
private pushedNextDeferred = plugins.smartpromise.defer();
|
private pushedNextDeferred = plugins.smartpromise.defer();
|
||||||
|
|
||||||
private readableStream = plugins.from2.obj(async (size, next) => {
|
constructor(options?: plugins.stream.ReadableOptions) {
|
||||||
|
super({ ...options, objectMode: true }); // Ensure that we are in object mode.
|
||||||
|
this.pushNextObservable.push('please push next');
|
||||||
|
}
|
||||||
|
|
||||||
|
_read(size: number): void {
|
||||||
// console.log('get next');
|
// console.log('get next');
|
||||||
// execute without backpressure
|
const pushChunk = (): void => {
|
||||||
while (this.chunkStore.length > 0) {
|
if (this.chunkStore.length > 0) {
|
||||||
next(null, this.chunkStore.shift());
|
// If push returns false, then we should stop reading
|
||||||
}
|
if (!this.push(this.chunkStore.shift())) {
|
||||||
if (this.signalEndBoolean) {
|
return;
|
||||||
next(null, null);
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// lets trigger backpressure handling
|
if (this.chunkStore.length === 0) {
|
||||||
this.pushNextObservable.push('please push next');
|
if (this.signalEndBoolean) {
|
||||||
await this.pushedNextDeferred.promise;
|
// If we're done, push null to signal the end of the stream
|
||||||
this.pushedNextDeferred = plugins.smartpromise.defer();
|
this.push(null);
|
||||||
|
} else {
|
||||||
|
// Ask for more data and wait
|
||||||
|
this.pushNextObservable.push('please push next');
|
||||||
|
this.pushedNextDeferred.promise.then(() => {
|
||||||
|
this.pushedNextDeferred = plugins.smartpromise.defer(); // Reset the deferred
|
||||||
|
pushChunk(); // Try pushing the next chunk
|
||||||
|
});
|
||||||
|
}
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
// execute with backpressure
|
pushChunk();
|
||||||
while (this.chunkStore.length > 0) {
|
|
||||||
next(null, this.chunkStore.shift());
|
|
||||||
}
|
|
||||||
if (this.signalEndBoolean) {
|
|
||||||
next(null, null);
|
|
||||||
}
|
|
||||||
});
|
|
||||||
|
|
||||||
constructor() {
|
|
||||||
this.pushNextObservable.push('please push next');
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* returns a new style readble stream
|
|
||||||
*/
|
|
||||||
public getReadable() {
|
|
||||||
const readable = new plugins.stream.Readable({
|
|
||||||
objectMode: true,
|
|
||||||
});
|
|
||||||
return readable.wrap(this.readableStream);
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* returns an oldstyle readble stream
|
|
||||||
*/
|
|
||||||
public getReadableStream() {
|
|
||||||
return this.readableStream;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public pushData(chunkData: T) {
|
public pushData(chunkData: T) {
|
||||||
|
@ -94,14 +94,3 @@ export class StreamWrapper {
|
|||||||
return done.promise;
|
return done.promise;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
export let cleanPipe = () => {
|
|
||||||
return plugins.through2.obj(
|
|
||||||
(file, enc, cb) => {
|
|
||||||
cb();
|
|
||||||
},
|
|
||||||
(cb) => {
|
|
||||||
cb();
|
|
||||||
}
|
|
||||||
);
|
|
||||||
};
|
|
||||||
|
@ -1,83 +0,0 @@
|
|||||||
import * as plugins from './smartstream.plugins.js';
|
|
||||||
|
|
||||||
export interface ITruncateFunc {
|
|
||||||
(): void;
|
|
||||||
}
|
|
||||||
|
|
||||||
export interface IPipeMoreFunc {
|
|
||||||
(pipeObject: any): void;
|
|
||||||
}
|
|
||||||
|
|
||||||
export interface IStreamTools {
|
|
||||||
truncate: ITruncateFunc;
|
|
||||||
pipeMore: IPipeMoreFunc;
|
|
||||||
}
|
|
||||||
|
|
||||||
export interface IStreamFunction<T, rT> {
|
|
||||||
(chunkArg: T, toolsArg: IStreamTools): Promise<rT>;
|
|
||||||
}
|
|
||||||
|
|
||||||
export interface IStreamEndFunction<rT> {
|
|
||||||
(toolsArg: IStreamTools): Promise<rT>;
|
|
||||||
}
|
|
||||||
|
|
||||||
export interface IStreamOptions {
|
|
||||||
objectMode?: boolean;
|
|
||||||
readableObjectMode?: boolean;
|
|
||||||
writableObjectMode?: boolean;
|
|
||||||
}
|
|
||||||
|
|
||||||
export let createDuplexStream = <T, rT>(
|
|
||||||
funcArg: IStreamFunction<T, rT>,
|
|
||||||
endFuncArg?: IStreamEndFunction<rT>,
|
|
||||||
optionsArg: IStreamOptions = {
|
|
||||||
objectMode: false,
|
|
||||||
readableObjectMode: true,
|
|
||||||
writableObjectMode: true,
|
|
||||||
}
|
|
||||||
) => {
|
|
||||||
return plugins.through2(
|
|
||||||
optionsArg,
|
|
||||||
function (chunk, enc, cb) {
|
|
||||||
let truncated = false;
|
|
||||||
const tools: IStreamTools = {
|
|
||||||
truncate: () => {
|
|
||||||
truncated = true;
|
|
||||||
cb(null, null);
|
|
||||||
},
|
|
||||||
pipeMore: (pipeObject) => {
|
|
||||||
this.push(pipeObject);
|
|
||||||
},
|
|
||||||
};
|
|
||||||
const asyncWrapper = async () => {
|
|
||||||
const resultChunk: rT = await funcArg(chunk, tools);
|
|
||||||
if (!truncated) {
|
|
||||||
cb(null, resultChunk);
|
|
||||||
}
|
|
||||||
};
|
|
||||||
asyncWrapper().catch((err) => {
|
|
||||||
console.log(err);
|
|
||||||
});
|
|
||||||
},
|
|
||||||
function (cb) {
|
|
||||||
const tools: IStreamTools = {
|
|
||||||
truncate: () => {
|
|
||||||
cb();
|
|
||||||
},
|
|
||||||
pipeMore: (pushArg) => {
|
|
||||||
this.push(pushArg);
|
|
||||||
},
|
|
||||||
};
|
|
||||||
const asyncWrapper = async () => {
|
|
||||||
if (endFuncArg) {
|
|
||||||
const result = await endFuncArg(tools);
|
|
||||||
this.push(result);
|
|
||||||
}
|
|
||||||
cb();
|
|
||||||
};
|
|
||||||
asyncWrapper().catch((err) => {
|
|
||||||
console.log(err);
|
|
||||||
});
|
|
||||||
}
|
|
||||||
);
|
|
||||||
};
|
|
21
ts/smartstream.functions.ts
Normal file
21
ts/smartstream.functions.ts
Normal file
@ -0,0 +1,21 @@
|
|||||||
|
import { Transform, type TransformCallback, type TransformOptions } from 'stream';
|
||||||
|
import { SmartDuplex } from './smartstream.classes.smartduplex.js';
|
||||||
|
|
||||||
|
export interface AsyncTransformFunction<TInput, TOutput> {
|
||||||
|
(chunkArg: TInput): Promise<TOutput>;
|
||||||
|
}
|
||||||
|
|
||||||
|
export function createTransformFunction<TInput, TOutput>(
|
||||||
|
asyncFunction: AsyncTransformFunction<TInput, TOutput>,
|
||||||
|
options?: TransformOptions
|
||||||
|
): SmartDuplex {
|
||||||
|
const smartDuplexStream = new SmartDuplex({
|
||||||
|
...options,
|
||||||
|
writeFunction: async (chunkArg, toolsArg) => {
|
||||||
|
const result = await asyncFunction(chunkArg);
|
||||||
|
return result;
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
return smartDuplexStream;
|
||||||
|
}
|
@ -9,8 +9,3 @@ import * as smartrx from '@push.rocks/smartrx';
|
|||||||
|
|
||||||
export { smartpromise, smartrx };
|
export { smartpromise, smartrx };
|
||||||
|
|
||||||
// thirdparty
|
|
||||||
import from2 from 'from2';
|
|
||||||
import through2 from 'through2';
|
|
||||||
|
|
||||||
export { from2, through2 };
|
|
||||||
|
Reference in New Issue
Block a user