aboutsummaryrefslogtreecommitdiffhomepage
path: root/lib
diff options
context:
space:
mode:
Diffstat (limited to 'lib')
-rw-r--r--lib/poolRequests.js211
-rw-r--r--lib/webTorrentNode.js27
-rw-r--r--lib/webtorrent.js6
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})()