Move launching apps to http /launch

This commit is contained in:
loki 2019-12-17 23:16:28 +01:00
parent da58021a1a
commit ae2eb217a5
5 changed files with 70 additions and 51 deletions

View File

@ -468,16 +468,52 @@ template<class T>
void launch(std::shared_ptr<typename SimpleWeb::ServerBase<T>::Response> response, std::shared_ptr<typename SimpleWeb::ServerBase<T>::Request> request) { void launch(std::shared_ptr<typename SimpleWeb::ServerBase<T>::Response> response, std::shared_ptr<typename SimpleWeb::ServerBase<T>::Request> request) {
print_req<T>(request); print_req<T>(request);
auto args = request->parse_query_string(); static std::int64_t current_appid { -1 };
pt::ptree tree;
auto g = util::fail_guard([&]() {
tree.put("root.gamesession", 0);
std::ostringstream data;
pt::write_xml(data, tree);
response->write(data.str());
});
auto args = request->parse_query_string();
auto appid = util::from_view(args.at("appid")) -2; auto appid = util::from_view(args.at("appid")) -2;
stream::launch_session_t launch_session; stream::launch_session_t launch_session;
if(appid >= 0) {
auto pos = std::begin(proc::proc.get_apps()); if(stream::has_session) {
tree.put("root.<xmlattr>.status_code", 503);
return;
}
if(!proc::proc.running()) {
current_appid = -1;
}
if(appid >= 0 && appid != current_appid) {
auto &apps = proc::proc.get_apps();
if(appid >= apps.size()) {
tree.put("root.<xmlattr>.status_code", 404);
return;
}
auto pos = std::begin(apps);
std::advance(pos, appid); std::advance(pos, appid);
launch_session.app_name = pos->first; auto err = proc::proc.execute(pos->first);
if(err) {
tree.put("root.<xmlattr>.status_code", 500);
return;
}
current_appid = appid;
} }
auto clientID = args.at("uniqueid"s); auto clientID = args.at("uniqueid"s);
@ -488,15 +524,6 @@ void launch(std::shared_ptr<typename SimpleWeb::ServerBase<T>::Response> respons
auto next = std::copy(prepend_iv_p, prepend_iv_p + sizeof(prepend_iv), std::begin(launch_session.iv)); auto next = std::copy(prepend_iv_p, prepend_iv_p + sizeof(prepend_iv), std::begin(launch_session.iv));
std::fill(next, std::end(launch_session.iv), 0); std::fill(next, std::end(launch_session.iv), 0);
pt::ptree tree;
auto g = util::fail_guard([&]() {
std::ostringstream data;
pt::write_xml(data, tree);
response->write(data.str());
});
stream::launch_event.raise(std::move(launch_session)); stream::launch_event.raise(std::move(launch_session));
/* /*
@ -509,6 +536,12 @@ void launch(std::shared_ptr<typename SimpleWeb::ServerBase<T>::Response> respons
tree.put("root.<xmlattr>.status_code", 200); tree.put("root.<xmlattr>.status_code", 200);
tree.put("root.gamesession", 1); tree.put("root.gamesession", 1);
std::ostringstream data;
pt::write_xml(data, tree);
response->write(data.str());
} }
template<class T> template<class T>

View File

@ -45,10 +45,8 @@ int exe(const std::string &cmd, bp::environment &env, file_t &file, std::error_c
int proc_t::execute(const std::string &name) { int proc_t::execute(const std::string &name) {
auto it = _name_to_proc.find(name); auto it = _name_to_proc.find(name);
std::cout << "Ensure clean slate"sv << std::endl;
// Ensure starting from a clean slate // Ensure starting from a clean slate
_undo_pre_cmd(); _undo_pre_cmd();
std::cout << "Clean slate"sv << std::endl;
if(it == std::end(_name_to_proc)) { if(it == std::end(_name_to_proc)) {
std::cout << "Error: Couldn't find ["sv << name << ']' << std::endl; std::cout << "Error: Couldn't find ["sv << name << ']' << std::endl;

View File

@ -115,9 +115,8 @@ struct session_t {
crypto::aes_t gcm_key; crypto::aes_t gcm_key;
crypto::aes_t iv; crypto::aes_t iv;
std::string app_name;
} session; } session;
std::atomic_bool has_session;
void free_host(ENetHost *host) { void free_host(ENetHost *host) {
std::for_each(host->peers, host->peers + host->peerCount, [](ENetPeer &peer_ref) { std::for_each(host->peers, host->peers + host->peerCount, [](ENetPeer &peer_ref) {
@ -155,6 +154,13 @@ host_t host_create(ENetAddress &addr, std::uint16_t port) {
void print_msg(PRTSP_MESSAGE msg); void print_msg(PRTSP_MESSAGE msg);
void cmd_not_found(host_t &host, peer_t peer, msg_t&& req); void cmd_not_found(host_t &host, peer_t peer, msg_t&& req);
void stop(session_t &session) {
session.video_packets->stop();
session.audio_packets->stop();
has_session.store(false);
}
class rtsp_server_t { class rtsp_server_t {
public: public:
rtsp_server_t(rtsp_server_t &&) noexcept = default; rtsp_server_t(rtsp_server_t &&) noexcept = default;
@ -522,8 +528,8 @@ void controlThread(video::idr_event_t idr_events) {
// something went wrong :( // something went wrong :(
std::cout << "failed to verify tag"sv << std::endl; std::cout << "failed to verify tag"sv << std::endl;
session.video_packets->stop();
session.audio_packets->stop(); stop(session);
} }
if(tagged_cipher_length >= 16 + session.iv.size()) { if(tagged_cipher_length >= 16 + session.iv.size()) {
@ -534,14 +540,14 @@ void controlThread(video::idr_event_t idr_events) {
input::passthrough(input, plaintext.data()); input::passthrough(input, plaintext.data());
}); });
while(session.video_packets->running()) { while(has_session) {
if(std::chrono::steady_clock::now() > session.pingTimeout) { if(std::chrono::steady_clock::now() > session.pingTimeout) {
std::cout << "ping timeout"sv << std::endl; std::cout << "ping timeout"sv << std::endl;
session.video_packets->stop();
session.audio_packets->stop(); stop(session);
} }
if(!session.app_name.empty() && !proc::proc.running()) { if(!proc::proc.running()) {
std::cout << "Process terminated"sv << std::endl; std::cout << "Process terminated"sv << std::endl;
std::uint16_t reason = 0x0100; std::uint16_t reason = 0x0100;
@ -552,8 +558,7 @@ void controlThread(video::idr_event_t idr_events) {
server.send(std::string_view {(char*)payload.data(), payload.size()}); server.send(std::string_view {(char*)payload.data(), payload.size()});
session.video_packets->stop(); stop(session);
session.audio_packets->stop();
} }
server.iterate(500ms); server.iterate(500ms);
@ -653,9 +658,7 @@ void audioThread() {
// std::cout << "Audio ["sv << frame << "] :: send..."sv << std::endl; // std::cout << "Audio ["sv << frame << "] :: send..."sv << std::endl;
} }
std::cout << "Audio: Joining()" << std::endl;
captureThread.join(); captureThread.join();
std::cout << "Audio: Joining()" << std::endl;
} }
void videoThread(video::idr_event_t idr_events) { void videoThread(video::idr_event_t idr_events) {
@ -760,9 +763,8 @@ void videoThread(video::idr_event_t idr_events) {
lowseq += shards.size(); lowseq += shards.size();
} }
std::cout << "Video: Joining()" << std::endl;
captureThread.join(); captureThread.join();
std::cout << "Video: Joined()" << std::endl;
} }
void respond(host_t &host, peer_t peer, msg_t &resp) { void respond(host_t &host, peer_t peer, msg_t &resp) {
@ -890,13 +892,13 @@ void cmd_announce(host_t &host, peer_t peer, msg_t &&req) {
auto seqn_str = std::to_string(req->sequenceNumber); auto seqn_str = std::to_string(req->sequenceNumber);
option.content = const_cast<char*>(seqn_str.c_str()); option.content = const_cast<char*>(seqn_str.c_str());
if(session.video_packets || !launch_event.peek()) { if(has_session || !launch_event.peek()) {
//Either already streaming or /launch has not been used //Either already streaming or /launch has not been used
respond(host, peer, &option, 503, "Service Unavailable", req->sequenceNumber, {}); respond(host, peer, &option, 503, "Service Unavailable", req->sequenceNumber, {});
return; return;
} }
auto launch_session { std::move(*launch_event.pop()) }; auto launch_session { *launch_event.pop() };
std::string_view payload { req->payload, (size_t)req->payloadLength }; std::string_view payload { req->payload, (size_t)req->payloadLength };
@ -961,26 +963,11 @@ void cmd_announce(host_t &host, peer_t peer, msg_t &&req) {
return; return;
} }
auto &app_name = launch_session.app_name; has_session.store(true);
auto &gcm_key = launch_session.gcm_key; auto &gcm_key = launch_session.gcm_key;
auto &iv = launch_session.iv; auto &iv = launch_session.iv;
if(!app_name.empty() && session.app_name != app_name ) {
if(auto err_code = proc::proc.execute(app_name)) {
if(err_code == 404) {
respond(host, peer, &option, 404, (app_name + " NOT FOUND").c_str(), req->sequenceNumber, {});
return;
}
else {
respond(host, peer, &option, 500, "INTERNAL ERROR", req->sequenceNumber, {});
return;
}
}
}
session.app_name = std::move(app_name);
std::copy(std::begin(gcm_key), std::end(gcm_key), std::begin(session.gcm_key)); std::copy(std::begin(gcm_key), std::end(gcm_key), std::begin(session.gcm_key));
std::copy(std::begin(iv), std::end(iv), std::begin(session.iv)); std::copy(std::begin(iv), std::end(iv), std::begin(session.iv));

View File

@ -5,6 +5,8 @@
#ifndef SUNSHINE_STREAM_H #ifndef SUNSHINE_STREAM_H
#define SUNSHINE_STREAM_H #define SUNSHINE_STREAM_H
#include <atomic>
#include "crypto.h" #include "crypto.h"
#include "thread_safe.h" #include "thread_safe.h"
namespace stream { namespace stream {
@ -12,10 +14,11 @@ namespace stream {
struct launch_session_t { struct launch_session_t {
crypto::aes_t gcm_key; crypto::aes_t gcm_key;
crypto::aes_t iv; crypto::aes_t iv;
std::string app_name;
}; };
extern safe::event_t<launch_session_t> launch_event; extern safe::event_t<launch_session_t> launch_event;
extern std::atomic_bool has_session;
void rtpThread(); void rtpThread();
} }

View File

@ -165,8 +165,6 @@ void encodeThread(
yuv_frame->pict_type = AV_PICTURE_TYPE_NONE; yuv_frame->pict_type = AV_PICTURE_TYPE_NONE;
} }
packets->stop();
} }
void capture_display(packet_queue_t packets, idr_event_t idr_events, config_t config) { void capture_display(packet_queue_t packets, idr_event_t idr_events, config_t config) {