diff options
author | Chocobozzz <florian.bigard@gmail.com> | 2016-01-31 11:23:52 +0100 |
---|---|---|
committer | Chocobozzz <florian.bigard@gmail.com> | 2016-01-31 11:23:52 +0100 |
commit | c45f7f84001c2731909db04dd82e1c1f290386eb (patch) | |
tree | b7e57420a1f65dfbbacc1a532bf489c9bea6125d /lib/poolRequests.js | |
parent | cda021079ff455cc0fd0eb95a5395fa808ab63d1 (diff) | |
download | PeerTube-c45f7f84001c2731909db04dd82e1c1f290386eb.tar.gz PeerTube-c45f7f84001c2731909db04dd82e1c1f290386eb.tar.zst PeerTube-c45f7f84001c2731909db04dd82e1c1f290386eb.zip |
Infile code reorganization
Diffstat (limited to 'lib/poolRequests.js')
-rw-r--r-- | lib/poolRequests.js | 211 |
1 files changed, 109 insertions, 102 deletions
diff --git a/lib/poolRequests.js b/lib/poolRequests.js index 9c7f3238b..53f47d629 100644 --- a/lib/poolRequests.js +++ b/lib/poolRequests.js | |||
@@ -2,29 +2,114 @@ | |||
2 | 'use strict' | 2 | 'use strict' |
3 | 3 | ||
4 | var async = require('async') | 4 | var async = require('async') |
5 | var pluck = require('lodash-node/compat/collection/pluck') | ||
5 | 6 | ||
6 | var constants = require('../initializers/constants') | 7 | var constants = require('../initializers/constants') |
7 | var logger = require('../helpers/logger') | ||
8 | var database = require('../initializers/database') | 8 | var database = require('../initializers/database') |
9 | var pluck = require('lodash-node/compat/collection/pluck') | 9 | var logger = require('../helpers/logger') |
10 | var PoolRequestsDB = database.PoolRequestsDB | ||
11 | var PodsDB = database.PodsDB | 10 | var PodsDB = database.PodsDB |
11 | var PoolRequestsDB = database.PoolRequestsDB | ||
12 | var utils = require('../helpers/utils') | 12 | var utils = require('../helpers/utils') |
13 | var VideosDB = database.VideosDB | 13 | var VideosDB = database.VideosDB |
14 | 14 | ||
15 | var poolRequests = {} | ||
16 | |||
17 | // ----------- Private ----------- | ||
18 | var timer = null | 15 | var timer = null |
19 | 16 | ||
20 | function removePoolRequestsFromDB (ids) { | 17 | var poolRequests = { |
21 | PoolRequestsDB.remove({ _id: { $in: ids } }, function (err) { | 18 | activate: activate, |
22 | if (err) { | 19 | addToPoolRequests: addToPoolRequests, |
23 | logger.error('Cannot remove requests from the pool requests database.', { error: err }) | 20 | deactivate: deactivate, |
24 | return | 21 | forceSend: forceSend |
22 | } | ||
23 | |||
24 | function deactivate () { | ||
25 | logger.info('Pool requests deactivated.') | ||
26 | clearInterval(timer) | ||
27 | } | ||
28 | |||
29 | function forceSend () { | ||
30 | logger.info('Force pool requests sending.') | ||
31 | makePoolRequests() | ||
32 | } | ||
33 | |||
34 | function activate () { | ||
35 | logger.info('Pool requests activated.') | ||
36 | timer = setInterval(makePoolRequests, constants.INTERVAL) | ||
37 | } | ||
38 | |||
39 | function addToPoolRequests (id, type, request) { | ||
40 | logger.debug('Add request to the pool requests.', { id: id, type: type, request: request }) | ||
41 | |||
42 | PoolRequestsDB.findOne({ id: id }, function (err, entity) { | ||
43 | if (err) logger.error(err) | ||
44 | |||
45 | if (entity) { | ||
46 | if (entity.type === type) { | ||
47 | logger.error(new Error('Cannot insert two same requests.')) | ||
48 | return | ||
49 | } | ||
50 | |||
51 | // Remove the request of the other type | ||
52 | PoolRequestsDB.remove({ id: id }, function (err) { | ||
53 | if (err) logger.error(err) | ||
54 | }) | ||
55 | } else { | ||
56 | PoolRequestsDB.create({ id: id, type: type, request: request }, function (err) { | ||
57 | if (err) logger.error(err) | ||
58 | }) | ||
25 | } | 59 | } |
60 | }) | ||
61 | } | ||
26 | 62 | ||
27 | logger.info('Pool requests flushed.') | 63 | // --------------------------------------------------------------------------- |
64 | |||
65 | module.exports = poolRequests | ||
66 | |||
67 | // --------------------------------------------------------------------------- | ||
68 | |||
69 | function makePoolRequest (type, requests, callback) { | ||
70 | if (!callback) callback = function () {} | ||
71 | |||
72 | PodsDB.find({}, { _id: 1, url: 1, publicKey: 1 }).exec(function (err, pods) { | ||
73 | if (err) throw err | ||
74 | |||
75 | var params = { | ||
76 | encrypt: true, | ||
77 | sign: true, | ||
78 | method: 'POST', | ||
79 | path: null, | ||
80 | data: requests | ||
81 | } | ||
82 | |||
83 | if (type === 'add') { | ||
84 | params.path = '/api/' + constants.API_VERSION + '/remotevideos/add' | ||
85 | } else if (type === 'remove') { | ||
86 | params.path = '/api/' + constants.API_VERSION + '/remotevideos/remove' | ||
87 | } else { | ||
88 | throw new Error('Unkown pool request type.') | ||
89 | } | ||
90 | |||
91 | var bad_pods = [] | ||
92 | var good_pods = [] | ||
93 | |||
94 | utils.makeMultipleRetryRequest(params, pods, callbackEachPodFinished, callbackAllPodsFinished) | ||
95 | |||
96 | function callbackEachPodFinished (err, response, body, url, pod, callback_each_pod_finished) { | ||
97 | if (err || (response.statusCode !== 200 && response.statusCode !== 204)) { | ||
98 | bad_pods.push(pod._id) | ||
99 | logger.error('Error sending secure request to %s pod.', url, { error: err || new Error('Status code not 20x') }) | ||
100 | } else { | ||
101 | good_pods.push(pod._id) | ||
102 | } | ||
103 | |||
104 | return callback_each_pod_finished() | ||
105 | } | ||
106 | |||
107 | function callbackAllPodsFinished (err) { | ||
108 | if (err) return callback(err) | ||
109 | |||
110 | updatePodsScore(good_pods, bad_pods) | ||
111 | callback(null) | ||
112 | } | ||
28 | }) | 113 | }) |
29 | } | 114 | } |
30 | 115 | ||
@@ -81,16 +166,6 @@ | |||
81 | }) | 166 | }) |
82 | } | 167 | } |
83 | 168 | ||
84 | function updatePodsScore (good_pods, bad_pods) { | ||
85 | logger.info('Updating %d good pods and %d bad pods scores.', good_pods.length, bad_pods.length) | ||
86 | |||
87 | PodsDB.update({ _id: { $in: good_pods } }, { $inc: { score: constants.PODS_SCORE.BONUS } }, { multi: true }).exec() | ||
88 | PodsDB.update({ _id: { $in: bad_pods } }, { $inc: { score: constants.PODS_SCORE.MALUS } }, { multi: true }, function (err) { | ||
89 | if (err) throw err | ||
90 | removeBadPods() | ||
91 | }) | ||
92 | } | ||
93 | |||
94 | function removeBadPods () { | 169 | function removeBadPods () { |
95 | PodsDB.find({ score: 0 }, { _id: 1, url: 1 }, function (err, pods) { | 170 | PodsDB.find({ score: 0 }, { _id: 1, url: 1 }, function (err, pods) { |
96 | if (err) throw err | 171 | if (err) throw err |
@@ -115,92 +190,24 @@ | |||
115 | }) | 190 | }) |
116 | } | 191 | } |
117 | 192 | ||
118 | function makePoolRequest (type, requests, callback) { | 193 | function removePoolRequestsFromDB (ids) { |
119 | if (!callback) callback = function () {} | 194 | PoolRequestsDB.remove({ _id: { $in: ids } }, function (err) { |
120 | 195 | if (err) { | |
121 | PodsDB.find({}, { _id: 1, url: 1, publicKey: 1 }).exec(function (err, pods) { | 196 | logger.error('Cannot remove requests from the pool requests database.', { error: err }) |
122 | if (err) throw err | 197 | return |
123 | |||
124 | var params = { | ||
125 | encrypt: true, | ||
126 | sign: true, | ||
127 | method: 'POST', | ||
128 | path: null, | ||
129 | data: requests | ||
130 | } | ||
131 | |||
132 | if (type === 'add') { | ||
133 | params.path = '/api/' + constants.API_VERSION + '/remotevideos/add' | ||
134 | } else if (type === 'remove') { | ||
135 | params.path = '/api/' + constants.API_VERSION + '/remotevideos/remove' | ||
136 | } else { | ||
137 | throw new Error('Unkown pool request type.') | ||
138 | } | ||
139 | |||
140 | var bad_pods = [] | ||
141 | var good_pods = [] | ||
142 | |||
143 | utils.makeMultipleRetryRequest(params, pods, callbackEachPodFinished, callbackAllPodsFinished) | ||
144 | |||
145 | function callbackEachPodFinished (err, response, body, url, pod, callback_each_pod_finished) { | ||
146 | if (err || (response.statusCode !== 200 && response.statusCode !== 204)) { | ||
147 | bad_pods.push(pod._id) | ||
148 | logger.error('Error sending secure request to %s pod.', url, { error: err || new Error('Status code not 20x') }) | ||
149 | } else { | ||
150 | good_pods.push(pod._id) | ||
151 | } | ||
152 | |||
153 | return callback_each_pod_finished() | ||
154 | } | 198 | } |
155 | 199 | ||
156 | function callbackAllPodsFinished (err) { | 200 | logger.info('Pool requests flushed.') |
157 | if (err) return callback(err) | ||
158 | |||
159 | updatePodsScore(good_pods, bad_pods) | ||
160 | callback(null) | ||
161 | } | ||
162 | }) | 201 | }) |
163 | } | 202 | } |
164 | 203 | ||
165 | // ----------- Public ----------- | 204 | function updatePodsScore (good_pods, bad_pods) { |
166 | poolRequests.activate = function () { | 205 | logger.info('Updating %d good pods and %d bad pods scores.', good_pods.length, bad_pods.length) |
167 | logger.info('Pool requests activated.') | ||
168 | timer = setInterval(makePoolRequests, constants.INTERVAL) | ||
169 | } | ||
170 | |||
171 | poolRequests.addToPoolRequests = function (id, type, request) { | ||
172 | logger.debug('Add request to the pool requests.', { id: id, type: type, request: request }) | ||
173 | |||
174 | PoolRequestsDB.findOne({ id: id }, function (err, entity) { | ||
175 | if (err) logger.error(err) | ||
176 | |||
177 | if (entity) { | ||
178 | if (entity.type === type) { | ||
179 | logger.error(new Error('Cannot insert two same requests.')) | ||
180 | return | ||
181 | } | ||
182 | 206 | ||
183 | // Remove the request of the other type | 207 | PodsDB.update({ _id: { $in: good_pods } }, { $inc: { score: constants.PODS_SCORE.BONUS } }, { multi: true }).exec() |
184 | PoolRequestsDB.remove({ id: id }, function (err) { | 208 | PodsDB.update({ _id: { $in: bad_pods } }, { $inc: { score: constants.PODS_SCORE.MALUS } }, { multi: true }, function (err) { |
185 | if (err) logger.error(err) | 209 | if (err) throw err |
186 | }) | 210 | removeBadPods() |
187 | } else { | ||
188 | PoolRequestsDB.create({ id: id, type: type, request: request }, function (err) { | ||
189 | if (err) logger.error(err) | ||
190 | }) | ||
191 | } | ||
192 | }) | 211 | }) |
193 | } | 212 | } |
194 | |||
195 | poolRequests.deactivate = function () { | ||
196 | logger.info('Pool requests deactivated.') | ||
197 | clearInterval(timer) | ||
198 | } | ||
199 | |||
200 | poolRequests.forceSend = function () { | ||
201 | logger.info('Force pool requests sending.') | ||
202 | makePoolRequests() | ||
203 | } | ||
204 | |||
205 | module.exports = poolRequests | ||
206 | })() | 213 | })() |