Home    News    Components    Get ViTooKi    Documentation    Community    Developers    @sf.net  

Info

Last update:
12 Feb 2007

Documentation

Doxygen LXR

ClientSession.cpp

 
/***********************************************************************
 *                                                                      *
 * ViTooKi                                                              *
 *                                                                      *
 * title:       ClientSession.cpp                                             *
 *                                                                      *
 *                                                                      *
 *                                                                      *
 * ITEC institute of the University of Klagenfurt (Austria)             *
 * http://www.itec.uni-klu.ac.at                                        *
 *                                                                      *
 *                                                                      *
 * For more information visit the ViTooKi homepage:                     *
 * http://ViTooKi.sourceforge.net                                       *
 * vitooki-user@lists.sourceforge.net                                   *
 * vitooki-devel@lists.sourceforge.net                                  *
 *                                                                      *
 * This file is part of ViTooKi, a free video toolkit.                  *
 * ViTooKi is free software; you can redistribute it and/or             *
 * modify it under the terms of the GNU General Public License          *
 * as published by the Free Software Foundation; either version 2       *
 * of the License, or (at your option) any later version.               *
 *                                                                      *
 * This program is distributed in the hope that it will be useful,      *
 * but WITHOUT ANY WARRANTY; without even the implied warranty of       *
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the         *
 * GNU General Public License for more details.                         *
 *                                                                      *
 * You should have received a copy of the GNU General Public License    *
 * along with this program; if not, write to the Free Software          *
 * Foundation, Inc., 59 Temple Place - Suite 330, Boston,               *
 * MA  02111-1307, USA.                                                 *
 *                                                                      *
 ***********************************************************************/

 /***********************************************************************
 *                                                                      *
 *  REVISION HISTORY:                                                   *
 *                                                                      *
 *                                                                      *
 *                                                                      *
 ***********************************************************************/



#include "ClientSession.hpp"

#include "RTSP.hpp"
#include "DataChannel.hpp"
#include "DataSink.hpp"
#include "ContainerInfo.hpp"
#include "ESInfo.hpp"
#include "Statistics.hpp"

#include "PacketizationLayer.hpp"
#include "Url.hpp"
#include "RFC3016.hpp"
#include "RFC2250.hpp"

#include "metadata/TerminalCapabilities.hpp"
#include "io/Rtp.hpp"
#include "io/SimpleRtp.hpp"
#include "SimpleStatistics.hpp"
#include "adaptors/MP4Decoder.hpp"
#include "adaptors/MP4audioDecoder.hpp"
#include "adaptors/Forwarder.hpp"
#include "cache/MetaObject.hpp"


#ifdef WIN32
#include <winsock2.h>
#else
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <netinet/in.h>
#endif

ClientSession::ClientSession(const char* uri, IO* vout, IO* aout, UserPreferences* up, bool enableRtx,bool visualize)
{
        printf("\033[31mXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX ClientSession UP: %p XXXXXXXXXXXXXXXXXXXXXXXXXXXXXX\n\033[0m", up );
        printf(uri);
        printf("\nX: min: %i, max: %i, best: %i\n", up->dimX.min, up->dimX.max, up->dimX.best);
        printf("B: min: %i, max: %i, best: %i\n", up->bitrate.min, up->bitrate.max, up->bitrate.best);
        printf("\n\n");
        rtxEnabled=enableRtx;
        videoOut=vout;
        audioOut=aout;
        mp4Stream=NULL;
        protocol=NULL;
        assert(up);
        usrPref = up;
//      assert(up->pay);
        termCap=NULL;
        this->construct(uri);
        decodeData=visualize;
}

