RTSP.cpp

#include "RTSP.h"
 
#include "Foundation/Logging/Logger.h"
#include "Foundation/Network/Address.h"
#include "Foundation/Network/Network.h"
#include "Foundation/Platform/Timer.h"
 
#include <base64.h>
 
#include <regex>
#include <sstream>
#include <cstring>
 
#include <curl/curl.h>
 
namespace {
  Cogs::Logging::Log logger = Cogs::Logging::getLogger("RTSP");
}
 
Cogs::RTSP::RTSP(std::string_view url, std::string_view userAgent):
  originalURL(url),
  has_describe(false),
  has_setup(false),
  has_play(false),
  has_pause(false),
  has_teardown(false),
  has_set_parameter(false),
  has_get_parameter(false),
  is_active(false),
  agentString(userAgent)
{
  incomingBuffer.resize(4096, false);
 
  enableAutoReconnect();
  setConnectTimeout(4000);
  breakdownURL(url);
 
  Cogs::Network::addConnection(this);
 
  sendOptions();
  sendDescribe();
}
 
Cogs::RTSP::~RTSP() {
  sendTeardown();
}
 
bool Cogs::RTSP::sendSetup(MediaDescription& mdt, const std::string& mediaURL, uint16_t client_port_a, uint16_t client_port_b) {
  if (has_setup) {
    Message*  message = new Message();
 
    message->userData = &mdt;
    message->firstLine = "SETUP " + mediaURL + " RTSP/1.0";
    message->headers["Transport"] = "RTP/AVP;unicast;client_port=" + std::to_string(client_port_a) + "-" + std::to_string(client_port_b);
 
    addStandardHeaders(*message);
    queueMessage(message);
    return true;
  }
  return false;
}
 
bool Cogs::RTSP::sendPlay() {
  if (has_play) {
    Message*  message = new Message();
 
    message->firstLine = buildURL("PLAY");
 
    addStandardHeaders(*message);
    queueMessage(message);
    return true;
  }
  return false;
}
 
bool Cogs::RTSP::sendPause() {
  if (has_pause) {
    Message*  message = new Message();
 
    message->firstLine = buildURL("PAUSE");
 
    addStandardHeaders(*message);
    queueMessage(message);
    return true;
  }
  return false;
}
 
bool Cogs::RTSP::sendTeardown() {
  if (has_teardown) {
    Message*  message = new Message();
 
    message->firstLine = buildURL("TEARDOWN");
 
    addStandardHeaders(*message);
    queueMessage(message);
    return true;
  }
  return false;
}
 
