4 files changed, 88 insertions, 94 deletions

diff --git a/server/lib/request/abstract-request-scheduler.ts b/server/lib/request/abstract-request-scheduler.ts
index e81ab9c36..dd77fddb7 100644
--- a/server/lib/request/abstract-request-scheduler.ts
+++ b/server/lib/request/abstract-request-scheduler.ts
@@ -1,15 +1,16 @@
-import * as eachLimit from 'async/eachLimit'
+import { isEmpty } from 'lodash'
+import * as Promise from 'bluebird'
 
 import { database as db } from '../../initializers/database'
 import { logger, makeSecureRequest } from '../../helpers'
-import { PodInstance } from '../../models'
+import { AbstractRequestClass, AbstractRequestToPodClass, PodInstance } from '../../models'
 import {
   API_VERSION,
   REQUESTS_IN_PARALLEL,
   REQUESTS_INTERVAL
 } from '../../initializers'
 
-abstract class AbstractRequestScheduler {
+abstract class AbstractRequestScheduler <T> {
   requestInterval: number
   limitPods: number
   limitPerPod: number
@@ -24,9 +25,9 @@ abstract class AbstractRequestScheduler {
     this.requestInterval = REQUESTS_INTERVAL
   }
 
-  abstract getRequestModel ()
-  abstract getRequestToPodModel ()
-  abstract buildRequestObjects (requests: any)
+  abstract getRequestModel (): AbstractRequestClass<T>
+  abstract getRequestToPodModel (): AbstractRequestToPodClass
+  abstract buildRequestObjects (requestsGrouped: T): {}
 
   activate () {
     logger.info('Requests scheduler activated.')
@@ -55,20 +56,18 @@ abstract class AbstractRequestScheduler {
     return REQUESTS_INTERVAL - (Date.now() - this.lastRequestTimestamp)
   }
 
-  remainingRequestsCount (callback: (err: Error, total: number) => void) {
-    return this.getRequestModel().countTotalRequests(callback)
+  remainingRequestsCount () {
+    return this.getRequestModel().countTotalRequests()
   }
 
-  flush (callback: (err: Error) => void) {
-    this.getRequestModel().removeAll(callback)
+  flush () {
+    return this.getRequestModel().removeAll()
   }
 
   // ---------------------------------------------------------------------------
 
   // Make a requests to friends of a certain type
-  protected makeRequest (toPod: PodInstance, requestEndpoint: string, requestsToMake: Object, callback) {
-    if (!callback) callback = function () { /* empty */ }
-
+  protected makeRequest (toPod: PodInstance, requestEndpoint: string, requestsToMake: Object) {
     const params = {
       toPod: toPod,
       sign: true, // Prove our identity
@@ -79,65 +78,64 @@ abstract class AbstractRequestScheduler {
 
     // Make multiple retry requests to all of pods
     // The function fire some useful callbacks
-    makeSecureRequest(params, (err, res) => {
-      if (err || (res.statusCode !== 200 && res.statusCode !== 201 && res.statusCode !== 204)) {
-        err = err ? err.message : 'Status code not 20x : ' + res.statusCode
+    return makeSecureRequest(params)
+      .then(({ response, body }) => {
+        if (response.statusCode !== 200 && response.statusCode !== 201 && response.statusCode !== 204) {
+          throw new Error('Status code not 20x : ' + response.statusCode)
+        }
+      })
+      .catch(err => {
         logger.error('Error sending secure request to %s pod.', toPod.host, { error: err })
 
-        return callback(err)
-      }
-
-      return callback(null)
-    })
+        throw err
+      })
   }
 
     // Make all the requests of the scheduler
   protected makeRequests () {
-    this.getRequestModel().listWithLimitAndRandom(this.limitPods, this.limitPerPod, (err, requests) => {
-      if (err) {
-        logger.error('Cannot get the list of "%s".', this.description, { err: err })
-        return // Abort
-      }
-
-      // If there are no requests, abort
-      if (requests.length === 0) {
-        logger.info('No "%s" to make.', this.description)
-        return
-      }
-
-      // We want to group requests by destinations pod and endpoint
-      const requestsToMakeGrouped = this.buildRequestObjects(requests)
-
-      logger.info('Making "%s" to friends.', this.description)
-
-      const goodPods = []
-      const badPods = []
-
-      eachLimit(Object.keys(requestsToMakeGrouped), REQUESTS_IN_PARALLEL, (hashKey, callbackEach) => {
-        const requestToMake = requestsToMakeGrouped[hashKey]
-        const toPod = requestToMake.toPod
-
-        this.makeRequest(toPod, requestToMake.endpoint, requestToMake.datas, (err) => {
-          if (err) {
-            badPods.push(requestToMake.toPod.id)
-            return callbackEach()
-          }
-
-          logger.debug('Removing requests for pod %s.', requestToMake.toPod.id, { requestsIds: requestToMake.ids })
-          goodPods.push(requestToMake.toPod.id)
-
-          // Remove the pod id of these request ids
-          this.getRequestToPodModel().removeByRequestIdsAndPod(requestToMake.ids, requestToMake.toPod.id, callbackEach)
+    return this.getRequestModel().listWithLimitAndRandom(this.limitPods, this.limitPerPod)
+      .then((requestsGrouped: T) => {
+        // We want to group requests by destinations pod and endpoint
+        const requestsToMake = this.buildRequestObjects(requestsGrouped)
+
+        // If there are no requests, abort
+        if (isEmpty(requestsToMake) === true) {
+          logger.info('No "%s" to make.', this.description)
+          return { goodPods: [], badPods: [] }
+        }
+
+        logger.info('Making "%s" to friends.', this.description)
+
+        const goodPods = []
+        const badPods = []
+
+        return Promise.map(Object.keys(requestsToMake), hashKey => {
+          const requestToMake = requestsToMake[hashKey]
+          const toPod: PodInstance = requestToMake.toPod
+
+          return this.makeRequest(toPod, requestToMake.endpoint, requestToMake.datas)
+            .then(() => {
+              logger.debug('Removing requests for pod %s.', requestToMake.toPod.id, { requestsIds: requestToMake.ids })
+              goodPods.push(requestToMake.toPod.id)
+
+              this.afterRequestHook()
+
+              // Remove the pod id of these request ids
+              return this.getRequestToPodModel().removeByRequestIdsAndPod(requestToMake.ids, requestToMake.toPod.id)
+            })
+            .catch(err => {
+              badPods.push(requestToMake.toPod.id)
+              logger.info('Cannot make request to %s.', toPod.host, { error: err })
+            })
+        }, { concurrency: REQUESTS_IN_PARALLEL }).then(() => ({ goodPods, badPods }))
+      })
+      .then(({ goodPods, badPods }) => {
+        this.afterRequestsHook()
 
-          this.afterRequestHook()
-        })
-      }, () => {
         // All the requests were made, we update the pods score
-        db.Pod.updatePodsScore(goodPods, badPods)
-
-        this.afterRequestsHook()
+        return db.Pod.updatePodsScore(goodPods, badPods)
       })
-    })
+      .catch(err => logger.error('Cannot get the list of "%s".', this.description, { error: err.stack }))
   }
 
   protected afterRequestHook () {
diff --git a/server/lib/request/request-scheduler.ts b/server/lib/request/request-scheduler.ts
index 575e0227c..0dd796fb0 100644
--- a/server/lib/request/request-scheduler.ts
+++ b/server/lib/request/request-scheduler.ts
@@ -3,10 +3,8 @@ import * as Sequelize from 'sequelize'
 import { database as db } from '../../initializers/database'
 import { AbstractRequestScheduler } from './abstract-request-scheduler'
 import { logger } from '../../helpers'
-import {
-  REQUESTS_LIMIT_PODS,
-  REQUESTS_LIMIT_PER_POD
-} from '../../initializers'
+import { REQUESTS_LIMIT_PODS, REQUESTS_LIMIT_PER_POD } from '../../initializers'
+import { RequestsGrouped } from '../../models'
 import { RequestEndpoint } from '../../../shared'
 
 export type RequestSchedulerOptions = {
@@ -17,7 +15,7 @@ export type RequestSchedulerOptions = {
   transaction: Sequelize.Transaction
 }
 
-class RequestScheduler extends AbstractRequestScheduler {
+class RequestScheduler extends AbstractRequestScheduler<RequestsGrouped> {
   constructor () {
     super()
 
@@ -36,11 +34,11 @@ class RequestScheduler extends AbstractRequestScheduler {
     return db.RequestToPod
   }
 
-  buildRequestObjects (requests: { [ toPodId: number ]: any }) {
+  buildRequestObjects (requestsGrouped: RequestsGrouped) {
     const requestsToMakeGrouped = {}
 
-    Object.keys(requests).forEach(toPodId => {
-      requests[toPodId].forEach(data => {
+    Object.keys(requestsGrouped).forEach(toPodId => {
+      requestsGrouped[toPodId].forEach(data => {
         const request = data.request
         const pod = data.pod
         const hashKey = toPodId + request.endpoint
@@ -62,12 +60,12 @@ class RequestScheduler extends AbstractRequestScheduler {
     return requestsToMakeGrouped
   }
 
-  createRequest ({ type, endpoint, data, toIds, transaction }: RequestSchedulerOptions, callback: (err: Error) => void) {
+  createRequest ({ type, endpoint, data, toIds, transaction }: RequestSchedulerOptions) {
     // TODO: check the setPods works
     const podIds = []
 
     // If there are no destination pods abort
-    if (toIds.length === 0) return callback(null)
+    if (toIds.length === 0) return undefined
 
     toIds.forEach(toPod => {
       podIds.push(toPod)
@@ -85,20 +83,18 @@ class RequestScheduler extends AbstractRequestScheduler {
       transaction
     }
 
-    return db.Request.create(createQuery, dbRequestOptions).asCallback((err, request) => {
-      if (err) return callback(err)
-
-      return request.setPods(podIds, dbRequestOptions).asCallback(callback)
-    })
+    return db.Request.create(createQuery, dbRequestOptions)
+      .then(request => {
+        return request.setPods(podIds, dbRequestOptions)
+      })
   }
 
   // ---------------------------------------------------------------------------
 
   afterRequestsHook () {
     // Flush requests with no pod
-    this.getRequestModel().removeWithEmptyTo(err => {
-      if (err) logger.error('Error when removing requests with no pods.', { error: err })
-    })
+    this.getRequestModel().removeWithEmptyTo()
+      .catch(err => logger.error('Error when removing requests with no pods.', { error: err }))
   }
 }
 
diff --git a/server/lib/request/request-video-event-scheduler.ts b/server/lib/request/request-video-event-scheduler.ts
index 4bb76f4c9..d4d714c02 100644
--- a/server/lib/request/request-video-event-scheduler.ts
+++ b/server/lib/request/request-video-event-scheduler.ts
@@ -7,6 +7,7 @@ import {
   REQUESTS_VIDEO_EVENT_LIMIT_PER_POD,
   REQUEST_VIDEO_EVENT_ENDPOINT
 } from '../../initializers'
+import { RequestsVideoEventGrouped } from '../../models'
 import { RequestVideoEventType } from '../../../shared'
 
 export type RequestVideoEventSchedulerOptions = {
@@ -16,7 +17,7 @@ export type RequestVideoEventSchedulerOptions = {
   transaction?: Sequelize.Transaction
 }
 
-class RequestVideoEventScheduler extends AbstractRequestScheduler {
+class RequestVideoEventScheduler extends AbstractRequestScheduler<RequestsVideoEventGrouped> {
   constructor () {
     super()
 
@@ -35,7 +36,7 @@ class RequestVideoEventScheduler extends AbstractRequestScheduler {
     return db.RequestVideoEvent
   }
 
-  buildRequestObjects (eventsToProcess: { [ toPodId: number ]: any }[]) {
+  buildRequestObjects (eventRequests: RequestsVideoEventGrouped) {
     const requestsToMakeGrouped = {}
 
     /* Example:
@@ -50,8 +51,8 @@ class RequestVideoEventScheduler extends AbstractRequestScheduler {
 
     // We group video events per video and per pod
     // We add the counts of the same event types
-    Object.keys(eventsToProcess).forEach(toPodId => {
-      eventsToProcess[toPodId].forEach(eventToProcess => {
+    Object.keys(eventRequests).forEach(toPodId => {
+      eventRequests[toPodId].forEach(eventToProcess => {
         if (!eventsPerVideoPerPod[toPodId]) eventsPerVideoPerPod[toPodId] = {}
 
         if (!requestsToMakeGrouped[toPodId]) {
@@ -97,7 +98,7 @@ class RequestVideoEventScheduler extends AbstractRequestScheduler {
     return requestsToMakeGrouped
   }
 
-  createRequest ({ type, videoId, count, transaction }: RequestVideoEventSchedulerOptions, callback: (err: Error) => void) {
+  createRequest ({ type, videoId, count, transaction }: RequestVideoEventSchedulerOptions) {
     if (count === undefined) count = 1
 
     const dbRequestOptions: Sequelize.CreateOptions = {}
@@ -109,7 +110,7 @@ class RequestVideoEventScheduler extends AbstractRequestScheduler {
       videoId
     }
 
-    return db.RequestVideoEvent.create(createQuery, dbRequestOptions).asCallback(callback)
+    return db.RequestVideoEvent.create(createQuery, dbRequestOptions)
   }
 }
 
diff --git a/server/lib/request/request-video-qadu-scheduler.ts b/server/lib/request/request-video-qadu-scheduler.ts
index d7169cc81..5ec7de9c2 100644
--- a/server/lib/request/request-video-qadu-scheduler.ts
+++ b/server/lib/request/request-video-qadu-scheduler.ts
@@ -9,6 +9,7 @@ import {
   REQUEST_VIDEO_QADU_ENDPOINT,
   REQUEST_VIDEO_QADU_TYPES
 } from '../../initializers'
+import { RequestsVideoQaduGrouped } from '../../models'
 import { RequestVideoQaduType } from '../../../shared'
 
 export type RequestVideoQaduSchedulerOptions = {
@@ -17,7 +18,7 @@ export type RequestVideoQaduSchedulerOptions = {
   transaction?: Sequelize.Transaction
 }
 
-class RequestVideoQaduScheduler extends AbstractRequestScheduler {
+class RequestVideoQaduScheduler extends AbstractRequestScheduler<RequestsVideoQaduGrouped> {
   constructor () {
     super()
 
@@ -36,7 +37,7 @@ class RequestVideoQaduScheduler extends AbstractRequestScheduler {
     return db.RequestVideoQadu
   }
 
-  buildRequestObjects (requests: { [ toPodId: number ]: any }[]) {
+  buildRequestObjects (requests: RequestsVideoQaduGrouped) {
     const requestsToMakeGrouped = {}
 
     Object.keys(requests).forEach(toPodId => {
@@ -105,20 +106,18 @@ class RequestVideoQaduScheduler extends AbstractRequestScheduler {
     return requestsToMakeGrouped
   }
 
-  createRequest ({ type, videoId, transaction }: RequestVideoQaduSchedulerOptions, callback: (err: Error) => void) {
+  createRequest ({ type, videoId, transaction }: RequestVideoQaduSchedulerOptions) {
     const dbRequestOptions: Sequelize.BulkCreateOptions = {}
     if (transaction) dbRequestOptions.transaction = transaction
 
     // Send the update to all our friends
-    db.Pod.listAllIds(transaction, function (err, podIds) {
-      if (err) return callback(err)
-
+    return db.Pod.listAllIds(transaction).then(podIds => {
       const queries = []
       podIds.forEach(podId => {
         queries.push({ type, videoId, podId })
       })
 
-      return db.RequestVideoQadu.bulkCreate(queries, dbRequestOptions).asCallback(callback)
+      return db.RequestVideoQadu.bulkCreate(queries, dbRequestOptions)
     })
   }
 }