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

Info

Last update:
12 Feb 2007

Documentation

Doxygen LXR

ProxySession.cpp

 /***********************************************************************
 *                                                                      *
 * ViTooKi                                                              *
 *                                                                      *
 * title: ProxySession.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 "ProxySession.hpp"
#include "RTSP.hpp"
#include "DataChannel.hpp"
#include "DataSink.hpp"
#include "../cache/MetaObject.hpp"
#include "VideoESInfo.hpp"
#include "../cache/AdmissionControl.hpp"
#include "IO.hpp"
#include "ServerSession.hpp"
#include "ClientSession.hpp"
#include "Protocol.hpp"
#include "Url.hpp"
#include "RFC3016.hpp"
#include "../cache/CacheManager.hpp"
#include "BitField.hpp"
#include "ReferenceCounter.hpp"
#include "Statistics.hpp"

#include "../metadata/MP21.hpp"
#include "../io/Rtp.hpp"
#include "../io/SimpleRtp.hpp"
#include "../io/MPGStreamIO.hpp"
#include "../io/IncompleteIO.hpp"
#include "../io/IOCreator.hpp"
#include "../adaptors/TemporalAdaptor.hpp"
#include "../adaptors/DataDump.hpp"
#include "../adaptors/MP4Decoder.hpp"
#include "../adaptors/MP4audioDecoder.hpp"
#include "AudioAdaptorChain.hpp"
#include "VideoAdaptorChain.hpp"

#ifndef WIN32
#include <arpa/inet.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netdb.h>
#include <netinet/in.h>
#else
#include <winsock2.h>
#include <ws2tcpip.h>
#ifndef WINCE
#include <direct.h>
#include <Wspiapi.h>
#endif
#endif


ProxyStatistics prxStat;

ProxySession::ProxySession(int socket, const struct sockaddr_in *clint,
                                                                                                         CacheManager * c,      AdmissionControl* adm, 
                                                                                                         bool allowRTX,SessionCounter* counter,
                                                                                                         TerminalCapabilities* defTC):Session(),
                                                                                                         rtxEnabled(allowRTX),neverForwardMetaData(true)
{
        assert(adm);
        dprintf_full("ProxySession::ProxySession\n");
        prxStat.requests++;
        this->counter = counter;
        admControl=adm;
        state = SESSION_CLOSED;
        rtxEnabledToServer=false;
        rtxEnabledToClient=false;
        delMp4Stream=false;
        dualMiss=false;
        canAdmitRequest=true;
        dontCacheDualMiss=false;
  cm = c;
  if (clint) {
                client = new sockaddr_in();
                memcpy(client, clint, sizeof(sockaddr_in));
                clientId=(u32)client->sin_addr.s_addr;
  } else {
                client = NULL;
                clientId=0;
        }
        serverSession=NULL;
        server=NULL;
  sessionControlChannel=socket;
  prot = NULL;
  url = NULL;
        termCap=defTC;
        userPref=NULL;
        suggestedAdaptor=NULL;
        mp4Stream=NULL;
        playerId=NULL;serverID=NULL;
        insertMp4StreamIntoCM=false;
        noDescribeSeenYet=true;
        resUsage.cpu=resUsage.disk=resUsage.network=0.0;
        origSrc=NULL;
        mp4StreamReserved = 0;
        origSrcReserved = 0;
}

ProxySession::~ProxySession()
{
        dprintf_full("ProxySession::~ProxySession()\n");
        prxStat.endedRequests++;
        admControl->freeRequest(&resUsage);
        if(origSrc)
                origSrc->getUsageCounter()->decrease();
        if(mp4Stream)
                mp4Stream->getUsageCounter()->decrease();

        if(dualMiss && origSrc && !dontCacheDualMiss) {
                cm->lockCache();
                cm->putVideo(origSrc,origSrcReserved);
                cm->unlockCache();
                origSrc=NULL;origSrcReserved=0;
        }
        if( insertMp4StreamIntoCM && mp4Stream) {       
                cm->lockCache();
                cm->putVideo(mp4Stream,mp4StreamReserved);
                cm->unlockCache();
                mp4StreamReserved=0;
                delMp4Stream=false;
                mp4Stream=NULL;
        }
        //TEST
        //assert(mp4StreamReserved == 0 && origSrcReserved == 0);

        if(mp4StreamReserved > 0)
        {
                cm->freeDiskSpace (mp4StreamReserved);
                mp4StreamReserved = 0;
        }
        if(origSrcReserved > 0)
        {
                cm->freeDiskSpace (origSrcReserved);
                origSrcReserved = 0;
        }
        

        if(origSrc && dualMiss && dontCacheDualMiss) {
                delete origSrc;
        }
        origSrc=NULL;
#ifdef WIN32
        closesocket(sessionControlChannel);
#else
        ::close(sessionControlChannel);
#endif
        tearDown(TEARDOWN_ALL,true,NULL,NULL);
        if (prot) {
                delete prot;
                prot = NULL;
  }
        if (client) {
                delete client;
                client = NULL;
        }
        if (url) {
                delete url;
                url = NULL;
        }
        if(serverSession) {
                delete serverSession;
        }
        if(playerId)
                delete playerId;

        std::list < rtx_group* >::iterator j;
        while(!rtxInfo.empty()) {
                j=rtxInfo.begin();
                delete *j;
                rtxInfo.pop_front();
        }
        std::list < rtx_info* >::iterator k;

        while(!rtxInfoServer.empty()) {
                k=rtxInfoServer.begin();
                delete *k;
                rtxInfoServer.pop_front();
        }       
        if(serverID)
                delete serverID;
        
        if(delMp4Stream && mp4Stream) {
                mp4Stream->getUsageCounter()->decrease();
                delete mp4Stream;
        }
        mp4Stream=NULL;
        if(termCap)
                delete termCap;
        if(userPref)
                delete userPref;
        if(server)
                delete server;
        
        if(suggestedAdaptor)
                delete suggestedAdaptor;
}


/***********************************************************************/
bool ProxySession::options(const Url* fileName, const char* remaining)
{
  prot->incSeqNr();
  return true;
}


/*
 *      -parse cmd describe, detect if client supports retransmission
 *  -check with CM: if not or only partially cached forward the command to the server
 *  and create a ServerSession object
 *  -read the reply from the server, detect if server supports retransmission
 */
