apps/astreamer/rtsp.cpp

Go to the documentation of this file.

#include "rtsp.h"
#include "helpers.h"

#include <string.h>
#include <vector>

using namespace std;

#undef RTSP_DEBUG
//#define RTSP_DEBUG

#ifdef RTSP_DEBUG
        #define D(a) a
#else
        #define D(a)
#endif

#define RTSP_PORT       "554"
#define RTSP_SIGN       "rtsp://"

// TODO
_Request::_Request(const string &request) {
        string s;
        int prev = 0;
        int i = 0;
        int l = request.length();
        const char *c = request.c_str();
        bool first = true;
        string name, value;

        while(i < l) {
                s = "";
                for(; c[i] != '\n' && c[i] != '\r' && c[i] != '\0'; i++);
                if(c[i] == '\n' || c[i] == '\r') {
                        if(i == prev) {
                                i++;
                                prev = i;
                                continue;
                        }
                        s.insert(0, request, prev, i - prev);
//                      cerr << "prev == " << prev << "; i == " << i << ";--> " << "\trequest string: \"" << s << "\"" << endl;
                        if(first) {
                                int x = s.find(" ");
                                method.insert(0, s, 0, x);
                                int y = s.find(" ", x + 1);
                                uri.insert(0, s, x + 1, y - x - 1);
//                              cerr  << "method == |" << method << "|; uri == |" << uri << "|" << endl;
                                first = false;
                        } else {
                                name = "";
                                value = "";
                                int x = s.find(": ");
                                name.insert(0, s, 0, x);
                                value.insert(0, s, x + 2, s.length() - x - 1);
                                fields[name] = value;
//                              cerr << "pair: name == |" << name << "|; value == |" << value << "|" << endl;
                        }
                }
                if(c[i] == '\0')
                        break;
                i++;
                prev = i;
        }
}

void _Responce::add_field(const string &name, const string &value) {
        fields[name] = value;
}

void _Responce::add_include(const string &_include) {
        include = _include;
}

string _Responce::serialize() {
        string rez;
        switch(_status) {
        case STATUS_OK:
                rez = "RTSP/1.0 200 OK\r\n";
                break;
        case STATUS_BUSY:
                rez = "RTSP/1.0 455 Server is busy\r\n";
                break;
        case STATUS_EMPTY:
                rez = "";
                return rez;
        }
        for(map<string, string>::iterator it = fields.begin(); it != fields.end(); it++) {
                rez += (*it).first + ": " + (*it).second + "\r\n";
        }
        rez += "\r\n";
        if(include.length() != 0) {
                rez += include;
        }
        return rez;
}

//RTSP_Server::RTSP_Server(void (*h)(void *, RTSP_Server *, RTSP_Server::event), void *handler_data, Session *_session) {
RTSP_Server::RTSP_Server(int (*h)(void *, RTSP_Server *, RTSP_Server::event), void *handler_data, Session *_session) {
        handler_f = h;
        this->handler_data = handler_data;
        session = _session;
        _busy = NULL;
}

void RTSP_Server::main(void) {
        list <Socket *> s;
        // create listen socket...

        Socket *socket_main = new Socket("", 554);
//      Socket *socket_1 = new Socket("", 554);
//      Socket *socket_2 = new Socket("", 8554);
//      Socket *socket_3 = new Socket("", 7070);
        s.push_back(socket_main);
        socket_main->listen(2);

        while(1) {
//              int p = Socket::poll(s);
                Socket::poll(s);
D(cerr << "incoming..." << endl;)
refresh:
D(cerr << "refresh..." << endl;)
                for(list<Socket *>::iterator it = s.begin(); it != s.end(); it++) {
                        Socket::state state = (*it)->state_refresh();
                        if(state == Socket::STATE_IN) {
                                if(*it == socket_main) {
                                        Socket *in = socket_main->accept();
                                        if(in)
                                                s.push_back(in);
                                } else {
                                        // check for remove closed socket !
                                        if(!process(*it)) {
                                                s.remove(*it);
                                                goto refresh;
                                        }
                                }
                        }
                }
        }
}

