🎩 You're Invited:Meet the Socket team at Black Hat in Las Vegas, August 3-6.RSVP →
Sign In

@kenyip/bull

Package Overview
Dependencies
Maintainers
1
Versions
5
Alerts
File Explorer

Advanced tools

License
Socket logo

Install Socket

Detect and block malicious and high-risk dependencies

Install

@kenyip/bull - npm Package Compare versions

Comparing version
1.0.2
 to
1.0.3
+18
bull.js
"use strict";
const bull = require("./src/bull");
/**
* These export configurations enable JS and TS developers
* to consume knex in whatever way best suits their needs.
* Some examples of supported import syntax includes:
* - `const knex = require('knex')`
* - `const { knex } = require('knex')`
* - `import * as knex from 'knex'`
* - `import { knex } from 'knex'`
* - `import knex from 'knex'`
*/
bull.knex = bull;
bull.default = bull;
module.exports = bull;
+232
src/bull.js
"use strict";
const fs = require("fs");
const _ = require("lodash");
const uuid = require("uuid");
const path = require("path");
const yaml = require("yaml");
const Queue = require("bull");
const p = require("@kenyip/p");
const moment = require("moment");
const Redis = require("ioredis");
const EventEmitter = require("eventemitter2");
const stringToUUID = require("uuid-by-string");
const util = require("util");
function Bull(options = {}) {
EventEmitter.call(this);
const CONFIG_FILE_PATH = options.configPath || process.env.BULL_CONFIG_FILE || "bull.yml";
if (!fs.existsSync(CONFIG_FILE_PATH) && !options.redis) {
throw new Error("Please either define the `BULL_CONFIG_FILE` environment variable or create a `bull.yml` file on the root directory");
}
const config = yaml.parse(fs.readFileSync(CONFIG_FILE_PATH, "utf-8"));
if (!config.redis && options.redis) {
config.redis = options.redis;
}
this.id = stringToUUID(config.service);
this._config = config;
this._jobs = config.jobs;
this._redis = config.redis;
this._client = createRedis(config.redis);
this._subscriber = createRedis(config.redis);
this._queue = new p.queue.default({ concurrency: 1 });
this._job_queues = {};
}
Bull.prototype.addJob = function (id, payload) {
this
._job_queues[id]
.add(
id,
payload || {},
{
jobId: uuid.v4()
}
);
};
Bull.prototype.listen = async function () {
const jobs = this._jobs;
await p.map(
jobs,
async job => {
const queue = new Queue(
job.id,
{
prefix: this._config.prefix,
createClient: type => {
switch (type) {
case "client":
return this._client;
case "subscriber":
return this._subscriber;
default:
return createRedis(this._redis);
}
}
}
);
await cleanQueue(queue);
const onDataChange = () => {
this._queue.add(
async () => {
const logs = await queue.getJobs(["completed", "active", "failed"]);
await p.map(
logs,
async log => {
const updates = await p.map(
logs,
async log => {
const status = await log.getState();
let message;
if (log.returnvalue) {
try {
message = JSON.parse(message)?.message;
} catch (error) { }
}
const result = {
id: log.id,
job_id: job.id,
status,
// progress: await log.progress()
message: status == "failed" ? log.failedReason : message,
input_data: JSON.stringify(log.data),
stacktrace: log.stacktrace?.join("\n"),
return_data: _.isNull(log.returnvalue) ? null : typeof log.returnvalue == "object" ? JSON.stringify(log.returnvalue) : log.returnvalue,
processed_on: log.processedOn ? moment(log.processedOn).toDate() : null,
finished_on: log.finishedOn ? moment(log.finishedOn).toDate() : null,
time_different: log.finishedOn ? moment(log.finishedOn).diff(moment(log.processedOn), "seconds") : null
};
if (["completed", "failed"].includes(status)) {
await log
.remove()
.catch(error => { });
}
return result;
}
);
if (updates.length > 0) {
this.emit("jobs_update", log);
}
}
);
}
);
};
queue
.on("active", onDataChange)
.on("completed", onDataChange)
.on("failed", onDataChange);
// .on("progress", onDataChange);
queue.process(job.id, 1, require(path.join(process.cwd(), job.path)));
if (job.repeat) {
queue.add(
job.id,
job.repeat.payload || {},
{
jobId: uuid.v4(),
repeat: job.repeat
}
);
if (job.repeat.invoke) {
queue.add(
job.id,
job.repeat.payload || {},
{
jobId: uuid.v4()
}
);
}
}
this._job_queues[job.id] = queue;
},
{
concurrency: 1
}
);
};
Bull.prototype.getJobs(){
return this._jobs;
};
function createRedis(config) {
const redis = new Redis({
...config,
maxRetriesPerRequest: null,
enableReadyCheck: false
});
redis.setMaxListeners(100);
return redis;
}
async function cleanQueue(queue) {
await p.map(
[
"completed",
"wait",
"active",
"delayed",
"failed",
"paused"
],
async (status) => {
await queue.clean(0, status);
}
);
await queue
.getRepeatableJobs()
.then(jobs =>
p.map(
jobs,
job => queue.removeRepeatableByKey(job.key)
),
{ concurrency: 2 }
);
return true;
}
function renameKeys(object, mapping) {
return _
.chain(mapping)
.map((value, key) => [mapping[key], object[key]])
.fromPairs()
.toJSON();
}
util.inherits(Bull, EventEmitter);
module.exports = Bull;
+4
-4
package.json
{
"name": "@kenyip/bull",
"version": "1.0.2",
"version": "1.0.3",
"description": "A task manager based on `bull` framework",
"main": "index.js",
"main": "bull.js",
"scripts": {
"test": "echo \"Error: no test specified\" && exit 1"
},
"author": "",
"license": "ISC",
"author": "ken.yip",
"license": "MIT",
"dependencies": {

@@ -12,0 +12,0 @@ "@kenyip/p": "^1.0.0",

-244
index.js
"use strict";
const fs = require("fs");
const _ = require("lodash");
const uuid = require("uuid");
const path = require("path");
const yaml = require("yaml");
const Queue = require("bull");
const p = require("@kenyip/p");
const moment = require("moment");
const Redis = require("ioredis");
const EventEmitter = require("eventemitter2");
const stringToUUID = require("uuid-by-string");
const util = require("util");
function Bull(options = {}) {
EventEmitter.call(this);
const CONFIG_FILE_PATH = options.configPath || process.env.BULL_CONFIG_FILE || "bull.yml";
if (!fs.existsSync(CONFIG_FILE_PATH) && !options.redis) {
throw new Error("Please either define the `BULL_CONFIG_FILE` environment variable or create a `bull.yml` file on the root directory");
}
const config = yaml.parse(fs.readFileSync(CONFIG_FILE_PATH, "utf-8"));
if (!config.redis && options.redis) {
config.redis = options.redis;
}
this.id = stringToUUID(config.service);
this._config = config;
this._jobs = config.jobs;
this._redis = config.redis;
this._client = createRedis(config.redis);
this._subscriber = createRedis(config.redis);
this._queue = new p.queue.default({ concurrency: 1 });
this._job_queues = {};
}
Bull.prototype.listen = async function () {
// Update the defintions
await this.updateDbConfig();
// Clear the unprocessed jobs and schedule new jobs.
await this.scheduleJobs();
return true;
};
Bull.prototype.addJob = function (id, payload) {
this
._job_queues[id]
.add(
id,
payload || {},
{
jobId: uuid.v4()
}
);
};
Bull.prototype.scheduleJobs = async function () {
const jobs = this._jobs;
await p.map(
jobs,
async job => {
const queue = new Queue(
job.id,
{
prefix: this._config.prefix,
createClient: type => {
switch (type) {
case "client":
return this._client;
case "subscriber":
return this._subscriber;
default:
return createRedis(this._redis);
}
}
}
);
await cleanQueue(queue);
const onDataChange = () => {
this._queue.add(
async () => {
const logs = await queue.getJobs(["completed", "active", "failed"]);
await p.map(
logs,
async log => {
const updates = await p.map(
logs,
async log => {
const status = await log.getState();
let message;
if (log.returnvalue) {
try {
message = JSON.parse(message)?.message;
} catch (error) { }
}
const result = {
id: log.id,
job_id: job.id,
status,
// progress: await log.progress()
message: status == "failed" ? log.failedReason : message,
input_data: JSON.stringify(log.data),
stacktrace: log.stacktrace?.join("\n"),
return_data: _.isNull(log.returnvalue) ? null : typeof log.returnvalue == "object" ? JSON.stringify(log.returnvalue) : log.returnvalue,
processed_on: log.processedOn ? moment(log.processedOn).toDate() : null,
finished_on: log.finishedOn ? moment(log.finishedOn).toDate() : null,
time_different: log.finishedOn ? moment(log.finishedOn).diff(moment(log.processedOn), "seconds") : null
};
if (["completed", "failed"].includes(status)) {
await log
.remove()
.catch(error => { });
}
return result;
}
);
if (updates.length > 0) {
this.emit("jobs_update", log);
}
}
);
}
);
};
queue
.on("active", onDataChange)
.on("completed", onDataChange)
.on("failed", onDataChange);
// .on("progress", onDataChange);
queue.process(job.id, 1, require(path.join(process.cwd(), job.path)));
if (job.repeat) {
queue.add(
job.id,
job.repeat.payload || {},
{
jobId: uuid.v4(),
repeat: job.repeat
}
);
if (job.repeat.invoke) {
queue.add(
job.id,
job.repeat.payload || {},
{
jobId: uuid.v4()
}
);
}
}
this._job_queues[job.id] = queue;
},
{
concurrency: 1
}
);
};
Bull.prototype.updateDbConfig = async function () {
this.emit("update_functions", this._jobs);
};
function createRedis(config) {
const redis = new Redis({
...config,
maxRetriesPerRequest: null,
enableReadyCheck: false
});
redis.setMaxListeners(100);
return redis;
}
async function cleanQueue(queue) {
await p.map(
[
"completed",
"wait",
"active",
"delayed",
"failed",
"paused"
],
async (status) => {
await queue.clean(0, status);
}
);
await queue
.getRepeatableJobs()
.then(jobs =>
p.map(
jobs,
job => queue.removeRepeatableByKey(job.key)
),
{ concurrency: 2 }
);
return true;
}
function renameKeys(object, mapping) {
return _
.chain(mapping)
.map((value, key) => [mapping[key], object[key]])
.fromPairs()
.toJSON();
}
util.inherits(Bull, EventEmitter);
module.exports = Bull;