void Cogs::RTSP::parseSDP(const std::string& sdp) {
  std::stringstream ss(sdp);
  std::string       str;
  uint8_t           ip[4] = {};
 
  mediaDescriptions.clear();
  has_video = false;
  has_audio = false;
 
  while (std::getline(ss, str, '\n')) {
    switch (str[0]) {
      case 'v': {
        break;
      }
      case 'o': {
        std::smatch m;
        std::regex r("IN IP4 ([0-9]*).([0-9]*).([0-9]*).([0-9]*)");
 
        if (std::regex_search(str, m, r)) {
          ip[0] = static_cast<uint8_t>(std::atoi(m[1].str().c_str()));
          ip[1] = static_cast<uint8_t>(std::atoi(m[2].str().c_str()));
          ip[2] = static_cast<uint8_t>(std::atoi(m[3].str().c_str()));
          ip[3] = static_cast<uint8_t>(std::atoi(m[4].str().c_str()));
        }
        break;
      }
      case 's':
      case 'i':
      case 'c':
      case 'b':
      case 't': {
        break;
      }
      case 'm': {
        MediaDescription  desc = {};
        std::smatch       m;
 
        desc.media_name = str.substr(2);
 
        if (std::regex_search(str, m, std::regex("([a-z]*) ([0-9]*) RTP/AVP ([0-9]*)"))) {
          desc.media_type = m[1].str();
 
          if (desc.media_type == "video") {
            has_video = true;
          }
          else if (desc.media_type == "audio") {
            has_audio = true;
          }
          desc.rtp_payload_type = std::atoi(m[3].str().c_str());
        }
        desc.ip[0] = ip[0];
        desc.ip[1] = ip[1];
        desc.ip[2] = ip[2];
        desc.ip[3] = ip[3];
        mediaDescriptions.push_back(desc);
        break;
      }
      case 'a': {
        if (!mediaDescriptions.empty()) {
          MediaDescription& desc = mediaDescriptions.back();
 
          if (str.find("rtpmap") == 2) {
            std::smatch m;
 
            if (std::regex_search(str, m, std::regex("rtpmap:([0-9]*) ([A-Za-z0-9-]*)/([0-9]*)/?([A-Za-z0-9-]*)"))) {
              uint32_t type = std::atoi(m[1].str().c_str());
 
              assert(type == desc.rtp_payload_type);
 
              desc.encoding_name = m[2].str();
              desc.rtp_clock_rate = std::atoi(m[3].str().c_str());
            }
            else {
              assert(false);
            }
          }
          else if (str.find("control") == 2) {
            std::smatch m;
 
            if (std::regex_search(str, m, std::regex("control:([0-9a-zA-Z=:/\\.\\-\\?]*)"))) {
              desc.control_id = m[1].str();
            }
          }
          else if (str.find("fmtp") == 2) {
            std::smatch m;
 
            if (std::regex_search(str, m, std::regex("fmtp:([0-9]*)"))) {
              uint32_t type = std::atoi(m[1].str().c_str());
              assert(type == desc.rtp_payload_type);
            }
            if (std::regex_search(str, m, std::regex("profile-level-id=([A-fa-f0-9]*)"))) {
              desc.profile_level_id = std::stol(m[1].str().c_str(), 0, 16);
            }
            if (std::regex_search(str, m, std::regex("packetization-mode=([0-9]*)"))) {
              desc.packetization_mode = std::atoi(m[1].str().c_str());
            }
            if (std::regex_search(str, m, std::regex("sprop-parameter-sets=([A-Za-z0-9+/]*)[=]*,[=]*([A-Za-z0-9+/]*)[=]*"))) {
              desc.sps = base64_decode(m[1].str());
              desc.pps = base64_decode(m[2].str());
            }
            if (std::regex_search(str, m, std::regex("sprop-vps=([A-Za-z0-9+/]*)[=]*"))) {
              desc.vps = base64_decode(m[1].str());
            }
            if (std::regex_search(str, m, std::regex("sprop-sps=([A-Za-z0-9+/]*)[=]*"))) {
              desc.sps = base64_decode(m[1].str());
            }
            if(std::regex_search(str, m, std::regex("sprop-pps=([A-Za-z0-9+/]*)[=]*"))){
              desc.pps = base64_decode(m[1].str());
            }
            if(std::regex_search(str, m, std::regex("sprop-max-don-diff"))){
              assert(false); // TODO: Add support for this in RTP.cpp
            }
            // TODO tx-mode=SRST;profile-id=1;level-id=186;tier-flag=0;profile-space=0;
          }
          else if(str.find("framerate") == 2){
            // TODO: a=framerate:25.000000
          }
          else if(str.find("transform") == 2){
            // TODO: a=transform:-1.000000,0.000000,0.000000;0.000000,-1.000000,0.000000;0.000000,0.000000,1.000000
          }
        }
        break;
      }
    }
  }
}
 
bool Cogs::RTSP::processOutgoing() {
  if (isConnected()) {
    if (!outgoingBuffer.unreadSize()) {
      Cogs::LockGuard lock(queueMutex);
 
      if (!queuedMessages.empty()) {
        Message*  message = queuedMessages.front();
 
        queuedMessages.pop_front();
 
        outgoingBuffer.clear();
        outgoingBuffer.write(message->firstLine.data(), message->firstLine.size());
        outgoingBuffer.write("\x0D\x0A", 2);
        for (auto& [key,value] : message->headers) {
          outgoingBuffer.write(key.data(), key.size());
          outgoingBuffer.write(": ", 2);
          outgoingBuffer.write(value.data(), value.size());
          outgoingBuffer.write("\x0D\x0A", 2);
        }
        outgoingBuffer.write("\x0D\x0A", 2);
        outgoingBuffer.write(message->data(), message->size());
        outgoingBuffer.seek(0, Memory::MemoryBuffer::Anchor::Start);
 
        sentMessages.push_back(message);
      }
    }
    if (outgoingBuffer.unreadSize()) {
      uint64_t  bytesSent = 0;
 
      if (send(outgoingBuffer.dataFromPos(), outgoingBuffer.unreadSize(), bytesSent)) {
        outgoingBuffer.seek(bytesSent, Memory::MemoryBuffer::Anchor::Current);
      }
    }
    if (keepAliveTime) {
      int64_t now = Timer::currentTimeMilliseconds();
 
      if (keepAliveTime < now) {
        Message*  message = new Message();
 
        if (has_set_parameter) {
          message->firstLine = buildURL("SET_PARAMETER");
        }
        else if (has_get_parameter) {
          message->firstLine = buildURL("GET_PARAMETER");
        }
        else {
          message->firstLine = buildURL("OPTIONS");
        }
 
        addStandardHeaders(*message);
        queueMessage(message);
        keepAliveTime = now + timeout - 1000;
        return true;
      }
    }
  }
  return false;
}
 
