fabio286/antares
1
1
Fork 0
mirror of https://github.com/Fabio286/antares.git synced 2025-06-05 21:59:22 +02:00
Code Issues Releases Wiki Activity

perf(MySQL): import tasks managed with async queue

Browse Source
This commit is contained in:
Giulio Ganci
2022-03-05 18:54:08 +01:00
parent f444746f46
commit bbe13f27dc
1 changed files with 16 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
src/main/libs/importers/sql/MysqlImporter.js
View File

@ -1,4 +1,5 @@
import fs from 'fs/promises'; import fs from 'fs/promises';
import { queue } from 'async';
import SqlParser from '../../../../common/libs/sqlParser'; import SqlParser from '../../../../common/libs/sqlParser';
import { BaseImporter } from '../BaseImporter'; import { BaseImporter } from '../BaseImporter';
@ -6,7 +7,6 @@ export default class MysqlImporter extends BaseImporter {
constructor (client, options) { constructor (client, options) {
super(options); super(options);
this._client = client; this._client = client;
this._queries = [];
} }
async import () { async import () {
@ -16,11 +16,17 @@ export default class MysqlImporter extends BaseImporter {
const parser = new SqlParser(); const parser = new SqlParser();
let readPosition = 0; let readPosition = 0;
let queryCount = 0; let queryCount = 0;
const q = queue(async (query) => await this._client.raw(query));
q.error((error, query) => {
this.emit('query-error', { sql: query, message: error.sqlMessage, sqlSnippet: error.sql, time: new Date().getTime() });
});
this.emitUpdate({ this.emitUpdate({
fileSize: totalFileSize, fileSize: totalFileSize,
readPosition: 0, readPosition: 0,
percentage: 0 percentage: 0,
queryCount: 0
}); });
await this._client.use(this._options.schema); await this._client.use(this._options.schema);
@ -35,25 +41,22 @@ export default class MysqlImporter extends BaseImporter {
parser.on('error', reject); parser.on('error', reject);
parser.on('finish', () => { parser.on('finish', async () => {
Promise.all(this._queries) console.log('TOTAL QUERIES', queryCount);
.then(() => { console.log('import end');
console.timeEnd('import'); await q.drain(); // not sure of this
console.log('TOTAL QUERIES', queryCount); resolve();
console.log('import end');
resolve();
})
.catch(reject);
}); });
parser.on('data', async query => { parser.on('data', async (query) => {
this._queries.push(this._client.raw(query, { split: false })); q.push(query);
queryCount++; queryCount++;
}); });
this._fileHandler.on('data', (chunk) => { this._fileHandler.on('data', (chunk) => {
readPosition += chunk.length; readPosition += chunk.length;
this.emitUpdate({ this.emitUpdate({
queryCount,
readPosition, readPosition,
percentage: readPosition / totalFileSize * 100 percentage: readPosition / totalFileSize * 100
}); });