PixelPiggy/CS_float

1const EventEmitter = require('events').EventEmitter;2const errors = require('../errors');3 4class Queue extends EventEmitter {5    constructor() {6        super();7 8        this.queue = [];9        this.users = {};10        this.running = false;11    }12 13    size() {14        return this.queue.length;15    }16 17    process(concurrency, controller, handler) {18        this.handler = handler;19        this.concurrency = concurrency;20        this.processing = 0;21 22        this.start();23 24        // Monkey patch to ensure queue processing size is roughly equal to amount of bots ready25        setInterval(() => {26            // Update concurrency level, possible bots went offline or otherwise27            const oldConcurrency = this.concurrency;28            this.concurrency = controller.getReadyAmount();29 30            if (this.concurrency > oldConcurrency) {31                for (let i = 0; i < this.concurrency - oldConcurrency; i++) {32                    this.checkQueue();33                }34            }35 36        }, 50);37    }38 39    addJob(job, max_attempts) {40        if (!(job.ip in this.users)) {41            this.users[job.ip] = 0;42        }43 44        for (const link of job.getRemainingLinks()) {45            this.queue.push({46                data: link,47                max_attempts: max_attempts,48                attempts: 0,49                ip: job.ip,50            });51 52            this.users[job.ip]++;53            this.checkQueue();54        }55    }56 57    checkQueue() {58        if (!this.running) return;59 60        if (this.queue.length > 0 && this.processing < this.concurrency) {61            // there is a free bot, process the job62            let job = this.queue.shift();63 64            this.processing += 1;65 66            this.handler(job).then((delay) => {67                if (!delay) delay = 0;68 69                // Allow users to request again before the promise resolve delay70                this.users[job.ip]--;71 72                return new Promise((resolve, reject) => {73                    setTimeout(() => {74                        resolve();75                    }, delay);76                });77            }).catch((err) => {78                if (err !== errors.NoBotsAvailable) {79                    job.attempts++;80                }81 82                if (job.attempts === job.max_attempts) {83                    // job failed84                    this.emit('job failed', job, err);85                    this.users[job.ip]--;86                }87                else {88                    // try again89                    this.queue.unshift(job);90                }91            }).then(() => {92                this.processing -= 1;93                this.checkQueue();94            });95        }96    }97 98    start() {99        if (!this.running) {100            this.running = true;101            this.checkQueue();102        }103    }104 105    pause() {106        if (this.running) this.running = false;107    }108 109    /**110     * Returns number of requests the ip currently has queued111     */112    getUserQueuedAmt(ip) {113        return this.users[ip] || 0;114    }115}116 117module.exports = Queue;118