bool ProxySession::connect(const Url* fileName, const char* remaining)
{
        assert(fileName && remaining && prot);
        bool ret=prot->parseCMDDescribe(this, fileName, remaining,&playerId);
        // illegal describe??
        if(!ret) {
                prot->generateBadCMD();
                const char* buf=prot->getBuffer();
                strcpy(buffer,buf);
                state=SESSION_ERR;
                prxStat.rejectedDueToErr++;
        }
        // problem: a 2nd describe arrives, don't allow changing termcaps in the same session!
        else {
                // we have a valid DESCRIBE
                // is this the first describe? then mp4Stream should be NULL
                TerminalCapabilities* dummyTermCap=MP21::createTerminalCapability(remaining);
                if(dummyTermCap) {
                        // overwrite the default with the new ones
                        dprintf_full("ProxySession::connect Found TermCaps\n");
                        if(termCap)
                                delete termCap;
                        // if termcaps are specified, they take precedence over default UserPrefs!
                        if(userPref)
                                delete userPref;
                        userPref=NULL;
                }
                if(dummyTermCap)
                        termCap=dummyTermCap;

                if(!termCap) {
                        // maybe UserPreferences?
                        // keep existing UserPref
                        UserPreferences* upDummy=MP21::createUserPreferences(remaining);
                        dprintf_full("ProxySession::connect Found UserPreferences %p in XX%sXX\n",upDummy,remaining);
                        
                        if(userPref && upDummy)
                                delete userPref;
                        if(upDummy)     {
                                userPref=upDummy;
                        }
                }
                setUrl(fileName); // sets insertMP4.. to true if adaptation is needed
                
                if(!mp4Stream) // no resUsage known at this time point
                        insertMp4StreamIntoCM=true;
                else if(resUsage.network < 0.0000001) { 
                        // mp4stream is set
                        canAdmitRequest=false;
                        fprintf(stderr,"ProxySession::connect: Request rejected due to AdmissionControl\n");
                        state=SESSION_ERR;
                        prot->generateNotEnoughBandwidth();
                        prot->incSeqNr();
                        insertMp4StreamIntoCM=false;
                        if(origSrc && mp4Stream) {
                                delete mp4Stream; // clone created in setUrl
                        }
                        origSrc=mp4Stream=NULL;
                        return true;
                }

                if(origSrc)
                        origSrc->getUsageCounter()->increase();
                if(mp4Stream)
                        mp4Stream->getUsageCounter()->increase();
                if(!mp4Stream) {
                        // serverConnection needed
                        dprintf_full("ProxySession::connect with url %s fileName %s\n",url->toString(),fileName->toString());
                        if(!establishConnectionToServer()) {
                                dprintf_full("ProxySession::connect with url %s fileName %s\n",url->toString(),fileName->toString());
                                // failed to connect to server
                                prot->generateDestUnreachable();
                                strcpy(buffer,prot->getBuffer());
                                ret=false;
                        }
                        else{
                                dprintf_full("ProxySession::connect with url %s fileName %s\n",url->toString(),fileName->toString());
                                ret=handleConnectMiss(fileName); //handle miss comm.
                                
                                if(resUsage.network < 0.00001) { 
                                        // mp4stream is set
                                        canAdmitRequest=false;
                                        fprintf(stderr,"ProxySession::connect: Request rejected due to AdmissionControl\n");
                                        state=SESSION_ERR;
                                        insertMp4StreamIntoCM=false;
                                        if(origSrc)
                                                delete origSrc;
                                        origSrc=NULL;
                                        if(mp4Stream)
                                                delete mp4Stream;
                                        mp4Stream=NULL;
                                        if(mp4StreamReserved > 0)
                                        {
                                                cm->freeDiskSpace (mp4StreamReserved);
                                                mp4StreamReserved = 0;
                                        }
                                        if(origSrcReserved > 0)
                                        {
                                                cm->freeDiskSpace (origSrcReserved);
                                                origSrcReserved = 0;
                                        }
                                        prot->generateNotEnoughBandwidth();
                                        prot->incSeqNr();
                                        const char* buf=prot->getBuffer();
                                        strcpy(buffer,buf);
                                        return true;
                                }
                                if(mp4Stream)
                                        mp4Stream->getUsageCounter()->increase();
                                if(origSrc)
                                        origSrc->getUsageCounter()->increase();
                                if(ret)
                                        ret=handleConnectHit(fileName);
                        }               
                }
                else {
                        ret=handleConnectHit(fileName);
                }
                detectRtxCapabilities();
                if(!mp4Stream) {
                        ret=false;
                }
                else if(ret) {
                        if(noDescribeSeenYet) {
                                /*
                                // insert video into cm! we do this in delete!
                                if(insertMp4StreamIntoCM && mp4Stream) {        
                                        mp4Stream->getUsageCounter()->increase();
                                        cm->putVideo(mp4Stream);
                                }
                                */
                                ;
                        }
                        if(state!=SESSION_ERR)
                                state=SESSION_INITIALIZED;
                        noDescribeSeenYet=false;
                }
        }               
        if(!ret) {
                state=SESSION_ERR;
        }
        prot->incSeqNr();
        return ret;
};

bool ProxySession::handleConnectHit(const Url* fileName)
{
        // no serverconnection is needed, no request forwarding is needed
        assert(mp4Stream);
        dprintf_full("ProxySession::handleConnectHit\n");

        bool ret=prot->generateCMDDescribeReply(this,fileName,fileName->getServerName(),playerId);
        const char* buf=prot->getBuffer();
        strcpy(buffer,buf);
        return ret;     
};

