hydrar/mistserver
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

RTP cleanup + RTSP analyser + RTP MPEG2/MP2/HEVC support

Browse source
This commit is contained in:
Thulinma 2017-05-21 13:52:14 +02:00
parent f5553f49f9
commit 54a8803f29
9 changed files with 1651 additions and 1216 deletions
Download patch file Download diff file Expand all files Collapse all files

298
src/analysers/analyser_rtsp.cpp
View file

@ -1,197 +1,119 @@
#include <cstdlib>
#include <iostream>
#include <fstream>
#include <string>
#include <string.h>
#include <vector>
#include "analyser_rtsp.h"
#include <mist/config.h>
#include <mist/rtp.h>
#include <mist/socket.h>
#include <mist/http_parser.h>
#include <sstream>
AnalyserRTSP *classPointer = 0;
namespace RtspRtp{
void incomingPacket(const DTSC::Packet &pkt){
classPointer->incoming(pkt);
}
int analyseRtspRtp(std::string rtspUrl){
/*//parse hostname
std::string hostname = rtspUrl.substr(7);
hostname = hostname.substr(0,hostname.find('/'));
std::cout << hostname << std::endl;
HTTP::Parser HTTP_R, HTTP_S;//HTTP Receiver en HTTP Sender.
Socket::Connection conn(hostname,554,false);//setting rtsp connection
bool optionsSent = false;
bool optionsRecvd = false;
bool descSent = false;
bool descRecvd = false;
bool setupComplete = false;
bool playSent = false;
int CSeq = 1;
while(conn.connected()){
if(!optionsSent){
HTTP_R.protocol="RTSP/1.0";
HTTP_R.method = "OPTIONS";
HTTP_R.url = rtspUrl;
HTTP_R.SetHeader("CSeq",CSeq);
CSeq++;
HTTP_R.SetHeader("User-Agent","mistANALyser");
HTTP_R.SendRequest(conn);
optionsSent = true;
}
if (optionsSent&& !optionsRecvd && (conn.Received().size() || conn.spool() )){
if(HTTP_S.Read(conn)){
std::cout << "recv opts" << std::endl;
std::cout << HTTP_S.BuildResponse(HTTP_S.method,HTTP_S.url);
optionsRecvd = true;
}
}
void AnalyserRTSP::init(Util::Config &conf){
Analyser::init(conf);
}
if(optionsRecvd && !descSent){
HTTP_S.Clean();
HTTP_R.protocol="RTSP/1.0";
HTTP_R.method = "DESCRIBE";
HTTP_R.url = rtspUrl;
HTTP_R.SetHeader("CSeq",CSeq);
CSeq++;
HTTP_R.SetHeader("User-Agent","mistANALyser");
HTTP_R.SendRequest(conn);
descSent = true;
}
std::vector<std::string> trackIds;
if (descSent&&!descRecvd && (conn.Received().size() || conn.spool() )){
if(HTTP_S.Read(conn)){
std::cout << "recv desc2" << std::endl;
std::cout << HTTP_S.BuildResponse(HTTP_S.method,HTTP_S.url);
size_t pos = HTTP_S.body.find("m=");
do{
//finding all track IDs
pos = HTTP_S.body.find("a=control:",pos);
if(pos !=std::string::npos){
trackIds.push_back(HTTP_S.body.substr(pos+10,HTTP_S.body.find("\r\n",pos)-pos-10 ) );//setting track IDs;
pos++;
}
}while(pos != std::string::npos);
//we have all the tracks
descRecvd = true;
}
}
unsigned int setupsSent = 0;
unsigned int setupsRecvd = 0;
Socket::UDPConnection connectors[trackIds.size()];
unsigned int setports[trackIds.size()];
uint32_t bport = 10000;
std::string sessionID = "";
std::stringstream setup;
if(descRecvd && !setupComplete){
//time to setup.
for(std::vector<std::string>::iterator it = trackIds.begin();it!=trackIds.end();it++){
std::cout << "setup " << setupsSent<< std::endl;
while(!connectors[setupsSent].SetConnection( bport,false) ){
bport +=2;//finding an available port
}
std::cout << "setup" << bport<< std::endl;
setports[setupsSent] = bport;
bport +=2;
if(setupsSent == setupsRecvd){
//send only one setup
HTTP_S.Clean();
HTTP_R.protocol="RTSP/1.0";
HTTP_R.method = "SETUP";
HTTP_R.url = rtspUrl+ '/' + *(it);
setup << "RTP/AVP/UDP;unicast;client_port="<< setports[setupsSent] <<"-" <<setports[setupsSent]+1 ;
HTTP_R.SetHeader("Transport",setup.str() );
std:: cout << setup.str()<<std::endl;
setup.str(std::string());
setup.clear();
HTTP_R.SetHeader("CSeq",CSeq);
CSeq++;
if(sessionID != ""){
HTTP_R.SetHeader("Session",sessionID);
}
HTTP_R.SetHeader("User-Agent","mistANALyser");
HTTP_R.SendRequest(conn);
setupsSent ++;
}
while(setupsSent == setupsRecvd+1){
//lets Assume we assume we always receive a response
if ( (conn.Received().size() || conn.spool() )){
if(HTTP_S.Read(conn)){
std::cout << "recv setup" << std::endl;
std::cout << HTTP_S.BuildResponse(HTTP_S.method,HTTP_S.url);
optionsRecvd = true;
sessionID = HTTP_S.GetHeader("Session");
setupsRecvd++;
}
}
}
//set up all parameters, and then after the for loop we have to listen to setups and all. sent if both are equal, and recv if one is sent
}
setupComplete = true;
}
if(setupComplete && !playSent){
//time to play
HTTP_S.Clean();
HTTP_R.protocol="RTSP/1.0";
HTTP_R.method = "PLAY";
HTTP_R.url = rtspUrl;
HTTP_R.SetHeader("CSeq",CSeq);
CSeq++;
HTTP_R.SetHeader("User-Agent","mistANALyser");
HTTP_R.SetHeader("Session",sessionID);
HTTP_R.SendRequest(conn);
playSent = true;
std::cout << "sent play" << std::endl;
char buffer[2000];
while(!connectors[0].iread((void*)buffer,2000)) {
std::cout << "buffer";
}
std::cout <<"buffer is not empty" << std::endl;
}
//streams set up
//time to read some packets
if(descRecvd){
conn.close();
}
void AnalyserRTSP::incoming(const DTSC::Packet &pkt){
char *dataPtr;
uint32_t dataSize;
pkt.getString("data", dataPtr, dataSize);
DETAIL_MED("Received %ub %sfor track %lu (%s) @ %llums", dataSize, pkt.getFlag("keyframe")?"keyframe ":"", pkt.getTrackId(),
myMeta.tracks[pkt.getTrackId()].getIdentifier().c_str(), pkt.getTime());
if (detail >= 8){
for (uint32_t i = 0; i < dataSize; ++i){
std::cout << std::hex << std::setw(2) << std::setfill('0') << (int)dataPtr[i] << " ";
if (i % 32 == 31){std::cout << std::endl;}
}
conn.close();*/
return 0;
std::cout << std::endl;
}
}
int main(int argc, char ** argv){
Util::Config conf = Util::Config(argv[0]);
conf.addOption("url",JSON::fromString("{\"arg\":\"string\",\"short\":\"u\",\"long\":\"url\",\"help\":\"URL To get.\", \"default\":\"rtsp://localhost/s1k\"}"));
conf.parseArgs(argc, argv);
return RtspRtp::analyseRtspRtp(conf.getString("url"));
AnalyserRTSP::AnalyserRTSP(Util::Config &conf) : Analyser(conf){
myConn = Socket::Connection(1, 0);
sdpState.myMeta = &myMeta;
sdpState.incomingPacketCallback = incomingPacket;
classPointer = this;
}
bool AnalyserRTSP::isOpen(){
return myConn;
}
bool AnalyserRTSP::parsePacket(){
do{
// No new data? Sleep and retry, if connection still open
if (!myConn.Received().size() || !myConn.Received().available(1)){
if (!myConn.spool() && isOpen()){Util::sleep(500);}
continue;
}
if (myConn.Received().copy(1) != "$"){
// not a TCP RTP packet, read RTSP commands
if (HTTP.Read(myConn)){
if (HTTP.hasHeader("Content-Type") && HTTP.GetHeader("Content-Type") == "application/sdp"){
sdpState.parseSDP(HTTP.body);
HTTP.Clean();
return true;
}
if (HTTP.hasHeader("Transport")){
uint32_t trackNo = sdpState.parseSetup(HTTP, "", "");
if (trackNo){
DETAIL_MED("Parsed transport for track: %lu", trackNo);
}else{
DETAIL_MED("Could not parse transport string!");
}
HTTP.Clean();
return true;
}
std::cout << HTTP.BuildRequest() << std::endl;
HTTP.Clean();
return true;
}else{
if (!myConn.spool() && isOpen()){Util::sleep(500);}
}
continue;
}
if (!myConn.Received().available(4)){
if (!myConn.spool() && isOpen()){Util::sleep(500);}
continue;
}// a TCP RTP packet, but not complete yet
// We have a TCP packet! Read it...
// Format: 1 byte '$', 1 byte channel, 2 bytes len, len bytes binary data
std::string tcpHead = myConn.Received().copy(4);
uint16_t len = ntohs(*(short *)(tcpHead.data() + 2));
if (!myConn.Received().available(len + 4)){
if (!myConn.spool() && isOpen()){Util::sleep(500);}
continue;
}// a TCP RTP packet, but not complete yet
// remove whole packet from buffer, including 4 byte header
std::string tcpPacket = myConn.Received().remove(len + 4);
RTP::Packet pkt(tcpPacket.data() + 4, len);
uint8_t chan = tcpHead.data()[1];
uint32_t trackNo = sdpState.getTrackNoForChannel(chan);
DETAIL_HI("Received %ub RTP packet #%u on channel %u, time %llu", len,
(unsigned int)pkt.getSequence(), chan, pkt.getTimeStamp());
if (!trackNo && (chan % 2) != 1){
DETAIL_MED("Received packet for unknown track number on channel %u", chan);
}
if (trackNo){
sdpState.tracks[trackNo].rtpSeq = pkt.getSequence();
}
if (detail >= 10){
char *pl = pkt.getPayload();
uint32_t payLen = pkt.getPayloadSize();
for (uint32_t i = 0; i < payLen; ++i){
std::cout << std::hex << std::setw(2) << std::setfill('0') << (int)pl[i] << " ";
if (i % 32 == 31){std::cout << std::endl;}
}
std::cout << std::endl;
}
sdpState.handleIncomingRTP(trackNo, pkt);
return true;
}while (isOpen());
// if needed, parse TCP packets, and cancel if it is not safe (yet) to read HTTP/RTSP packets
}

40
src/analysers/analyser_rtsp.h
View file

@ -1,27 +1,23 @@
#include <mist/config.h>
#pragma once
#include "analyser.h"
#include <cstdlib>
#include <iomanip>
#include <vector>
#include <sstream>
#include <mist/socket.h>
#include <mist/rtp.h>
#include <mist/h264.h>
#include <mist/http_parser.h>
#include <mist/rtp.h>
#include <mist/sdp.h>
class rtpAnalyser : public analysers
{
Socket::Connection conn;
HTTP::Parser HTTP_R, HTTP_S;//HTTP Receiver en HTTP Sender.
int step;
std::vector<std::string> tracks;
std::vector<Socket::UDPConnection> connections;
unsigned int trackIt;
class AnalyserRTSP : public Analyser{
public:
AnalyserRTSP(Util::Config &conf);
static void init(Util::Config &cfg);
bool parsePacket();
void incoming(const DTSC::Packet &pkt);
bool isOpen();
public:
rtpAnalyser(Util::Config config);
bool packetReady();
void PreProcessing();
//int Analyse();
int doAnalyse();
void doValidate();
private:
Socket::Connection myConn;
HTTP::Parser HTTP;
DTSC::Meta myMeta;
SDP::State sdpState;
};

793
src/output/output_rtsp.cpp
View file

File diff suppressed because it is too large Load diff

232
src/output/output_rtsp.h
View file

@ -1,214 +1,36 @@
#pragma once
#include "output.h"
#include <mist/socket.h>
#include <mist/rtp.h>
#include <mist/http_parser.h>
#include <mist/encode.h>
#include <mist/h264.h>
#include <mist/http_parser.h>
#include <mist/rtp.h>
#include <mist/sdp.h>
#include <mist/socket.h>
namespace Mist {
///Structure used to keep track of selected tracks.
class RTPTrack {
public:
Socket::UDPConnection data;
Socket::UDPConnection rtcp;
RTP::Packet pack;
long long rtcpSent;
uint64_t firstTime;
int channel;/// Channel number, used in TCP sending
uint64_t packCount;
uint16_t rtpSeq;
std::map<uint16_t, RTP::Packet> packBuffer;
uint32_t cPort;
std::string transportString;
std::string control;
std::string fmtp;
std::string spsData;
std::string ppsData;
uint64_t fpsTime;
double fpsMeta;
double fps;
RTPTrack(){
rtcpSent = 0;
channel = -1;
firstTime = 0;
packCount = 0;
cPort = 0;
rtpSeq = 0;
fpsTime = 0;
fpsMeta = 0;
fps = 0;
}
std::string getParamString(const std::string & param) const{
if (!fmtp.size()){return "";}
size_t pos = fmtp.find(param);
if (pos == std::string::npos){return "";}
pos += param.size()+1;
size_t ePos = fmtp.find_first_of(" ;", pos);
return fmtp.substr(pos, ePos-pos);
}
uint64_t getParamInt(const std::string & param) const{
return atoll(getParamString(param).c_str());
}
std::string mediaDescription(const DTSC::Track & trk){
std::stringstream mediaDesc;
if (trk.codec == "H264") {
MP4::AVCC avccbox;
avccbox.setPayload(trk.init);
mediaDesc << "m=video 0 RTP/AVP 97\r\n"
"a=rtpmap:97 H264/90000\r\n"
"a=cliprect:0,0," << trk.height << "," << trk.width << "\r\n"
"a=framesize:97 " << trk.width << '-' << trk.height << "\r\n"
"a=fmtp:97 packetization-mode=1;profile-level-id="
<< std::hex << std::setw(2) << std::setfill('0') << (int)trk.init.data()[1] << std::dec << "E0"
<< std::hex << std::setw(2) << std::setfill('0') << (int)trk.init.data()[3] << std::dec << ";"
"sprop-parameter-sets="
<< Encodings::Base64::encode(std::string(avccbox.getSPS(), avccbox.getSPSLen()))
<< ","
<< Encodings::Base64::encode(std::string(avccbox.getPPS(), avccbox.getPPSLen()))
<< "\r\n"
"a=framerate:" << ((double)trk.fpks)/1000.0 << "\r\n"
"a=control:track" << trk.trackID << "\r\n";
} else if (trk.codec == "AAC") {
mediaDesc << "m=audio 0 RTP/AVP 96" << "\r\n"
"a=rtpmap:96 mpeg4-generic/" << trk.rate << "/" << trk.channels << "\r\n"
"a=fmtp:96 streamtype=5; profile-level-id=15; config=";
for (unsigned int i = 0; i < trk.init.size(); i++) {
mediaDesc << std::hex << std::setw(2) << std::setfill('0') << (int)trk.init[i] << std::dec;
}
//these values are described in RFC 3640
mediaDesc << "; mode=AAC-hbr; SizeLength=13; IndexLength=3; IndexDeltaLength=3;\r\n"
"a=control:track" << trk.trackID << "\r\n";
}else if (trk.codec == "MP3") {
mediaDesc << "m=" << trk.type << " 0 RTP/AVP 14" << "\r\n"
"a=rtpmap:14 MPA/90000/" << trk.channels << "\r\n"
"a=control:track" << trk.trackID << "\r\n";
}else if ( trk.codec == "AC3") {
mediaDesc << "m=audio 0 RTP/AVP 100" << "\r\n"
"a=rtpmap:100 AC3/" << trk.rate << "/" << trk.channels << "\r\n"
"a=control:track" << trk.trackID << "\r\n";
}else if ( trk.codec == "ALAW") {
if (trk.channels == 1 && trk.rate == 8000){
mediaDesc << "m=audio 0 RTP/AVP 8" << "\r\n";
}else{
mediaDesc << "m=audio 0 RTP/AVP 101" << "\r\n";
mediaDesc << "a=rtpmap:101 PCMA/" << trk.rate << "/" << trk.channels << "\r\n";
}
mediaDesc << "a=control:track" << trk.trackID << "\r\n";
}else if ( trk.codec == "PCM") {
if (trk.size == 16 && trk.channels == 2 && trk.rate == 44100){
mediaDesc << "m=audio 0 RTP/AVP 10" << "\r\n";
} else if (trk.size == 16 && trk.channels == 1 && trk.rate == 44100){
mediaDesc << "m=audio 0 RTP/AVP 11" << "\r\n";
}else{
mediaDesc << "m=audio 0 RTP/AVP 103" << "\r\n";
mediaDesc << "a=rtpmap:103 L" << trk.size << "/" << trk.rate << "/" << trk.channels << "\r\n";
}
mediaDesc << "a=control:track" << trk.trackID << "\r\n";
}else if ( trk.codec == "opus") {
mediaDesc << "m=audio 0 RTP/AVP 102" << "\r\n"
"a=rtpmap:102 opus/" << trk.rate << "/" << trk.channels << "\r\n"
"a=control:track" << trk.trackID << "\r\n";
}
return mediaDesc.str();
}
bool parseTransport(const std::string & transport, const std::string & host, const std::string & source, const DTSC::Track & trk){
unsigned int SSrc = rand();
if (trk.codec == "H264") {
pack = RTP::Packet(97, 1, 0, SSrc);
}else if(trk.codec == "AAC"){
pack = RTP::Packet(96, 1, 0, SSrc);
}else if(trk.codec == "AC3"){
pack = RTP::Packet(100, 1, 0, SSrc);
}else if(trk.codec == "MP3"){
pack = RTP::Packet(14, 1, 0, SSrc);
}else if(trk.codec == "ALAW"){
if (trk.channels == 1 && trk.rate == 8000){
pack = RTP::Packet(8, 1, 0, SSrc);
}else{
pack = RTP::Packet(101, 1, 0, SSrc);
}
}else if ( trk.codec == "PCM") {
if (trk.size == 16 && trk.channels == 2 && trk.rate == 44100){
pack = RTP::Packet(10, 1, 0, SSrc);
} else if (trk.size == 16 && trk.channels == 1 && trk.rate == 44100){
pack = RTP::Packet(11, 1, 0, SSrc);
}else{
pack = RTP::Packet(103, 1, 0, SSrc);
}
}else if(trk.codec == "opus"){
pack = RTP::Packet(102, 1, 0, SSrc);
}else{
ERROR_MSG("Unsupported codec %s for RTSP on track %u", trk.codec.c_str(), trk.trackID);
return false;
}
if (transport.find("TCP") != std::string::npos) {
std::string chanE = transport.substr(transport.find("interleaved=") + 12, (transport.size() - transport.rfind('-') - 1)); //extract channel ID
channel = atol(chanE.c_str());
rtcpSent = 0;
transportString = transport;
} else {
channel = -1;
size_t port_loc = transport.rfind("client_port=") + 12;
cPort = atol(transport.substr(port_loc, transport.rfind('-') - port_loc).c_str());
uint32_t portA, portB;
//find available ports locally;
int sendbuff = 4*1024*1024;
data.SetDestination(host, cPort);
portA = data.bind(0);
setsockopt(data.getSock(), SOL_SOCKET, SO_SNDBUF, &sendbuff, sizeof(sendbuff));
rtcp.SetDestination(host, cPort + 1);
portB = rtcp.bind(0);
setsockopt(rtcp.getSock(), SOL_SOCKET, SO_SNDBUF, &sendbuff, sizeof(sendbuff));
std::stringstream tStr;
tStr << "RTP/AVP/UDP;unicast;client_port=" << cPort << '-' << cPort + 1 << ";";
if (source.size()){
tStr << "source=" << source << ";";
}
tStr << "server_port=" << portA << "-" << portB << ";ssrc=" << std::hex << SSrc << std::dec;
transportString = tStr.str();
INFO_MSG("Transport string: %s", transportString.c_str());
}
return true;
}
std::string rtpInfo(const DTSC::Track & trk, const std::string & source, uint64_t currentTime){
unsigned int timeMultiplier = 1;
timeMultiplier = ((double)trk.rate / 1000.0);
if (trk.codec == "H264") {
timeMultiplier = 90;
}
std::stringstream rInfo;
rInfo << "url=" << source << "/track" << trk.trackID << ";"; //get the current url, not localhost
rInfo << "sequence=" << pack.getSequence() << ";rtptime=" << currentTime * timeMultiplier;
return rInfo.str();
}
};
class OutRTSP : public Output {
public:
OutRTSP(Socket::Connection & myConn);
static void init(Util::Config * cfg);
void sendNext();
void onRequest();
void requestHandler();
bool onFinish();
private:
void parseSDP(const std::string & sdp);
long long connectedAt;///< The timestamp the connection was made, as reference point for RTCP packets.
std::map<int, RTPTrack> tracks;///< List of selected tracks with RTSP-specific session data.
unsigned int pausepoint;///< Position to pause at, when reached
HTTP::Parser HTTP_R, HTTP_S;
std::string source;
uint64_t lastTimeSync;
bool expectTCP;
bool handleTCP();
void handleUDP();
void handleIncomingRTP(const uint64_t track, const RTP::Packet & pkt);
void h264MultiParse(uint64_t ts, const uint64_t track, char * buffer, const uint32_t len);
void h264Packet(uint64_t ts, const uint64_t track, const char * buffer, const uint32_t len, bool isKey);
void updateH264Init(uint64_t trackNo);
namespace Mist{
class OutRTSP : public Output{
public:
OutRTSP(Socket::Connection &myConn);
static void init(Util::Config *cfg);
void sendNext();
void onRequest();
void requestHandler();
bool onFinish();
void incomingPacket(const DTSC::Packet &pkt);
private:
long long connectedAt; ///< The timestamp the connection was made, as reference point for RTCP
///packets.
unsigned int pausepoint; ///< Position to pause at, when reached
SDP::State sdpState;
HTTP::Parser HTTP_R, HTTP_S;
std::string source;
uint64_t lastTimeSync;
bool expectTCP;
bool handleTCP();
void handleUDP();
};
}
typedef Mist::OutRTSP mistOut;