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