ClientSession::ClientSession(const char* videoUri, const char* proxyUri, 
                          IO* vout, IO* aout, UserPreferences* up, 
                          bool enableRtx,bool visualize)
{
        assert(videoUri);
        assert(proxyUri);
        rtxEnabled=enableRtx;
        videoOut=vout;
        audioOut=aout;
        mp4Stream=NULL;
        protocol=NULL;
        assert(up);
        usrPref = up;
        termCap=NULL;
        decodeData=visualize;
        url=new char[strlen(videoUri)+1];
        strcpy(url,videoUri);
        // connect to proxy
        Url u(proxyUri);
        protocol = new RTSP();
        const char* proxyName=u.getServerName();
        int port=u.hasPort()?u.getPort():protocol->getDefaultPort();
        server = new sockaddr_in();
        dprintf_full("ClientSession::ClientSession: getting %s from %s:%i\n",videoUri,proxyName,port);
        if(!Url::convertUrlToSocket(proxyName,port,server,sessionControlChannel)) {
                // ignore proxy setting
                dprintf_err("Proxy connect failed! Ignoring Proxy... direct connect to server\n");
                delete url;
                url=NULL;
                construct(videoUri);
        }
        else
                state = SESSION_CLOSED;
};
ClientSession::ClientSession(const char* uri, IO* vout, IO* aout, TerminalCapabilities* tc,bool enableRtx,bool visualize)
{
        rtxEnabled=enableRtx;
        videoOut=vout;
        audioOut=aout;
        termCap=tc;
        mp4Stream=NULL;
        protocol = NULL;
        usrPref = NULL;
        decodeData=visualize;
        this->construct(uri);

}

void ClientSession::construct(const char* uri)
{
        // establish a connection to the server
        if(!uri) {
                dprintf_err("ClientSession::ClientSession: NULL url\r\n");
                state=SESSION_ERR;
                return;
        }
        else {
                url=new char[strlen(uri)+1];
                strcpy(url,uri);
        }
        
        char* pUrl=strstr(url,"rtsp://");
        pUrl+=strlen("rtsp://");
        if(!pUrl) {
                dprintf_err("ClientSession::ClientSession: invalid RTSP url %s\r\n",url);
                state=SESSION_ERR;
                return;
        }

        Url u(url);
        protocol = new RTSP();
        const char* serverName=u.getServerName();
        int port=u.hasPort()?u.getPort():protocol->getDefaultPort();
        server = new sockaddr_in();
        if(!Url::convertUrlToSocket(serverName,port,server,sessionControlChannel)) {
                state=SESSION_ERR;
                return;
        }
        state = SESSION_CLOSED;
};

ClientSession::ClientSession(int socket,  const struct sockaddr_in *serv, const char* uri,
                 IO* vout, IO* aout,TerminalCapabilities* tc, bool enableRtx,bool visualize)
{
        rtxEnabled=enableRtx;
        state = SESSION_CLOSED;
        videoOut=vout;
        audioOut=aout;
        mp4Stream=NULL;
        usrPref = NULL; 
        sessionControlChannel = socket;
        
        if (serv) {
                server = new sockaddr_in();
                memcpy(server, serv, sizeof(sockaddr_in));
        } else {
                server = NULL;
        }
        
        termCap=tc;
        
        protocol = new RTSP();
        if(!uri)
                url = NULL;
        else {
                url=new char[strlen(uri)+1];
                strcpy(url,uri);
        }
        decodeData=visualize;
};

ClientSession::~ClientSession()
{

#ifdef WIN32
        closesocket(sessionControlChannel);
#else
        ::close(sessionControlChannel);
#endif
        tearDown(INVALID_SESSIONID);
        if (protocol) {
                delete protocol;
                protocol = NULL;
        }

        

        if ( server ) {
                delete server;
                server = NULL;
        }
        if (url) {
                delete url;
                url = NULL;
        }


        if(this->mp4Stream)
                delete mp4Stream;
}

bool ClientSession::sendResponse(const char *str,int bytes)
{
        if ( state == SESSION_ERR )
                return false;
        
        int n=0;
        int total=0;
        bool ret=false;
        dprintf_full("ClientSession::sendResponse: Writing %i bytes: XxX\033[33m%s\033[0mXxX\r\n",bytes,str);

        while(bytes>0) {
                printf(">>>>>>>>>>>>>>SOCKET: %i with SessionState: %i<<<<<<<<<<<<<\n", sessionControlChannel, state);
                fflush(stdout);
                n=send(sessionControlChannel,str+total, bytes,0);
                if(n==-1)
                        break;
                bytes-=n;
                total+=n;
        }
        if(bytes==0)
                ret=true;
        dprintf_full("ClientSession::sendResponse exit %i\r\n",(int)ret);
        return ret;
};