bool ProxySession::handleConnectMiss(const Url* fileName)
{
        // we need a serverconnection
        assert(serverSession);
        dprintf_full("ProxySession::handleConnectMiss fn%p (%s) s%p\n", fileName,fileName->toString(),serverSession);
        if(!serverSession->getProtocol())
                serverSession->determineProtocol(buffer);
        if(serverID)
                delete serverID;
        // reset rtx_groups
        std::list < rtx_group* >::iterator j;
        while(!rtxInfo.empty()) {
                j=rtxInfo.begin();
                delete *j;
                rtxInfo.pop_front();
        }       

        // if playerId is not our player and the server is our server
        // regenerate the describe, so that we can enable rtx
        // Problem: we don't know with which kind of server we are dealing at the moment
        // so regenerate the DESCRIBE always!!!
        // replace the User Agent string with our id
        // no foreign player supports termcaps so far, set to NULL
        if(neverForwardMetaData)
                serverSession->getProtocol()->generateDescribeForUrl(url->toString(),PLAYER_ID,NULL);
        else {
                if(this->userPref) 
                        serverSession->getProtocol()->generateDescribeForUrl(url->toString(),PLAYER_ID,NULL,userPref);
                else if(this->termCap)
                        serverSession->getProtocol()->generateDescribeForUrl(url->toString(),PLAYER_ID,termCap);
                else
                        serverSession->getProtocol()->generateDescribeForUrl(url->toString(),PLAYER_ID,NULL);
        }
        strcpy(buffer,serverSession->getProtocol()->getBuffer());
        
        //fwd the cmd to the server
        dprintf_full("ProxySession::handleConnectMiss: responding...\n");
        serverSession->sendResponse(buffer,strlen(buffer));
        serverSession->readRequest(buffer,MSG_BUFFER_SIZE);
        dprintf_full("ProxySession::handleConnectMiss: after request forwarding %p\n",serverSession->getProtocol());
        serverSession->getProtocol()->incSeqNr();
        dprintf_full("ProxySession::handleConnectMiss: after incSeqNr \n");
        ContainerInfo* dummy=NULL;
        if(mp4Stream) {
                //FIXME?  filename->toString or url->toString.... why fileName not url??? seems the same....
                dummy=prot->parseResponseDescribe(buffer, fileName->toString(),&serverID, &this->rtxInfo);
                assert(dummy);
                delete dummy; //just use the old mp4stream
        }
        else
                //FIXME?  filename->toString or url->toString.... why fileName not url??? seems the same....
                mp4Stream=prot->parseResponseDescribe(buffer, fileName->toString(),&serverID, &this->rtxInfo);
        dprintf_full("ProxySession::handleConnectMiss connect to server >%s<, url #%s#  fileName #%s#\n",serverID,url->toString(), fileName->toString());
        if(!mp4Stream) {
                // forward the buffer from the server
                return false;
        }
        
        //FIXME: THESE TWO BLOCKS ARE HACKS!
        // correct voptimeincrement
        VideoESInfo* ves=mp4Stream->getFirstVisualES();
        if(ves) {
                MP4Decoder* mp4d=new MP4Decoder(ves,true);
                mp4d->initialize();
                delete mp4d;
        }
        
        AudioESInfo* aes=mp4Stream->getFirstAudioES();
        if(aes) {
                MP4audioDecoder* mp4a=new MP4audioDecoder(aes,true);
                mp4a->initialize();
                delete mp4a;
        }

        // check for transcoding!!!!!
        prxStat.bytesRequested+=mp4Stream->getEstimatedNetworkFileSize();
        ContainerInfo* pCI=mp4Stream->clone();
        MetaObject* meta=new MetaObject(pCI);
        if(userPref) {
                // FIXME: hardcode delay values to 100ms
                dprintf_full("ServerSession::handleConnectMiss  meta && userPref\n");
                if(suggestedAdaptor)
                        delete suggestedAdaptor;
                suggestedAdaptor=admControl->makeBestMatch(userPref,100,meta,&pCI,&resUsage);

        }
        else if(termCap) {
                // bandwidth should be the minimum of network and decodeBandwidth
                // if one of them is zero, ignore the zero param
                u32 bandwidth=termCap->getNetworkCapacityInByte()*8;
                if(bandwidth==0)
                        bandwidth=termCap->getMaxDecoderBitRateInBit();
                else if(termCap->getMaxDecoderBitRateInBit() && 
                        termCap->getMaxDecoderBitRateInBit() < bandwidth) {
                        bandwidth=termCap->getMaxDecoderBitRateInBit();
                }
                // the returned mp4stream is equal to the previous mp4Stream here
                // we have only one in MetaObject
                bool exactMatch=false;
                //              ContainerInfo* dummyMp4Stream = 
                meta->findMp4Stream(0,termCap->getDisplayWidth(),0,termCap->getDisplayHeight(),
                                                                                                0,bandwidth,(termCap->getColorDisplay()==0),
                                                                                                (float)termCap->getDisplayRefreshRate() ,&exactMatch);
                if(!exactMatch) {
                        insertMp4StreamIntoCM=true;
                        suggestedAdaptor=admControl->makeExactMatch(termCap->getDisplayWidth(),
                                                                        termCap->getDisplayHeight(),
                                                                        bandwidth,
                                                                        (termCap->getColorDisplay()==0),
                                                                        (float)termCap->getDisplayRefreshRate(),
                                                                        mp4Stream,&resUsage);
                }

        }       

        if (!suggestedAdaptor) {// no transcoding
                VideoESInfo* vis=mp4Stream->getFirstVisualES();
                if(vis) {
                        Feature src(vis);
                        src.isCached=false;
                        ResourceUsage* tmpRU=CostFunction::calcResourceUsage(&src,&src,admControl->getResourceLimit());
                        if(tmpRU) {
                                resUsage.network=tmpRU->network;
                                resUsage.disk=tmpRU->disk;
                                resUsage.cpu=tmpRU->cpu;
                                delete tmpRU;
                        }
                }
                insertMp4StreamIntoCM=true;
        }

        if(suggestedAdaptor && mp4Stream)
                suggestedAdaptor->setESInfo(mp4Stream->getFirstVisualES());

        delete meta;

        // check for admission but not for userPrefs+termcaps which are already inserted!
        if(!userPref && !termCap && !admControl->insertRequest(&resUsage)) {
                resUsage.cpu=resUsage.disk=resUsage.network=0.0;
                if(mp4Stream)
                        delete mp4Stream;
                mp4Stream=NULL;
                if(suggestedAdaptor)
                        delete suggestedAdaptor;
                suggestedAdaptor=NULL;
        }

        if(mp4Stream && suggestedAdaptor &&
                        strcmp(suggestedAdaptor->getName(),"Forwarder")!=0 && 
                        !strstr(suggestedAdaptor->getName(),"Temporal")) { // transcoding
                                // we must insert two ContainerInfos!
                                dualMiss=true;
                                origSrc=mp4Stream;
                                mp4Stream=mp4Stream->clone();
                                insertMp4StreamIntoCM=true;
        }       else if(mp4Stream && suggestedAdaptor &&
                        (strcmp(suggestedAdaptor->getName(),"Forwarder")==0 ||
                        strstr(suggestedAdaptor->getName(),"Temporal"))) { 
                insertMp4StreamIntoCM=false;
        }

        if(insertMp4StreamIntoCM && mp4Stream) {
                u64 durInMS=0;
                if(mp4Stream->getFirstVisualES())
                        durInMS=mp4Stream->getFirstVisualES()->getDurationInMs();
                else if(mp4Stream->getFirstAudioES())
                        durInMS=mp4Stream->getFirstAudioES()->getDurationInMs();
                u32 byteSize=0;
                if(userPref) {
                        byteSize=userPref->bitrate.best/8*durInMS/1000;
                }
                else if(termCap) {
                        u32 bandwidth=termCap->getNetworkCapacityInByte()*8 -1;
                        if(bandwidth==0)
                                bandwidth=termCap->getMaxDecoderBitRateInBit();
                        else if(termCap->getMaxDecoderBitRateInBit() && 
                                termCap->getMaxDecoderBitRateInBit() < bandwidth) {
                                bandwidth=termCap->getMaxDecoderBitRateInBit();
                        }
                        byteSize=bandwidth/8*durInMS/1000;
                        if(byteSize > mp4Stream->getEstimatedNetworkFileSize())
                                byteSize=mp4Stream->getEstimatedNetworkFileSize();
                }
                else
                        byteSize=mp4Stream->getEstimatedNetworkFileSize();
                cm->lockCache();
                assert(mp4StreamReserved == 0);
                if(cm->reserveDiskSpace(byteSize))
                        mp4StreamReserved+=byteSize;
                else {
                        insertMp4StreamIntoCM=false;
                }
                cm->unlockCache();
        }

        if(dualMiss && origSrc) {
                dprintf_full("ProxySession::handleConnectMiss dualMiss!\n"); 
                cm->lockCache();
                assert(origSrcReserved == 0);
                if(cm->reserveDiskSpace(origSrc->getEstimatedNetworkFileSize()))
                        origSrcReserved=origSrc->getEstimatedNetworkFileSize();
                else {
                        origSrcReserved=0;
                        // we can't cache the original but maybe we can cache the transcoded!
                        // DON't change insertmp4streamintocm!!!
                        dontCacheDualMiss=true;
                        /*insertMp4StreamIntoCM=false;
                        if(mp4StreamReserved > 0)
                        {
                                cm->freeDiskSpace (mp4StreamReserved);
                                mp4StreamReserved = 0;
                        } */
                }
                cm->unlockCache();
        }

        return true;    
};


void ProxySession::detectRtxCapabilities()
{
        if(serverID && strstr(serverID,SERVER_ID)) {
                rtxEnabledToServer=true&rtxEnabled;
        }
        else {
                rtxEnabledToServer=false;
        }
        if(playerId && strstr(playerId,PLAYER_ID)) {
                rtxEnabledToClient=true&rtxEnabled;
        }
        else {
                rtxEnabledToClient=false;
        }
        dprintf_full("ProxySession::detectRtxCapabilities() rtxToClient %i rtxToServer %i\n",rtxEnabledToClient,rtxEnabledToServer);
};

