Добавлена очередь скачивания и конвертирования

server/core/Reader/BookConverter/ConvertBase.js

@@ -3,11 +3,10 @@ const iconv = require('iconv-lite');
 const chardet = require('chardet');
 const he = require('he');
 
+const LimitedQueue = require('../../LimitedQueue');
 const textUtils = require('./textUtils');
 const utils = require('../../utils');
 
-let execConverterCounter = 0;
-
 class ConvertBase {
     constructor(config) {
         this.config = config;
@@ -15,6 +14,7 @@ class ConvertBase {
         this.calibrePath = `${config.dataDir}/calibre/ebook-convert`;
         this.sofficePath = '/usr/bin/soffice';
         this.pdfToHtmlPath = '/usr/bin/pdftohtml';
+        this.queue = new LimitedQueue(2, 20, 3);
     }
 
     async run(data, opts) {// eslint-disable-line no-unused-vars
@@ -33,11 +33,14 @@ class ConvertBase {
     }
 
     async execConverter(path, args, onData) {
-        execConverterCounter++;
+        let q = null;
         try {
-            if (execConverterCounter > 10)
+            q = await this.queue.get(() => {onData();});
-                throw new Error('Слишком большая очередь конвертирования. Пожалуйста, попробуйте позже.');
+        } catch (e) {
+            throw new Error('Слишком большая очередь конвертирования. Пожалуйста, попробуйте позже.');
+        }
 
+        try {
             const result = await utils.spawnProcess(path, {args, onData});
             if (result.code != 0) {
                 let error = result.code;
@@ -54,7 +57,7 @@ class ConvertBase {
                 throw new Error(e);
             }
         } finally {
-            execConverterCounter--;
+            q.ret();
         }
     }
 

server/core/Reader/ReaderWorker.js

@@ -1,6 +1,7 @@
 const fs = require('fs-extra');
 const path = require('path');
 
+const LimitedQueue = require('../LimitedQueue');
 const WorkerState = require('../WorkerState');//singleton
 const FileDownloader = require('../FileDownloader');
 const FileDecompressor = require('../FileDecompressor');
@@ -26,6 +27,7 @@ class ReaderWorker {
             this.config.tempPublicDir = `${config.publicDir}/tmp`;
             fs.ensureDirSync(this.config.tempPublicDir);
 
+            this.queue = new LimitedQueue(5, 100, 3);
             this.workerState = new WorkerState();
             this.down = new FileDownloader(config.maxUploadFileSize);
             this.decomp = new FileDecompressor(2*config.maxUploadFileSize);
@@ -53,7 +55,21 @@ class ReaderWorker {
         let downloadedFilename = '';
         let isUploaded = false;
         let convertFilename = '';
+
+        let q = null;
         try {
+            wState.set({state: 'queue', step: 1, totalSteps: 1});
+            try {
+                let qSize = 0;
+                q = await this.queue.get((place) => {
+                    wState.set({place, progress: (qSize ? Math.round((qSize - place)/qSize*100) : 0)});
+                    if (!qSize)
+                        qSize = place;
+                });
+            } catch (e) {
+                throw new Error('Слишком большая очередь загрузки. Пожалуйста, попробуйте позже.');
+            }
+
             wState.set({state: 'download', step: 1, totalSteps: 3, url});
 
             const tempFilename = utils.randomHexString(30);
@@ -123,6 +139,8 @@ class ReaderWorker {
             wState.set({state: 'error', error: e.message});
         } finally {
             //clean
+            if (q)
+                q.ret();
             if (decompDir)
                 await fs.remove(decompDir);
             if (downloadedFilename && !isUploaded)