Skip to content

Commit

Permalink
Added stream tagging support
Browse files Browse the repository at this point in the history
  • Loading branch information
@Thulinma
Thulinma committed Dec 20, 2023
1 parent 4d50364 commit 6bec406
Show file tree
Hide file tree
Showing 8 changed files with 293 additions and 35 deletions.
37 changes: 37 additions & 0 deletions lib/stream.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -471,6 +471,43 @@ bool Util::streamAlive(std::string &streamname){
}
}

/// Returns active tags for an exact-matching (already sanitized) streamname
std::set<std::string> Util::streamTags(const std::string &streamname){
std::set<std::string> ret;

IPC::sharedPage shmStreams(SHM_STATE_STREAMS, 0, false, false);
// Abort silently if page cannot be loaded
if (!shmStreams){return ret;}

Util::RelAccX rlxStreams(shmStreams.mapped);
// Abort silently if page cannot be loaded
if (!rlxStreams.isReady()){return ret;}

uint64_t startPos = rlxStreams.getDeleted();
uint64_t endPos = rlxStreams.getEndPos();
for (uint64_t cPos = startPos; cPos < endPos; ++cPos){
const std::string & strm = rlxStreams.getPointer("stream", cPos);
if (strm != streamname){continue;}

// Found it! Fill and break, since only one match can exist.
std::string tags = rlxStreams.getPointer("tags", cPos);
while (tags.size()){
size_t endPos = tags.find(' ');
if (!endPos){
//extra space, ignore
tags.erase(0, 1);
continue;
}
if (endPos == std::string::npos){endPos = tags.size();}
ret.insert(tags.substr(0, endPos));
if (endPos == tags.size()){break;}
tags.erase(0, endPos+1);
}
break;
}
return ret;
}