IO* ProxySession::handleSetupMiss(const Url* fileName, int esId, int* sessionId)
{
        assert(mp4Stream && serverSession && serverSession->getProtocol() && sessionId);
        // always regenerate the setup request to catch scenarios where the client supports
        // retransmission and the server doesn't
        ESInfo* es=mp4Stream->getES((u32)esId);
        if(!es) {
                // stream not found
                prot->generateFileNotFound();
                strcpy(buffer,prot->getBuffer());
                return NULL;
        }
        RTSP* serverProt=serverSession->getProtocol();
        IO *input=NULL;
        int proxyPortToServer=0,serverPortToProxy=0, proxyRtxPortToServer=0;
        int retProxyPortToServer=0;
        bool bUnicast=true,retBUnicast=false;
        char* ssrc=NULL;
        rtx_info* esRtxInfo=new rtx_info();
        proxyPortToServer=portGen.getNextFreePortPair();
        if(rtxEnabledToServer) {
                dprintf_full("ProxySession::setup rtx to server is enabled\n");
                proxyRtxPortToServer=portGen.getNextFreePortPair();
        }
        serverProt->generateSetup(fileName->toString(),esId,proxyPortToServer,bUnicast,proxyRtxPortToServer,*sessionId);

        const char* serverBuffer=serverProt->getBuffer();

        //fwd the SETUP to the server
        serverSession->sendResponse(serverBuffer,strlen(serverBuffer));
        serverSession->readRequest(buffer,MSG_BUFFER_SIZE); // reply from server is stored in buffer
        serverSession->getProtocol()->incSeqNr();
        // parse the reply from the server
        int retVal=prot->parseResponseSetup(buffer,sessionId,&retProxyPortToServer,
                &serverPortToProxy,&retBUnicast,&ssrc,esRtxInfo);
        if(retVal<2 || retProxyPortToServer!=proxyPortToServer || retProxyPortToServer<=0 || 
                serverPortToProxy<=0 || retBUnicast!=bUnicast || retBUnicast==false) {
                // generate error
                prot->generateBadCMD();
                portGen.closePortPair(proxyPortToServer);
                if(rtxEnabledToServer)
                        portGen.closePortPair(proxyRtxPortToServer);
                
                strcpy(buffer,prot->getBuffer());
                input= NULL;
        }
        else {
                // succeeded in parsing reply
                // we can now create the Input IO class
                // but first check if the server returned any rtx info
                if(rtxEnabledToServer && (esRtxInfo->localPort<=0 || esRtxInfo->remotePort<=0)) {
                        portGen.closePortPair(proxyRtxPortToServer);
                        esRtxInfo->localPort=0; 
                        esRtxInfo->remotePort=0;
                }
                // IncompleteIO is too difficult to handle
                // for now just assume, that it's a full miss

                ESInfo* es=mp4Stream->getES((u32)esId);
                // build Rtp as input
                PacketizationLayer* pl=new RFC3016();
                
                const char* urlServerName=url->getServerName();

#ifndef WINCE
                if(rtxEnabled) {
#endif
                        Statistics *stats = new Statistics(es,false);
                        input=new Rtp(fileName->getPath(),serverPortToProxy,urlServerName,
                                        proxyPortToServer,es,pl,false,termCap,NULL,stats);
#ifndef WINCE
                }
                else {
                        char *logName = new char[MAX_STR_LEN];
                        struct timeval t;
                        gettimeofday(&t,NULL);
                        sprintf(logName,"log%10i.%3i_proxy.in",(int)t.tv_sec,(int)(t.tv_usec/1000));
                        input=new SimpleRtp(fileName->getPath(),serverPortToProxy,proxyPortToServer,
                                        urlServerName,es,pl,false,logName,true,SimpleRtp::MIXED_CTS_BITRATE,true,true);
                        delete [] logName;
                }
#endif


                if(rtxEnabledToServer) {
                        dprintf_full("Proxysession::setup rtx for input is enabled\n");
                        input->setRtxInfo(esRtxInfo);
                }
                rtxInfoServer.push_front(esRtxInfo); // save it for garbage collection          
                esRtxInfo=NULL;
        }
        // cleanup
        if(ssrc)
                delete ssrc;
        if(esRtxInfo)
                delete esRtxInfo;
        return input;
};
DataSink* ProxySession::handleSetupHit(const Url* fileName, int esId, 
                                                                                                                                                         int remoteClientPort, int remoteUpperClientPort,
                                                                                                                                                         int remoteClientRtxPort,
                                                                                                                                                         int sesKey, const char* protocol, const char* unicast)
{
        int localPortToClient=portGen.getNextFreePortPair();
        int proxyRtxPortToClient=0;
        if(remoteClientRtxPort>0 && rtxEnabledToClient) {
                proxyRtxPortToClient=portGen.getNextFreePortPair();
        }
        else {
                remoteClientRtxPort=proxyRtxPortToClient=0;
        }
        
        // create an RTP with RFC3016 output channel
        // create DataChannel
        // open RTP session

        ESInfo* es=mp4Stream->getES((u32)esId);
        if(!es) {
                // stream not found
                prot->generateFileNotFound();
                strcpy(buffer,prot->getBuffer());
                return NULL;
        }
        
  PacketizationLayer * p = new RFC3016();
  char *clientAddress = inet_ntoa(client->sin_addr);    // points to a static array in inet..
        Statistics *stats = NULL;
        IO* output=NULL;
        // no probs with inet_ntoa: rtp deep-copies the clientaddress

#ifndef WINCE
        if(rtxEnabled) {
#endif
                stats = new Statistics(es,true);
                output = new Rtp(fileName->getPath(), remoteClientPort, clientAddress, localPortToClient,
                                                                                         es, p, true,termCap,NULL,stats);
#ifndef WINCE
        }
        else {
                char *logName = new char[MAX_STR_LEN];
                struct timeval t;
                gettimeofday(&t,NULL);
                sprintf(logName,"log%10i.%3i_proxy.out",(int)t.tv_sec,(int)(t.tv_usec/1000));
                output=new SimpleRtp(fileName->getPath(),remoteClientPort,localPortToClient,
                clientAddress,es,p,true,logName,true,SimpleRtp::MIXED_CTS_BITRATE,false,false);
                delete [] logName;
        }
#endif

        this->prot->generateSetupReply(this,fileName,sesKey,protocol,unicast,clientAddress,
                esId,remoteClientPort,remoteUpperClientPort,localPortToClient,remoteClientRtxPort,proxyRtxPortToClient);

        strcpy(buffer,prot->getBuffer());
        
        rtx_group* group=getRtxGroup(esId);
        if(rtxEnabledToClient && proxyRtxPortToClient>0 && group && group->rtx) {
                dprintf_full("Proxysession::setup rtx for output is enabled\n");
                group->rtx->remotePort=remoteClientRtxPort;
                group->rtx->localPort=proxyRtxPortToClient;
                output->setRtxInfo(group->rtx);
        }
        else {
                dprintf_err("Ses has stored %i groups \n",rtxInfo.size());
                dprintf_full("ProxySession::setup: No rtx_port for es %i [%u,%u] g:%p\n", 
                        esId,remoteClientRtxPort,proxyRtxPortToClient,group);
        }
        output->open();
/*      
        if(     playerId && suggestedAdaptor==NULL && es && es->isVisualStream() && 
                        strstr(playerId, QT_PLAYER_ID) && 
                        strstr(playerId, OLD_BFRAMES_DISABLED_QT_PLAYER) ) {
                suggestedAdaptor=new TemporalAdaptor((VideoESInfo*)es);
        }
*/
        
        DataSink* clientObj =
                new DataSink((u32)client->sin_addr.s_addr,output, clientAddress, 
                        ntohl(client->sin_port));
        return clientObj;
};

