yujunjun/strapi
1
0
Fork 0
mirror of https://github.com/strapi/strapi.git synced 2025-09-24 16:04:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

send chunks of data instead of 1 at a time

Browse Source
This commit is contained in:
Bassel 2023-02-20 18:04:39 +02:00
parent 1472a5369c
commit 4c6a4d97dc
3 changed files with 77 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

66
packages/core/data-transfer/src/strapi/providers/remote-destination/index.ts
View File

@ -151,7 +151,6 @@ class RemoteStrapiDestinationProvider implements IDestinationProvider {
}
const wsProtocol = url.protocol === 'https:' ? 'wss:' : 'ws:';
const wsUrl = `${wsProtocol}//${url.host}${url.pathname}${TRANSFER_PATH}`;
// No auth defined, trying public access for transfer
if (!auth) {
ws = new WebSocket(wsUrl);
@ -215,35 +214,58 @@ class RemoteStrapiDestinationProvider implements IDestinationProvider {
}
createEntitiesWriteStream(): Writable {
const chunkSize = 100;
let entities: IEntity[] = [];
const startEntitiesTransferOnce = this.#startStepOnce('entities');
return new Writable({
objectMode: true,
final: async (callback) => {
if (entities.length > 0) {
const streamError = await this.#streamStep('entities', entities);
entities = [];
if (streamError) {
return callback(streamError);
}
}
const e = await this.#endStep('entities');
callback(e);
},
write: async (entity: IEntity, _encoding, callback) => {
const startError = await startEntitiesTransferOnce();
if (startError) {
return callback(startError);
}
const streamError = await this.#streamStep('entities', entity);
callback(streamError);
entities.push(entity);
if (entities.length === chunkSize) {
const streamError = await this.#streamStep('entities', entities);
if (streamError) {
return callback(streamError);
}
entities = [];
}
callback();
},
});
}
createLinksWriteStream(): Writable {
const chunkSize = 100;
let links: ILink[] = [];
const startLinksTransferOnce = this.#startStepOnce('links');
return new Writable({
objectMode: true,
final: async (callback) => {
if (links.length > 0) {
const streamError = await this.#streamStep('links', links);
if (streamError) {
return callback(streamError);
}
links = [];
}
const e = await this.#endStep('links');
callback(e);
@ -255,19 +277,35 @@ class RemoteStrapiDestinationProvider implements IDestinationProvider {
return callback(startError);
}
const streamError = await this.#streamStep('links', link);
if (links.length === chunkSize) {
const streamError = await this.#streamStep('links', links);
callback(streamError);
if (streamError) {
return callback(streamError);
}
links = [];
}
callback();
},
});
}
createConfigurationWriteStream(): Writable {
const chunkSize = 100;
let configurations: IConfiguration[] = [];
const startConfigurationTransferOnce = this.#startStepOnce('configuration');
return new Writable({
objectMode: true,
final: async (callback) => {
if (configurations.length > 0) {
const streamError = await this.#streamStep('configuration', configurations);
if (streamError) {
return callback(streamError);
}
configurations = [];
}
const e = await this.#endStep('configuration');
callback(e);
@ -279,9 +317,17 @@ class RemoteStrapiDestinationProvider implements IDestinationProvider {
return callback(startError);
}
const streamError = await this.#streamStep('configuration', configuration);
configurations.push(configuration);
if (configurations.length === chunkSize) {
const streamError = await this.#streamStep('configuration', configurations);
callback(streamError);
if (streamError) {
return callback(streamError);
}
configurations = [];
}
callback();
},
});
}

27
packages/core/data-transfer/src/strapi/remote/controllers/push.ts
View File

@ -70,28 +70,37 @@ const createPushController = (options: ILocalStrapiDestinationProviderOptions):
},
transfer: {
async entities(entity) {
async entities(entities) {
if (!streams.entities) {
streams.entities = provider.createEntitiesWriteStream();
}
await writeAsync(streams.entities, entity);
entities.map(async (entity) => {
if (streams.entities) {
await writeAsync(streams.entities, entity);
}
});
},
async links(link) {
async links(links) {
if (!streams.links) {
streams.links = await provider.createLinksWriteStream();
}
await writeAsync(streams.links, link);
links.map(async (link) => {
if (streams.links) {
await writeAsync(streams.links, link);
}
});
},
async configuration(config) {
async configuration(configs) {
if (!streams.configuration) {
streams.configuration = await provider.createConfigurationWriteStream();
}
await writeAsync(streams.configuration, config);
configs.map(async (config) => {
if (streams.configuration) {
await writeAsync(streams.configuration, config);
}
});
},
async assets(payload) {

6
packages/core/data-transfer/types/remote/protocol/client/transfer/push.d.ts vendored
View File

@ -3,9 +3,9 @@ import type { IEntity, ILink, IConfiguration, IAsset } from '../../../../common-
export type TransferPushMessage = CreateTransferMessage<
'step',
| TransferStepCommands<'entities', IEntity>
| TransferStepCommands<'links', ILink>
| TransferStepCommands<'configuration', IConfiguration>
| TransferStepCommands<'entities', IEntity[]>
| TransferStepCommands<'links', ILink[]>
| TransferStepCommands<'configuration', IConfiguration[]>
| TransferStepCommands<'assets', TransferAssetFlow | null>
>;