2 files changed, 26 insertions, 30 deletions

diff --git a/server/lib/activitypub/inbox-manager.ts b/server/lib/activitypub/inbox-manager.ts
index f2785d6ce..27778cc9d 100644
--- a/server/lib/activitypub/inbox-manager.ts
+++ b/server/lib/activitypub/inbox-manager.ts
@@ -1,4 +1,4 @@
-import { queue, QueueObject } from 'async'
+import PQueue from 'p-queue'
 import { logger } from '@server/helpers/logger'
 import { SCHEDULER_INTERVALS_MS } from '@server/initializers/constants'
 import { MActorDefault, MActorSignature } from '@server/types/models'
@@ -6,42 +6,33 @@ import { Activity } from '@shared/models'
 import { StatsManager } from '../stat-manager'
 import { processActivities } from './process'
 
-type QueueParam = {
-  activities: Activity[]
-  signatureActor?: MActorSignature
-  inboxActor?: MActorDefault
-}
-
 class InboxManager {
 
   private static instance: InboxManager
-
-  private readonly inboxQueue: QueueObject<QueueParam>
+  private readonly inboxQueue: PQueue
 
   private constructor () {
-    this.inboxQueue = queue<QueueParam, Error>((task, cb) => {
-      const options = { signatureActor: task.signatureActor, inboxActor: task.inboxActor }
-
-      processActivities(task.activities, options)
-        .then(() => cb())
-        .catch(err => {
-          logger.error('Error in process activities.', { err })
-          cb()
-        })
-    })
+    this.inboxQueue = new PQueue({ concurrency: 1 })
 
     setInterval(() => {
       StatsManager.Instance.updateInboxWaiting(this.getActivityPubMessagesWaiting())
     }, SCHEDULER_INTERVALS_MS.UPDATE_INBOX_STATS)
   }
 
-  addInboxMessage (options: QueueParam) {
-    this.inboxQueue.push(options)
-      .catch(err => logger.error('Cannot add options in inbox queue.', { options, err }))
+  addInboxMessage (param: {
+    activities: Activity[]
+    signatureActor?: MActorSignature
+    inboxActor?: MActorDefault
+  }) {
+    this.inboxQueue.add(() => {
+      const options = { signatureActor: param.signatureActor, inboxActor: param.inboxActor }
+
+      return processActivities(param.activities, options)
+    }).catch(err => logger.error('Error with inbox queue.', { err }))
   }
 
   getActivityPubMessagesWaiting () {
-    return this.inboxQueue.length() + this.inboxQueue.running()
+    return this.inboxQueue.size + this.inboxQueue.pending
   }
 
   static get Instance () {
diff --git a/server/middlewares/async.ts b/server/middlewares/async.ts
index 9d0193536..7e131257d 100644
--- a/server/middlewares/async.ts
+++ b/server/middlewares/async.ts
@@ -1,21 +1,26 @@
-import { eachSeries } from 'async'
+import Bluebird from 'bluebird'
 import { NextFunction, Request, RequestHandler, Response } from 'express'
 import { ValidationChain } from 'express-validator'
 import { ExpressPromiseHandler } from '@server/types/express-handler'
 import { retryTransactionWrapper } from '../helpers/database-utils'
 
-// Syntactic sugar to avoid try/catch in express controllers
-// Thanks: https://medium.com/@Abazhenov/using-async-await-in-express-with-node-8-b8af872c0016
+// Syntactic sugar to avoid try/catch in express controllers/middlewares
 
 export type RequestPromiseHandler = ValidationChain | ExpressPromiseHandler
 
 function asyncMiddleware (fun: RequestPromiseHandler | RequestPromiseHandler[]) {
   return (req: Request, res: Response, next: NextFunction) => {
     if (Array.isArray(fun) === true) {
-      return eachSeries(fun as RequestHandler[], (f, cb) => {
-        Promise.resolve(f(req, res, err => cb(err)))
-          .catch(err => next(err))
-      }, next)
+      return Bluebird.each(fun as RequestPromiseHandler[], f => {
+        return new Promise<void>((resolve, reject) => {
+          return asyncMiddleware(f)(req, res, err => {
+            if (err) return reject(err)
+
+            return resolve()
+          })
+        })
+      }).then(() => next())
+        .catch(err => next(err))
     }
 
     return Promise.resolve((fun as RequestHandler)(req, res, next))