bool ProxySession::setup(const Url* fileName, const char* remaining)
{
        assert(fileName && remaining && prot);
        dprintf_full("ProxySession::setup %s\n",url->toString());
        bool hasSessionKey=false,hasRtxPort=false;
        char* protocol=NULL;
        char* unicast=NULL;

        int sesKey=0,esId=0,rPort=0,rPort2=0;
        u32 clientRtxPort=0;
        //u32 proxyRtxPortToClient=0, proxyRtxPortToServer=0;

        bool err=!prot->parseCMDSetup(this,fileName,remaining,&hasSessionKey,&sesKey,&protocol,
                &unicast,&esId,&rPort,&rPort2,&hasRtxPort,&clientRtxPort);
        if(!err) {
                if(!prot->isSupportedTransport(protocol)) {
                        prot->generateIllegalTransport();
                        err=true;
                }
        } 
        else {
                prot->generateBadCMD();
        }

        if(err) {
                dprintf_err("ProxySession::setup error in parsing setup request\n");
                const char* buf=prot->getBuffer();
                strcpy(buffer,buf);
                if(protocol)
                        delete protocol;
                if(unicast)
                        delete unicast;
                prot->incSeqNr();
                return false;
        }
        if(hasSessionKey==false) {
                sesKey=prot->generateSessionKey();
        }

        // now check: is this ES cached?
        ESInfo* es=mp4Stream->getES((u32)esId);
        IO* input=NULL;
        DataSink* clientObj=NULL;
        // we accept as soon as we have more than 90%
        
        u32 uiSize=es->getCurrentDemuxedMediaSize();
        if(uiSize>0) {
                // cache hit, hurrahhh!
                // build an MPGStreamIO as input
                dprintf_full("ProxySession::setup HIT\n");
                input=IOCreator::createInput(es, true);
                clientObj=handleSetupHit(fileName,esId,rPort,rPort2,clientRtxPort,sesKey,protocol,unicast);
        }
        else {
                // cache miss
                dprintf_full("ProxySession::setup MISS\n");
                int sessionId=-1; // the sessionid from the server
                if(sessionTranslator.find(sesKey)!=sessionTranslator.end()) {
                        // we already have a key
                        sessionId=sessionTranslator[sesKey];
                }
                int bakSessionId=sessionId;
                input=handleSetupMiss(fileName,esId,&sessionId);
                if(input) {
                        if(bakSessionId!=sessionId) // a new one was generated
                                sessionTranslator[sesKey]=sessionId;
                        clientObj=handleSetupHit(fileName,esId,rPort,rPort2,clientRtxPort,sesKey,protocol,unicast);
                }
        }
        if(input && clientObj) {
                // now create the DataChannel
                // make sure we ve got the right es!
                if(suggestedAdaptor && es->isVisualStream())
                        suggestedAdaptor->setESInfo(es);
                if(dualMiss && origSrc && !dontCacheDualMiss) {
                        ESInfo* esFullQ=origSrc->getES((u32)esId);

                        if(esFullQ) {
                                
                                // save the full quality to a file and set the new input at origSrc
                                // add a DataDumper to suggestedAdaptor
                                char* pFileName=Url::transformUrlToLocalFile(origSrc->getUrl());
                                FILE* fp=NULL;
                                char *out = new char[4096];
                                strcpy(out,"demux");
#ifdef WIN32
                                strcat(out,"\\");
#else
                                strcat(out,"/");
#endif
                                strcat(out,pFileName);
                                delete[] pFileName;
                                while((fp=fopen(out,"r")) != 0) {
                                        // merde! file exists
                                        sprintf(out,"%s%u",out,(unsigned int)(rand()%10));
                                        fclose(fp);fp=NULL;
                                }
                                if(fp)
                                        fclose(fp);
                                // create a writer, use esFullQ to get correct FrameStatistic
                                MPGStreamIO* output = new MPGStreamIO(esFullQ, out, true, false);
                                output->open();
                                AdaptorChain* pAc;
                                if(es->isAudioStream())
                                        pAc=new AudioAdaptorChain();
                                else if(es->isVisualStream())
                                        pAc=new VideoAdaptorChain();
                                if(es->isAudioStream()||es->isVisualStream()) {
                                        pAc->addAdaptor(new DataDump(es,output));
                                        pAc->addAdaptor(suggestedAdaptor);
                                        suggestedAdaptor=pAc;
                                        esFullQ->setInput(out);
                                }

                                delete [] out;
                        }
                }
                DataChannel * d = new DataChannel(input, es,(es->isVisualStream()?this->suggestedAdaptor:NULL));
                suggestedAdaptor=NULL;
                d->setSessionId(sesKey);
                d->insert(clientObj); // add the client to the DC
                
                if(insertMp4StreamIntoCM) {
                        char* outCacheName=d->enableCaching(false,"demux",true,fileName);
                        // we must delete the old file and set the new one but without incompleteio there is no old file :-)
                        // set the new one here because other clients could use the same ESInfo
                        dprintf_full("ProxySession.setup: prior calling setInput\n");
                        es->setInput(outCacheName);
                        delete[] outCacheName;
                }

                // set the DataChannel at the Session
                setDataChannel(d);
                d=NULL;
                clientObj=NULL;
                input=NULL;
                err=false;
        }
        else {
                err=true;
                prot->generateInternalServerError();
                strcpy(buffer,prot->getBuffer());
        }
        
        //cleanup       
        if(protocol)
                delete protocol;
        if(unicast)
                delete unicast;
        if(input)
                delete input;
        if(clientObj)
                delete clientObj;
        prot->incSeqNr();
        return !err;
}