int ClientSession::readRequest(char* str, const int MAX_BYTES)
{
        dprintf_full("\033[32mClientSession::readRequest( %p, %i )\n\033[0m", str, MAX_BYTES);
        if ( state == SESSION_ERR )
                return -1;

        int n=0;
        int total=0;
        memset(str,0,MAX_BYTES);

        do {
                // n == 0 if connection closed
                // n < 0 if error
                n=recv(sessionControlChannel,str+total,MAX_BYTES-1-total,0);
                dprintf_full("ClientSession::readRequest: partial %i total %i\r\n",n,total);
                dprintf_full("ClientSession::readRequest:X%sX\r\n",str+total);
                if(n<=0)
                        break;
                total+=n;
        }
        while(n>0 && !(total>3 && str[total-4]=='\r' && str[total-3]=='\n' && 
                                str[total-2]=='\r' && str[total-1]=='\n'));
        
        dprintf_full("ClientSession::readRequest: %i\r\n",total);fflush(stdout);
        return total;
};

bool ClientSession::connect()
{
        if(!protocol || !url)
                return false;
        char* serverID;
        
        protocol->generateDescribeForUrl(url,PLAYER_ID,this->termCap,this->usrPref);
        
        const char* req=protocol->getBuffer();

        if(!sendResponse(req,strlen(req))) {
                return false;
        }
        
        
        char *buffer = new char[MSG_BUFFER_SIZE];
        buffer[0]=0;
        // read answer to describe
        int bytesRead=readRequest(buffer,MSG_BUFFER_SIZE);
        if(bytesRead<=0) {
                delete [] buffer;
                return false;
        }
        // FIXME: check RTSP 404
        
        ContainerInfo* mp4=protocol->parseResponseDescribe(buffer,url,&serverID,&rtxInfo);
        if(this->mp4Stream)
                delete mp4Stream;
        mp4Stream=mp4;
        if(serverID)
                //printf(">>> %s\n",serverID);
                //delete serverID;

        if(mp4) {
                state=SESSION_CONNECTED;
        }

        delete [] buffer;
        
        return state==SESSION_CONNECTED;
};

bool ClientSession::play(bool with_caching)
{

        dprintf_err("\033[32mClientSession::play(), mp4Stream is: %i\n\033[0m", (int)mp4Stream);

        if(mp4Stream) 
                {
                // search for the session
                int sessionId=0;
                u32 streamId=0;
                bool res = true;
                // start it
                for (list < DataChannel * >::iterator i = this->channels.begin();
                                 i != this->channels.end(); i++) {
                        streamId=(*i)->getESInfo()->getStreamId();
                        if(!clientObj[streamId]) {
                                dprintf_err("ClientSession::play Failed to add NULL client???\r\n");
                                tearDown(TEARDOWN_ALL);
                                state=SESSION_ERR;
                                return false;
                        }
                        if ((*i)->getESInfo()->isVisualStream() || (*i)->getESInfo()->isAudioStream()) {
                                sessionId=(*i)->getSessionId();
                                double duration=(double)(*i)->getESInfo()->getDurationInMs();
                                duration/=1000.0;
                                if(rtxEnabled)
                                        protocol->generatePlay(this->url,sessionId,0.0,duration,RTP_PREFETCH_SECS);
                                else
                                        protocol->generatePlay(this->url,sessionId,0.0,duration,3);
                                const char* req=protocol->getBuffer();
                                if(!this->sendResponse(req,strlen(req))) {
                                        return false;
                                }
                                char *buffer = new char[MSG_BUFFER_SIZE];
                                int len=readRequest(buffer,MSG_BUFFER_SIZE);
                                if(len<=0)
                                {
                                        delete [] buffer;
                                        return false;
                                }
                                
                                if(!strstr(buffer,"200")) // status code for OK
                                        state=SESSION_ERR;
                                if(state==SESSION_ERR) {
                                        delete [] buffer;
                                        return false;
                                }

                                // insert client
                                (*i)->insert(clientObj[streamId]);
                                res &= (*i)->play(clientObj[streamId]->getDataSinkId(),RTP_PREFETCH_SECS);
                                dprintf_full("ClientSession::play Successfully added client %s is %i\r\n",
                                                                                         clientObj[streamId]->getAddress(), res);
                                        
                                
                                //IO* input=(*i)->getInput();
                                // if not yet started -> start it
                                if (!(*i)->running()) {
                                        if ( with_caching )
                                        {
                                                dprintf_full("\033[32mClientSession::enabled caching\n\033[32m");
                                                (*i)->enableCaching(true,".");
                                        }
                                        dprintf_full("\033[32mClientSession::thread started\n\033[0m");
                                        (*i)->start();
                                        state=SESSION_PLAYING;
                                } else
                                        res= res & false;

                                delete [] buffer;
                        } // a visual stream
                } // iterator
                return res;
        } //mp4Stream
        else {
          dprintf_err("ClientSession::play: FAILED. No ContainerInfo object\r\n");
        }

        dprintf_err("ClientSession::play: FAILED. No DataChannels found\r\n");  
        return false;
};

