'use strict'
var async = require('async')
+ var pluck = require('lodash-node/compat/collection/pluck')
var constants = require('../initializers/constants')
- var logger = require('../helpers/logger')
var database = require('../initializers/database')
- var pluck = require('lodash-node/compat/collection/pluck')
- var PoolRequestsDB = database.PoolRequestsDB
+ var logger = require('../helpers/logger')
var PodsDB = database.PodsDB
+ var PoolRequestsDB = database.PoolRequestsDB
var utils = require('../helpers/utils')
var VideosDB = database.VideosDB
- var poolRequests = {}
-
- // ----------- Private -----------
var timer = null
- function removePoolRequestsFromDB (ids) {
- PoolRequestsDB.remove({ _id: { $in: ids } }, function (err) {
- if (err) {
- logger.error('Cannot remove requests from the pool requests database.', { error: err })
- return
+ var poolRequests = {
+ activate: activate,
+ addToPoolRequests: addToPoolRequests,
+ deactivate: deactivate,
+ forceSend: forceSend
+ }
+
+ function deactivate () {
+ logger.info('Pool requests deactivated.')
+ clearInterval(timer)
+ }
+
+ function forceSend () {
+ logger.info('Force pool requests sending.')
+ makePoolRequests()
+ }
+
+ function activate () {
+ logger.info('Pool requests activated.')
+ timer = setInterval(makePoolRequests, constants.INTERVAL)
+ }
+
+ function addToPoolRequests (id, type, request) {
+ logger.debug('Add request to the pool requests.', { id: id, type: type, request: request })
+
+ PoolRequestsDB.findOne({ id: id }, function (err, entity) {
+ if (err) logger.error(err)
+
+ if (entity) {
+ if (entity.type === type) {
+ logger.error(new Error('Cannot insert two same requests.'))
+ return
+ }
+
+ // Remove the request of the other type
+ PoolRequestsDB.remove({ id: id }, function (err) {
+ if (err) logger.error(err)
+ })
+ } else {
+ PoolRequestsDB.create({ id: id, type: type, request: request }, function (err) {
+ if (err) logger.error(err)
+ })
}
+ })
+ }
- logger.info('Pool requests flushed.')
+ // ---------------------------------------------------------------------------
+
+ module.exports = poolRequests
+
+ // ---------------------------------------------------------------------------
+
+ function makePoolRequest (type, requests, callback) {
+ if (!callback) callback = function () {}
+
+ PodsDB.find({}, { _id: 1, url: 1, publicKey: 1 }).exec(function (err, pods) {
+ if (err) throw err
+
+ var params = {
+ encrypt: true,
+ sign: true,
+ method: 'POST',
+ path: null,
+ data: requests
+ }
+
+ if (type === 'add') {
+ params.path = '/api/' + constants.API_VERSION + '/remotevideos/add'
+ } else if (type === 'remove') {
+ params.path = '/api/' + constants.API_VERSION + '/remotevideos/remove'
+ } else {
+ throw new Error('Unkown pool request type.')
+ }
+
+ var bad_pods = []
+ var good_pods = []
+
+ utils.makeMultipleRetryRequest(params, pods, callbackEachPodFinished, callbackAllPodsFinished)
+
+ function callbackEachPodFinished (err, response, body, url, pod, callback_each_pod_finished) {
+ if (err || (response.statusCode !== 200 && response.statusCode !== 204)) {
+ bad_pods.push(pod._id)
+ logger.error('Error sending secure request to %s pod.', url, { error: err || new Error('Status code not 20x') })
+ } else {
+ good_pods.push(pod._id)
+ }
+
+ return callback_each_pod_finished()
+ }
+
+ function callbackAllPodsFinished (err) {
+ if (err) return callback(err)
+
+ updatePodsScore(good_pods, bad_pods)
+ callback(null)
+ }
})
}
})
}
- function updatePodsScore (good_pods, bad_pods) {
- logger.info('Updating %d good pods and %d bad pods scores.', good_pods.length, bad_pods.length)
-
- PodsDB.update({ _id: { $in: good_pods } }, { $inc: { score: constants.PODS_SCORE.BONUS } }, { multi: true }).exec()
- PodsDB.update({ _id: { $in: bad_pods } }, { $inc: { score: constants.PODS_SCORE.MALUS } }, { multi: true }, function (err) {
- if (err) throw err
- removeBadPods()
- })
- }
-
function removeBadPods () {
PodsDB.find({ score: 0 }, { _id: 1, url: 1 }, function (err, pods) {
if (err) throw err
})
}
- function makePoolRequest (type, requests, callback) {
- if (!callback) callback = function () {}
-
- PodsDB.find({}, { _id: 1, url: 1, publicKey: 1 }).exec(function (err, pods) {
- if (err) throw err
-
- var params = {
- encrypt: true,
- sign: true,
- method: 'POST',
- path: null,
- data: requests
- }
-
- if (type === 'add') {
- params.path = '/api/' + constants.API_VERSION + '/remotevideos/add'
- } else if (type === 'remove') {
- params.path = '/api/' + constants.API_VERSION + '/remotevideos/remove'
- } else {
- throw new Error('Unkown pool request type.')
- }
-
- var bad_pods = []
- var good_pods = []
-
- utils.makeMultipleRetryRequest(params, pods, callbackEachPodFinished, callbackAllPodsFinished)
-
- function callbackEachPodFinished (err, response, body, url, pod, callback_each_pod_finished) {
- if (err || (response.statusCode !== 200 && response.statusCode !== 204)) {
- bad_pods.push(pod._id)
- logger.error('Error sending secure request to %s pod.', url, { error: err || new Error('Status code not 20x') })
- } else {
- good_pods.push(pod._id)
- }
-
- return callback_each_pod_finished()
+ function removePoolRequestsFromDB (ids) {
+ PoolRequestsDB.remove({ _id: { $in: ids } }, function (err) {
+ if (err) {
+ logger.error('Cannot remove requests from the pool requests database.', { error: err })
+ return
}
- function callbackAllPodsFinished (err) {
- if (err) return callback(err)
-
- updatePodsScore(good_pods, bad_pods)
- callback(null)
- }
+ logger.info('Pool requests flushed.')
})
}
- // ----------- Public -----------
- poolRequests.activate = function () {
- logger.info('Pool requests activated.')
- timer = setInterval(makePoolRequests, constants.INTERVAL)
- }
-
- poolRequests.addToPoolRequests = function (id, type, request) {
- logger.debug('Add request to the pool requests.', { id: id, type: type, request: request })
-
- PoolRequestsDB.findOne({ id: id }, function (err, entity) {
- if (err) logger.error(err)
-
- if (entity) {
- if (entity.type === type) {
- logger.error(new Error('Cannot insert two same requests.'))
- return
- }
+ function updatePodsScore (good_pods, bad_pods) {
+ logger.info('Updating %d good pods and %d bad pods scores.', good_pods.length, bad_pods.length)
- // Remove the request of the other type
- PoolRequestsDB.remove({ id: id }, function (err) {
- if (err) logger.error(err)
- })
- } else {
- PoolRequestsDB.create({ id: id, type: type, request: request }, function (err) {
- if (err) logger.error(err)
- })
- }
+ PodsDB.update({ _id: { $in: good_pods } }, { $inc: { score: constants.PODS_SCORE.BONUS } }, { multi: true }).exec()
+ PodsDB.update({ _id: { $in: bad_pods } }, { $inc: { score: constants.PODS_SCORE.MALUS } }, { multi: true }, function (err) {
+ if (err) throw err
+ removeBadPods()
})
}
-
- poolRequests.deactivate = function () {
- logger.info('Pool requests deactivated.')
- clearInterval(timer)
- }
-
- poolRequests.forceSend = function () {
- logger.info('Force pool requests sending.')
- makePoolRequests()
- }
-
- module.exports = poolRequests
})()
projects / github / Chocobozzz / PeerTube.git / blobdiff