bool ProxySession::play(const Url* fileName, const char* remaining)
{
        assert(fileName && remaining);

        int sessionId=0;u32 startFrame=0, endFrame=0;
        bool res=false;
        bool remote=false;  
        double start=0.0, end=0.0;
        double prefetch=0.0;

        if(!prot->parseCMDPlay(this,fileName,remaining,&sessionId,&start,&end,&prefetch)) {
                prot->generateSessionNotFound();
                const char* buf=prot->getBuffer();
                strcpy(buffer,buf);
                prot->incSeqNr();
                return false;
        }
        if(end<=0.00001) {
                end=mp4Stream->getFirstVisualES()->getDurationInMs()/1000.0;
        }
        // if we can find an entry in sessiontranslator then we have to fwd the play
        remote=(sessionTranslator.find(sessionId)!=sessionTranslator.end());
        if(remote) {
                // fwd,don't allow proxy to introduce such a high delay, reduce prefetch
                serverSession->getProtocol()->generatePlay(url->toString(),sessionTranslator[sessionId],start,end,prefetch/2);
                const char* playForServer=serverSession->getProtocol()->getBuffer();
                serverSession->sendResponse(playForServer,strlen(playForServer));

                // get reply
                serverSession->readRequest(bufferBackup,MSG_BUFFER_SIZE);
                serverSession->getProtocol()->incSeqNr();
                if(!RTSP::isCmdOk(bufferBackup)) {
                        dprintf_err("ProxySession::play: Server sent an error\n");
                        prot->incSeqNr();
                        strcpy(buffer,bufferBackup);
                        return false;
                }
        }
        // if we came that far we have an okay
        const ESInfo* es=NULL;
        for (list < DataChannel * >::iterator i = this->channels.begin(); i != channels.end();
        i++) {
                dprintf_full("ProxySession::play: Looking for sesId %i, found %i\n",sessionId,(*i)->getSessionId());
                if ((*i)->getSessionId() == (u32)sessionId) {
                        es=(*i)->getESInfo();
                        
                        if(es) {
                                startFrame=es->getFrameNumber(start);
                                endFrame=es->getFrameNumber(end);
                                DataSink* clientOut=(*i)->getDataSink(clientId);
                                IO* clientOutput=NULL;

                                if(clientOut && (clientOutput=clientOut->getOutput())) {                                        
                                        // flush the Rtp output buffer!!!
                                        clientOutput->setToFrameNumber(startFrame);
                                }
                                if(!remote)
                                        (*i)->play(clientId,prefetch);
                                else
                                        (*i)->play(clientId,prefetch/2);
                                IO* tmpInput=(*i)->getInput();
                                dprintf_full("ProxySession::play: got input %p\n",tmpInput);
                                if(tmpInput->getState()==IO::OPEN && !tmpInput->setToFrameNumber(startFrame)) {
                                                dprintf_err("ProxySession::play Failed to set input to frame %i\r\n",startFrame);
                                }
                                else
                                        res=true;

                                // if not yet started -> start it
                                if (!(*i)->running()) {
                                        (*i)->start();
                                }
                        }
                        else {
                                startFrame=0;
                                dprintf_err("RTSP::handleCMDPlay Missing ES for request? PANIC!\r\n");
                        }
                }
        }
        if(remote)
                strcpy(buffer,bufferBackup);
        else {
                if(res)
                        prot->generateOk();
                else
                        prot->generateSessionNotFound();

                const char* buf=prot->getBuffer();
                strcpy(buffer,buf);
        }
        prot->incSeqNr();
        return res;
}

bool ProxySession::pause(const Url* fileName, const char* remaining)
{
        bool found=false;
        assert(fileName && remaining);
        int sessionId=prot->extractSessionKeyFromCMD(remaining);
        // if we can find an entry in sessiontranslator then we have to fwd the pause
        bool remote=(sessionTranslator.find(sessionId)!=sessionTranslator.end());       
        if(remote) {
                serverSession->getProtocol()->generatePause(url->toString(),sessionTranslator[sessionId]);
                const char* pauseForServer=serverSession->getProtocol()->getBuffer();
                serverSession->sendResponse(pauseForServer,strlen(pauseForServer));
                // get reply
                serverSession->readRequest(bufferBackup,MSG_BUFFER_SIZE);
                serverSession->getProtocol()->incSeqNr();
        }
  if (sessionId > 0 && ((u32)sessionId)!=INVALID_SESSIONID) {
    // search for the session
    // start it
    for (list < DataChannel * >::iterator i = this->channels.begin();
         i != this->channels.end(); i++) {
      if ((*i)->getSessionId() == (unsigned)sessionId) {
                                found=true;
                                (*i)->pause(clientId);
                                
      }
    }
  }
        if(!remote && !found) {
                prot->generateSessionNotFound();
                dprintf_small("ProxySession::pause session %i not found\n",sessionId);
                strcpy(buffer,prot->getBuffer());
        }
        else {
                if(RTSP::isCmdOk(bufferBackup))
                        found=true;
                strcpy(buffer,bufferBackup);
        }
        prot->incSeqNr();       
        return found;
}


bool ProxySession::tearDown(int sessionId, bool immediate, const Url* fileName, const char* remaining)
{
        dprintf_full("ProxySession::tearDown(%i, %i)\r\n",sessionId,immediate);
        if(fileName && remaining) {
                sessionId=RTSP::extractSessionId(remaining);
        }
        // serverSession->tearDown(sessionId,immediate,fileName,remaining);
        if( ((u32)sessionId)==TEARDOWN_ALL) {
                // teardown all
                state = SESSION_CLOSED;
                list < DataChannel * >::iterator dcI;
                while (!channels.empty()) {
                        dcI=channels.begin();
                        // pause the caching client
                        (*dcI)->pause(0);
                        DataSink* ds=(*dcI)->getDataSink(0);
                        if(ds) {
                                IO* dsOut=ds->getOutput();
                                if(dsOut) {
                                        ESInfo* es=(ESInfo*) (*dcI)->getESInfo();
                                        
                                        float completeStatePercent=es->getFrameStatistic()->getPercentageSetBits(); //expensive
                                        if(completeStatePercent>0.75) {
                                                es->setCompleteState(true);
                                                dprintf_full("ProxySession::teardown: stream %s is complete with %f%%\r\n",es->getInput(),completeStatePercent);
                                        }
                                        dprintf_full("ProxySession::teardown: stream %s is cached with %f%%\r\n",es->getInput(),completeStatePercent);
                                        es->getUsageCounter()->decrease();
                                }
                        }
                        (*dcI)->teardown(clientId,immediate);
                        // 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); // decrease RefCounter
                        }
                        channels.pop_front();
                }
                // generate reply cmd
                prot->generateOk();
                strcpy(buffer,prot->getBuffer());
                prot->incSeqNr();
                return true;
        }
        else {
                // teardown one session
                list < DataChannel * >::iterator dcI;
                if (!channels.empty()) {
                        dcI = channels.begin();
                        while (dcI != channels.end()) {
                                if ((*dcI)->getSessionId()== (unsigned)sessionId) {
                                        (*dcI)->pause(0);
                                        DataSink* ds=(*dcI)->getDataSink(0);
                                        if(ds) {
                                                IO* dsOut=ds->getOutput();
                                                if(dsOut) {
                                                        ESInfo* es=(ESInfo*) (*dcI)->getESInfo();
                                                        
                                                        float completeStatePercent=es->getFrameStatistic()->getPercentageSetBits(); //expensive
                                                        if(completeStatePercent>0.75) {
                                                                es->setCompleteState(true);
                                                                dprintf_full("ProxySession::teardown: stream %s is complete with %f%%\r\n",es->getInput(),completeStatePercent);
                                                        }
                                                        dprintf_full("ProxySession::teardown: stream %s is cached with %f%%\r\n",es->getInput(),completeStatePercent);
                                                        es->getUsageCounter()->decrease();
                                                }
                                        }
                                        (*dcI)->teardown(clientId,immediate);
                                        // 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;
                                        prot->generateOk();
                                        strcpy(buffer,prot->getBuffer());
                                        prot->incSeqNr();       
                                        return true;
                                }
                                else {
                                        ++dcI;
                                }
                        }
                }
                
        }
        // session was not found
        prot->generateSessionNotFound();
        dprintf_small("ProxySession::teardown: session %i not found\n",sessionId);
        strcpy(buffer,prot->getBuffer());
        prot->incSeqNr();       
        return false;
};

//is teardown on streams or sessions ???