bool Cogs::RTSP::processIncoming() {
  uint64_t  remainingSpace = incomingBuffer.unreadSize() - 1;
  uint64_t  bytesReceived = incomingBuffer.unreadSize();
  char*     bufferStart = static_cast<char*>( incomingBuffer.data() );
 
  if (receive(bufferStart + incomingBuffer.pos(), remainingSpace, bytesReceived)) {
    incomingBuffer.seek(bytesReceived, Cogs::Memory::MemoryBuffer::Anchor::Current);
    *static_cast<char*>(incomingBuffer.dataFromPos()) = 0;
  }
  if (!incomingMessage.finishedHeaders) {
    char* lineStart = bufferStart;
 
    for( ;; ) {
      char* lineEnd = std::strstr(lineStart, "\x0D\x0A");
 
      if( lineEnd ) {
        *lineEnd = 0;
 
        if (lineStart == lineEnd) {
          incomingMessage.finishedHeaders = true;
          if (!incomingMessage.size()) {
            incomingMessage.finishedBody = true;
          }
          lineStart = lineEnd + 2;
          break;
        }
        else {
          if (incomingMessage.firstLine.empty()) {
            incomingMessage.firstLine = lineStart;
          }
          else {
            char* colon = strchr(lineStart, ':');
            if (colon) {
              *colon++ = 0;
              incomingMessage.headers[lineStart] = ++colon;
 
              if (!strcmp(lineStart, "Content-Length")) {
                                  incomingMessage.resize(std::atoi(colon));
              }
            }
          }
        }
        lineStart = lineEnd + 2;
      }
      else {
        break;
      }
    }
    uint64_t  remainingData = (static_cast<char*>(incomingBuffer.data()) + incomingBuffer.size()) - lineStart;
 
    memmove(incomingBuffer.data(), lineStart, remainingData);
    incomingBuffer.seek(static_cast<char*>(incomingBuffer.data()) - lineStart, Cogs::Memory::MemoryBuffer::Anchor::Current);
  }
  if (!incomingMessage.finishedBody) {
    memcpy(incomingMessage.dataFromPos(), incomingBuffer.data(), incomingBuffer.pos());
    incomingMessage.seek(incomingBuffer.pos(), Cogs::Memory::MemoryBuffer::Anchor::Current);
    incomingBuffer.seek(0, Cogs::Memory::MemoryBuffer::Anchor::Start);
 
    if (incomingMessage.pos() == incomingMessage.size()) {
      incomingMessage.finishedBody = true;
    }
  }
  if (incomingMessage.finishedBody) {
    incomingMessage.seek(0, Cogs::Memory::MemoryBuffer::Anchor::Start);
    processIncomingMessage(incomingMessage);
 
    incomingMessage.reset();
  }
  return true;
}
 
