diff options
Diffstat (limited to 'lib')
-rw-r--r-- | lib/poolRequests.js | 211 | ||||
-rw-r--r-- | lib/webTorrentNode.js | 27 | ||||
-rw-r--r-- | lib/webtorrent.js | 6 |
3 files changed, 128 insertions, 116 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 | })() |
diff --git a/lib/webTorrentNode.js b/lib/webTorrentNode.js index 8827c68c5..69fa6b012 100644 --- a/lib/webTorrentNode.js +++ b/lib/webTorrentNode.js | |||
@@ -10,22 +10,21 @@ | |||
10 | 10 | ||
11 | var host = config.get('webserver.host') | 11 | var host = config.get('webserver.host') |
12 | var port = config.get('webserver.port') | 12 | var port = config.get('webserver.port') |
13 | |||
14 | var nodeKey = 'webtorrentnode' + port | 13 | var nodeKey = 'webtorrentnode' + port |
15 | var processKey = 'webtorrent' + port | 14 | var processKey = 'webtorrent' + port |
16 | |||
17 | ipc.config.silent = true | 15 | ipc.config.silent = true |
18 | ipc.config.id = nodeKey | 16 | ipc.config.id = nodeKey |
19 | 17 | ||
20 | var webtorrentnode = {} | 18 | var webtorrentnode = { |
21 | 19 | add: add, | |
22 | // Useful for beautiful tests | 20 | app: null, // Pid of the app |
23 | webtorrentnode.silent = false | 21 | create: create, |
24 | 22 | remove: remove, | |
25 | // Useful to kill it | 23 | seed: seed, |
26 | webtorrentnode.app = null | 24 | silent: false // Useful for beautiful tests |
25 | } | ||
27 | 26 | ||
28 | webtorrentnode.create = function (options, callback) { | 27 | function create (options, callback) { |
29 | if (typeof options === 'function') { | 28 | if (typeof options === 'function') { |
30 | callback = options | 29 | callback = options |
31 | options = {} | 30 | options = {} |
@@ -75,7 +74,7 @@ | |||
75 | ipc.server.start() | 74 | ipc.server.start() |
76 | } | 75 | } |
77 | 76 | ||
78 | webtorrentnode.seed = function (path, callback) { | 77 | function seed (path, callback) { |
79 | var extension = pathUtils.extname(path) | 78 | var extension = pathUtils.extname(path) |
80 | var basename = pathUtils.basename(path, extension) | 79 | var basename = pathUtils.basename(path, extension) |
81 | var data = { | 80 | var data = { |
@@ -104,7 +103,7 @@ | |||
104 | ipc.server.broadcast(processKey + '.seed', data) | 103 | ipc.server.broadcast(processKey + '.seed', data) |
105 | } | 104 | } |
106 | 105 | ||
107 | webtorrentnode.add = function (magnetUri, callback) { | 106 | function add (magnetUri, callback) { |
108 | var data = { | 107 | var data = { |
109 | _id: magnetUri, | 108 | _id: magnetUri, |
110 | args: { | 109 | args: { |
@@ -131,7 +130,7 @@ | |||
131 | ipc.server.broadcast(processKey + '.add', data) | 130 | ipc.server.broadcast(processKey + '.add', data) |
132 | } | 131 | } |
133 | 132 | ||
134 | webtorrentnode.remove = function (magnetUri, callback) { | 133 | function remove (magnetUri, callback) { |
135 | var data = { | 134 | var data = { |
136 | _id: magnetUri, | 135 | _id: magnetUri, |
137 | args: { | 136 | args: { |
@@ -156,5 +155,7 @@ | |||
156 | ipc.server.broadcast(processKey + '.remove', data) | 155 | ipc.server.broadcast(processKey + '.remove', data) |
157 | } | 156 | } |
158 | 157 | ||
158 | // --------------------------------------------------------------------------- | ||
159 | |||
159 | module.exports = webtorrentnode | 160 | module.exports = webtorrentnode |
160 | })() | 161 | })() |
diff --git a/lib/webtorrent.js b/lib/webtorrent.js index b72bc500d..41e60499f 100644 --- a/lib/webtorrent.js +++ b/lib/webtorrent.js | |||
@@ -1,7 +1,7 @@ | |||
1 | ;(function () { | 1 | ;(function () { |
2 | 'use strict' | 2 | 'use strict' |
3 | 3 | ||
4 | module.exports = function (args) { | 4 | function webtorrent (args) { |
5 | var WebTorrent = require('webtorrent') | 5 | var WebTorrent = require('webtorrent') |
6 | var ipc = require('node-ipc') | 6 | var ipc = require('node-ipc') |
7 | 7 | ||
@@ -88,4 +88,8 @@ | |||
88 | ipc.of[nodeKey].emit(processKey + '.exception', { exception: e }) | 88 | ipc.of[nodeKey].emit(processKey + '.exception', { exception: e }) |
89 | }) | 89 | }) |
90 | } | 90 | } |
91 | |||
92 | // --------------------------------------------------------------------------- | ||
93 | |||
94 | module.exports = webtorrent | ||
91 | })() | 95 | })() |