void ProxySession::run()
{
        // while not end of session
        //check for control command
        // do control command
        state = SESSION_NEW;
        /* the buffer for incoming rtsp/http commands */
        if(this->counter != NULL)
                this->counter->increment();
        while (state!=SESSION_ERR && state!=SESSION_CLOSED) {
                buffer[0] = '\0';
                bufferBackup[0]='\0';
                dprintf_full("ProxySession::run: Listening for data on ControlChannel\r\n");
                int bytesRead = readRequest(buffer, MSG_BUFFER_SIZE);
                strcpy(bufferBackup, buffer);
                dprintf_full("ProxySession::run: readBlock ended. read %i bytes\r\n",bytesRead);
                if (bytesRead == 0) {
                        //no data read
                        // the other side has closed the connection, teardown all
                        dprintf_full("ProxySession::run: No data read from ControlChannel\r\n");
                        tearDown(TEARDOWN_ALL,true, NULL, NULL);
                } else if (bytesRead < 0) {
                        // error:
                        dprintf_err("ProxySession::run Error reading from controlchannel");
                        tearDown(TEARDOWN_ALL,true,NULL, NULL);
                        state = SESSION_ERR;
                } else {                // we got a command
                        dprintf_full("ProxySession::run: got command\r\n");
                        // check for protocol if not yet set
                        if (!prot) {
                                if ( determineProtocol(buffer) == PROTO_UNKNOWN) {
                                        dprintf_err("ProxySession::run: Unknown protocol: %s\n",
                                        buffer);
                                }
                        }
                        if (prot) {
                                switch (prot->getID()) {
                                case PROTO_RTSP:
                                case PROTO_HTTP:
                                        if (bytesRead >= MSG_BUFFER_SIZE) {
                                                dprintf_err("received command exceeded buffer size");
                                                break;
                                        }
                                        break;
                                default:
                                        dprintf_err("ProxySession::run: Unknown protocol: %s\n",buffer);
                                        break;
                                }

                                if (!prot->applyReqToSession(buffer, bytesRead, this)) {
                                        dprintf_err("ProxySession::run: Failed to parse request >%s<\n",        buffer);
                                }
                                if(strlen(buffer)==0) {
                                        prot->generateInternalServerError();
                                        strcpy(buffer,prot->getBuffer());
                                }
                                sendResponse(buffer,strlen(buffer));
                        }       
                }
  }                             // while valid     
  if(this->counter != NULL)
                this->counter->decrement();
        this->exit();

};


void ProxySession::setUrl(const Url *uri, bool makeExactMatch)
{
        bool insertResUsage=false;
        // problem: a 2nd describe arrives, don't allow that in the same session!
  if (cm && uri && !mp4Stream) {
                if (url) {
                   delete url;
                }
                dprintf_full("ProxySession::setUrl: %s\r\n", uri->toString());
                url = new Url(uri->toString());
                cm->lockCache();
                MetaObject* meta=cm->findMetaObject(url);
                

                bool exactMatch=false;
                if( (!termCap && !userPref)) {
                        mp4Stream = cm->getVideo(url);
                        
                        // calc resUsage 
                        VideoESInfo* vis=NULL;
                        if(mp4Stream)
                                vis=mp4Stream->getFirstVisualES();
                        
                        if(vis) {
                                Feature src(vis);
                                src.isCached=true;
                                ResourceUsage* tmpRU=CostFunction::calcResourceUsage(&src,&src,admControl->getResourceLimit());
                                if(tmpRU) {
                                        resUsage.network=tmpRU->network;
                                        resUsage.disk=tmpRU->disk;
                                        resUsage.cpu=tmpRU->cpu;
                                        delete tmpRU;
                                        insertResUsage=true;
                                }
                        }
                        insertMp4StreamIntoCM=false;
                        dprintf_full("ProxySession::setUrl found video (no termcaps or userpref)\n");
                }
                else if(meta && userPref) {
                        // FIXME: hardcode delay values to 100ms
                        dprintf_full("ProxySession::setUrl meta && userPref\n");
                        suggestedAdaptor=admControl->makeBestMatch(userPref,100,meta,&mp4Stream,&resUsage);
                        dprintf_full("ProxySession::setUrl (N=%f,C=%f,D=%f)\n",
                                resUsage.network,resUsage.cpu,resUsage.disk);

                        if(mp4Stream && suggestedAdaptor && resUsage.network>=0.0000001 &&
                                strcmp(suggestedAdaptor->getName(),"Forwarder")!=0 && 
                                !strstr(suggestedAdaptor->getName(),"Temporal")){ // transcoding
                                        // a hit with transcoding!
                                        origSrc=mp4Stream;
                                        mp4Stream=mp4Stream->clone();
                                        insertMp4StreamIntoCM=true;

                                        u64 durInMS=0;
                                        if(mp4Stream->getFirstVisualES())
                                                durInMS=mp4Stream->getFirstVisualES()->getDurationInMs();
                                        else if(mp4Stream->getFirstAudioES())
                                                durInMS=mp4Stream->getFirstAudioES()->getDurationInMs();
                                        u32 byteSize=0;
                                        if(userPref) {
                                                byteSize=userPref->bitrate.best/8*durInMS/1000;
                                        }
                                        assert (mp4StreamReserved == 0);
                                        if(cm->reserveDiskSpace(byteSize))
                                                mp4StreamReserved+=byteSize;
                                        else {
                                                insertMp4StreamIntoCM=false;
                                        }
                                        prxStat.bytesHit+=mp4Stream->getFileSize();
                                        prxStat.bytesRequested+=mp4Stream->getFileSize();
                                        prxStat.objHits++;
                        }
                        else if(mp4Stream && resUsage.network>=0.0000001 ) {
                                // never cache temporal transcoding or no transcoding
                                origSrc=mp4Stream;
                                insertMp4StreamIntoCM=false;
                                mp4Stream=mp4Stream->clone();
                                this->delMp4Stream=true;
                                prxStat.bytesHit+=mp4Stream->getFileSize();
                                prxStat.bytesRequested+=mp4Stream->getFileSize();
                                prxStat.objHits++;
                                prxStat.qualHits++;
                        }
                        else if(mp4Stream) {
                                // happens in MISS! prxStat.bytesRequested+=mp4Stream->getFileSize();
                                mp4Stream=NULL;
                        }
                }
                else if(meta && termCap) {
                        // bandwidth should be the minimum of network and decodeBandwidth
                        // if one of them is zero, ignore the zero param
                        u32 bandwidth=termCap->getNetworkCapacityInByte()*8;
                        if(bandwidth==0)
                                bandwidth=termCap->getMaxDecoderBitRateInBit();
                        else if(termCap->getMaxDecoderBitRateInBit() && 
                                termCap->getMaxDecoderBitRateInBit() < bandwidth) {
                                bandwidth=termCap->getMaxDecoderBitRateInBit();
                        }
                        mp4Stream=meta->findMp4Stream(0,termCap->getDisplayWidth(),
                                                        0,termCap->getDisplayHeight(),
                                                        0,bandwidth,
                                                        (termCap->getColorDisplay()==0),(float)termCap->getDisplayRefreshRate() ,&exactMatch);
                        
                        if(makeExactMatch && !exactMatch) {
                                // clone the mp4Stream
                                origSrc=mp4Stream;
                                mp4Stream=mp4Stream->clone();
                                mp4Stream->getUsageCounter()->increase();
                                insertMp4StreamIntoCM=true;
                                suggestedAdaptor=admControl->makeExactMatch(termCap->getDisplayWidth(),
                                                                                termCap->getDisplayHeight(),
                                                                                bandwidth,
                                                                                (termCap->getColorDisplay()==0),
                                                                                (float)termCap->getDisplayRefreshRate(),
                                                                                mp4Stream,&resUsage);
                                // check if we have a TemporalAdaptor, do not cache results of temporaladaptor
                                if(strstr(suggestedAdaptor->getName(),"Temporal") ||
                                         strcmp(suggestedAdaptor->getName(),"Forwarder")) {
                                        insertMp4StreamIntoCM=false;
                                        delMp4Stream=true;
                                }
                                prxStat.bytesHit+=mp4Stream->getFileSize();
                                prxStat.bytesRequested+=mp4Stream->getFileSize();
                                prxStat.objHits++;                      
                        }
                        else {
                                // calc the resource usage for the non transcoding case
                                origSrc=mp4Stream;
                                mp4Stream=mp4Stream->clone();
                                mp4Stream->getUsageCounter()->increase();
                                insertMp4StreamIntoCM=false;
                                delMp4Stream=true;
                                if(mp4Stream) {
                                        VideoESInfo* vis=mp4Stream->getFirstVisualES();
                                        if(vis) {
                                                Feature src(vis);
                                                src.isCached=true;
                                                ResourceUsage* tmpRU=CostFunction::calcResourceUsage(&src,&src,admControl->getResourceLimit());
                                                if(tmpRU) {
                                                        resUsage.network=tmpRU->network;
                                                        resUsage.disk=tmpRU->disk;
                                                        resUsage.cpu=tmpRU->cpu;
                                                        delete tmpRU;
                                                        insertResUsage=true;
                                                }
                                        }
                                }
                        }
                }
                cm->unlockCache();
  }
        // now insert the resource usage for those other cases (all non-transcoding!)
        if(insertResUsage) {
                if(!admControl->insertRequest(&resUsage)) {
                        resUsage.cpu=resUsage.disk=resUsage.network=0.0;
                        prxStat.bytesRequested+=mp4Stream->getFileSize();
                        if(delMp4Stream) {
                                mp4Stream->getUsageCounter()->decrease();
                                delete mp4Stream;
                                mp4Stream=NULL;
                                delMp4Stream=false;
                        }
                        if(origSrc)
                                origSrc->getUsageCounter()->decrease();
                        origSrc=NULL;
                        insertMp4StreamIntoCM=false;
                        if(suggestedAdaptor)
                                delete suggestedAdaptor;
                        suggestedAdaptor=NULL;
                }
                else {
                        prxStat.bytesHit+=mp4Stream->getFileSize();
                        prxStat.bytesRequested+=mp4Stream->getFileSize();
                        prxStat.objHits++;
                        prxStat.qualHits++;
                }
        }
        if(!mp4Stream)
                prxStat.objMiss++;
        if(suggestedAdaptor && mp4Stream)
                suggestedAdaptor->setESInfo(mp4Stream->getFirstVisualES());
};


