refs @4550 concurrency, logging improved, fixes, code clean, refactor

config.yml

@@ -1,4 +1,10 @@
-debug: true
+debug: false
+log: true
+dryPrint: false
+concurrency: 4
+reconnectTimeout: 10
+refreshRate: 1000
+tmpDir: /dev/shm/printnatura
 db:
   host: localhost
   port: 3306
@@ -9,5 +15,3 @@ salix:
   url: http://localhost:3000
   user: user
   password: password
-reconnectTimeout: 30
-refreshRate: 1000

main.js

@@ -5,7 +5,6 @@ async function main() {
 	await printServer.start();
 
 	process.on('SIGINT', async function() {
-		console.log(`\nBye ( ◕ ‿ ◕ )っ`);
 		try {
 			await printServer.stop();
 		} catch (err) {

print-server.js

@@ -10,7 +10,6 @@ const selectQuery = fs.readFileSync(`sql/selectQueued.sql`).toString();
 const jobDataQuery = fs.readFileSync(`sql/jobData.sql`).toString();
 const jobArgsQuery = fs.readFileSync(`sql/jobArgs.sql`).toString();
 const updateQuery = fs.readFileSync(`sql/updateState.sql`).toString();
-const appDir = path.dirname(require.main.filename);
 
 class PrintServer {
     async start() {
@@ -20,22 +19,41 @@ class PrintServer {
             conf = Object.assign({}, conf, yml(localConfFile));
         this.conf = conf;
 
-        const decoration = '△▽'.repeat(10)
         console.clear();
-        console.log(decoration, `${colors.bgBlack.white.bold(' Print')}${colors.bgBlack.green.bold('Natura ')}`, decoration, '\n')
-        await this.getToken();
+        const decoration = '△▽'.repeat(10);
+        const printnatura = colors.bgBlack.bold(' Print'.white + 'Natura '.green);
+        console.log(`${decoration} ${printnatura} ${decoration}`);
+        if (this.conf.dryPrint)
+            this.serverLog('log', 'Running in dry print mode! Documents won\'t be printed and PDFs not removed.'.yellow);
+
         await this.init();
-    }
-    async init() {
-        this.pool = await mysql.createPool(this.conf.db);
-        console.log('Connected to DB successfully.'.green);
-        await this.poll();
+        this.rejectionHandler = (err, p) => this.onRejection(err, p);
+        process.on('unhandledRejection', this.rejectionHandler);
+
+        this.serverLog('log', 'Ready to print'.green);
+        setTimeout(() => this.poll());
     }
     async stop() {
+        process.off('unhandledRejection', this.rejectionHandler);
+        this.serverLog('log', 'Bye ( ◕ ‿ ◕ )っ'.green);
         await this.end();
-        await axios.post(`${this.conf.salix.url}/api/Accounts/logout?access_token=${this.token}`);
+    }
+    async init() {
+        const conf = this.conf;
+        const api = this.api = axios.create({
+            baseURL: `${conf.salix.url}/api/`
+        });
+        api.interceptors.request.use(config => {
+            if (this.token)
+                config.headers['Authorization'] = this.token
+            return config;
+        });
+        await this.getToken();
+
+        this.pool = await mysql.createPool(conf.db);
     }
     async end() {
+        await this.api.post(`Accounts/logout`);
         if (this.pollTimeout) {
             clearTimeout(this.pollTimeout);
             this.pollTimeout = null;
@@ -44,62 +62,124 @@ class PrintServer {
     }
     async getToken() {
         const salix = this.conf.salix;
-        let response = await axios.post(`${salix.url}/api/Accounts/login`, {
+        let response = await this.api.post(`Accounts/login`, {
             user: salix.user,
             password: salix.password
         });
         this.token = response.data.token;
     }
-    async poll() {
-        let delay = this.conf.refreshRate;
-        this.pollTimeout = null;
-
-        try {
-            const conn = await this.pool.getConnection();
-
-            try {
-                if (this.dbDisconnected) {
-                    await conn.ping();
-                    this.dbDisconnected = false;
-                    console.log('DB connection recovered.'.green);
+    serverLog(realm, message) {
+        this.log(`Server`, realm, message);
     }
+    log(classMsg, realm, message) {
+        classMsg = `${classMsg}:`;
 
-                if (await this.printJob(conn))
-                    delay = 0;
-            } finally {
-                await conn.release();
+        switch(realm) {
+            case 'debug':
+                if (this.conf.debug) console.debug(classMsg, message.magenta);
+                break;
+            case 'error':
+                console.error(classMsg, message.red);
+                break;
+            default:
+                if (this.conf.log) console.log(classMsg, message);
         }
-        } catch (err) {
+    }
+    onRejection(err, p) {
+        console.debug('unhandledRejection');
+        this.errorHandler(err);
+    }
+    errorHandler(err) {
         if (err.code === 'ETIMEDOUT') {
-                delay = this.conf.reconnectTimeout;
-                if (!this.dbDisconnected) {
-                    this.dbDisconnected = true;
-                    console.log(`DB connection lost: ${err.message}`.red);
+            if (!this.dbDown) {
+                this.dbDown = true;
+                this.serverLog('error', `DB connection lost: ${err.message}`);
             }
         } else
             console.error(err);
     }
+    async poll() {
+        let conf = this.conf;
+        this.pollTimeout = null;
 
+        if (this.dbDown) {
+            let conn;
+            try {
+                try {
+                    conn = await this.pool.getConnection();
+                    await conn.ping();
+                    this.dbDown = false;
+                    this.serverLog('log', 'DB connection recovered'.green);
+                } catch (err) {
+                    conn.release();
+                }
+            } catch(e) {}
+        }
+
+        if (!this.dbDown) {
+            try {
+                let jobs;
+                let nJobs = 0;
+                do {
+                    jobs = [];
+                    for (let i = 0; i < conf.concurrency; i++) {
+                        const jobId = await this.getJob();
+                        if (jobId)
+                            jobs.push(this.printJob(jobId));
+                        else
+                            break;
+                    }
+
+                    nJobs += jobs.length;
+                    await Promise.all(jobs);
+                } while (jobs.length);
+
+                if (nJobs > 0)
+                    this.serverLog('debug', `${nJobs} jobs printed`);
+            } catch (err) {
+                this.errorHandler(err);
+            }
+        }
+
+        let delay = this.conf.refreshRate;
+        if (this.dbDown) delay = this.conf.reconnectTimeout;
         this.pollTimeout = setTimeout(() => this.poll(), delay);
     }
-    async printJob(conn) {
-        const conf = this.conf;
+    async getJob() {
         let jobId;
 
+        const conn = await this.pool.getConnection();
         try {
+            await conn.beginTransaction();
+            try {
+                const [[printJob]] = await conn.query(selectQuery);
+                if (printJob) {
+                    jobId = printJob.id;
+                    await conn.query(updateQuery, ['printing', null, jobId]);
+                    await conn.commit();
+                    this.jobLog(jobId, 'debug', 'get: printing');
+                }
+            } catch (err) {
+                await conn.rollback();
+                if (jobId)
+                    this.jobLog(jobId, 'error', err.message);
+                throw err;
+            }
+        } finally {
+            conn.release();
+        }
+
+        return jobId;
+    }
+    async printJob(jobId) {
+        const conf = this.conf;
         let jobData;
         const args = {};
 
+        const conn = await this.pool.getConnection();
         try {
             await conn.beginTransaction();
-                const [[printJob]] = await conn.query(selectQuery);
-                if (!printJob) {
-                    await conn.rollback();
-                    return;
-                }
-
-                jobId = printJob.id;
-
+            try {
                 // Job data
                 const [[data]] = await conn.query(jobDataQuery, jobId);
                 jobData = data;
@@ -109,7 +189,6 @@ class PrintServer {
                 for (const row of res)
                     args[row.name] = row.value;
 
-                await conn.query(updateQuery, ['printing', null, jobId]);
                 await conn.commit();
             } catch (err) {
                 await conn.rollback();
@@ -126,6 +205,7 @@ class PrintServer {
             });
 
             let pdfData;
+            let url;
             for (let attempts = 0; !pdfData && attempts < 2; attempts++) {
                 // URL params
                 const params = {userFk: jobData.userFk};
@@ -135,16 +215,16 @@ class PrintServer {
                 }
                 const urlParams = new URLSearchParams(params);
 
+                url = `${methodPath}?${urlParams.toString()}`;
+                this.jobLog(jobId, 'debug', `api: ${url}`);
+
                 // Request
                 try {
-                    const response = await axios({
+                    const response = await this.api({
                         method: 'get',
-                        url: `${conf.salix.url}/api/${methodPath}?${urlParams.toString()}`,
+                        url,
                         responseType: 'arraybuffer',
-                        headers: {
-                            'Accept': 'application/pdf',
-                            'Authorization': this.token
-                        }
+                        headers: {'Accept': 'application/pdf'}
                     });
                     pdfData = response.data;  
                 }
@@ -158,26 +238,26 @@ class PrintServer {
 
             // Save PDF to disk
             const printer = jobData.printer;
-            const tmpPath = path.join(appDir, 'tmp')
+            const tmpPath = conf.tmpDir;
             if (!fs.existsSync(tmpPath))
                 fs.mkdirSync(tmpPath)
-            const tmpFilePath = path.join(tmpPath, `${Math.random().toString(36).substring(7)}.pdf`);
+            const tmpFilePath = path.join(tmpPath, `job-${jobId}.pdf`);
             await fs.writeFile(tmpFilePath, pdfData, 'binary');
 
             // Print PDF
+            const printCommand = `lp -d "${printer}" "${tmpFilePath}"`;
+            this.jobLog(jobId, 'debug', `print: ${printCommand}`);
             try {
-                await pExec(`lp -d "${printer}" "${tmpFilePath}"`);
+                if (!conf.dryPrint) await pExec(printCommand);
             } catch(err) {
                 await fs.unlink(tmpFilePath);
                 throw new Error(`Print error: ${err.message}`);
             } 
 
             await conn.query(updateQuery, ['printed', null, jobId]);
+            this.jobLog(jobId, 'log', `report: ${jobData.report}, printer: ${printer}, get: ${url}`);
 
-            if (conf.debug)
-                console.debug(`(${colors.yellow(jobId)}) Document has been printed`, `[${args.collectionFk}, ${jobData.report}, ${printer}]`.green);
-
-            await fs.unlink(tmpFilePath);
+            if (!conf.dryPrint) await fs.unlink(tmpFilePath);
         } catch (err) {
             let message = err.message;
             if (err.name === 'AxiosError' && err.code === 'ERR_BAD_REQUEST') {
@@ -186,10 +266,17 @@ class PrintServer {
             }
 
             await conn.query(updateQuery, ['error', message, jobId]);
-            throw new Error(`(${jobId}) ${message}`);
-        }
+            this.jobLog(jobId, 'error', message);
 
-        return jobId;
+            const jobErr = new Error(`(${jobId}) ${message}`);
+            jobErr.stack = err.stack;
+            throw jobErr;
+        } finally {
+            conn.release();
+        }
+    }
+    jobLog(jobId, realm, message) {
+        this.log(`Job[${colors.yellow(jobId)}]`, realm, message);
     }
 }