refactor: worker threads to export sql dump instead of process

This commit is contained in:
Fabio286 2023-12-02 09:31:54 +01:00
parent 329246e2d8
commit c176841b75
2 changed files with 28 additions and 24 deletions

View File

@ -4,6 +4,7 @@ import * as workers from 'common/interfaces/workers';
import { dialog, ipcMain } from 'electron';
import * as fs from 'fs';
import * as path from 'path';
import { Worker } from 'worker_threads';
import { validateSender } from '../libs/misc/validateSender';
@ -11,7 +12,7 @@ const isDevelopment = process.env.NODE_ENV !== 'production';
const isFlatpak = process.platform === 'linux' && process.env.DISTRIBUTION === 'flatpak';
export default (connections: {[key: string]: antares.Client}) => {
let exporter: ChildProcess = null;
let exporter: Worker = null;
let importer: ChildProcess = null;
ipcMain.handle('create-schema', async (event, params) => {
@ -203,16 +204,16 @@ export default (connections: {[key: string]: antares.Client}) => {
if (!validateSender(event.senderFrame)) return { status: 'error', response: 'Unauthorized process' };
if (exporter !== null) {
exporter.kill();
exporter.terminate();
return;
}
return new Promise((resolve/*, reject */) => {
(async () => {
if (isFlatpak) {
resolve({ status: 'error', response: 'Temporarily unavailable on Flatpak' });
return;
}
// if (isFlatpak) {
// resolve({ status: 'error', response: 'Temporarily unavailable on Flatpak' });
// return;
// }
if (fs.existsSync(rest.outputFile)) { // If file exists ask for replace
const result = await dialog.showMessageBox({
@ -233,12 +234,10 @@ export default (connections: {[key: string]: antares.Client}) => {
}
}
// Init exporter process
exporter = fork(isDevelopment ? './dist/exporter.js' : './exporter.js', [], {
execArgv: isDevelopment ? ['--inspect=9224'] : undefined
});
// Init exporter thread
exporter = new Worker(isDevelopment ? './dist/exporter.js' : './exporter.js');
exporter.send({
exporter.postMessage({
type: 'init',
client: {
name: type,
@ -255,19 +254,19 @@ export default (connections: {[key: string]: antares.Client}) => {
event.sender.send('export-progress', payload);
break;
case 'end':
setTimeout(() => { // Ensures that writing process has finished
exporter.kill();
setTimeout(() => { // Ensures that writing thread has finished
exporter.terminate();
exporter = null;
}, 2000);
resolve({ status: 'success', response: payload });
break;
case 'cancel':
exporter.kill();
exporter.terminate();
exporter = null;
resolve({ status: 'error', response: 'Operation cancelled' });
break;
case 'error':
exporter.kill();
exporter.terminate();
exporter = null;
resolve({ status: 'error', response: payload });
break;
@ -298,7 +297,7 @@ export default (connections: {[key: string]: antares.Client}) => {
if (result.response === 1) {
willAbort = true;
exporter.send({ type: 'cancel' });
exporter.postMessage({ type: 'cancel' });
}
}

View File

@ -1,6 +1,7 @@
import * as antares from 'common/interfaces/antares';
import * as log from 'electron-log/main';
import * as fs from 'fs';
import { parentPort } from 'worker_threads';
import { MySQLClient } from '../libs/clients/MySQLClient';
import { PostgreSQLClient } from '../libs/clients/PostgreSQLClient';
@ -13,7 +14,9 @@ log.transports.file.fileName = 'workers.log';
log.errorHandler.startCatching();
// eslint-disable-next-line @typescript-eslint/no-explicit-any
process.on('message', async ({ type, client, tables, options }: any) => {
const exportHandler = async (data: any) => {
const { type, client, tables, options } = data;
if (type === 'init') {
try {
const connection = await ClientsFactory.getClient({
@ -32,7 +35,7 @@ process.on('message', async ({ type, client, tables, options }: any) => {
exporter = new PostgreSQLExporter(connection as PostgreSQLClient, tables, options);
break;
default:
process.send({
parentPort.postMessage({
type: 'error',
payload: `"${client.name}" exporter not aviable`
});
@ -41,14 +44,14 @@ process.on('message', async ({ type, client, tables, options }: any) => {
exporter.once('error', err => {
log.error(err.toString());
process.send({
parentPort.postMessage({
type: 'error',
payload: err.toString()
});
});
exporter.once('end', () => {
process.send({
parentPort.postMessage({
type: 'end',
payload: { cancelled: exporter.isCancelled }
});
@ -57,11 +60,11 @@ process.on('message', async ({ type, client, tables, options }: any) => {
exporter.once('cancel', () => {
fs.unlinkSync(exporter.outputFile);
process.send({ type: 'cancel' });
parentPort.postMessage({ type: 'cancel' });
});
exporter.on('progress', state => {
process.send({
parentPort.postMessage({
type: 'export-progress',
payload: state
});
@ -71,7 +74,7 @@ process.on('message', async ({ type, client, tables, options }: any) => {
}
catch (err) {
log.error(err.toString());
process.send({
parentPort.postMessage({
type: 'error',
payload: err.toString()
});
@ -79,4 +82,6 @@ process.on('message', async ({ type, client, tables, options }: any) => {
}
else if (type === 'cancel')
exporter.cancel();
});
};
parentPort.on('message', exportHandler);