rainbownapkin/canopy
1
0
Fork 0
Code Issues 2 Pull requests Actions Packages Projects Releases Wiki Activity

Pushback Schedule mode now works when ending livestreams.

Browse source
This commit is contained in:
rainbow napkin 2025-05-19 04:11:25 -04:00
parent 92edd74aaa
commit b5e54afe99
1 changed files with 191 additions and 60 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

163
src/app/channel/media/queue.js
View file

@ -51,6 +51,8 @@ module.exports = class{
this.nowPlaying = null;
//Create variable to hold item that was playing during the last liveStream (can't check against full duration since it might've been stopped for other reasons)
this.liveRemainder = null;
//Create variable to hold current live mode
this.liveMode = null;
//Create variable to lock standard queuing functions during livestreams
this.streamLock = false;
@ -314,6 +316,8 @@ module.exports = class{
new Date().getTime()
);
//Validate mode input, and default to overwrite
this.liveMode = (data.mode == "pushback") ? "pushback" : "overwrite";
//Throw stream lock
this.streamLock = true;
@ -463,7 +467,7 @@ module.exports = class{
}
}
async rescheduleMedia(uuid, start = new Date().getTime(), socket){
async rescheduleMedia(uuid, start = new Date().getTime(), socket, chanDB){
//If we're streamlocked
if(this.streamLock){
//If an originating socket was provided for this request
@ -475,6 +479,19 @@ module.exports = class{
return;
}
try{
//If we wheren't handed a channel
if(chanDB == null){
//DO everything ourselves since we don't have a fance end() function to do it
chanDB = await channelModel.findOne({name:this.channel.name});
}
//If we couldn't find the channel
if(chanDB == null){
//FUCK
throw loggerUtils.exceptionSmith(`Unable to find channel document ${this.channel.name} while queue item!`, "queue");
}
//Find our media, don't remove it yet since we want to do some more testing first
const media = this.getItemByUUID(uuid);
@ -493,7 +510,6 @@ module.exports = class{
if(media.startTime < new Date().getTime()){
//If an originating socket was provided for this request
if(socket != null){
//If the item is currently playing
if(media.getEndTime() > new Date().getTime()){
//Yell at the user for being an asshole
@ -511,7 +527,7 @@ module.exports = class{
}
//Remove the media from the schedule
await this.removeMedia(uuid);
await this.removeMedia(uuid, socket, chanDB);
//Grab the old start time for safe keeping
const oldStart = media.startTime;
@ -523,20 +539,32 @@ module.exports = class{
media.startTimeStamp = 0;
//Attempt to schedule media at given time
//Otherwise, if it returns false for fuckup
if(!(await this.scheduleMedia([media], socket))){
//Otherwise, if it returns false for fuckup, with noSave enabled
if(!(await this.scheduleMedia([media], socket, chanDB))){
//Reset start time
media.startTime = oldStart;
//Reset the start time stamp for re-calculation
media.startTimeStamp = 0;
//Schedule in old slot
this.scheduleMedia([media], socket, null, true);
//Schedule in old slot with noSave enabled
await this.scheduleMedia([media], socket, chanDB, true);
}
}catch(err){
//If this was originated by someone
if(socket != null){
//Bitch at them
loggerUtils.socketExceptionHandler(socket, err);
//If not
}else{
//Bitch to the console
loggerUtils.localExceptionHandler(err);
}
}
}
async removeMedia(uuid, socket, chanDB){
async removeMedia(uuid, socket, chanDB, noScheduling = false){
//If we're streamlocked
if(this.streamLock){
//If an originating socket was provided for this request
@ -585,7 +613,6 @@ module.exports = class{
}else{
//Broadcast changes
this.broadcastQueue(chanDB);
//Save changes to the DB
await chanDB.save();
}
@ -608,13 +635,18 @@ module.exports = class{
//Take the item out of the schedule map
this.schedule.delete(media.startTime);
if(!noScheduling){
//Refresh next timer
this.refreshNextTimer();
}
//If we're currently playing the requested item.
if(this.nowPlaying != null && this.nowPlaying.uuid == uuid){
//If scheduling is enabled
if(!noScheduling){
//End playback
this.end(false, true);
}
//otherwise
}else{
try{
@ -635,7 +667,10 @@ module.exports = class{
return record.uuid != uuid;
});
//If saving is enabled (seperate from all DB transactions since caller function may want modifications but handle saving on its own)
if(!noScheduling){
await chanDB.save();
}
//Broadcast the channel
this.broadcastQueue(chanDB);
@ -660,7 +695,7 @@ module.exports = class{
return media;
}
async scheduleMedia(media, socket, chanDB, force = false, volatile = false, startVolatile = false, saveLate = false){
async scheduleMedia(media, socket, chanDB, force = false, volatile = false, startVolatile = false, saveLate = false, noSave = false){
/* This is a fun method and I think it deserves it's own little explination...
Since we're working with a time based schedule, using start epochs as keys for our iterable seemed the best option
I don't want to store everything in a sparse array because that *feels* icky, and would probably be a pain in the ass.
@ -811,8 +846,11 @@ module.exports = class{
//If we fucked with the DB during the main loop
if(chanDB != null && !volatile){
try{
//If saving is enabled (seperate from all DB transactions since caller function may want modifications but handle saving on its own)
if(!noSave){
//Save the database
chanDB.save();
await chanDB.save();
}
//If something fucked up
}catch(err){
//If this was originated by someone
@ -877,7 +915,7 @@ module.exports = class{
const chanDB = await channelModel.findOne({name: this.channel.name});
//If nowPlaying isn't null and isn't what we're about to throw on
if(chanDB.media.nowPlaying != null && chanDB.media.nowPlaying.uuid.toString != mediaObj.uuid){
if(chanDB.media.nowPlaying != null && chanDB.media.nowPlaying.uuid.toString() != mediaObj.uuid){
//Archive whats already in there since we're about to clobber the fuck out of it
chanDB.media.archived.push(chanDB.media.nowPlaying);
}
@ -1034,15 +1072,30 @@ module.exports = class{
//Disable stream lock
this.streamLock = false;
//We don't have to here since someone else will do it for us :)
//We don't have to save here since someone else will do it for us :)
chanDB.media.liveRemainder = null;
//Get current epoch
const now = new Date().getTime()
//Set duration from start and end time
wasPlaying.duration = (now - wasPlaying.startTime) / 1000;
//If we're in pushback mode
if(this.liveMode == "pushback"){
await this.livestreamPushbackSchedule(wasPlaying, chanDB);
//Otherwise
}else{
//This is where I'd stick the IF statetement I'd add to switch between overwrite
await this.livestreamOverwriteSchedule(wasPlaying, chanDB)
}
//Refresh next timer
this.refreshNextTimer();
//Null out live mode
this.liveMode = null;
//Broadcast Queue
this.broadcastQueue();
//ACK
@ -1075,9 +1128,6 @@ module.exports = class{
//while the other needs to run regardless of this.liveRemainders definition
let finished = false;
//Set duration from start and end time
wasPlaying.duration = (now - wasPlaying.startTime) / 1000;
//Throw the livestream into the archive
chanDB.media.archived.push(wasPlaying);
@ -1141,6 +1191,87 @@ module.exports = class{
}
async livestreamPushbackSchedule(wasPlaying, chanDB){
try{
//Get current epoch
const now = new Date().getTime()
//If we wheren't handed a channel
if(chanDB == null){
//Now that everything is clean, we can take our time with the DB :P
chanDB = await channelModel.findOne({name:this.channel.name});
}
//If we couldn't find the channel
if(chanDB == null){
//FUCK
throw loggerUtils.exceptionSmith(`Unable to find channel document ${this.channel.name} while ending queue item!`, "queue");
}
//Throw the livestream into the archive
chanDB.media.archived.push(wasPlaying);
//Set the current place to schedule items at 5ms after the end of the live stream
let curPlace = wasPlaying.getEndTime() + 5;
const newSched = [];
//if we have a live remainder
if(this.liveRemainder != null){
//Set item to continue where it left off
this.liveRemainder.startTimeStamp = this.liveRemainder.earlyEnd;
//Rip out the early end so it finish up
this.liveRemainder.earlyEnd = null;
//Generate new UUID for uniqueness
this.liveRemainder.genUUID();
//Set start time to the end of the stream
this.liveRemainder.startTime = curPlace;
//Reset starter time to end of current item + 5ms
curPlace = this.liveRemainder.getEndTime(true) + 5;
//Throw live remainder into the new schedule
newSched.push(this.liveRemainder);
//Null out live remainder for the next stream
this.liveRemainder = null;
chanDB.liveRemainder = null;
}
//Iterate through objects in schedule
for(const entry of this.schedule){
//Pull media object from map entry
const mediaObj = entry[1];
//Remove media from queue without calling chanDB.save() to make room before we move everything
await this.removeMedia(mediaObj.uuid, null, chanDB, true);
mediaObj.genUUID();
//Change start time to current starter place
mediaObj.startTime = curPlace;
//Throw item into the temp sched
newSched.push(mediaObj);
//Set cur place to 5ms after the item we just queued
curPlace = mediaObj.getEndTime() + 5;
}
//Schedule the moved schedule, letting scheduleMedia save our changes for us, starting w/o saves to prevent over-saving
await this.scheduleMedia(newSched, null, chanDB);
}catch(err){
//Null out live remainder for the next stream
this.liveRemainder = null;
//Handle the error
loggerUtils.localExceptionHandler(err);
}
}
stop(socket){
//If we're not currently playing anything
if(this.nowPlaying == null){