Use bee-queue instead of Kue

docs/manage.en.md

@@ -1,11 +1,7 @@
 # Management guide
 
 ## Check the status of the job queue
-In the directory of Misskey:
-``` shell
-node_modules/kue/bin/kue-dashboard -p 3050
-```
-When you access port 3050, you will see the UI.
+coming soon
 
 ## Mark as 'admin' user
 ``` shell

docs/manage.ja.md

@@ -1,11 +1,7 @@
 # 運営ガイド
 
 ## ジョブキューの状態を調べる
-Misskeyのディレクトリで:
-``` shell
-node_modules/kue/bin/kue-dashboard -p 3050
-```
-ポート3050にアクセスするとUIが表示されます
+coming soon
 
 ## 管理者ユーザーを設定する
 ``` shell

package.json

@@ -55,7 +55,6 @@
 		"@types/koa-send": "4.1.1",
 		"@types/koa-views": "2.0.3",
 		"@types/koa__cors": "2.2.2",
-		"@types/kue": "0.11.9",
 		"@types/minio": "6.0.2",
 		"@types/mkdirp": "0.5.2",
 		"@types/mocha": "5.2.3",
@@ -86,6 +85,7 @@
 		"autosize": "4.0.2",
 		"autwh": "0.1.0",
 		"bcryptjs": "2.4.3",
+		"bee-queue": "1.2.2",
 		"bootstrap-vue": "2.0.0-rc.11",
 		"cafy": "11.3.0",
 		"chalk": "2.4.1",
@@ -144,7 +144,6 @@
 		"koa-send": "5.0.0",
 		"koa-slow": "2.1.0",
 		"koa-views": "6.1.4",
-		"kue": "0.11.6",
 		"loader-utils": "1.1.0",
 		"mecab-async": "0.1.2",
 		"minio": "6.0.0",

src/index.ts

@@ -31,9 +31,6 @@ if (process.env.NODE_ENV != 'production') {
 	process.env.DEBUG = 'misskey:*';
 }
 
-// https://github.com/Automattic/kue/issues/822
-require('events').EventEmitter.prototype._maxListeners = 512;
-
 // Start app
 main();
 

src/queue/index.ts

@@ -1,52 +1,36 @@
-import { createQueue } from 'kue';
+import * as Queue from 'bee-queue';
 
 import config from '../config';
 import http from './processors/http';
 import { ILocalUser } from '../models/user';
 
-const queue = createQueue({
+const queue = new Queue('misskey', {
 	redis: {
 		port: config.redis.port,
 		host: config.redis.host,
-		auth: config.redis.pass
-	}
+		password: config.redis.pass
+	},
+
+	removeOnSuccess: true,
+	removeOnFailure: true
 });
 
-process.once('SIGTERM', () => {
-	queue.shutdown(5000, (err: any) => {
-		console.log('Kue shutdown: ', err || '');
-		process.exit(0);
-	});
-});
-
-export function createHttp(data: any) {
-	return queue
-		.create('http', data)
-		.removeOnComplete(true)
-		.events(false)
-		.attempts(8)
-		.backoff({ delay: 16384, type: 'exponential' });
+export function createHttpJob(data: any) {
+	return queue.createJob(data)
+		.retries(4)
+		.backoff('exponential', 16384) // 16s
+		.save();
 }
 
 export function deliver(user: ILocalUser, content: any, to: any) {
-	createHttp({
-		title: 'deliver',
+	createHttpJob({
 		type: 'deliver',
 		user,
 		content,
 		to
-	}).save();
+	});
 }
 
 export default function() {
-	/*
-		256 is the default concurrency limit of Mozilla Firefox and Google
-		Chromium.
-		a8af215e691f3a2205a3758d2d96e9d328e100ff - chromium/src.git - Git at Google
-		https://chromium.googlesource.com/chromium/src.git/+/a8af215e691f3a2205a3758d2d96e9d328e100ff
-		Network.http.max-connections - MozillaZine Knowledge Base
-		http://kb.mozillazine.org/Network.http.max-connections
-	*/
-	//queue.process('http', 256, http);
-	queue.process('http', 128, http);
+	queue.process(8, http);
 }

src/queue/processors/http/deliver.ts

@@ -1,8 +1,8 @@
-import * as kue from 'kue';
+import * as bq from 'bee-queue';
 
 import request from '../../../remote/activitypub/request';
 
-export default async (job: kue.Job, done: any): Promise<void> => {
+export default async (job: bq.Job, done: any): Promise<void> => {
 	try {
 		await request(job.data.user, job.data.to, job.data.content);
 		done();

src/queue/processors/http/process-inbox.ts

@@ -1,4 +1,4 @@
-import * as kue from 'kue';
+import * as bq from 'bee-queue';
 import * as debug from 'debug';
 
 const httpSignature = require('http-signature');
@@ -10,7 +10,7 @@ import { resolvePerson } from '../../../remote/activitypub/models/person';
 const log = debug('misskey:queue:inbox');
 
 // ユーザーのinboxにアクティビティが届いた時の処理
-export default async (job: kue.Job, done: any): Promise<void> => {
+export default async (job: bq.Job, done: any): Promise<void> => {
 	const signature = job.data.signature;
 	const activity = job.data.activity;
 

src/server/activitypub.ts

@@ -3,7 +3,7 @@ import * as Router from 'koa-router';
 const json = require('koa-json-body');
 const httpSignature = require('http-signature');
 
-import { createHttp } from '../queue';
+import { createHttpJob } from '../queue';
 import pack from '../remote/activitypub/renderer';
 import Note from '../models/note';
 import User, { isLocalUser, ILocalUser, IUser } from '../models/user';
@@ -30,11 +30,11 @@ function inbox(ctx: Router.IRouterContext) {
 		return;
 	}
 
-	createHttp({
+	createHttpJob({
 		type: 'processInbox',
 		activity: ctx.request.body,
 		signature
-	}).save();
+	});
 
 	ctx.status = 202;
 }