server/lib/request/abstract-request-scheduler.ts

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