bool ProxySession::establishConnectionToServer()
{
        // get the url
        // extract the ip+port
        // connect and set the sessionControlChannel
        // extract the url
        assert(url!=NULL);
        dprintf_full("ProxySession::establishConnectionToServer: url %s\n",url->toString());
        const char* ip=url->getServerName();

        if(!url || !ip || !prot) {
                dprintf_err("ProxySession::establishConnectionToServer() failed\r\n");
                state=SESSION_ERR;
                return false;
        }

        int port=-1;
        char *realIP = NULL;
        struct hostent *ipAddr;

        /*
        //depricated gethostbyname
        ipAddr=gethostbyname(ip);
        if (!ipAddr) {
                dprintf_err("ProxySession::establishConnectionToServer: DNS resolution failed for host %s\r\n",ip);
                state=SESSION_ERR;
                return false;
        } else {
                realIP = new char[strlen(inet_ntoa( *((struct in_addr*)ipAddr->h_addr)))+1];
                strcpy(realIP, inet_ntoa( *((struct in_addr*)ipAddr->h_addr)));
                dprintf_full("ProxySession::establishConnectionToServer: DNS returned %s\r\n", realIP);
        }
        */

        struct addrinfo ai, *res_ai=NULL;
        int ret =0;

        memset(&ai,0,sizeof(struct addrinfo));
        ai.ai_flags = AI_CANONNAME;
        ai.ai_family =  PF_UNSPEC;
        ai.ai_socktype = 0;
        ai.ai_protocol = 0;
        dprintf_full("ProxySession::establishConnectionToServer: getaddrinfo\n");
        if ((ret = getaddrinfo(ip, NULL, &ai, &res_ai)) < 0)
                {
                        dprintf_err("ERROR %s\n",gai_strerror(ret));
                        ::exit(1);
                }
        assert(res_ai->ai_canonname);
        realIP = new char[strlen(res_ai->ai_canonname)+1];
        strcpy(realIP, res_ai->ai_canonname);
        freeaddrinfo(res_ai);
        assert((ipAddr=gethostbyname(ip)));
        dprintf_full("ProxySession::establishConnectionToServer: real server name: #%s#\n",ip);

        if(url->hasPort())
                port=url->getPort();
        else
                port=prot->getDefaultPort();

        int socketToServer=::socket(AF_INET,SOCK_STREAM,0);
        if(socketToServer==-1) {
                dprintf_full("ProxySession::establishConnectionToServer: Couldn't create socket\r\n");
                state=SESSION_ERR;
                return false;
        }
                
        if(!server)
                server=new sockaddr_in();
 
        server->sin_family=AF_INET;
        //      server->sin_port=htons(((u16)port));
        server->sin_port=htons(port);
        server->sin_addr.s_addr=inet_addr(ipAddr->h_addr);
        memset(&(server->sin_zero),'\0',8);
        bcopy(ipAddr->h_addr, (char *)&server->sin_addr, ipAddr->h_length);
//      bcopy(ipAddr->h_addr, realIP, strlen(realIP));
        
        if(::connect(socketToServer,(struct sockaddr*)server,sizeof(struct sockaddr)) < 0) {
                dprintf_full("ProxySession::establishConnectionToServer: Failed to connect to orig. server %s\r\n",realIP);
                state=SESSION_ERR;
                ::exit(2);
                return false;
        }
        
        dprintf_full("ProxySession::establishConnectionToServer: Could connect to server\r\n");

 
        if(serverSession)
                delete serverSession;
        serverSession=new ServerSession(socketToServer,NULL,cm,admControl,rtxEnabled);
        dprintf_full("ProxySession::establishConnectionToServer: Successfully created serverSession to %s\n",ip);
        return true;
};

bool ProxySession::videoFullyCached(ContainerInfo* stream) const
{
        if(!stream)
                return false;
        if(stream->getESList()->empty())
                return false;

        for (list < ESInfo * >::iterator i = stream->getESList()->begin(); i != stream->getESList()->end();++i) {
                if(!esFullyCached(*i))
                        return false;
        }
        return true;
};

bool ProxySession::esFullyCached(const ESInfo* es) const
{
        if(!es)
                return false;

        const char* esInput=NULL;
         FILE* esFP=NULL;
        bool ret=false;
        esInput=es->getInput();
        ret=(esInput && (esFP=fopen(esInput,"rb")) && es->getCompleteState());
        if(esFP)
                fclose(esFP);
        return ret;
}

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