]>
Commit | Line | Data |
---|---|---|
1 | import PQueue from 'p-queue' | |
2 | import { logger } from '@server/helpers/logger' | |
3 | import { SCHEDULER_INTERVALS_MS } from '@server/initializers/constants' | |
4 | import { MActorDefault, MActorSignature } from '@server/types/models' | |
5 | import { Activity } from '@shared/models' | |
6 | import { StatsManager } from '../stat-manager' | |
7 | import { processActivities } from './process' | |
8 | ||
9 | class InboxManager { | |
10 | ||
11 | private static instance: InboxManager | |
12 | private readonly inboxQueue: PQueue | |
13 | ||
14 | private constructor () { | |
15 | this.inboxQueue = new PQueue({ concurrency: 1 }) | |
16 | ||
17 | setInterval(() => { | |
18 | StatsManager.Instance.updateInboxWaiting(this.getActivityPubMessagesWaiting()) | |
19 | }, SCHEDULER_INTERVALS_MS.UPDATE_INBOX_STATS) | |
20 | } | |
21 | ||
22 | addInboxMessage (param: { | |
23 | activities: Activity[] | |
24 | signatureActor?: MActorSignature | |
25 | inboxActor?: MActorDefault | |
26 | }) { | |
27 | this.inboxQueue.add(() => { | |
28 | const options = { signatureActor: param.signatureActor, inboxActor: param.inboxActor } | |
29 | ||
30 | return processActivities(param.activities, options) | |
31 | }).catch(err => logger.error('Error with inbox queue.', { err })) | |
32 | } | |
33 | ||
34 | getActivityPubMessagesWaiting () { | |
35 | return this.inboxQueue.size + this.inboxQueue.pending | |
36 | } | |
37 | ||
38 | static get Instance () { | |
39 | return this.instance || (this.instance = new this()) | |
40 | } | |
41 | } | |
42 | ||
43 | // --------------------------------------------------------------------------- | |
44 | ||
45 | export { | |
46 | InboxManager | |
47 | } |