winamp/Src/Plugins/DSP/sc_serv3/protocol_relay.cpp
2024-09-24 14:54:57 +02:00

544 lines
18 KiB
C++

#ifdef _WIN32
#include <winsock2.h>
#endif
#include <assert.h>
#include "protocol_relay.h"
#include "protocol_backup.h"
#include "protocol_relay_shoutcast.h"
#include "protocol_relay_uvox.h"
#include "bandwidth.h"
#include "services/stdServiceImpl.h"
using namespace std;
using namespace stringUtil;
using namespace uniString;
// g_streamSourceRelayIsActive
// This is a map of sid to int, where the int is a bitmap of flags, the bits being
// 0 - registered, some runnable is working with this.
// 1 - shutdown of relay requested.
// 2 - primary relay in use
// 3 - backup runnable started
#define LOGNAME "RELAY"
#define DEBUG_LOG(...) do { if (gOptions.relayDebug()) DLOG(__VA_ARGS__); } while (0)
protocol_relay::protocol_relay(const config::streamConfig &info, const bool retry) throw()
: m_redirectCount(0), m_outBuffer(0), m_outBufferSize(0), m_relayWaitingToReconnect(false),
m_relaySentConnectWait(false), m_retryRelay(retry), m_skip(false),
m_backupStarted(false), m_registered(false), m_relayInfo(info),
m_originalRelayInfo(info), m_retryCount(0), m_relayReconnectStartTime(::time(NULL)),
m_state(&protocol_relay::state_Initial), m_nextState(0)
{
streamData::setRelayActive (info.m_streamID, -1); // make sure entry exists
m_srcPort = m_relayInfo.m_relayUrl.port();
m_srcAddrName = m_relayInfo.m_relayUrl.server();
m_srcURLpart = m_relayInfo.m_relayUrl.path();
m_srcLogString = "[RELAY " + m_srcAddrName + ":" + tos(m_srcPort) +
(m_srcURLpart == "/" ? "" : m_srcURLpart) +
" sid=" + tos(m_relayInfo.m_streamID) + "] ";
DEBUG_LOG(m_srcLogString + __FUNCTION__, LOGNAME, m_originalRelayInfo.m_streamID);
}
protocol_relay::~protocol_relay() throw()
{
DEBUG_LOG(m_srcLogString + __FUNCTION__, LOGNAME, m_originalRelayInfo.m_streamID);
socketOps::forgetTCPSocket(m_socket);
if (m_registered)
{
bool noEntry = false;
streamData::setRelayActiveFlags (m_originalRelayInfo.m_streamID, noEntry, 0, 7);
}
}
void protocol_relay::timeSlice() throw(exception)
{
if (iskilled())
{
m_result.done();
return;
}
// normal running
try
{
if (m_registered && m_relayWaitingToReconnect && (::time(NULL) < m_relayReconnectStartTime))
{
// an error occured in the past and we are waiting for the appropriate time interval before we do anything
bool noEntry = false;
int status = streamData::isRelayActive(m_originalRelayInfo.m_streamID, noEntry);
if ((status & 2) == 0) // if no shutdown req
{
m_result.schedule(400);
return;
}
DEBUG_LOG (m_srcLogString + "relay shutdown req", LOGNAME, m_originalRelayInfo.m_streamID);
throwEx<runtime_error>("");
}
(this->*m_state)();
}
catch(const exception &ex)
{
// close socket and move into waiting state for reconnect
socketOps::forgetTCPSocket(m_socket);
// but first see if we've hit the retry limit (which can be set as zero to keep on going)
++m_retryCount;
int retryLimit = gOptions.relayConnectRetries();
bool noEntry = false;
DEBUG_LOG(m_srcLogString + __FUNCTION__ + utf8(" m_retryCount:") + tos(m_retryCount) + " retryLimit:" + tos(retryLimit), LOGNAME, m_originalRelayInfo.m_streamID);
int status = streamData::setRelayActiveFlags (m_originalRelayInfo.m_streamID, noEntry, 0, 4);
#if defined(_DEBUG) || defined(DEBUG)
bool relayActive = status & 4 ? true : false;
DLOG(m_srcLogString + __FUNCTION__ + " a: " + tos(retryLimit) +
" " + tos(m_retryCount) + " "+tos(relayActive) + " " +
tos(m_retryCount < retryLimit && retryLimit > 0) + " " +
tos(((m_retryCount < retryLimit && retryLimit > 0)) && relayActive),LOGNAME, m_originalRelayInfo.m_streamID);
#endif
if (noEntry == false && (status & 2) == 0) // is relay still configured
{
if ((status & 8) == 0)
startBackupConnection (""); // kick backup off if present, in case of repeated failure
utf8 msg = ex.what();
if (!msg.empty())
{
ELOG (ex.what(), LOGNAME, m_originalRelayInfo.m_streamID);
}
if (retryLimit == 0 || (m_retryCount < retryLimit && retryLimit > 0))
{
m_relayWaitingToReconnect = true;
m_state = &protocol_relay::state_Initial;
m_relayInfo = m_originalRelayInfo;
m_relayReconnectStartTime = ::time(NULL) + gOptions.relayReconnectTime();
m_result.schedule (550);
DLOG (m_srcLogString + "reconnect time " + tos ((long)m_relayReconnectStartTime), LOGNAME, m_originalRelayInfo.m_streamID);
return;
}
}
else
{
utf8 ss = m_srcLogString + "Abort trying to ";
if (m_retryCount > 0)
ss += "re";
ss += "connect to the source relay";
ILOG (ss, LOGNAME, m_originalRelayInfo.m_streamID);
}
m_result.done();
}
}
// parse out relayInfo object
void protocol_relay::state_Initial() throw(exception)
{
DEBUG_LOG(m_srcLogString + __FUNCTION__, LOGNAME, m_originalRelayInfo.m_streamID);
// if we have a moved stream then now we have the stream id
// then we need to check and block the source as applicable
utf8 movedUrl = gOptions.stream_movedUrl(m_relayInfo.m_streamID);
if (!movedUrl.empty())
{
ELOG (m_srcLogString + "Relay connection aborted. Stream is configured as having moved.", LOGNAME, m_originalRelayInfo.m_streamID);
m_result.done();
return;
}
bool noEntry = false;
if (m_registered == false)
{
int state = streamData::setRelayActiveFlags (m_originalRelayInfo.m_streamID, noEntry, 1);
if (state < 0)
{
ILOG (m_srcLogString + "waiting on other relay termination", LOGNAME, m_originalRelayInfo.m_streamID);
streamData::setRelayActiveFlags (m_originalRelayInfo.m_streamID, noEntry, 2); // request shutdown of other relay
m_result.schedule (100);
return;
}
m_registered = true;
}
int state = streamData::setRelayActiveFlags (m_originalRelayInfo.m_streamID, noEntry, 4);
if (noEntry || (state & 2) == 2)
{
ILOG (m_srcLogString + "relay shutting down", LOGNAME, m_originalRelayInfo.m_streamID);
m_result.done();
return;
}
// if m_retryCount is over 1 then pull the stream relay url from
// the config details as we could be in a re-try or having done
// a config reload and the stream relay url has then sinc changed
// - this primarily aids a config change for a pending relay join
if (m_retryCount > 0)
{
m_relayInfo.m_relayUrl = gOptions.stream_relayURL(m_originalRelayInfo.m_streamID);
}
m_relayWaitingToReconnect = false;
m_srcAddrName = m_relayInfo.m_relayUrl.server();
m_srcPort = m_relayInfo.m_relayUrl.port();
m_srcURLpart = m_relayInfo.m_relayUrl.path();
m_srcLogString = "[RELAY " + m_srcAddrName + ":" + tos(m_srcPort) +
(m_srcURLpart == "/" ? "" : m_srcURLpart) +
" sid=" + tos(m_relayInfo.m_streamID) + "] ";
m_state = &protocol_relay::state_ResolveServer;
m_result.run();
}
// resolve server name to numeric address
void protocol_relay::state_ResolveServer() throw(exception)
{
DEBUG_LOG(m_srcLogString + __FUNCTION__, LOGNAME, m_originalRelayInfo.m_streamID);
assert(m_socket == socketOps::cINVALID_SOCKET);
m_socket = socketOps::createTCPSocketTHROW();
socketOps::setNonblock(m_socket, true);
m_srcAddrNumeric = socketOps::hostNameToAddress(m_srcAddrName.hideAsString(), m_srcPort);
if (m_srcAddrNumeric.empty())
{
ELOG (m_srcLogString + "Could not resolve host address", LOGNAME, m_originalRelayInfo.m_streamID);
throwEx<runtime_error>("");
}
m_state = &protocol_relay::state_Connect;
m_result.run();
}
// TCP connect
void protocol_relay::state_Connect() throw(exception)
{
DEBUG_LOG(m_srcLogString + __FUNCTION__, LOGNAME, m_originalRelayInfo.m_streamID);
assert(m_socket != socketOps::cINVALID_SOCKET);
if (!m_skip)
{
utf8 ss = m_srcLogString;
if (m_retryCount > 0)
{
ss += "Trying to restore connection to source relay [attempt #";
ss += tos(m_retryCount+1) + "]";
}
else
ss += "Connecting to source relay";
ILOG (ss, LOGNAME, m_originalRelayInfo.m_streamID);
}
m_skip = false;
socketOps::connectTHROW (m_socket, m_srcAddrNumeric, m_srcPort);
m_lastActivityTime = ::time(NULL);
m_state = &protocol_relay::state_ConnectWait;
m_result.run();
}
// wait for connect to complete
void protocol_relay::state_ConnectWait() throw(exception)
{
DEBUG_LOG(m_srcLogString + __FUNCTION__, LOGNAME, m_originalRelayInfo.m_streamID);
assert(m_socket != socketOps::cINVALID_SOCKET);
time_t cur_time;
const int autoDumpSourceTime = detectAutoDumpTimeout (cur_time, m_originalRelayInfo.m_streamID, (m_srcLogString + "Timeout trying to connect"));
bool noEntry = false, relayActive = ((streamData::isRelayActive(m_originalRelayInfo.m_streamID, noEntry) & 6) == 4);
if (!relayActive && !noEntry)
{
throwEx<runtime_error>("");
}
string error;
socketOps::nonBlockConnect_t connectResult = socketOps::nonBlockingConnectWait(m_socket, error);
switch (connectResult)
{
case socketOps::NBC_ERROR:
{
ELOG (m_srcLogString + error, LOGNAME, m_originalRelayInfo.m_streamID);
throwEx<runtime_error>("");
break;
}
case socketOps::NBC_INPROGRESS:
{
// try again but wait a bit
m_result.schedule(100);
m_result.timeout((autoDumpSourceTime - (int)(cur_time - m_lastActivityTime)));
break;
}
case socketOps::NBC_CONNECTED:
{
m_lastActivityTime = ::time(NULL);
m_state = &protocol_relay::state_SendGreeting;
m_result.run();
break;
}
default:
{
ELOG (m_srcLogString + "Unknown non-blocking connect state.", LOGNAME, m_originalRelayInfo.m_streamID);
throwEx<runtime_error>("");
}
}
}
void protocol_relay::state_SendGreeting() throw(exception)
{
DEBUG_LOG(m_srcLogString + __FUNCTION__, LOGNAME, m_originalRelayInfo.m_streamID);
m_HTTPHeaders.clear();
m_HTTPGreetingResponse.clear();
string cdn;
if (isCDNSlave(m_relayInfo.m_streamID))
{
cdn = "cdn-slave:1\r\n";
}
m_lineBuffer = "GET " + m_srcURLpart + " " + "HTTP/1.1\r\n" +
"Host:" + stripHTTPprefix(m_srcAddrName) + ":" + tos(m_srcPort) + "\r\n" +
"User-Agent:" + g_userAgent + " Relay\r\n" +
"Ultravox transport type:TCP\r\n" +
"Accept:*/*\r\n" +
"icy-metadata:1\r\n" +
cdn +
"icy-host:" + metrics::metrics_verifyDestIP(gOptions) + "\r\n\r\n";
DEBUG_LOG(m_srcLogString + "Sending request [" + eol() + stripWhitespace(m_lineBuffer) + eol() + "]", LOGNAME, m_originalRelayInfo.m_streamID);
m_outBuffer = &(m_lineBuffer[0]);
bandWidth::updateAmount(bandWidth::RELAY_MISC_RECV, (m_outBufferSize = (int)m_lineBuffer.size()));
m_lastActivityTime = ::time(NULL);
m_state = &protocol_relay::state_Send;
m_nextState = &protocol_relay::state_GetGreetingResponse;
m_result.write();
m_result.timeoutSID(m_originalRelayInfo.m_streamID);
}
// send whatever is in outBuffer
void protocol_relay::state_Send() throw(exception)
{
DEBUG_LOG(m_srcLogString + __FUNCTION__, LOGNAME, m_originalRelayInfo.m_streamID);
if (sendDataBuffer(m_relayInfo.m_streamID, m_outBuffer, m_outBufferSize, m_srcLogString))
{
m_state = m_nextState;
m_lineBuffer.clear();
}
}
void protocol_relay::state_GetLine() throw(exception)
{
DEBUG_LOG(m_srcLogString + __FUNCTION__, LOGNAME, m_originalRelayInfo.m_streamID);
if (getHTTPStyleHeaderLine(m_relayInfo.m_streamID, m_lineBuffer, m_srcLogString))
{
m_state = m_nextState;
}
}
void protocol_relay::state_GetGreetingResponse() throw(exception)
{
DEBUG_LOG(m_srcLogString + __FUNCTION__, LOGNAME, m_originalRelayInfo.m_streamID);
m_state = &protocol_relay::state_GetLine;
m_nextState = &protocol_relay::state_AnalyzeGreetingResponse;
m_lastActivityTime = ::time(NULL);
m_result.read();
m_result.timeoutSID(m_originalRelayInfo.m_streamID);
}
// analyze header lines in greeting response
void protocol_relay::state_AnalyzeGreetingResponse() throw(exception)
{
DEBUG_LOG(m_srcLogString + __FUNCTION__, LOGNAME, m_originalRelayInfo.m_streamID);
m_lastActivityTime = ::time(NULL);
if ((int)m_HTTPHeaders.size() >= gOptions.maxHeaderLineCount())
{
throwEx<runtime_error>(m_srcLogString + "Max HTTP header lines exceeded");
}
m_lineBuffer = stripWhitespace(m_lineBuffer);
if (m_lineBuffer.empty())
{
m_state = &protocol_relay::state_DetermineProtocol;
m_result.run();
}
else
{
if (m_HTTPGreetingResponse.empty())
{
m_HTTPGreetingResponse = m_lineBuffer;
}
else
{
// find the colon that divides header lines into key/value fields
utf8::size_type pos = m_lineBuffer.find(utf8(":"));
if (pos == utf8::npos)
{
throwEx<runtime_error>(m_srcLogString + "Connection rejected. Bad HTTP header string [" + m_lineBuffer + "]");
}
utf8 key = toLower(stripWhitespace(m_lineBuffer.substr(0, pos)));
utf8 value = stripWhitespace(m_lineBuffer.substr(pos + 1));
// allow empty values. (for urls and what-not)
if (key.empty())
{
throwEx<runtime_error>(m_srcLogString + "Connection rejected. Bad HTTP header string [" + m_lineBuffer + "]");
}
m_HTTPHeaders[key] = value;
}
m_state = &protocol_relay::state_GetLine;
m_nextState = &protocol_relay::state_AnalyzeGreetingResponse;
m_result.read();
m_result.timeoutSID(m_originalRelayInfo.m_streamID);
m_lineBuffer.clear();
}
}
void protocol_relay::state_DetermineProtocol() throw(exception)
{
DEBUG_LOG(m_srcLogString + __FUNCTION__, LOGNAME, m_originalRelayInfo.m_streamID);
if (m_HTTPGreetingResponse.empty())
{
throwEx<runtime_error>(m_srcLogString + "Empty greeting response");
}
// parse into three fields
utf8 s = m_HTTPGreetingResponse;
utf8::size_type pos = (!s.empty() ? s.find(utf8(" ")) : utf8::npos);
if (pos == utf8::npos)
{
ELOG (m_srcLogString + "Badly formed response line [" + m_HTTPGreetingResponse + "]", LOGNAME, m_originalRelayInfo.m_streamID);
throwEx<runtime_error>("");
}
s = stripWhitespace(s.substr(pos));
pos = (!s.empty() ? s.find(utf8(" ")) : utf8::npos);
if (pos == utf8::npos)
{
ELOG (m_srcLogString + "Badly formed response line [" + m_HTTPGreetingResponse + "]", LOGNAME, m_originalRelayInfo.m_streamID);
throwEx<runtime_error>("");
}
int resultCode = utf8(s.substr(0, pos)).toInt();
if (resultCode == 200)
{
utf8 diag;
for (httpHeaderMap_t::const_iterator i = m_HTTPHeaders.begin(); i != m_HTTPHeaders.end(); ++i)
{
diag += (*i).first + ": " + (*i).second + eol();
}
DEBUG_LOG(m_srcLogString + "Received headers [" + eol() + stripWhitespace(diag) + eol() + "]", LOGNAME, m_originalRelayInfo.m_streamID);
const socketOps::tSOCKET s = m_socket;
m_result.done();
// changed order of this in build 19 so that all uvox2 is reported as misc/ultravox
// and we then base things off the user-agent containing 'Ultravox/2.1'
// we look at content-type to determine the protocol
if ((m_HTTPHeaders["content-type"] == "misc/ultravox") &&
(m_HTTPHeaders["server"].find(utf8("Ultravox/2.1")) != utf8::npos))
{
// uvox 2.1
threadedRunner::scheduleRunnable(new protocol_relay_uvox(s, m_originalRelayInfo, m_srcAddrName, m_srcAddrNumeric,
m_srcPort, (m_srcURLpart == "/" ? "" : m_srcURLpart), m_HTTPHeaders));
}
else
{
// shoutcast
threadedRunner::scheduleRunnable(new protocol_relay_shoutcast(s, m_originalRelayInfo, m_srcAddrName, m_srcAddrNumeric,
m_srcPort, (m_srcURLpart == "/" ? "" : m_srcURLpart), m_HTTPHeaders));
}
m_socket = socketOps::cINVALID_SOCKET;
m_backupStarted = false; // any backup client should terminate automatically
}
else if ((resultCode >= 300) && (resultCode < 400))
{
utf8 location = m_HTTPHeaders["location"];
// do we maybe have a /stream/x url and gotten a /index.html?sid=# redirect?
// if so then we should attempt to access a relay on /stream/x/ (note end / )
utf8::size_type pos = (!location.empty() ? location.find(utf8("/index.html?sid=")) : utf8::npos);
if (pos != utf8::npos)
{
streamData::streamID_t foundID = atoi((const char *)location.substr(pos + 16).c_str());
utf8::size_type pos2 = m_relayInfo.m_relayUrl.path().find(utf8("/stream/"));
if (pos2 != utf8::npos)
{
streamData::streamID_t origID = atoi((const char *)m_relayInfo.m_relayUrl.path().substr(pos2 + 8).c_str());
if (foundID == origID)
{
m_relayInfo.m_relayUrl = m_relayInfo.m_relayUrl.url() + "/";
gOptions.setOption(utf8("streamrelayurl_"+tos(m_originalRelayInfo.m_streamID)),
m_relayInfo.m_relayUrl.url());
m_skip = true;
m_state = &protocol_relay::state_Initial;
socketOps::forgetTCPSocket(m_socket);
m_result.run();
WLOG(m_srcLogString + "Received an invalid redirect to `" + location + "' - trying " + m_relayInfo.m_relayUrl.url(), LOGNAME, m_originalRelayInfo.m_streamID);
return;
}
}
}
WLOG(m_srcLogString + "Received redirect to " + location, LOGNAME, m_originalRelayInfo.m_streamID);
if (++m_redirectCount > gOptions.maxHTTPRedirects())
{
ELOG ("Max redirects exceeded", LOGNAME, m_originalRelayInfo.m_streamID);
throwEx<runtime_error>("");
}
m_skip = true;
m_relayInfo.m_relayUrl = location;
m_state = &protocol_relay::state_Initial;
socketOps::forgetTCPSocket(m_socket);
m_result.run();
}
else
{
ELOG ((resultCode >= 400 ? "Source responded with error [" : "Unsupported HTTP response code [") + m_HTTPGreetingResponse + "]");
throwEx<runtime_error>("");
}
}
#ifdef INCLUDE_BACKUP_STREAMS
void protocol_relay::startBackupConnection(uniString::utf8 errorMessage) throw(exception)
{
if (!m_relayInfo.m_backupUrl.url().empty() && m_backupStarted == false)
{
if (errorMessage.empty() == false)
ELOG(m_srcLogString + errorMessage, LOGNAME, m_originalRelayInfo.m_streamID);
// this should pass through the bitrate of the original stream but as we do not
// know what it is then we have to effectively force through to allow the backup
bool noEntry = false;
streamData::setRelayActiveFlags (m_originalRelayInfo.m_streamID, noEntry, 8);
threadedRunner::scheduleRunnable(new protocol_backup(m_relayInfo, 0, ""));
m_backupStarted = true;
return;
}
}
#else
void protocol_relay::startBackupConnection(uniString::utf8) throw(exception)
{
}
#endif