summaryrefslogtreecommitdiff
path: root/src/queue
diff options
context:
space:
mode:
authorsyuilo <Syuilotan@yahoo.co.jp>2021-10-22 05:36:48 +0900
committerGitHub <noreply@github.com>2021-10-22 05:36:48 +0900
commit4e4c559db6964cbf17fcadf38d55fc79c995ca42 (patch)
tree0e136ef1bf75c201b5805e667129082db3abcb61 /src/queue
parentリモートノートで意図せずローカルカスタム絵文字が使... (diff)
downloadsharkey-4e4c559db6964cbf17fcadf38d55fc79c995ca42.tar.gz
sharkey-4e4c559db6964cbf17fcadf38d55fc79c995ca42.tar.bz2
sharkey-4e4c559db6964cbf17fcadf38d55fc79c995ca42.zip
Migrate to Chart.js v3 (#7896)
* wip * wip * wip * wip * wip * wip * wip * 定期的にresync * Update overview.vue * wip * wip
Diffstat (limited to 'src/queue')
-rw-r--r--src/queue/index.ts28
-rw-r--r--src/queue/processors/system/index.ts12
-rw-r--r--src/queue/processors/system/resync-charts.ts21
-rw-r--r--src/queue/queues.ts1
4 files changed, 55 insertions, 7 deletions
diff --git a/src/queue/index.ts b/src/queue/index.ts
index 43c062bae7..37eb809604 100644
--- a/src/queue/index.ts
+++ b/src/queue/index.ts
@@ -10,7 +10,7 @@ import procesObjectStorage from './processors/object-storage/index';
import { queueLogger } from './logger';
import { DriveFile } from '@/models/entities/drive-file';
import { getJobInfo } from './get-job-info';
-import { dbQueue, deliverQueue, inboxQueue, objectStorageQueue } from './queues';
+import { systemQueue, dbQueue, deliverQueue, inboxQueue, objectStorageQueue } from './queues';
import { ThinUser } from './types';
import { IActivity } from '@/remote/activitypub/type';
@@ -22,11 +22,20 @@ function renderError(e: Error): any {
};
}
+const systemLogger = queueLogger.createSubLogger('system');
const deliverLogger = queueLogger.createSubLogger('deliver');
const inboxLogger = queueLogger.createSubLogger('inbox');
const dbLogger = queueLogger.createSubLogger('db');
const objectStorageLogger = queueLogger.createSubLogger('objectStorage');
+systemQueue
+ .on('waiting', (jobId) => systemLogger.debug(`waiting id=${jobId}`))
+ .on('active', (job) => systemLogger.debug(`active id=${job.id}`))
+ .on('completed', (job, result) => systemLogger.debug(`completed(${result}) id=${job.id}`))
+ .on('failed', (job, err) => systemLogger.warn(`failed(${err}) id=${job.id}`, { job, e: renderError(err) }))
+ .on('error', (job: any, err: Error) => systemLogger.error(`error ${err}`, { job, e: renderError(err) }))
+ .on('stalled', (job) => systemLogger.warn(`stalled id=${job.id}`));
+
deliverQueue
.on('waiting', (jobId) => deliverLogger.debug(`waiting id=${jobId}`))
.on('active', (job) => deliverLogger.debug(`active ${getJobInfo(job, true)} to=${job.data.to}`))
@@ -220,12 +229,17 @@ export function createCleanRemoteFilesJob() {
}
export default function() {
- if (!envOption.onlyServer) {
- deliverQueue.process(config.deliverJobConcurrency || 128, processDeliver);
- inboxQueue.process(config.inboxJobConcurrency || 16, processInbox);
- processDb(dbQueue);
- procesObjectStorage(objectStorageQueue);
- }
+ if (envOption.onlyServer) return;
+
+ deliverQueue.process(config.deliverJobConcurrency || 128, processDeliver);
+ inboxQueue.process(config.inboxJobConcurrency || 16, processInbox);
+ processDb(dbQueue);
+ procesObjectStorage(objectStorageQueue);
+
+ systemQueue.add('resyncCharts', {
+ }, {
+ repeat: { cron: '0 0 * * *' }
+ });
}
export function destroy() {
diff --git a/src/queue/processors/system/index.ts b/src/queue/processors/system/index.ts
new file mode 100644
index 0000000000..52b7868105
--- /dev/null
+++ b/src/queue/processors/system/index.ts
@@ -0,0 +1,12 @@
+import * as Bull from 'bull';
+import { resyncCharts } from './resync-charts';
+
+const jobs = {
+ resyncCharts,
+} as Record<string, Bull.ProcessCallbackFunction<{}> | Bull.ProcessPromiseFunction<{}>>;
+
+export default function(dbQueue: Bull.Queue<{}>) {
+ for (const [k, v] of Object.entries(jobs)) {
+ dbQueue.process(k, v);
+ }
+}
diff --git a/src/queue/processors/system/resync-charts.ts b/src/queue/processors/system/resync-charts.ts
new file mode 100644
index 0000000000..b36b024cfb
--- /dev/null
+++ b/src/queue/processors/system/resync-charts.ts
@@ -0,0 +1,21 @@
+import * as Bull from 'bull';
+
+import { queueLogger } from '../../logger';
+import { driveChart, notesChart, usersChart } from '@/services/chart/index';
+
+const logger = queueLogger.createSubLogger('resync-charts');
+
+export default async function resyncCharts(job: Bull.Job<{}>, done: any): Promise<void> {
+ logger.info(`Resync charts...`);
+
+ // TODO: ユーザーごとのチャートも更新する
+ // TODO: インスタンスごとのチャートも更新する
+ await Promise.all([
+ driveChart.resync(),
+ notesChart.resync(),
+ usersChart.resync(),
+ ]);
+
+ logger.succ(`All charts successfully resynced.`);
+ done();
+}
diff --git a/src/queue/queues.ts b/src/queue/queues.ts
index d8c09ef86e..a66a7ca451 100644
--- a/src/queue/queues.ts
+++ b/src/queue/queues.ts
@@ -2,6 +2,7 @@ import config from '@/config/index';
import { initialize as initializeQueue } from './initialize';
import { DeliverJobData, InboxJobData, DbJobData, ObjectStorageJobData } from './types';
+export const systemQueue = initializeQueue<{}>('system');
export const deliverQueue = initializeQueue<DeliverJobData>('deliver', config.deliverJobPerSec || 128);
export const inboxQueue = initializeQueue<InboxJobData>('inbox', config.inboxJobPerSec || 16);
export const dbQueue = initializeQueue<DbJobData>('db');