server/lib/base-request-scheduler.js

   1 'use strict'
   2
   3 const eachLimit = require('async/eachLimit')
   4
   5 const constants = require('../initializers/constants')
   6 const db = require('../initializers/database')
   7 const logger = require('../helpers/logger')
   8 const requests = require('../helpers/requests')
   9
  10 module.exports = class BaseRequestScheduler {
  11
  12   constructor (options) {
  13     this.lastRequestTimestamp = 0
  14     this.timer = null
  15     this.requestInterval = constants.REQUESTS_INTERVAL
  16   }
  17
  18   activate () {
  19     logger.info('Requests scheduler activated.')
  20     this.lastRequestTimestamp = Date.now()
  21
  22     this.timer = setInterval(() => {
  23       this.lastRequestTimestamp = Date.now()
  24       this.makeRequests()
  25     }, this.requestInterval)
  26   }
  27
  28   deactivate () {
  29     logger.info('Requests scheduler deactivated.')
  30     clearInterval(this.timer)
  31     this.timer = null
  32   }
  33
  34   forceSend () {
  35     logger.info('Force requests scheduler sending.')
  36     this.makeRequests()
  37   }
  38
  39   remainingMilliSeconds () {
  40     if (this.timer === null) return -1
  41
  42     return constants.REQUESTS_INTERVAL - (Date.now() - this.lastRequestTimestamp)
  43   }
  44
  45   remainingRequestsCount (callback) {
  46     return this.getRequestModel().countTotalRequests(callback)
  47   }
  48
  49   // ---------------------------------------------------------------------------
  50
  51   // Make a requests to friends of a certain type
  52   makeRequest (toPod, requestEndpoint, requestsToMake, callback) {
  53     if (!callback) callback = function () {}
  54
  55     const params = {
  56       toPod: toPod,
  57       sign: true, // Prove our identity
  58       method: 'POST',
  59       path: '/api/' + constants.API_VERSION + '/remote/' + requestEndpoint,
  60       data: requestsToMake // Requests we need to make
  61     }
  62
  63     // Make multiple retry requests to all of pods
  64     // The function fire some useful callbacks
  65     requests.makeSecureRequest(params, (err, res) => {
  66       if (err || (res.statusCode !== 200 && res.statusCode !== 201 && res.statusCode !== 204)) {
  67         err = err ? err.message : 'Status code not 20x : ' + res.statusCode
  68         logger.error('Error sending secure request to %s pod.', toPod.host, { error: err })
  69
  70         return callback(false)
  71       }
  72
  73       return callback(true)
  74     })
  75   }
  76
  77     // Make all the requests of the scheduler
  78   makeRequests () {
  79     this.getRequestModel().listWithLimitAndRandom(this.limitPods, this.limitPerPod, (err, requests) => {
  80       if (err) {
  81         logger.error('Cannot get the list of "%s".', this.description, { err: err })
  82         return // Abort
  83       }
  84
  85       // If there are no requests, abort
  86       if (requests.length === 0) {
  87         logger.info('No "%s" to make.', this.description)
  88         return
  89       }
  90
  91       // We want to group requests by destinations pod and endpoint
  92       const requestsToMakeGrouped = this.buildRequestObjects(requests)
  93
  94       logger.info('Making "%s" to friends.', this.description)
  95
  96       const goodPods = []
  97       const badPods = []
  98
  99       eachLimit(Object.keys(requestsToMakeGrouped), constants.REQUESTS_IN_PARALLEL, (hashKey, callbackEach) => {
 100         const requestToMake = requestsToMakeGrouped[hashKey]
 101         const toPod = requestToMake.toPod
 102
 103         this.makeRequest(toPod, requestToMake.endpoint, requestToMake.datas, (success) => {
 104           if (success === false) {
 105             badPods.push(requestToMake.toPod.id)
 106             return callbackEach()
 107           }
 108
 109           logger.debug('Removing requests for pod %s.', requestToMake.toPod.id, { requestsIds: requestToMake.ids })
 110           goodPods.push(requestToMake.toPod.id)
 111
 112           // Remove the pod id of these request ids
 113           this.getRequestToPodModel().removeByRequestIdsAndPod(requestToMake.ids, requestToMake.toPod.id, callbackEach)
 114
 115           this.afterRequestHook()
 116         })
 117       }, () => {
 118         // All the requests were made, we update the pods score
 119         db.Pod.updatePodsScore(goodPods, badPods)
 120
 121         this.afterRequestsHook()
 122       })
 123     })
 124   }
 125
 126   flush (callback) {
 127     this.getRequestModel().removeAll(callback)
 128   }
 129
 130   afterRequestHook () {
 131    // Nothing to do, let children reimplement it
 132   }
 133
 134   afterRequestsHook () {
 135    // Nothing to do, let children reimplement it
 136   }
 137 }