server/lib/request/request-scheduler.ts

   1 import * as Sequelize from 'sequelize'
   2
   3 import { database as db } from '../../initializers/database'
   4 import { AbstractRequestScheduler } from './abstract-request-scheduler'
   5 import { logger } from '../../helpers'
   6 import {
   7   REQUESTS_LIMIT_PODS,
   8   REQUESTS_LIMIT_PER_POD
   9 } from '../../initializers'
  10
  11 export type RequestSchedulerOptions = {
  12   type: string
  13   endpoint: string
  14   data: Object
  15   toIds: number[]
  16   transaction: Sequelize.Transaction
  17 }
  18
  19 class RequestScheduler extends AbstractRequestScheduler {
  20   constructor () {
  21     super()
  22
  23     // We limit the size of the requests
  24     this.limitPods = REQUESTS_LIMIT_PODS
  25     this.limitPerPod = REQUESTS_LIMIT_PER_POD
  26
  27     this.description = 'requests'
  28   }
  29
  30   getRequestModel () {
  31     return db.Request
  32   }
  33
  34   getRequestToPodModel () {
  35     return db.RequestToPod
  36   }
  37
  38   buildRequestObjects (requests: { [ toPodId: number ]: any }) {
  39     const requestsToMakeGrouped = {}
  40
  41     Object.keys(requests).forEach(toPodId => {
  42       requests[toPodId].forEach(data => {
  43         const request = data.request
  44         const pod = data.pod
  45         const hashKey = toPodId + request.endpoint
  46
  47         if (!requestsToMakeGrouped[hashKey]) {
  48           requestsToMakeGrouped[hashKey] = {
  49             toPod: pod,
  50             endpoint: request.endpoint,
  51             ids: [], // request ids, to delete them from the DB in the future
  52             datas: [] // requests data,
  53           }
  54         }
  55
  56         requestsToMakeGrouped[hashKey].ids.push(request.id)
  57         requestsToMakeGrouped[hashKey].datas.push(request.request)
  58       })
  59     })
  60
  61     return requestsToMakeGrouped
  62   }
  63
  64   createRequest ({ type, endpoint, data, toIds, transaction }: RequestSchedulerOptions, callback: (err: Error) => void) {
  65     // TODO: check the setPods works
  66     const podIds = []
  67
  68     // If there are no destination pods abort
  69     if (toIds.length === 0) return callback(null)
  70
  71     toIds.forEach(toPod => {
  72       podIds.push(toPod)
  73     })
  74
  75     const createQuery = {
  76       endpoint,
  77       request: {
  78         type: type,
  79         data: data
  80       }
  81     }
  82
  83     const dbRequestOptions: Sequelize.CreateOptions = {
  84       transaction
  85     }
  86
  87     return db.Request.create(createQuery, dbRequestOptions).asCallback((err, request) => {
  88       if (err) return callback(err)
  89
  90       return request.setPods(podIds, dbRequestOptions).asCallback(callback)
  91     })
  92   }
  93
  94   // ---------------------------------------------------------------------------
  95
  96   afterRequestsHook () {
  97     // Flush requests with no pod
  98     this.getRequestModel().removeWithEmptyTo(err => {
  99       if (err) logger.error('Error when removing requests with no pods.', { error: err })
 100     })
 101   }
 102 }
 103
 104 // ---------------------------------------------------------------------------
 105
 106 export {
 107   RequestScheduler
 108 }