bool ClientSession::pause(u32 sessionId)
{
        protocol->generatePause(this->url,sessionId);
        const char* req=protocol->getBuffer();
        u32 streamId=0;
        // search for the session
        // pause it
        for (list < DataChannel * >::iterator i = this->channels.begin();
                         i != this->channels.end(); i++) {

                if ((*i)->getSessionId() == sessionId) {
                        if(!this->sendResponse(req,strlen(req))) {
                                return false;
                        }

                        char *buffer = new char[MSG_BUFFER_SIZE];
                        int len=readRequest(buffer,MSG_BUFFER_SIZE);
                        if(len<0) {
                                delete [] buffer;
                                return false;
                        }
                        buffer[len]=0;
                        if(!strstr(buffer,"200")) // status code for OK
                                state=SESSION_ERR;
                        else
                                state=SESSION_PAUSED;
                        streamId=(*i)->getESInfo()->getStreamId();
                        (*i)->pause(clientObj[streamId]->getDataSinkId());

                        delete [] buffer;
                        return state==SESSION_PAUSED;
                }
        }
        
        return false;
};

bool ClientSession::tearDown(u32 sessionId)
{
        if(sessionId==INVALID_SESSIONID) {
                // teardown all
                state = SESSION_CLOSED;
                list < DataChannel * >::iterator dcI;
                while (!channels.empty()) {
                        dcI=channels.begin();
                        protocol->generateTeardown(this->url,(*dcI)->getSessionId());
                        const char* req=protocol->getBuffer();
                        this->sendResponse(req,strlen(req));
                        (*dcI)->teardown(clientObj[(*dcI)->getESInfo()->getStreamId()]->getDataSinkId());
                        // check if we have to delete it, might be shared with other sessions
                        if((*dcI)->getNumberOfActiveDataSinks()==0) {
                                (*dcI)->wait(); // wait for thread to finish
                                delete (*dcI);
                        }
                        channels.pop_front();
                }
                return true;
        }
        else {
                // teardown one session
                list < DataChannel * >::iterator dcI;
                if (!channels.empty()) {
                        dcI = channels.begin();
                        while (dcI != channels.end()) {
                                if ((*dcI)->getSessionId()== sessionId) {
                                        protocol->generateTeardown(this->url,sessionId);
                                        const char* req=protocol->getBuffer();
                                        this->sendResponse(req,strlen(req));
                                        char *t = new char[MSG_BUFFER_SIZE];
                                        // read response -> overwrite req
                                        readRequest(t,MSG_BUFFER_SIZE);
                                        (*dcI)->teardown(clientObj[(*dcI)->getESInfo()->getStreamId()]->getDataSinkId());
                                        // check if we have to delete it, might be shared with other sessions
                                        if((*dcI)->getNumberOfActiveDataSinks()==0)
                                        {
                                                (*dcI)->wait(); // wait for thread to finish
                                                delete (*dcI);
                                                                                
                                        }
                                        channels.erase(dcI);
                                        if(channels.empty())
                                                state = SESSION_CLOSED;

                                        delete [] t;
                                        return true;
                                }
                                else {
                                        ++dcI;
                                }
                        }
                }
                return false; // delete on empty session
        }
        
        return false;
};