bool RTSP_Server::process(Socket *s) {

        string req;
        _Request *request;
        _Responce *responce = NULL;
        bool r = true;
        _Responce::status status = _Responce::STATUS_EMPTY;

//      cerr << "process()..." << endl;
//      int err = s->recv(req);
        if(!s->recv(req))
                return false;
        if(req == "") {
D(cerr << "request is empty, close client..." << endl;)
                return false;
        }
        if(!session->rtp_out.multicast) {
                if(req == "") { // socket has closed - close stream.
                        if(_busy == s) {
                                handler(TEARDOWN);
                                _busy = NULL;
                        }
                        return false;
D( cerr << "--==-->> reset _busy !!! <<--==--" << endl;)
                }
        }
D(cerr << __FILE__<< ":"<< __FUNCTION__ << ":" <<__LINE__ << " part_REQUEST: " << endl << "\"" << req << "\"" << endl;)
        // check for partial request...
        part_of_request += req;
        req = part_of_request;
        const char *c = req.c_str();
        string end;
        int e = req.length();
        e -= 4;
        if(e <= 0)
                e = 0;
        for(int i = req.length() - 1; i >= e; i--) {
                end += c[i];
//cerr << "..." << (int)c[i] << endl;
        }
        if(end.find("\n\n") != 0 && end.find("\n\r\n\r") != 0) {
//cerr << "____________________" << endl << "|" << part_of_request << "|" << endl;
                return true;
        } else
                part_of_request = "";
        // process...
//cerr << "process request !!!!!!!!!!!!!!!!!!!!!!!!!!!!!" << endl;
D(      cerr << __FILE__<< ":"<< __FUNCTION__ << ":" <<__LINE__ << "\tREQUEST: " << endl << "\"" << req << "\"" << endl;)
        request = new _Request(req);

        responce = new _Responce();
        responce->add_field("CSeq", (*request->get_fields().find("CSeq")).second);
//      responce->add_field("CSeq", (*request->get_fields().find("CSeq")).second);
        if(request->get_method() == "OPTIONS") {
               if (handler(IS_DAEMON_ENABLED)) {
                    status = _Responce::STATUS_OK;
                    responce->add_field("Public", "DESCRIBE, SETUP, TEARDOWN, PLAY, PAUSE");
                // check transport...
                    if(!session->rtp_out.multicast)
                        session->rtp_out.ip = s->source_ip();
               } else {
                        status = _Responce::STATUS_BUSY;
               }
        }
        if(request->get_method() == "DESCRIBE") {
                if(handler(IS_DAEMON_ENABLED)  && (session->rtp_out.multicast || !_busy)) {
                        handler(DESCRIBE); 
                        status = _Responce::STATUS_OK;
                        char buf[8];
                        string sdp = make_sdp(request->get_uri());
                        responce->add_field("Content-Type", "application/sdp");
                        sprintf(buf, "%d", (int)sdp.length());
                        responce->add_field("Content-Length", buf);
                        responce->add_include(sdp);
                } else {
                        status = _Responce::STATUS_BUSY;
                }
        }
        if(request->get_method() == "SETUP") {
                if (handler(IS_DAEMON_ENABLED)  && (session->rtp_out.multicast || !_busy)) {
                        status = _Responce::STATUS_OK;
                        responce->add_field("Session", "47112344");
                        responce->add_field("Transport", make_transport((*request->get_fields().find("Transport")).second));
                } else
                        status = _Responce::STATUS_BUSY;
        }
        if(request->get_method() == "TEARDOWN") {
                status = _Responce::STATUS_OK;
                handler(TEARDOWN);
                _busy = NULL;
        }
        if(request->get_method() == "PLAY") {
D(cerr << "--==-->> transport: " << session->rtp_out.multicast << "; _busy == " << (unsigned long)_busy << endl;)
//              if(session->rtp_out.multicast || !_busy) {
                if (handler(IS_DAEMON_ENABLED) && (session->rtp_out.multicast || !_busy)) {
                        status = _Responce::STATUS_OK;
                        handler(PLAY);
                        if(!session->rtp_out.multicast)
                                _busy = s;
                } else
                        status = _Responce::STATUS_BUSY;
D(cerr << "--==-->> transport: " << session->rtp_out.multicast << "; _busy == " << (unsigned long)_busy << " status=" << status << endl;)
        }
        if(request->get_method() == "PAUSE") {
                status = _Responce::STATUS_OK;
                handler(PAUSE);
        }
        delete request;
        responce->set_status(status);
D(              cerr << "\tRESPONSE: " << endl << "\"" << responce->serialize() << "\"" << endl;)
        r = s->send(responce->serialize());
        delete responce;
        return r;
}

