1 import { AsyncQueue, queue } from 'async'
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 { processActivities } from './process'
7 import { StatsManager } from '../stat-manager'
10 activities: Activity[]
11 signatureActor?: MActorSignature
12 inboxActor?: MActorDefault
17 private static instance: InboxManager
19 private readonly inboxQueue: AsyncQueue<QueueParam>
21 private messagesProcessed = 0
23 private constructor () {
24 this.inboxQueue = queue<QueueParam, Error>((task, cb) => {
25 const options = { signatureActor: task.signatureActor, inboxActor: task.inboxActor }
27 this.messagesProcessed++
29 processActivities(task.activities, options)
32 logger.error('Error in process activities.', { err })
38 StatsManager.Instance.updateInboxStats(this.messagesProcessed, this.getActivityPubMessagesWaiting())
39 }, SCHEDULER_INTERVALS_MS.updateInboxStats)
42 addInboxMessage (options: QueueParam) {
43 this.inboxQueue.push(options)
44 .catch(err => logger.error('Cannot add options in inbox queue.', { options, err }))
47 getActivityPubMessagesWaiting () {
48 return this.inboxQueue.length() + this.inboxQueue.running()
51 static get Instance () {
52 return this.instance || (this.instance = new this())
56 // ---------------------------------------------------------------------------