src/poolRequests.js

   1 ;(function () {
   2   'use strict'
   3
   4   var async = require('async')
   5
   6   var logger = require('./logger')
   7   var database = require('./database')
   8   var PoolRequestsDB = database.PoolRequestsDB
   9   var PodsDB = database.PodsDB
  10   var utils = require('./utils')
  11
  12   var poolRequests = {}
  13
  14   // ----------- Constants -----------
  15
  16   // Time to wait between requests to the friends
  17   var INTERVAL = utils.isTestInstance() ? 10000 : 60000
  18   var PODS_SCORE = {
  19     MALUS: -10,
  20     BONUS: 10
  21   }
  22
  23   // ----------- Private -----------
  24   var timer = null
  25
  26   function makePoolRequests () {
  27     logger.info('Making pool requests to friends.')
  28
  29     PoolRequestsDB.find({}, { type: 1, request: 1 }, function (err, pool_requests) {
  30       if (err) throw err
  31
  32       if (pool_requests.length === 0) return
  33
  34       var requests = {
  35         add: [],
  36         remove: []
  37       }
  38
  39       async.each(pool_requests, function (pool_request, callback_each) {
  40         if (pool_request.type === 'add') {
  41           requests.add.push(pool_request.request)
  42         } else if (pool_request.type === 'remove') {
  43           requests.remove.push(pool_request.request)
  44         } else {
  45           throw new Error('Unkown pool request type.')
  46         }
  47
  48         callback_each()
  49       }, function () {
  50         makePoolRequest('add', requests.add)
  51         makePoolRequest('remove', requests.remove)
  52         logger.info('Pool requests to friends sent.')
  53       })
  54     })
  55   }
  56
  57   function updatePodsScore (good_pods, bad_pods) {
  58     logger.info('Updating %d good pods and %d bad pods scores.', good_pods.length, bad_pods.length)
  59
  60     PodsDB.update({ _id: { $in: good_pods } }, { $inc: { score: PODS_SCORE.BONUS } }, { multi: true }).exec()
  61     PodsDB.update({ _id: { $in: bad_pods } }, { $inc: { score: PODS_SCORE.MALUS } }, { multi: true }, function (err) {
  62       if (err) throw err
  63       removeBadPods()
  64     })
  65   }
  66
  67   function removeBadPods () {
  68     PodsDB.remove({ score: 0 }, function (err, result) {
  69       if (err) throw err
  70
  71       var number_removed = result.result.n
  72       if (number_removed !== 0) logger.info('Removed %d pod.', number_removed)
  73     })
  74   }
  75
  76   function makePoolRequest (type, requests) {
  77     logger.debug('Make pool requests scheduled.')
  78     PodsDB.find({}, { _id: 1, url: 1, publicKey: 1 }).exec(function (err, pods) {
  79       if (err) throw err
  80
  81       var params = {
  82         encrypt: true,
  83         sign: true,
  84         method: 'POST',
  85         path: null,
  86         data: requests
  87       }
  88
  89       if (type === 'add') {
  90         params.path = '/api/' + global.API_VERSION + '/remotevideos/add'
  91       } else if (type === 'remove') {
  92         params.path = '/api/' + global.API_VERSION + '/remotevideos/remove'
  93       } else {
  94         throw new Error('Unkown pool request type.')
  95       }
  96
  97       var bad_pods = []
  98       var good_pods = []
  99
 100       utils.makeMultipleRetryRequest(params, pods, callbackEachPodFinished, callbackAllPodsFinished)
 101
 102       function callbackEachPodFinished (err, response, body, url, pod, callback_each_pod_finished) {
 103         if (err || response.statusCode !== 200) {
 104           bad_pods.push(pod._id)
 105           logger.error('Error sending secure request to %s pod.', url, { error: err })
 106         } else {
 107           good_pods.push(pod._id)
 108         }
 109
 110         return callback_each_pod_finished()
 111       }
 112
 113       function callbackAllPodsFinished (err) {
 114         if (err) {
 115           logger.error('There was some errors when sending the video meta data.', { error: err })
 116         }
 117
 118         updatePodsScore(good_pods, bad_pods)
 119         PoolRequestsDB.remove().exec()
 120       }
 121     })
 122   }
 123
 124   // ----------- Public -----------
 125   poolRequests.activate = function () {
 126     logger.info('Pool requests activated.')
 127     timer = setInterval(makePoolRequests, INTERVAL)
 128   }
 129
 130   poolRequests.addToPoolRequests = function (id, type, request) {
 131     logger.debug('Add request to the pool requests.', { id: id, type: type, request: request })
 132
 133     PoolRequestsDB.findOne({ id: id }, function (err, entity) {
 134       if (err) logger.error(err)
 135
 136       if (entity) {
 137         if (entity.type === type) {
 138           logger.error(new Error('Cannot insert two same requests.'))
 139           return
 140         }
 141
 142         // Remove the request of the other type
 143         PoolRequestsDB.remove({ id: id }, function (err) {
 144           if (err) logger.error(err)
 145         })
 146       } else {
 147         PoolRequestsDB.create({ id: id, type: type, request: request }, function (err) {
 148           if (err) logger.error(err)
 149         })
 150       }
 151     })
 152   }
 153
 154   poolRequests.deactivate = function () {
 155     logger.info('Pool requests deactivated.')
 156     clearInterval(timer)
 157   }
 158
 159   module.exports = poolRequests
 160 })()