string RTSP_Server::make_sdp(string uri) {
        char buf[256];

D(cerr << "make SDP" << endl;)
        // make video description
        // m = - port + PAYLOAD
        string rez = "m=video ";
D(cerr << "_____________________" << endl;)
D(cerr << "_____________________" << endl;)
D(cerr << "_____________________" << endl;)
        sprintf(buf, "%d", session->rtp_out.port_video);
        if(session->process_audio == false) {
D(cerr << "+++++++++++++++++++++" << endl;)
                if(session->rtp_out.multicast) {
                        rez += buf;
                } else {
                        rez += "0";
                }
        } else {
                rez += buf;
        }
        rez += " RTP/AVP ";
        sprintf(buf, "%d", session->video.type);
        rez += buf;
        rez += "\r\n";  // port
        if(session->rtp_out.multicast) {
                rez += "a=type:multicast\r\n";
                rez += "c=IN IP4 " + session->rtp_out.ip + "/" + session->rtp_out.ttl + "\r\n"; // IP + TTL
        } else {
                rez += "a=type:unicast\r\n";
                rez += "c=IN IP4 0.0.0.0\r\n";  // IP + TTL
        }
        sprintf(buf, "%.4f", session->video.fps);
        rez += "a=framerate:";
        rez += buf;
        rez += "\r\na=x-framerate:";
        rez += buf;
        if(session->video.width > 0) {
                sprintf(buf, "%d", session->video.width);
                rez += "\r\na=width:";
                rez += buf;
                rez += "\r\na=x-width:";
                rez += buf;
        }
        if(session->video.height > 0) {
                sprintf(buf, "%d", session->video.height);
                rez += "\r\na=height:";
                rez += buf;
                rez += "\r\na=x-height:";
                rez += buf;
        }
        if(session->video.width > 0 && session->video.height > 0) {
                sprintf(buf, "%d,%d", session->video.width, session->video.height);
                rez += "\r\na=x-dimensions:";
                rez += buf;
        }

        rez += "\r\n";
        // make audio description, if present
        if(session->process_audio) {
//              sprintf(buf, "m=audio %d RTP/AVP %d\r\n", atoi(session->rtp_out.port.c_str()) + 2, session->audio.type);
                sprintf(buf, "m=audio %d RTP/AVP %d\r\n", session->rtp_out.port_audio, session->audio.type);
                rez += buf;
                sprintf(buf, "a=rtpmap:%d L16/%d/%d\r\n", session->audio.type, session->audio.sample_rate, session->audio.channels);
                rez += buf;
        }
        rez += "a=control:" + uri + "\r\n";
D(cerr << "make SDP - ok!" << endl;)
        return rez;
}

string RTSP_Server::make_transport(string req) {
        map<string, string> m = String::split_list_to_map(String::split_to_list(req, ';'), '=');
        string client_port = m["client_port"];
        int port_req = session->rtp_out.port_video;
//D(cerr << "client_port == " << client_port << endl;)
        if(client_port != "") {
                string first, second;
                String::split(client_port, '-', first, second);
                port_req = atol(first.c_str());
//D(cerr << "client_port == |" << client_port << "|; first == |" << first << "|; second == |" << "|" << endl;)
//              session->rtp_out.port = first;
        }

        if(!session->rtp_out.multicast) {
                if(!session->process_audio) {
                        session->rtp_out.port_video = port_req;
                }
        }

        string rez = "RTP/AVP;";
        if(session->rtp_out.multicast)
                rez += "multicast;";
        else
                rez += "unicast;";
        char buf[128];
        sprintf(buf, "%d", port_req);
        rez += "destination=" + session->rtp_out.ip + ";port=" + buf;
        if(session->process_audio) {
                sprintf(buf, "-%d", port_req + 1);
//              rez += "-";
//              long p = atol(session->rtp_out.port.c_str());
//              p++;
//              sprintf(buf, "%d", p);
                rez += buf;
        }
        if(session->rtp_out.multicast)
                rez += ";ttl=" + session->rtp_out.ttl;
//      rez += "destination=" + transport.ip + ";port=" + transport.port;
//      if(transport.multicast)
//              rez += ";ttl=" + transport.ttl;
        return rez;

/*
        if(!session->rtp_out.multicast) {
                if(!session->process_audio) {
                        if(session->rtp_out)
                }
        }
*/
/*
D(      cerr << "transport request: |" << req << "|" << endl;)
        if(!session->rtp_out.multicast) {
                // unicast
                map<string, string> m = String::split_list_to_map(String::split_to_list(req, ';'), '=');
                string client_port = m["client_port"];
D(cerr << "client_port == " << client_port << endl;)
                if(client_port != "") {
                        string first, second;
                        String::split(client_port, '-', first, second);
D(cerr << "client_port == |" << client_port << "|; first == |" << first << "|; second == |" << "|" << endl;)
                        session->rtp_out.port = first;
                }
        }

        string rez = "RTP/AVP;";
        if(session->rtp_out.multicast)
                rez += "multicast;";
        else
                rez += "unicast;";
        rez += "destination=" + session->rtp_out.ip + ";port=" + session->rtp_out.port;
        if(session->process_audio) {
                rez += "-";
                long p = atol(session->rtp_out.port.c_str());
                p++;
                char buf[128];
                sprintf(buf, "%d", p);
                rez += buf;
        }
        if(session->rtp_out.multicast)
                rez += ";ttl=" + session->rtp_out.ttl;
//      rez += "destination=" + transport.ip + ";port=" + transport.port;
//      if(transport.multicast)
//              rez += ";ttl=" + transport.ttl;
        return rez;
*/
}

RTSP_Server::~RTSP_Server() {
        cerr << "destroy RTSP_Server object" << endl;
//      delete socket_main;
//      if (request)  delete request;
//      if (responce) delete responce;
}

---