/// Assures the input for the given stream name is active.
/// Does stream name sanitation first, followed by a stream name length check (<= 100 chars).
/// Then, checks if an input is already active by running streamAlive(). If yes, return true.
Expand Down
1 change: 1 addition & 0 deletions lib/stream.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ namespace Util{
std::string getTmpFolder();
void sanitizeName(std::string &streamname);
bool streamAlive(std::string &streamname);
std::set<std::string> streamTags(const std::string &streamname);
bool startInput(std::string streamname, std::string filename = "", bool forkFirst = true,
bool isProvider = false,
const std::map<std::string, std::string> &overrides = std::map<std::string, std::string>(),
Expand Down
9 changes: 9 additions & 0 deletions lib/triggers.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -203,6 +203,15 @@ namespace Triggers{
if ((streamName.size() == stringLen || splitter == stringLen) &&
strncmp(strPtr + bPos + 4, streamName.data(), stringLen) == 0){
isHandled = true;
break;
}
// Tag-based? Check tags for this stream
if (strPtr[bPos + 4] == '#'){
std::set<std::string> tags = Util::streamTags(streamName);
if (tags.count(std::string(strPtr + bPos + 5, stringLen - 1))){
isHandled = true;
break;
}
}
bPos += stringLen + 4;
}
Expand Down
69 changes: 68 additions & 1 deletion src/controller/controller_api.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -175,15 +175,17 @@ class streamStat{
viewers = rlx.getInt("viewers", entry);
inputs = rlx.getInt("inputs", entry);
outputs = rlx.getInt("outputs", entry);
tags = rlx.getPointer("tags", entry);
}
bool operator==(const streamStat &b) const{
return (status == b.status && viewers == b.viewers && inputs == b.inputs && outputs == b.outputs);
return (status == b.status && viewers == b.viewers && inputs == b.inputs && outputs == b.outputs && tags == b.tags);
}
bool operator!=(const streamStat &b) const{return !(*this == b);}
uint8_t status;
uint64_t viewers;
uint64_t inputs;
uint64_t outputs;
std::string tags;
};

void Controller::handleWebSocket(HTTP::Parser &H, Socket::Connection &C){
Expand Down Expand Up @@ -292,6 +294,7 @@ void Controller::handleWebSocket(HTTP::Parser &H, Socket::Connection &C){
tmp[1u].append(tmpStat.viewers);
tmp[1u].append(tmpStat.inputs);
tmp[1u].append(tmpStat.outputs);
tmp[1u].append(tmpStat.tags);
W.sendFrame(tmp.toString());
}
}
Expand All @@ -305,6 +308,7 @@ void Controller::handleWebSocket(HTTP::Parser &H, Socket::Connection &C){
tmp[1u].append(0u);
tmp[1u].append(0u);
tmp[1u].append(0u);
tmp[1u].append("");
W.sendFrame(tmp.toString());
strmRemove.erase(strm);
lastStrmStat.erase(strm);
Expand Down Expand Up @@ -1164,6 +1168,69 @@ void Controller::handleAPICommands(JSON::Value &Request, JSON::Value &Response){
}
}

if (Request.isMember("tag_stream")){
if (Request["tag_stream"].isObject()){
jsonForEach(Request["tag_stream"], it){
if (it->isString()){
Controller::stream_tag(it.key(), it->asStringRef());
}else if (it->isArray()){
jsonForEach(*it, jt){
if (jt->isString()){
Controller::stream_tag(it.key(), jt->asStringRef());
}
}
}
}
}
}

if (Request.isMember("untag_stream")){
if (Request["untag_stream"].isObject()){
jsonForEach(Request["untag_stream"], it){
if (it->isString()){
Controller::stream_untag(it.key(), it->asStringRef());
}else if (it->isArray()){
jsonForEach(*it, jt){
if (jt->isString()){
Controller::stream_untag(it.key(), jt->asStringRef());
}
}
}
}
}
}

if (Request.isMember("stream_tags")){
JSON::Value & rT = Response["stream_tags"];
if (Request["stream_tags"].isArray()){
jsonForEach(Request["stream_tags"], it){
if (it->isString()){
std::set<std::string> tags = Controller::stream_tags(it->asStringRef());
JSON::Value & tRef = rT[it->asStringRef()];
for (std::set<std::string>::iterator ti = tags.begin(); ti != tags.end(); ++ti){tRef.append(*ti);}
}
}
}else if (Request["stream_tags"].isObject()){
jsonForEach(Request["stream_tags"], it){
std::set<std::string> tags = Controller::stream_tags(it.key());
JSON::Value & tRef = rT[it.key()];
for (std::set<std::string>::iterator ti = tags.begin(); ti != tags.end(); ++ti){tRef.append(*ti);}
}
}else if (Request["stream_tags"].isString() && Request["stream_tags"].asStringRef().size()){
std::set<std::string> tags = Controller::stream_tags(Request["stream_tags"].asStringRef());
JSON::Value & tRef = rT[Request["stream_tags"].asStringRef()];
for (std::set<std::string>::iterator ti = tags.begin(); ti != tags.end(); ++ti){tRef.append(*ti);}
}else{
JSON::Value nullPkt, resp;
Controller::fillActive(nullPkt, resp);
jsonForEach(resp, it){
std::set<std::string> tags = Controller::stream_tags(it->asStringRef());
JSON::Value & tRef = rT[it->asStringRef()];
for (std::set<std::string>::iterator ti = tags.begin(); ti != tags.end(); ++ti){tRef.append(*ti);}
}
}
}

if (Request.isMember("push_start")){
std::string stream;
std::string target;
Expand Down
30 changes: 13 additions & 17 deletions src/controller/controller_push.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -333,19 +333,17 @@ namespace Controller{
for (std::set<std::string>::iterator jt = activeStreams.begin();
jt != activeStreams.end(); ++jt){
std::string streamname = *jt;
if (stream == streamname || (*stream.rbegin() == '+' && streamname.substr(0, stream.size()) == stream)){
if (!isPushActive(streamname, target)){
if (waitingPushes[streamname][target]++ >= waittime && (curCount < maxspeed || !maxspeed)){
waitingPushes[streamname].erase(target);
if (!waitingPushes[streamname].size()){waitingPushes.erase(streamname);}
MEDIUM_MSG("Conditions of push `%s->%s` evaluate to true. Starting push...", stream.c_str(), target.c_str());
startPush(streamname, target);
curCount++;
// If no end time is given but there is a start time, remove the push after starting it
if (startTime && !endTime){
removePush(*it);
break;
}
if (!isPushActive(streamname, target)){
if (waitingPushes[streamname][target]++ >= waittime && (curCount < maxspeed || !maxspeed)){
waitingPushes[streamname].erase(target);
if (!waitingPushes[streamname].size()){waitingPushes.erase(streamname);}
MEDIUM_MSG("Conditions of push `%s->%s` evaluate to true. Starting push...", stream.c_str(), target.c_str());
startPush(streamname, target);
curCount++;
// If no end time is given but there is a start time, remove the push after starting it
if (startTime && !endTime){
removePush(*it);
break;
}
}
}
Expand Down Expand Up @@ -537,9 +535,7 @@ namespace Controller{
for (std::set<std::string>::iterator jt = activeStreams.begin();
jt != activeStreams.end(); ++jt){
std::string streamname = *jt;
if (stream == streamname || (*stream.rbegin() == '+' && streamname.substr(0, stream.size()) == stream)){
startPush(streamname, target);
}
startPush(streamname, target);
}
}
// Return push list
Expand Down Expand Up @@ -588,7 +584,7 @@ namespace Controller{
jsonForEach(Controller::Storage["autopushes"], it){
if ((*it)[2u].asInt() && (*it)[2u].asInt() < Util::epoch()){continue;}
const std::string &pStr = (*it)[0u].asStringRef();
if (pStr == streamname || (*pStr.rbegin() == '+' && streamname.substr(0, pStr.size()) == pStr)){
if (Controller::streamMatches(streamname, pStr)){
std::string stream = streamname;
Util::sanitizeName(stream);
// Check variable condition if it exists
Expand Down
Loading

0 comments on commit 6bec406

Please sign in to comment.