241 lines
8.2 KiB
JavaScript
241 lines
8.2 KiB
JavaScript
const WebSocket = require ('ws');
|
|
const _ = require('lodash');
|
|
|
|
const ReaderWorker = require('../core/Reader/ReaderWorker');//singleton
|
|
const JembaReaderStorage = require('../core/Reader/JembaReaderStorage');//singleton
|
|
const WorkerState = require('../core/WorkerState');//singleton
|
|
const BUCClient = require('../core/BookUpdateChecker/BUCClient');//singleton
|
|
const log = new (require('../core/AppLogger'))().log;//singleton
|
|
const utils = require('../core/utils');
|
|
|
|
const cleanPeriod = 1*60*1000;//1 минута
|
|
const closeSocketOnIdle = 5*60*1000;//5 минут
|
|
|
|
class WebSocketController {
|
|
constructor(wss, config) {
|
|
this.config = config;
|
|
this.isDevelopment = (config.branch == 'development');
|
|
|
|
this.readerStorage = new JembaReaderStorage();
|
|
this.readerWorker = new ReaderWorker(config);
|
|
this.workerState = new WorkerState();
|
|
|
|
this.configHash = '';
|
|
|
|
if (config.bucEnabled) {
|
|
this.bucClient = new BUCClient(config);
|
|
}
|
|
|
|
this.wss = wss;
|
|
|
|
wss.on('connection', (ws) => {
|
|
ws.on('message', (message) => {
|
|
this.onMessage(ws, message.toString());
|
|
});
|
|
|
|
ws.on('error', (err) => {
|
|
log(LM_ERR, err);
|
|
});
|
|
});
|
|
|
|
setTimeout(() => { this.periodicClean(); }, cleanPeriod);
|
|
}
|
|
|
|
periodicClean() {
|
|
try {
|
|
const now = Date.now();
|
|
this.wss.clients.forEach((ws) => {
|
|
if (!ws.lastActivity || now - ws.lastActivity > closeSocketOnIdle - 50) {
|
|
ws.terminate();
|
|
}
|
|
});
|
|
} finally {
|
|
setTimeout(() => { this.periodicClean(); }, cleanPeriod);
|
|
}
|
|
}
|
|
|
|
async onMessage(ws, message) {
|
|
let req = {};
|
|
try {
|
|
if (this.isDevelopment) {
|
|
log(`WebSocket-IN: ${message.substr(0, 4000)}`);
|
|
}
|
|
|
|
req = JSON.parse(message);
|
|
|
|
ws.lastActivity = Date.now();
|
|
|
|
//pong for WebSocketConnection
|
|
this.send({_rok: 1}, req, ws);
|
|
|
|
switch (req.action) {
|
|
case 'test':
|
|
await this.test(req, ws); break;
|
|
case 'get-config':
|
|
await this.getConfig(req, ws); break;
|
|
case 'load-book':
|
|
await this.loadBook(req, ws); break;
|
|
case 'worker-get-state':
|
|
await this.workerGetState(req, ws); break;
|
|
case 'worker-get-state-finish':
|
|
await this.workerGetStateFinish(req, ws); break;
|
|
case 'reader-storage':
|
|
await this.readerStorageDo(req, ws); break;
|
|
case 'upload-file-buf':
|
|
await this.uploadFileBuf(req, ws); break;
|
|
case 'upload-file-touch':
|
|
await this.uploadFileTouch(req, ws); break;
|
|
case 'check-buc':
|
|
await this.checkBuc(req, ws); break;
|
|
|
|
default:
|
|
throw new Error(`Action not found: ${req.action}`);
|
|
}
|
|
} catch (e) {
|
|
this.send({error: e.message}, req, ws);
|
|
}
|
|
}
|
|
|
|
send(res, req, ws) {
|
|
if (ws.readyState == WebSocket.OPEN) {
|
|
ws.lastActivity = Date.now();
|
|
let r = res;
|
|
if (req.requestId)
|
|
r = Object.assign({requestId: req.requestId}, r);
|
|
|
|
const message = JSON.stringify(r);
|
|
ws.send(message);
|
|
|
|
if (this.isDevelopment) {
|
|
log(`WebSocket-OUT: ${message.substr(0, 4000)}`);
|
|
}
|
|
|
|
}
|
|
}
|
|
|
|
//Actions ------------------------------------------------------------------
|
|
async test(req, ws) {
|
|
this.send({message: 'Liberama project is awesome'}, req, ws);
|
|
}
|
|
|
|
async getConfig(req, ws) {
|
|
if (Array.isArray(req.params)) {
|
|
const paramsSet = new Set(req.params);
|
|
const _configHash = req._configHash;
|
|
|
|
let response = {_useCached: true};
|
|
|
|
//оптимизация, чтобы не отдавал большой конфиг каждый раз при обновлении страницы
|
|
if (!_configHash || _configHash !== this.configHash) {
|
|
if (!this.configHash) {
|
|
const webConfig = _.pick(this.config, this.config.webConfigParams);
|
|
this.configHash = await utils.getBufHash(Buffer.from(JSON.stringify(webConfig)), 'sha256', 'hex');
|
|
}
|
|
|
|
response = _.pick(this.config, this.config.webConfigParams.filter(x => paramsSet.has(x)));
|
|
response._configHash = this.configHash;
|
|
}
|
|
|
|
this.send(response, req, ws);
|
|
} else {
|
|
throw new Error('params is not an array');
|
|
}
|
|
}
|
|
|
|
async loadBook(req, ws) {
|
|
const workerId = this.readerWorker.loadBookUrl({
|
|
url: req.url,
|
|
enableSitesFilter: (_.has(req, 'enableSitesFilter') ? req.enableSitesFilter : true),
|
|
skipHtmlCheck: (_.has(req, 'skipHtmlCheck') ? req.skipHtmlCheck : false),
|
|
isText: (_.has(req, 'isText') ? req.isText : false),
|
|
uploadFileName: (_.has(req, 'uploadFileName') ? req.uploadFileName : false),
|
|
djvuQuality: (_.has(req, 'djvuQuality') ? req.djvuQuality : false),
|
|
pdfAsText: (_.has(req, 'pdfAsText') ? req.pdfAsText : false),
|
|
pdfQuality: (_.has(req, 'pdfQuality') ? req.pdfQuality : false),
|
|
});
|
|
const state = this.workerState.getState(workerId);
|
|
|
|
this.send((state ? state : {}), req, ws);
|
|
}
|
|
|
|
async workerGetState(req, ws) {
|
|
if (!req.workerId)
|
|
throw new Error(`key 'workerId' is wrong`);
|
|
|
|
const state = this.workerState.getState(req.workerId);
|
|
this.send((state ? state : {}), req, ws);
|
|
}
|
|
|
|
async workerGetStateFinish(req, ws) {
|
|
if (!req.workerId)
|
|
throw new Error(`key 'workerId' is wrong`);
|
|
|
|
const refreshPause = 200;
|
|
let i = 0;
|
|
let state = {};
|
|
while (1) {// eslint-disable-line no-constant-condition
|
|
const prevProgress = state.progress || -1;
|
|
const prevState = state.state || '';
|
|
const lastModified = state.lastModified || 0;
|
|
state = this.workerState.getState(req.workerId);
|
|
|
|
this.send((state && lastModified != state.lastModified ? state : {}), req, ws);
|
|
if (!state) break;
|
|
|
|
if (state.state != 'finish' && state.state != 'error')
|
|
await utils.sleep(refreshPause);
|
|
else
|
|
break;
|
|
|
|
i++;
|
|
if (i > 3*60*1000/refreshPause) {//3 мин ждем телодвижений воркера
|
|
this.send({state: 'error', error: 'Время ожидания процесса истекло'}, req, ws);
|
|
break;
|
|
}
|
|
i = (prevProgress != state.progress || prevState != state.state ? 1 : i);
|
|
}
|
|
}
|
|
|
|
async readerStorageDo(req, ws) {
|
|
if (!req.body)
|
|
throw new Error(`key 'body' is empty`);
|
|
if (!req.body.action)
|
|
throw new Error(`key 'action' is empty`);
|
|
if (!req.body.items || Array.isArray(req.body.data))
|
|
throw new Error(`key 'items' is empty`);
|
|
|
|
this.send(await this.readerStorage.doAction(req.body), req, ws);
|
|
}
|
|
|
|
async uploadFileBuf(req, ws) {
|
|
if (!req.buf)
|
|
throw new Error(`key 'buf' is empty`);
|
|
|
|
this.send({url: await this.readerWorker.saveFileBuf(req.buf)}, req, ws);
|
|
}
|
|
|
|
async uploadFileTouch(req, ws) {
|
|
if (!req.url)
|
|
throw new Error(`key 'url' is empty`);
|
|
|
|
this.send({url: await this.readerWorker.uploadFileTouch(req.url)}, req, ws);
|
|
}
|
|
|
|
async checkBuc(req, ws) {
|
|
if (!this.config.bucEnabled)
|
|
throw new Error('BookUpdateChecker disabled');
|
|
|
|
if (!req.bookUrls)
|
|
throw new Error(`key 'bookUrls' is empty`);
|
|
|
|
if (!Array.isArray(req.bookUrls))
|
|
throw new Error(`key 'bookUrls' must be array`);
|
|
|
|
const data = await this.bucClient.checkBuc(req.bookUrls);
|
|
|
|
this.send({state: 'success', data}, req, ws);
|
|
}
|
|
}
|
|
|
|
module.exports = WebSocketController;
|