void ClientSession::run() {

        dprintf_full("\033[32mClientSession::run()\033[0m\n");
        
        if(!url) {
                dprintf_err("ClientSession::run(): No URL set. Can't continue\r\n");
                this->exit();
                return;
        }
        // url is set, check the session state
        if(this->state==SESSION_CLOSED) {
                if(!this->connect())
                {
                        printf("\033[33m!!! REJECTED REQUEST !!!\033[0m\n");

                        state=SESSION_ERR;
                        }
                else
                {
                        printf("$$$$$$$$$$ REQUEST OK ... START STREAMING $$$$$$$$$$$$$$$\n");
                }
        }
        if(this->state==SESSION_CONNECTED) {
                // do setup
                
                if(mp4Stream==NULL) {
                        tearDown(TEARDOWN_ALL);
                        state=SESSION_ERR;
                        this->exit();
                        return;
                }
                // FIXME: do SETUP for all ES of type visual & audio
                list<ESInfo*> *allES=mp4Stream->getESList();
                if(allES->empty()) {
                        tearDown(TEARDOWN_ALL);
                        state=SESSION_ERR;
                        this->exit();
                        return;
                }
                u32 streamId = 0;
                bool visualSet=false;
                bool audioSet=false;
                for(list<ESInfo*>::iterator esi=allES->begin();
                                esi!=allES->end(); ++esi) {
                        
                        streamId = (*esi)->getStreamId();


                        
                        if( ((*esi)->isAudioStream() && !audioSet && audioOut) || 
                                        ((*esi)->isVisualStream() && !visualSet && videoOut)) {
                                dprintf_full("ClientSession::run(): streamID is %i\n",streamId);
                                const char* buffer;
                                if(rtxEnabled)
                                        protocol->generateSetup(url,streamId,portGen.getNextFreePortPair(),true,
                                                                                                                                         portGen.getNextFreePortPair());
                                else
                                        protocol->generateSetup(url,streamId,portGen.getNextFreePortPair(),true);
                                buffer=protocol->getBuffer();
                                                                
                                if(!sendResponse(buffer,strlen(buffer)))
                                        state=SESSION_ERR;
                                else
                                        state=SESSION_INITIALIZED;
                                
                                if(state==SESSION_INITIALIZED) {
                                        // parse the response and create the RTP class
                                        char *response = new char[MSG_BUFFER_SIZE];
                                        int len=readRequest(response,MSG_BUFFER_SIZE);
                                        if(len<=0) {
                                                state=SESSION_ERR;
                                                this->exit();
                                                delete [] response;
                                                return;
                                        }
                                        int clientPort=0;
                                        int serverPort=0;
                                        bool unicast=true;
                                        char* ssrc=NULL;
                                        int sessionID;
                                        rtx_info* rtx=NULL;
                                        if(rtxEnabled && !rtxInfo.empty()) {
                                                list<rtx_group*>::iterator li=rtxInfo.begin();
                                                while(li!=rtxInfo.end()) {
                                                        if( (*li)->es && (*li)->es->getStreamId()==streamId) {
                                                                rtx=(*li)->rtx;
                                                                li=rtxInfo.end();
                                                        }
                                                        else
                                                                ++li;
                                                }
                                        }
                                        if(protocol->parseResponseSetup(response,&sessionID,
                                                                &clientPort,&serverPort,&unicast,&ssrc,rtx)<4) {
                                                tearDown(TEARDOWN_ALL);
                                                state=SESSION_ERR;
                                                dprintf_err("parseResponseSetup failed\r\n");
                                                if(ssrc)
                                                        delete ssrc;
                                                this->exit();

                                                delete [] response;
                                                return;
                                        }
                                        dprintf_full("Session is %i\n",sessionID);
                                        char localhost[10]="127.0.0.1";
                                        
                                        PacketizationLayer * p;
                                        if ((*esi)->isAudioStream())
                                                  p = new RFC2250();
                                        else
                                                  p = new RFC3016();
                                        

                                        IO* input = NULL;

#ifndef WINCE
                                        if ( !rtxEnabled )
                                        {
                                                char logName[MAX_STR_LEN];
                                                struct timeval t;
                                                gettimeofday(&t,NULL);
                                                sprintf(logName,"log%10i.%3i_%s_%s%llu_client.in",
                                                        (int)t.tv_sec,(int)(t.tv_usec/1000),
                                                        mp4Stream->getUrl()->getFileName(),
                                                        ((*esi)->isAudioStream()?"aud":"vid"),(*esi)->getStreamId());
                                                input= (IO*)new SimpleRtp( url, serverPort, clientPort, "127.0.0.1", *esi , p, false, logName, true, SimpleRtp::CTS_BASED, true, false);
                                        }
                                        else
                                        {               
#endif
                                                Statistics *stat = new Statistics(*esi,false);
                                                input=(IO*)new Rtp(localhost, serverPort, inet_ntoa(server->sin_addr), clientPort,
                                                                *esi, p, false, this->termCap, NULL, stat);
                                                                
                                                if(rtx)
                                                        ((Rtp*)input)->setRtxInfo(rtx);
#ifndef WINCE
                                        }
#endif

                                        DataSink* clientTmp=0;
                                        Adaptor* a=NULL;
                                        if( (*esi)->isVisualStream()) {
                                                visualSet=true;
                                                videoOut->setESInfo( (*esi) );
                                                
                                                if(decodeData)
                                                        a=new MP4Decoder(((VideoESInfo*)(*esi)),true, UncompressedVideoFrame::ColorSpaceRGB32);
                                                else {
                                                        a=new Forwarder();
                                                        a->setESInfo(*esi);
                                                }
                                                clientTmp= new DataSink(0,videoOut,localhost,0,a);
                                        }
                                        else if( (*esi)->isAudioStream()) {
                                                audioSet=true;
                                                audioOut->setESInfo((*esi));

                                                if(decodeData)
                                                        a=new MP4audioDecoder(((AudioESInfo*)(*esi)),true);
                                                else {
                                                        a=new Forwarder();
                                                        a->setESInfo(*esi);
                                                }
                                                clientTmp= new DataSink(0,audioOut,localhost,0,a);
                                        }

                                        // map clientObj to streamId
                                        clientObj[streamId]=clientTmp;
                                        // create DataChannel
                                        DataChannel* dc=new DataChannel(input,(*esi));
                                        dc->setSessionId(sessionID);
                                        if(ssrc)
                                                delete ssrc;
                                        
                                        setDataChannel(dc);

                                        delete [] response;
                                } // is audio or video
                        } // for
                }
                allES->clear();
        }
        if(this->state==SESSION_INITIALIZED ||
                this->state==SESSION_PAUSED) {
                // do a play
                if(!this->play()) 
                        state=SESSION_ERR;
        }
        if(this->state==SESSION_PLAYING) {
                // we have tried to play
                // but we have to check the datachannels to be sure
                for (list < DataChannel * >::iterator i = this->channels.begin();
                                 i != this->channels.end(); i++) {
                        if( (*i)->running())
                                (*i)->wait();
                }
                // teardown all
#ifdef WIN32
                closesocket(sessionControlChannel);
#else
                ::close(sessionControlChannel);
#endif
                tearDown(INVALID_SESSIONID);
                state=SESSION_PAUSED;
        }
        this->exit();
}


bool ClientSession::setRtxInfoForES(u32 esId, rtx_info* rtx)
{
        const ESInfo* es=this->mp4Stream->getES(esId);
        if(!es) {
                dprintf("ClientSession::setRtxInfoForES: invalid esId as input\r\n");
                return false;
        }

        rtx_group* grp=new rtx_group(ACK,3000);
        grp->es=es;
        grp->rtx=rtx;
        rtx->mId=rtxInfo.size()+1;
        grp->midESInfo=esId;
        rtxInfo.push_back(grp);
        return true;
};


rtx_group* ClientSession::getRtxGroup(u32 esId)
{
        
        for(    list<rtx_group*>::iterator li=rtxInfo.begin();
        li!=rtxInfo.end(); li++) {
                if(esId==(*li)->es->getStreamId())
                        return *li;
        }
        return NULL;

};

© ViTooKi 2003 :: [Lite version] Created with Doxygen