void Cogs::RTSP::processIncomingMessage(Message& message) {
  std::string seq = message.headers["CSeq"];
 
  // We don't support these (optional) requests from the server...
  if (message.firstLine.starts_with("OPTIONS") ||
      message.firstLine.starts_with("GET_PARAMETER" )) {
    Message*  response = new Message();
 
    // Do NOT add standard headers here, because we need the original sequence number.
    response->firstLine = "RTSP/1.0 501 Not Implemented";
    response->headers["CSeq"] = seq;
    response->headers["User-Agent"] = agentString;
    queueMessage(response);
    return;
  }
 
  LOG_INFO(logger, "Received: %s", message.firstLine.c_str());
 
  Message*  request = nullptr;
 
  // Find the request we sent that matches the sequence number of the incoming message.
  for (auto i = sentMessages.begin(), e = sentMessages.end(); i != e; ++i) {
    if ((*i)->headers["CSeq"] == seq) {
      request = *i;
      sentMessages.erase(i);
      break;
    }
  }
  if (request) {
    if (message.firstLine.starts_with("RTSP/1.0")) {
      switch (std::atoi(message.firstLine.c_str() + 9)) {
        case 301:     // Moved Permanently
        case 302:     // Found
        case 305: {   // Use Proxy
          breakdownURL(message.headers["Location"]);
 
          request->firstLine = buildURL(request->firstLine.substr(0, request->firstLine.find(' ')));
          request->headers["CSeq"] = std::to_string(sequenceNumber++);
          queueMessage(request);
          return;
        }
        case 401: {   // Authorisation Needed.
          // The original request should be retried with a new sequence number and the username/password
          // encoded per the server's requirements as an "Authorization" header.
          assert(false);
          break;
        }
      }
    }
 
    if (request->firstLine.starts_with("OPTIONS")) {
      auto  i = message.headers.find("Public");
 
      if (i != message.headers.end()) {
        has_describe = i->second.find("DESCRIBE") != std::string::npos;
        has_setup = i->second.find("SETUP") != std::string::npos;
        has_play = i->second.find("PLAY") != std::string::npos;
        has_pause = i->second.find("PAUSE") != std::string::npos;
        has_teardown = i->second.find("TEARDOWN") != std::string::npos;
        has_set_parameter = i->second.find("SET_PARAMETER") != std::string::npos;
        has_get_parameter = i->second.find("GET_PARAMETER") != std::string::npos;
      }
    }
    else if (request->firstLine.starts_with("DESCRIBE")) {
      parseSDP(std::string(static_cast<char*>(message.data()), message.size()));
      receivedMediaDescriptions();
    }
    else if (request->firstLine.starts_with("SETUP")) {
      sessionID = message.headers["Session"];
 
      MediaDescription* md = static_cast<MediaDescription*>(request->userData);
      size_t            semicolon = sessionID.find(";timeout=");
      std::string&      s = message.headers["Transport"];
      std::smatch       m;
 
      if (semicolon != std::string::npos) {
        timeout = std::atoi(sessionID.c_str() + semicolon + 9) * 1000;
        sessionID.erase(semicolon);
      }
      keepAliveTime = Timer::currentTimeMilliseconds() + timeout - 1000;
 
      if (std::regex_search(s, m, std::regex("server_port=([0-9]+)-([0-9]+)"))) {
        md->server_port[0] = static_cast<uint16_t>(std::stoi(m[1].str()));
        md->server_port[1] = static_cast<uint16_t>(std::stoi(m[2].str()));
      }
      if (std::regex_search(s, m, std::regex("client_port=([0-9]+)-([0-9]+)"))) {
        md->client_port[0] = static_cast<uint16_t>(std::stoi(m[1].str()));
        md->client_port[1] = static_cast<uint16_t>(std::stoi(m[2].str()));
      }
      if (std::regex_search(s, m, std::regex("ssrc=([A-Fa-f0-9]+)"))) {
        md->ssrc = std::stoul(m[1].str(), 0, 16);
      }
      receivedStreamSetup(*md);
    }
    else if (request->firstLine.starts_with("TEARDOWN")) {
      sessionID.clear();
      timeout = 60000;
      keepAliveTime = 0;
    }
    delete request;
  }
}
 
void Cogs::RTSP::queueMessage(Message* message) {
  if (message) {
    Cogs::LockGuard lock(queueMutex);
 
    LOG_INFO(logger, "Sending %s.", message->firstLine.c_str());
    queuedMessages.push_back(message);
  }
}
 
void Cogs::RTSP::addStandardHeaders(Message& message) {
  message.headers["CSeq"] = std::to_string(sequenceNumber++);
  message.headers["User-Agent"] = agentString;
 
  if (!sessionID.empty()) {
    message.headers["Session"] = sessionID;
  }
}
 
bool Cogs::RTSP::breakdownURL(std::string_view url) {
  std::string oldHostname = hostname;
  uint16_t    oldPort = hostPort;
  size_t      start = (url.starts_with("rtsp://")) ? 7 : 0;
  size_t      end = url.find('/', start);
 
  hostname = std::string(url.substr(start, end - start));
  if (end != std::string::npos) {
    resourcePath = std::string(url.substr(end + 1));
  }
 
  end = hostname.find('@');
  if (end != std::string::npos) {
    userInfo = std::string(hostname.substr(0, end++));
    hostname.erase(0, end);
  }
  start = hostname.find(':');
  if (start != std::string::npos) {
    hostPort = static_cast<uint16_t>(std::atoi(hostname.c_str() + start + 1));
    hostname.erase(start);
  }
  else {
    hostPort = 554;
  }
  if ((hostname != oldHostname) || (hostPort != oldPort)) {
    sockAddr = Network::SockaddrIn(Network::AddrIn(hostname), hostPort);
    return true;
  }
  return false;
}
 
std::string Cogs::RTSP::buildURL(std::string_view method) {
  std::string url;
 
  url.reserve(400);
  url = method;
  url += " rtsp://";
 
  if (!userInfo.empty()) {
    url += userInfo;
    url += "@";
  }
  url += hostname;
  url += "/";
 
  if (!resourcePath.empty()) {
    url += resourcePath;
  }
  url += " RTSP/1.0";
 
  return url;
}
 
void Cogs::RTSP::sendOptions() {
  Message*  message = new Message();
 
  message->firstLine = buildURL("OPTIONS");
 
  addStandardHeaders(*message);
  queueMessage(message);
}
 
void Cogs::RTSP::sendDescribe() {
  Message*  message = new Message();
 
  message->firstLine = buildURL("DESCRIBE");
  message->headers["Accept"] = "application/sdp";
 
  addStandardHeaders(*message);
  queueMessage(message);
}