Compare commits

...

4 Commits

4 changed files with 23 additions and 2 deletions

View File

@ -1,5 +1,16 @@
# Changelog
## 2024-10-16 - 3.1.6 - fix(VirtualStream)
Fix backpressure handling in VirtualStream workOnQueue method
- Resolved an issue in the workOnQueue method of VirtualStream where concurrent execution was not properly managed.
- Introduced a workingDeferred promise to ensure proper queue handling and resolve potential race conditions.
## 2024-10-16 - 3.1.5 - fix(virtualstream)
Add console log for debugging backpressure feedback loop
- Inserted a console log message to provide insight when waiting due to backpressure in the workOnQueue method.
## 2024-10-16 - 3.1.4 - fix(VirtualStream)
Corrected the logic for backpressure handling in response

View File

@ -1,6 +1,6 @@
{
"name": "@api.global/typedrequest",
"version": "3.1.4",
"version": "3.1.6",
"private": false,
"description": "A TypeScript library for making typed requests towards APIs, including facilities for handling requests, routing, and virtual stream handling.",
"main": "dist_ts/index.js",

View File

@ -3,6 +3,6 @@
*/
export const commitinfo = {
name: '@api.global/typedrequest',
version: '3.1.4',
version: '3.1.6',
description: 'A TypeScript library for making typed requests towards APIs, including facilities for handling requests, routing, and virtual stream handling.'
}

View File

@ -143,10 +143,17 @@ export class VirtualStream<T = Uint8Array> implements plugins.typedRequestInterf
constructor() {}
workingDeferred: plugins.smartpromise.Deferred<void>;
/**
* takes care of sending
*/
private async workOnQueue() {
if (this.workingDeferred) {
return this.workingDeferred.promise;
} else {
this.workingDeferred = plugins.smartpromise.defer();
}
if(this.side === 'requesting') {
let thisSideIsBackpressured = !this.receiveBackpressuredArray.checkSpaceAvailable();
let otherSideHasNext = false;
@ -180,6 +187,7 @@ export class VirtualStream<T = Uint8Array> implements plugins.typedRequestInterf
while (this.sendBackpressuredArray.data.length > 0 || otherSideHasNext) {
if (otherSideIsBackpressured) {
while (otherSideIsBackpressured) {
console.log('waiting for feedback because of backpressure...');
await plugins.smartdelay.delayFor(50);
await getFeedback();
}
@ -218,6 +226,8 @@ export class VirtualStream<T = Uint8Array> implements plugins.typedRequestInterf
}
}
this.workingDeferred.resolve();
this.workingDeferred = null;
}
/**