yujunjun/strapi
1
0
Fork 0
mirror of https://github.com/strapi/strapi.git synced 2025-07-22 16:37:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: Data transfer stream backpressure from providers

Browse Source
This commit is contained in:
Christian 2023-10-30 11:43:20 +01:00 committed by GitHub
parent 9e14912c58
commit c76dc9cb2b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 27 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

61
packages/core/data-transfer/src/strapi/remote/handlers/push.ts
View File

@ -383,44 +383,37 @@ export const createPushController = handlerControllerFactory<Partial<PushHandler
}
for (const item of payload) {
/**
* This queues the given callback function to be executed in the next iteration
* of the event loop, immediately after the current operation completes.
*/
process.nextTick(async () => {
const { action, assetID } = item;
const { action, assetID } = item;
if (!assetsStream) {
throw new Error('Stream not defined');
}
if (!assetsStream) {
throw new Error('Stream not defined');
}
if (action === 'start') {
this.assets[assetID] = { ...item.data, stream: new PassThrough() };
writeAsync(assetsStream, this.assets[assetID]);
}
if (action === 'start') {
this.assets[assetID] = { ...item.data, stream: new PassThrough() };
writeAsync(assetsStream, this.assets[assetID]);
}
if (action === 'stream') {
// The buffer has gone through JSON operations and is now of shape { type: "Buffer"; data: UInt8Array }
// We need to transform it back into a Buffer instance
const rawBuffer = item.data as unknown as { type: 'Buffer'; data: Uint8Array };
const chunk = Buffer.from(rawBuffer.data);
if (action === 'stream') {
// The buffer has gone through JSON operations and is now of shape { type: "Buffer"; data: UInt8Array }
// We need to transform it back into a Buffer instance
const rawBuffer = item.data as unknown as { type: 'Buffer'; data: Uint8Array };
const chunk = Buffer.from(rawBuffer.data);
await writeAsync(this.assets[assetID].stream, chunk);
}
await writeAsync(this.assets[assetID].stream, chunk);
}
if (action === 'end') {
await new Promise<void>((resolve, reject) => {
const { stream: assetStream } = this.assets[assetID];
assetStream
.on('close', () => {
delete this.assets[assetID];
resolve();
})
.on('error', reject)
.end();
});
}
});
if (action === 'end') {
await new Promise<void>((resolve, reject) => {
const { stream: assetStream } = this.assets[assetID];
assetStream
.on('close', () => {
delete this.assets[assetID];
resolve();
})
.on('error', reject)
.end();
});
}
}
},