]>
Commit | Line | Data |
---|---|---|
7a9e420a | 1 | import PQueue from 'p-queue' |
99afa081 C |
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' | |
99afa081 | 6 | import { StatsManager } from '../stat-manager' |
543442a3 | 7 | import { processActivities } from './process' |
99afa081 | 8 | |
99afa081 C |
9 | class InboxManager { |
10 | ||
11 | private static instance: InboxManager | |
7a9e420a | 12 | private readonly inboxQueue: PQueue |
99afa081 C |
13 | |
14 | private constructor () { | |
7a9e420a | 15 | this.inboxQueue = new PQueue({ concurrency: 1 }) |
99afa081 C |
16 | |
17 | setInterval(() => { | |
543442a3 | 18 | StatsManager.Instance.updateInboxWaiting(this.getActivityPubMessagesWaiting()) |
61953742 | 19 | }, SCHEDULER_INTERVALS_MS.UPDATE_INBOX_STATS) |
99afa081 C |
20 | } |
21 | ||
7a9e420a C |
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 })) | |
99afa081 C |
32 | } |
33 | ||
fae6e4da | 34 | getActivityPubMessagesWaiting () { |
7a9e420a | 35 | return this.inboxQueue.size + this.inboxQueue.pending |
fae6e4da C |
36 | } |
37 | ||
99afa081 C |
38 | static get Instance () { |
39 | return this.instance || (this.instance = new this()) | |
40 | } | |
41 | } | |
42 | ||
43 | // --------------------------------------------------------------------------- | |
44 | ||
45 | export { | |
46 | InboxManager | |
47 | } |