Merge
This commit is contained in:
commit
e6234297a1
20 changed files with 305 additions and 257 deletions
|
|
@ -12,39 +12,65 @@ import * as ChannelStore from '../channel-storage/channelstore';
|
|||
import { ChannelStateSizeError } from '../errors';
|
||||
import Promise from 'bluebird';
|
||||
|
||||
/**
|
||||
* Previously, async channel functions were riddled with race conditions due to
|
||||
* an event causing the channel to be unloaded while a pending callback still
|
||||
* needed to reference it.
|
||||
*
|
||||
* This solution should be better than constantly checking whether the channel
|
||||
* has been unloaded in nested callbacks. The channel won't be unloaded until
|
||||
* nothing needs it anymore. Conceptually similar to a reference count.
|
||||
*/
|
||||
function ActiveLock(channel) {
|
||||
this.channel = channel;
|
||||
this.count = 0;
|
||||
}
|
||||
class ReferenceCounter {
|
||||
constructor(channel) {
|
||||
this.channel = channel;
|
||||
this.channelName = channel.name;
|
||||
this.refCount = 0;
|
||||
this.references = {};
|
||||
}
|
||||
|
||||
ActiveLock.prototype = {
|
||||
lock: function () {
|
||||
this.count++;
|
||||
},
|
||||
ref(caller) {
|
||||
if (caller) {
|
||||
if (this.references.hasOwnProperty(caller)) {
|
||||
this.references[caller]++;
|
||||
} else {
|
||||
this.references[caller] = 1;
|
||||
}
|
||||
}
|
||||
|
||||
release: function () {
|
||||
this.count--;
|
||||
if (this.count === 0) {
|
||||
/* sanity check */
|
||||
if (this.channel.users.length > 0) {
|
||||
Logger.errlog.log("Warning: ActiveLock count=0 but users.length > 0 (" +
|
||||
"channel: " + this.channel.name + ")");
|
||||
this.count = this.channel.users.length;
|
||||
this.refCount++;
|
||||
}
|
||||
|
||||
unref(caller) {
|
||||
if (caller) {
|
||||
if (this.references.hasOwnProperty(caller)) {
|
||||
this.references[caller]--;
|
||||
if (this.references[caller] === 0) {
|
||||
delete this.references[caller];
|
||||
}
|
||||
} else {
|
||||
Logger.errlog.log("ReferenceCounter::unref() called by caller [" +
|
||||
caller + "] but this caller had no active references! " +
|
||||
`(channel: ${this.channelName})`);
|
||||
}
|
||||
}
|
||||
|
||||
this.refCount--;
|
||||
this.checkRefCount();
|
||||
}
|
||||
|
||||
checkRefCount() {
|
||||
if (this.refCount === 0) {
|
||||
if (Object.keys(this.references).length > 0) {
|
||||
Logger.errlog.log("ReferenceCounter::refCount reached 0 but still had " +
|
||||
"active references: " +
|
||||
JSON.stringify(Object.keys(this.references)) +
|
||||
` (channel: ${this.channelName})`);
|
||||
for (var caller in this.references) {
|
||||
this.refCount += this.references[caller];
|
||||
}
|
||||
} else if (this.channel.users.length > 0) {
|
||||
Logger.errlog.log("ReferenceCounter::refCount reached 0 but still had " +
|
||||
this.channel.users.length + " active users" +
|
||||
` (channel: ${this.channelName})`);
|
||||
this.refCount = this.channel.users.length;
|
||||
} else {
|
||||
this.channel.emit("empty");
|
||||
}
|
||||
}
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
function Channel(name) {
|
||||
MakeEmitter(this);
|
||||
|
|
@ -54,7 +80,7 @@ function Channel(name) {
|
|||
this.logger = new Logger.Logger(path.join(__dirname, "..", "..", "chanlogs",
|
||||
this.uniqueName + ".log"));
|
||||
this.users = [];
|
||||
this.activeLock = new ActiveLock(this);
|
||||
this.refCounter = new ReferenceCounter(this);
|
||||
this.flags = 0;
|
||||
var self = this;
|
||||
db.channels.load(this, function (err) {
|
||||
|
|
@ -238,15 +264,16 @@ Channel.prototype.saveState = function () {
|
|||
};
|
||||
|
||||
Channel.prototype.checkModules = function (fn, args, cb) {
|
||||
var self = this;
|
||||
const self = this;
|
||||
const refCaller = `Channel::checkModules/${fn}`;
|
||||
this.waitFlag(Flags.C_READY, function () {
|
||||
self.activeLock.lock();
|
||||
self.refCounter.ref(refCaller);
|
||||
var keys = Object.keys(self.modules);
|
||||
var next = function (err, result) {
|
||||
if (result !== ChannelModule.PASSTHROUGH) {
|
||||
/* Either an error occured, or the module denied the user access */
|
||||
cb(err, result);
|
||||
self.activeLock.release();
|
||||
self.refCounter.unref(refCaller);
|
||||
return;
|
||||
}
|
||||
|
||||
|
|
@ -254,7 +281,7 @@ Channel.prototype.checkModules = function (fn, args, cb) {
|
|||
if (m === undefined) {
|
||||
/* No more modules to check */
|
||||
cb(null, ChannelModule.PASSTHROUGH);
|
||||
self.activeLock.release();
|
||||
self.refCounter.unref(refCaller);
|
||||
return;
|
||||
}
|
||||
|
||||
|
|
@ -278,13 +305,13 @@ Channel.prototype.notifyModules = function (fn, args) {
|
|||
};
|
||||
|
||||
Channel.prototype.joinUser = function (user, data) {
|
||||
var self = this;
|
||||
const self = this;
|
||||
|
||||
self.activeLock.lock();
|
||||
self.refCounter.ref("Channel::user");
|
||||
self.waitFlag(Flags.C_READY, function () {
|
||||
/* User closed the connection before the channel finished loading */
|
||||
if (user.socket.disconnected) {
|
||||
self.activeLock.release();
|
||||
self.refCounter.unref("Channel::user");
|
||||
return;
|
||||
}
|
||||
|
||||
|
|
@ -293,7 +320,7 @@ Channel.prototype.joinUser = function (user, data) {
|
|||
if (err) {
|
||||
Logger.errlog.log("user.refreshAccount failed at Channel.joinUser");
|
||||
Logger.errlog.log(err.stack);
|
||||
self.activeLock.release();
|
||||
self.refCounter.unref("Channel::user");
|
||||
return;
|
||||
}
|
||||
|
||||
|
|
@ -304,8 +331,10 @@ Channel.prototype.joinUser = function (user, data) {
|
|||
}
|
||||
|
||||
function afterAccount() {
|
||||
if (self.dead || user.socket.disconnected) {
|
||||
if (self.activeLock) self.activeLock.release();
|
||||
if (user.socket.disconnected) {
|
||||
self.refCounter.unref("Channel::user");
|
||||
return;
|
||||
} else if (self.dead) {
|
||||
return;
|
||||
}
|
||||
|
||||
|
|
@ -318,9 +347,7 @@ Channel.prototype.joinUser = function (user, data) {
|
|||
} else {
|
||||
user.account.channelRank = 0;
|
||||
user.account.effectiveRank = user.account.globalRank;
|
||||
if (self.activeLock) {
|
||||
self.activeLock.release();
|
||||
}
|
||||
self.refCounter.unref("Channel::user");
|
||||
}
|
||||
});
|
||||
}
|
||||
|
|
@ -408,7 +435,7 @@ Channel.prototype.partUser = function (user) {
|
|||
});
|
||||
this.sendUsercount(this.users);
|
||||
|
||||
this.activeLock.release();
|
||||
this.refCounter.unref("Channel::user");
|
||||
user.die();
|
||||
};
|
||||
|
||||
|
|
@ -555,20 +582,20 @@ Channel.prototype.sendUserJoin = function (users, user) {
|
|||
};
|
||||
|
||||
Channel.prototype.readLog = function (cb) {
|
||||
var maxLen = 102400;
|
||||
var file = this.logger.filename;
|
||||
this.activeLock.lock();
|
||||
var self = this;
|
||||
const maxLen = 102400;
|
||||
const file = this.logger.filename;
|
||||
this.refCounter.ref("Channel::readLog");
|
||||
const self = this;
|
||||
fs.stat(file, function (err, data) {
|
||||
if (err) {
|
||||
self.activeLock.release();
|
||||
self.refCounter.unref("Channel::readLog");
|
||||
return cb(err, null);
|
||||
}
|
||||
|
||||
var start = Math.max(data.size - maxLen, 0);
|
||||
var end = data.size - 1;
|
||||
const start = Math.max(data.size - maxLen, 0);
|
||||
const end = data.size - 1;
|
||||
|
||||
var read = fs.createReadStream(file, {
|
||||
const read = fs.createReadStream(file, {
|
||||
start: start,
|
||||
end: end
|
||||
});
|
||||
|
|
@ -579,7 +606,7 @@ Channel.prototype.readLog = function (cb) {
|
|||
});
|
||||
read.on("end", function () {
|
||||
cb(null, buffer);
|
||||
self.activeLock.release();
|
||||
self.refCounter.unref("Channel::readLog");
|
||||
});
|
||||
});
|
||||
};
|
||||
|
|
@ -648,7 +675,7 @@ Channel.prototype.packInfo = function (isAdmin) {
|
|||
}
|
||||
|
||||
if (isAdmin) {
|
||||
data.activeLockCount = this.activeLock.count;
|
||||
data.activeLockCount = this.refCounter.refCount;
|
||||
}
|
||||
|
||||
var self = this;
|
||||
|
|
|
|||
|
|
@ -73,9 +73,10 @@ KickBanModule.prototype.onUserPostJoin = function (user) {
|
|||
return;
|
||||
}
|
||||
|
||||
var chan = this.channel;
|
||||
const chan = this.channel;
|
||||
const refCaller = "KickBanModule::onUserPostJoin";
|
||||
user.waitFlag(Flags.U_LOGGED_IN, function () {
|
||||
chan.activeLock.lock();
|
||||
chan.refCounter.ref(refCaller);
|
||||
db.channels.isNameBanned(chan.name, user.getName(), function (err, banned) {
|
||||
if (!err && banned) {
|
||||
user.kick("You are banned from this channel.");
|
||||
|
|
@ -84,7 +85,7 @@ KickBanModule.prototype.onUserPostJoin = function (user) {
|
|||
"name is banned)");
|
||||
}
|
||||
}
|
||||
chan.activeLock.release();
|
||||
chan.refCounter.unref(refCaller);
|
||||
});
|
||||
});
|
||||
|
||||
|
|
@ -222,10 +223,10 @@ KickBanModule.prototype.handleCmdBan = function (user, msg, meta) {
|
|||
var name = args.shift().toLowerCase();
|
||||
var reason = args.join(" ");
|
||||
|
||||
var chan = this.channel;
|
||||
chan.activeLock.lock();
|
||||
const chan = this.channel;
|
||||
chan.refCounter.ref("KickBanModule::handleCmdBan");
|
||||
this.banName(user, name, reason, function (err) {
|
||||
chan.activeLock.release();
|
||||
chan.refCounter.unref("KickBanModule::handleCmdBan");
|
||||
});
|
||||
};
|
||||
|
||||
|
|
@ -249,10 +250,10 @@ KickBanModule.prototype.handleCmdIPBan = function (user, msg, meta) {
|
|||
}
|
||||
var reason = args.join(" ");
|
||||
|
||||
var chan = this.channel;
|
||||
chan.activeLock.lock();
|
||||
const chan = this.channel;
|
||||
chan.refCounter.ref("KickBanModule::handleCmdIPBan");
|
||||
this.banAll(user, name, range, reason, function (err) {
|
||||
chan.activeLock.release();
|
||||
chan.refCounter.unref("KickBanModule::handleCmdIPBan");
|
||||
});
|
||||
};
|
||||
|
||||
|
|
@ -416,9 +417,10 @@ KickBanModule.prototype.handleUnban = function (user, data) {
|
|||
}
|
||||
|
||||
var self = this;
|
||||
this.channel.activeLock.lock();
|
||||
this.channel.refCounter.ref("KickBanModule::handleUnban");
|
||||
db.channels.unbanId(this.channel.name, data.id, function (err) {
|
||||
if (err) {
|
||||
self.channel.refCounter.unref("KickBanModule::handleUnban");
|
||||
return user.socket.emit("errorMsg", {
|
||||
msg: err
|
||||
});
|
||||
|
|
@ -431,7 +433,7 @@ KickBanModule.prototype.handleUnban = function (user, data) {
|
|||
self.channel.modules.chat.sendModMessage(user.getName() + " unbanned " +
|
||||
data.name, banperm);
|
||||
}
|
||||
self.channel.activeLock.release();
|
||||
self.channel.refCounter.unref("KickBanModule::handleUnban");
|
||||
});
|
||||
};
|
||||
|
||||
|
|
|
|||
|
|
@ -51,16 +51,19 @@ LibraryModule.prototype.handleUncache = function (user, data) {
|
|||
return;
|
||||
}
|
||||
|
||||
var chan = this.channel;
|
||||
chan.activeLock.lock();
|
||||
const chan = this.channel;
|
||||
chan.refCounter.ref("LibraryModule::handleUncache");
|
||||
db.channels.deleteFromLibrary(chan.name, data.id, function (err, res) {
|
||||
if (chan.dead || err) {
|
||||
if (chan.dead) {
|
||||
return;
|
||||
} else if (err) {
|
||||
chan.refCounter.unref("LibraryModule::handleUncache");
|
||||
return;
|
||||
}
|
||||
|
||||
chan.logger.log("[library] " + user.getName() + " deleted " + data.id +
|
||||
"from the library");
|
||||
chan.activeLock.release();
|
||||
chan.refCounter.unref("LibraryModule::handleUncache");
|
||||
});
|
||||
};
|
||||
|
||||
|
|
|
|||
|
|
@ -44,6 +44,15 @@ MediaRefresherModule.prototype.onPreMediaChange = function (data, cb) {
|
|||
}
|
||||
};
|
||||
|
||||
MediaRefresherModule.prototype.unload = function () {
|
||||
try {
|
||||
clearInterval(this._interval);
|
||||
this._interval = null;
|
||||
} catch (error) {
|
||||
Logger.errlog.log(error.stack);
|
||||
}
|
||||
};
|
||||
|
||||
MediaRefresherModule.prototype.initGoogleDocs = function (data, cb) {
|
||||
var self = this;
|
||||
self.refreshGoogleDocs(data, cb);
|
||||
|
|
@ -63,23 +72,26 @@ MediaRefresherModule.prototype.initVimeo = function (data, cb) {
|
|||
return;
|
||||
}
|
||||
|
||||
var self = this;
|
||||
self.channel.activeLock.lock();
|
||||
const self = this;
|
||||
self.channel.refCounter.ref("MediaRefresherModule::initVimeo");
|
||||
Vimeo.extract(data.id).then(function (direct) {
|
||||
if (self.dead || self.channel.dead)
|
||||
if (self.dead || self.channel.dead) {
|
||||
self.unload();
|
||||
return;
|
||||
}
|
||||
|
||||
if (self._media === data) {
|
||||
data.meta.direct = direct;
|
||||
self.channel.logger.log("[mediarefresher] Refreshed vimeo video with ID " +
|
||||
data.id);
|
||||
}
|
||||
self.channel.activeLock.release();
|
||||
|
||||
if (cb) cb();
|
||||
}).catch(function (err) {
|
||||
Logger.errlog.log("Unexpected vimeo::extract() fail: " + err.stack);
|
||||
if (cb) cb();
|
||||
}).finally(() => {
|
||||
self.channel.refCounter.unref("MediaRefresherModule::initVimeo");
|
||||
});
|
||||
};
|
||||
|
||||
|
|
@ -87,10 +99,11 @@ MediaRefresherModule.prototype.refreshGoogleDocs = function (media, cb) {
|
|||
var self = this;
|
||||
|
||||
if (self.dead || self.channel.dead) {
|
||||
self.unload();
|
||||
return;
|
||||
}
|
||||
|
||||
self.channel.activeLock.lock();
|
||||
self.channel.refCounter.ref("MediaRefresherModule::refreshGoogleDocs");
|
||||
InfoGetter.getMedia(media.id, "gd", function (err, data) {
|
||||
if (self.dead || self.channel.dead) {
|
||||
return;
|
||||
|
|
@ -108,7 +121,7 @@ MediaRefresherModule.prototype.refreshGoogleDocs = function (media, cb) {
|
|||
self.channel.logger.log("[mediarefresher] Google Docs refresh failed " +
|
||||
"(likely redirect to login page-- make sure it is shared " +
|
||||
"correctly)");
|
||||
self.channel.activeLock.release();
|
||||
self.channel.refCounter.unref("MediaRefresherModule::refreshGoogleDocs");
|
||||
if (cb) cb();
|
||||
return;
|
||||
case "Access Denied":
|
||||
|
|
@ -119,7 +132,7 @@ MediaRefresherModule.prototype.refreshGoogleDocs = function (media, cb) {
|
|||
case "Google Drive videos must be shared publicly":
|
||||
self.channel.logger.log("[mediarefresher] Google Docs refresh failed: " +
|
||||
err);
|
||||
self.channel.activeLock.release();
|
||||
self.channel.refCounter.unref("MediaRefresherModule::refreshGoogleDocs");
|
||||
if (cb) cb();
|
||||
return;
|
||||
default:
|
||||
|
|
@ -128,14 +141,14 @@ MediaRefresherModule.prototype.refreshGoogleDocs = function (media, cb) {
|
|||
err);
|
||||
Logger.errlog.log("Google Docs refresh failed for ID " + media.id +
|
||||
": " + err);
|
||||
self.channel.activeLock.release();
|
||||
self.channel.refCounter.unref("MediaRefresherModule::refreshGoogleDocs");
|
||||
if (cb) cb();
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
if (media !== self._media) {
|
||||
self.channel.activeLock.release();
|
||||
self.channel.refCounter.unref("MediaRefresherModule::refreshGoogleDocs");
|
||||
if (cb) cb();
|
||||
return;
|
||||
}
|
||||
|
|
@ -143,7 +156,7 @@ MediaRefresherModule.prototype.refreshGoogleDocs = function (media, cb) {
|
|||
self.channel.logger.log("[mediarefresher] Refreshed Google Docs video with ID " +
|
||||
media.id);
|
||||
media.meta = data.meta;
|
||||
self.channel.activeLock.release();
|
||||
self.channel.refCounter.unref("MediaRefresherModule::refreshGoogleDocs");
|
||||
if (cb) cb();
|
||||
});
|
||||
};
|
||||
|
|
@ -152,10 +165,11 @@ MediaRefresherModule.prototype.initGooglePlus = function (media, cb) {
|
|||
var self = this;
|
||||
|
||||
if (self.dead || self.channel.dead) {
|
||||
self.unload();
|
||||
return;
|
||||
}
|
||||
|
||||
self.channel.activeLock.lock();
|
||||
self.channel.refCounter.ref("MediaRefresherModule::initGooglePlus");
|
||||
InfoGetter.getMedia(media.id, "gp", function (err, data) {
|
||||
if (self.dead || self.channel.dead) {
|
||||
return;
|
||||
|
|
@ -177,7 +191,7 @@ MediaRefresherModule.prototype.initGooglePlus = function (media, cb) {
|
|||
"and is shared publicly"):
|
||||
self.channel.logger.log("[mediarefresher] Google+ refresh failed: " +
|
||||
err);
|
||||
self.channel.activeLock.release();
|
||||
self.channel.refCounter.unref("MediaRefresherModule::initGooglePlus");
|
||||
if (cb) cb();
|
||||
return;
|
||||
default:
|
||||
|
|
@ -186,14 +200,14 @@ MediaRefresherModule.prototype.initGooglePlus = function (media, cb) {
|
|||
err);
|
||||
Logger.errlog.log("Google+ refresh failed for ID " + media.id +
|
||||
": " + err);
|
||||
self.channel.activeLock.release();
|
||||
self.channel.refCounter.unref("MediaRefresherModule::initGooglePlus");
|
||||
if (cb) cb();
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
if (media !== self._media) {
|
||||
self.channel.activeLock.release();
|
||||
self.channel.refCounter.unref("MediaRefresherModule::initGooglePlus");
|
||||
if (cb) cb();
|
||||
return;
|
||||
}
|
||||
|
|
@ -201,7 +215,7 @@ MediaRefresherModule.prototype.initGooglePlus = function (media, cb) {
|
|||
self.channel.logger.log("[mediarefresher] Refreshed Google+ video with ID " +
|
||||
media.id);
|
||||
media.meta = data.meta;
|
||||
self.channel.activeLock.release();
|
||||
self.channel.refCounter.unref("MediaRefresherModule::initGooglePlus");
|
||||
if (cb) cb();
|
||||
});
|
||||
};
|
||||
|
|
|
|||
|
|
@ -11,6 +11,13 @@ var CustomEmbedFilter = require("../customembed").filter;
|
|||
var XSS = require("../xss");
|
||||
|
||||
const MAX_ITEMS = Config.get("playlist.max-items");
|
||||
// Limit requestPlaylist to once per 60 seconds
|
||||
const REQ_PLAYLIST_THROTTLE = {
|
||||
burst: 1,
|
||||
sustained: 0,
|
||||
cooldown: 60
|
||||
};
|
||||
|
||||
|
||||
const TYPE_QUEUE = {
|
||||
id: "string,boolean",
|
||||
|
|
@ -216,9 +223,7 @@ PlaylistModule.prototype.onUserPostJoin = function (user) {
|
|||
user.socket.on("playerReady", function () {
|
||||
self.sendChangeMedia([user]);
|
||||
});
|
||||
user.socket.on("requestPlaylist", function () {
|
||||
self.sendPlaylist([user]);
|
||||
});
|
||||
user.socket.on("requestPlaylist", this.handleRequestPlaylist.bind(this, user));
|
||||
user.on("login", function () {
|
||||
self.sendPlaylist([user]);
|
||||
});
|
||||
|
|
@ -447,15 +452,15 @@ PlaylistModule.prototype.queueStandard = function (user, data) {
|
|||
});
|
||||
};
|
||||
|
||||
var self = this;
|
||||
this.channel.activeLock.lock();
|
||||
const self = this;
|
||||
this.channel.refCounter.ref("PlaylistModule::queueStandard");
|
||||
this.semaphore.queue(function (lock) {
|
||||
var lib = self.channel.modules.library;
|
||||
if (lib && self.channel.is(Flags.C_REGISTERED) && !util.isLive(data.type)) {
|
||||
lib.getItem(data.id, function (err, item) {
|
||||
if (err && err !== "Item not in library") {
|
||||
error(err+"");
|
||||
self.channel.activeLock.release();
|
||||
self.channel.refCounter.unref("PlaylistModule::queueStandard");
|
||||
return lock.release();
|
||||
}
|
||||
|
||||
|
|
@ -464,7 +469,7 @@ PlaylistModule.prototype.queueStandard = function (user, data) {
|
|||
data.shouldAddToLibrary = false;
|
||||
self._addItem(item, data, user, function () {
|
||||
lock.release();
|
||||
self.channel.activeLock.release();
|
||||
self.channel.refCounter.unref("PlaylistModule::queueStandard");
|
||||
});
|
||||
} else {
|
||||
handleLookup();
|
||||
|
|
@ -479,25 +484,13 @@ PlaylistModule.prototype.queueStandard = function (user, data) {
|
|||
InfoGetter.getMedia(data.id, data.type, function (err, media) {
|
||||
if (err) {
|
||||
error(XSS.sanitizeText(String(err)));
|
||||
if (self.channel && self.channel.activeLock) {
|
||||
self.channel.activeLock.release();
|
||||
} else {
|
||||
Logger.errlog.log("Attempted release of channel lock after " +
|
||||
"channel was already unloaded in queueStandard: " +
|
||||
channelName + " " + data.type + ":" + data.id);
|
||||
}
|
||||
self.channel.refCounter.unref("PlaylistModule::queueStandard");
|
||||
return lock.release();
|
||||
}
|
||||
|
||||
self._addItem(media, data, user, function () {
|
||||
lock.release();
|
||||
if (self.channel && self.channel.activeLock) {
|
||||
self.channel.activeLock.release();
|
||||
} else {
|
||||
Logger.errlog.log("Attempted release of channel lock after " +
|
||||
"channel was already unloaded in queueStandard: " +
|
||||
channelName + " " + data.type + ":" + data.id);
|
||||
}
|
||||
self.channel.refCounter.unref("PlaylistModule::queueStandard");
|
||||
});
|
||||
});
|
||||
}
|
||||
|
|
@ -536,12 +529,12 @@ PlaylistModule.prototype.queueYouTubePlaylist = function (user, data) {
|
|||
}
|
||||
}
|
||||
|
||||
self.channel.activeLock.lock();
|
||||
self.channel.refCounter.ref("PlaylistModule::queueYouTubePlaylist");
|
||||
vids.forEach(function (media) {
|
||||
data.link = util.formatLink(media.id, media.type);
|
||||
self._addItem(media, data, user);
|
||||
});
|
||||
self.channel.activeLock.release();
|
||||
self.channel.refCounter.unref("PlaylistModule::queueYouTubePlaylist");
|
||||
|
||||
lock.release();
|
||||
});
|
||||
|
|
@ -560,7 +553,7 @@ PlaylistModule.prototype.handleDelete = function (user, data) {
|
|||
}
|
||||
|
||||
var plitem = this.items.find(data);
|
||||
self.channel.activeLock.lock();
|
||||
self.channel.refCounter.ref("PlaylistModule::handleDelete");
|
||||
this.semaphore.queue(function (lock) {
|
||||
if (self._delete(data)) {
|
||||
self.channel.logger.log("[playlist] " + user.getName() + " deleted " +
|
||||
|
|
@ -568,7 +561,7 @@ PlaylistModule.prototype.handleDelete = function (user, data) {
|
|||
}
|
||||
|
||||
lock.release();
|
||||
self.channel.activeLock.release();
|
||||
self.channel.refCounter.unref("PlaylistModule::handleDelete");
|
||||
});
|
||||
};
|
||||
|
||||
|
|
@ -602,27 +595,27 @@ PlaylistModule.prototype.handleMoveMedia = function (user, data) {
|
|||
return;
|
||||
}
|
||||
|
||||
var self = this;
|
||||
self.channel.activeLock.lock();
|
||||
const self = this;
|
||||
self.channel.refCounter.ref("PlaylistModule::handleMoveMedia");
|
||||
self.semaphore.queue(function (lock) {
|
||||
if (!self.items.remove(data.from)) {
|
||||
self.channel.activeLock.release();
|
||||
self.channel.refCounter.unref("PlaylistModule::handleMoveMedia");
|
||||
return lock.release();
|
||||
}
|
||||
|
||||
if (data.after === "prepend") {
|
||||
if (!self.items.prepend(from)) {
|
||||
self.channel.activeLock.release();
|
||||
self.channel.refCounter.unref("PlaylistModule::handleMoveMedia");
|
||||
return lock.release();
|
||||
}
|
||||
} else if (data.after === "append") {
|
||||
if (!self.items.append(from)) {
|
||||
self.channel.activeLock.release();
|
||||
self.channel.refCounter.unref("PlaylistModule::handleMoveMedia");
|
||||
return lock.release();
|
||||
}
|
||||
} else {
|
||||
if (!self.items.insertAfter(from, data.after)) {
|
||||
self.channel.activeLock.release();
|
||||
self.channel.refCounter.unref("PlaylistModule::handleMoveMedia");
|
||||
return lock.release();
|
||||
}
|
||||
}
|
||||
|
|
@ -633,7 +626,7 @@ PlaylistModule.prototype.handleMoveMedia = function (user, data) {
|
|||
from.media.title +
|
||||
(after ? " after " + after.media.title : ""));
|
||||
lock.release();
|
||||
self.channel.activeLock.release();
|
||||
self.channel.refCounter.unref("PlaylistModule::handleMoveMedia");
|
||||
});
|
||||
};
|
||||
|
||||
|
|
@ -1128,55 +1121,6 @@ PlaylistModule.prototype._leadLoop = function() {
|
|||
}
|
||||
};
|
||||
|
||||
PlaylistModule.prototype.refreshGoogleDocs = function (cb) {
|
||||
var self = this;
|
||||
|
||||
if (self.dead || !self.channel || self.channel.dead) {
|
||||
return;
|
||||
}
|
||||
|
||||
var abort = function () {
|
||||
if (self.current) {
|
||||
self.current.media.meta.object = self.current.media.meta.object || null;
|
||||
self.current.media.meta.failed = true;
|
||||
}
|
||||
if (cb) {
|
||||
cb();
|
||||
}
|
||||
};
|
||||
|
||||
if (!this.current || this.current.media.type !== "gd") {
|
||||
return abort();
|
||||
}
|
||||
|
||||
self.channel.activeLock.lock();
|
||||
InfoGetter.getMedia(this.current.media.id, "gd", function (err, media) {
|
||||
if (err) {
|
||||
Logger.errlog.log("Google Docs autorefresh failed: " + err);
|
||||
Logger.errlog.log("ID was: " + self.current.media.id);
|
||||
if (self.current) {
|
||||
self.current.media.meta.object = self.current.media.meta.object || null;
|
||||
self.current.media.meta.failed = true;
|
||||
}
|
||||
if (cb) {
|
||||
cb();
|
||||
}
|
||||
self.channel.activeLock.release();
|
||||
} else {
|
||||
if (!self.current || self.current.media.type !== "gd") {
|
||||
self.channel.activeLock.release();
|
||||
return abort();
|
||||
}
|
||||
|
||||
self.current.media.meta = media.meta;
|
||||
self.current.media.meta.expiration = Date.now() + 3600000;
|
||||
self.channel.logger.log("[playlist] Auto-refreshed Google Doc video");
|
||||
cb && cb();
|
||||
self.channel.activeLock.release();
|
||||
}
|
||||
});
|
||||
};
|
||||
|
||||
PlaylistModule.prototype._playNext = function () {
|
||||
if (!this.current) {
|
||||
return;
|
||||
|
|
@ -1335,10 +1279,11 @@ PlaylistModule.prototype.handleQueuePlaylist = function (user, data) {
|
|||
pos: data.pos
|
||||
};
|
||||
|
||||
var self = this;
|
||||
self.channel.activeLock.lock();
|
||||
const self = this;
|
||||
self.channel.refCounter.ref("PlaylistModule::handleQueuePlaylist");
|
||||
db.getUserPlaylist(user.getName(), data.name, function (err, pl) {
|
||||
if (err) {
|
||||
self.channel.refCounter.unref("PlaylistModule::handleQueuePlaylist");
|
||||
return user.socket.emit("errorMsg", {
|
||||
msg: "Playlist load failed: " + err
|
||||
});
|
||||
|
|
@ -1369,7 +1314,6 @@ PlaylistModule.prototype.handleQueuePlaylist = function (user, data) {
|
|||
var m = new Media(item.id, item.title, item.seconds, item.type, item.meta);
|
||||
self._addItem(m, qdata, user);
|
||||
});
|
||||
self.channel.activeLock.release();
|
||||
} catch (e) {
|
||||
Logger.errlog.log("Loading user playlist failed!");
|
||||
Logger.errlog.log("PL: " + user.getName() + "-" + data.name);
|
||||
|
|
@ -1378,9 +1322,22 @@ PlaylistModule.prototype.handleQueuePlaylist = function (user, data) {
|
|||
msg: "Internal error occurred when loading playlist.",
|
||||
link: null
|
||||
});
|
||||
self.channel.activeLock.release();
|
||||
} finally {
|
||||
self.channel.refCounter.unref("PlaylistModule::handleQueuePlaylist");
|
||||
}
|
||||
});
|
||||
};
|
||||
|
||||
PlaylistModule.prototype.handleRequestPlaylist = function (user) {
|
||||
if (user.reqPlaylistLimiter.throttle(REQ_PLAYLIST_THROTTLE)) {
|
||||
user.socket.emit("errorMsg", {
|
||||
msg: "Get Playlist URLs is limited to 1 usage every 60 seconds. " +
|
||||
"Please try again later.",
|
||||
code: "REQ_PLAYLIST_LIMIT_REACHED"
|
||||
});
|
||||
} else {
|
||||
this.sendPlaylist([user]);
|
||||
}
|
||||
};
|
||||
|
||||
module.exports = PlaylistModule;
|
||||
|
|
|
|||
|
|
@ -50,7 +50,8 @@ var defaults = {
|
|||
io: {
|
||||
domain: "http://localhost",
|
||||
"default-port": 1337,
|
||||
"ip-connection-limit": 10
|
||||
"ip-connection-limit": 10,
|
||||
"per-message-deflate": false
|
||||
},
|
||||
mail: {
|
||||
enabled: false,
|
||||
|
|
|
|||
|
|
@ -25,11 +25,20 @@ Socket.prototype.packet = function () {
|
|||
exports.add('socket.io:packet');
|
||||
};
|
||||
|
||||
function getConnectedSockets() {
|
||||
var sockets = io.instance.sockets.sockets;
|
||||
if (typeof sockets.length === 'number') {
|
||||
return sockets.length;
|
||||
} else {
|
||||
return Object.keys(sockets).length;
|
||||
}
|
||||
}
|
||||
|
||||
setInterval(function () {
|
||||
try {
|
||||
counters['memory:rss'] = process.memoryUsage().rss / 1048576;
|
||||
counters['load:1min'] = os.loadavg()[0];
|
||||
counters['socket.io:count'] = io.instance.sockets.sockets.length;
|
||||
counters['socket.io:count'] = getConnectedSockets();
|
||||
counterLog.log(JSON.stringify(counters));
|
||||
} catch (e) {
|
||||
Logger.errlog.log(e.stack);
|
||||
|
|
|
|||
|
|
@ -583,7 +583,7 @@ module.exports.loadAnnouncement = function () {
|
|||
var sv = Server.getServer();
|
||||
sv.announcement = announcement;
|
||||
for (var id in sv.ioServers) {
|
||||
sv.ioServers[id].sockets.emit("announcement", announcement);
|
||||
sv.ioServers[id].emit("announcement", announcement);
|
||||
}
|
||||
});
|
||||
};
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ var Config = require("./config");
|
|||
var ffmpeg = require("./ffmpeg");
|
||||
var mediaquery = require("cytube-mediaquery");
|
||||
var YouTube = require("cytube-mediaquery/lib/provider/youtube");
|
||||
var Vimeo = require("cytube-mediaquery/lib/provider/vimeo");
|
||||
|
||||
/*
|
||||
* Preference map of quality => youtube formats.
|
||||
|
|
@ -158,50 +159,11 @@ var Getters = {
|
|||
return Getters.vi_oauth(id, callback);
|
||||
}
|
||||
|
||||
var options = {
|
||||
host: "vimeo.com",
|
||||
port: 443,
|
||||
path: "/api/v2/video/" + id + ".json",
|
||||
method: "GET",
|
||||
dataType: "jsonp",
|
||||
timeout: 1000
|
||||
};
|
||||
|
||||
urlRetrieve(https, options, function (status, data) {
|
||||
switch (status) {
|
||||
case 200:
|
||||
break; /* Request is OK, skip to handling data */
|
||||
case 400:
|
||||
return callback("Invalid request", null);
|
||||
case 403:
|
||||
return callback("Private video", null);
|
||||
case 404:
|
||||
return callback("Video not found", null);
|
||||
case 500:
|
||||
case 503:
|
||||
return callback("Service unavailable", null);
|
||||
default:
|
||||
return callback("HTTP " + status, null);
|
||||
}
|
||||
|
||||
try {
|
||||
data = JSON.parse(data);
|
||||
data = data[0];
|
||||
var seconds = data.duration;
|
||||
var title = data.title;
|
||||
var media = new Media(id, title, seconds, "vi");
|
||||
callback(false, media);
|
||||
} catch(e) {
|
||||
var err = e;
|
||||
/**
|
||||
* This should no longer be necessary as the outer handler
|
||||
* checks for HTTP 404
|
||||
*/
|
||||
if (buffer.match(/not found/))
|
||||
err = "Video not found";
|
||||
|
||||
callback(err, null);
|
||||
}
|
||||
Vimeo.lookup(id).then(video => {
|
||||
video = new Media(video.id, video.title, video.duration, "vi");
|
||||
callback(null, video);
|
||||
}).catch(error => {
|
||||
callback(error.message);
|
||||
});
|
||||
},
|
||||
|
||||
|
|
|
|||
|
|
@ -242,6 +242,9 @@ function handleConnection(sock) {
|
|||
module.exports = {
|
||||
init: function (srv, webConfig) {
|
||||
var bound = {};
|
||||
const ioOptions = {
|
||||
perMessageDeflate: Config.get("io.per-message-deflate")
|
||||
};
|
||||
var io = sio.instance = sio();
|
||||
|
||||
io.use(handleAuth);
|
||||
|
|
@ -259,7 +262,7 @@ module.exports = {
|
|||
}
|
||||
|
||||
if (id in srv.servers) {
|
||||
io.attach(srv.servers[id]);
|
||||
io.attach(srv.servers[id], ioOptions);
|
||||
} else {
|
||||
var server = require("http").createServer().listen(bind.port, bind.ip);
|
||||
server.on("clientError", function (err, socket) {
|
||||
|
|
@ -268,7 +271,7 @@ module.exports = {
|
|||
} catch (e) {
|
||||
}
|
||||
});
|
||||
io.attach(server);
|
||||
io.attach(server, ioOptions);
|
||||
}
|
||||
|
||||
bound[id] = null;
|
||||
|
|
|
|||
|
|
@ -207,6 +207,24 @@ Server.prototype.unloadChannel = function (chan) {
|
|||
chan.notifyModules("unload", []);
|
||||
Object.keys(chan.modules).forEach(function (k) {
|
||||
chan.modules[k].dead = true;
|
||||
/*
|
||||
* Automatically clean up any timeouts/intervals assigned
|
||||
* to properties of channel modules. Prevents a memory leak
|
||||
* in case of forgetting to clear the timer on the "unload"
|
||||
* module event.
|
||||
*/
|
||||
Object.keys(chan.modules[k]).forEach(function (prop) {
|
||||
if (chan.modules[k][prop] && chan.modules[k][prop]._onTimeout) {
|
||||
Logger.errlog.log("Warning: detected non-null timer when unloading " +
|
||||
"module " + k + ": " + prop);
|
||||
try {
|
||||
clearTimeout(chan.modules[k][prop]);
|
||||
clearInterval(chan.modules[k][prop]);
|
||||
} catch (error) {
|
||||
Logger.errlog.log(error.stack);
|
||||
}
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
for (var i = 0; i < this.channels.length; i++) {
|
||||
|
|
@ -220,7 +238,9 @@ Server.prototype.unloadChannel = function (chan) {
|
|||
// Empty all outward references from the channel
|
||||
var keys = Object.keys(chan);
|
||||
for (var i in keys) {
|
||||
delete chan[keys[i]];
|
||||
if (keys[i] !== "refCounter") {
|
||||
delete chan[keys[i]];
|
||||
}
|
||||
}
|
||||
chan.dead = true;
|
||||
};
|
||||
|
|
|
|||
|
|
@ -21,6 +21,7 @@ function User(socket) {
|
|||
self.channel = null;
|
||||
self.queueLimiter = util.newRateLimiter();
|
||||
self.chatLimiter = util.newRateLimiter();
|
||||
self.reqPlaylistLimiter = util.newRateLimiter();
|
||||
self.awaytimer = false;
|
||||
|
||||
var announcement = Server.getServer().announcement;
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue