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

Info

Last update:
12 Feb 2007

Documentation

Doxygen LXR

DataChannel.cpp

 /***********************************************************************
 *                                                                      *
 * ViTooKi                                                              *
 *                                                                      *
 * title:       DataChannel.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 "DataChannel.hpp"

#include "DataSink.hpp"
#include "Frame.hpp"
#include "ContainerInfo.hpp"
#include "ReferenceCounter.hpp"

#include "Adaptor.hpp"
#include "AdaptorChain.hpp"
#include "adaptors/Forwarder.hpp"
#include "adaptors/Visualizer.hpp"
#include "adaptors/MP4Decoder.hpp"
#include "VideoAdaptorChain.hpp"
#include "adaptors/FrameRotate.hpp"

#include "IO.hpp"
#include "io/MPGStreamIO.hpp"
#include "io/DevNull.hpp"



DataChannel::DataChannel(IO * in, ESInfo * es)
{
    Adaptor *a = NULL;
    init(in, es, a);
};

DataChannel::DataChannel(IO * in, ESInfo * es, Adaptor * adapt)
{
    init(in, es, adapt);
};

void
 DataChannel::init(IO * &in, ESInfo * &e, Adaptor * &adapt)
{
        assert(e->getMediaTimeScale()!=0);
        es = e;
        sessionId = 0;
        input = in;
        channelOpen = true;
        immediateClose=false;
        tearDown=false;
        lockInput.initialize("lockInput");
        lockDataSink.initialize("lockDataSink");
        lockAdaptor.initialize("lockAdaptor");
        localViewer=NULL;
        adaptor = adapt;
        if (!adaptor) {
                adaptor = new Forwarder();
        }
        if(es->getUsageCounter())
                es->getUsageCounter()->increase();
        outstats = NULL;

        currentFrameNumber = 0;
        finalFrameNumber = -1; //so there is no limit
};


/* is always called unter &lockInput */
void DataChannel::deleteInput() {
        if(input) {
                input->close();
                if(input->running())
                    input->wait();
        }
        // empty send queue
        list <Frame*>::iterator si;
        while(!sendQueue.empty()) {
                si=sendQueue.begin();
                if(*si) {
                        delete (*si);
                        (*si)=NULL;
                }
                sendQueue.pop_front();
        }
}

DataChannel::~DataChannel()
{
        closeAllClients(true);

        lockDataSink.lock();
        this->tearDown=true;

        lockInput.lock();
        deleteInput();
        lockInput.release();

  // delete clients
  dprintf_full("DataChannel Destruct: delete active clients\n");
  list < DataSink * >::iterator li;
  while (!active.empty()) {
                li = active.begin();
                if (*li) {
                        (*li)->close(true);
                        delete (*li);
                        (*li) = NULL;
                }
                active.pop_front();
  }
  dprintf_full("DataChannel Destruct: delete paused clients\n");
        while (!paused.empty()) {
                li = paused.begin();
                if (*li) {
                        delete (*li);
                        (*li) = NULL;
                }
                paused.pop_front();
  }
        lockDataSink.release();

        lockInput.destroy();
        lockDataSink.destroy();
    lockAdaptor.destroy();



  dprintf_full("DataChannel Destruct: delete viewer\n");
        if(localViewer)
                delete localViewer;
  dprintf_full("DataChannel Destruct: delete input\n");
        if(input)
                delete input;
  
        if(es && es->getUsageCounter())
                es->getUsageCounter()->decrease();
        
        if(adaptor) {
                delete adaptor;
                dprintf_full("DataChannel Destruct: delete adaptor\n");
        }
        dprintf_full("DataChannel Destruct: done\n");
};

// what happens if the last client is paused?
// we will wait till play or teardown arrives

void DataChannel::run()
{
        bool resetSendHeader;

  dprintf_full("DataChannel::run: DC.Thread started: %s\n", input->getURL());
        resetSendHeader=false;

  if (!active.empty() || localViewer) {
                if (input->getState()!=IO::OPEN && !input->open()) {
                        dprintf_err("DataChannel::run: failed to open input\r\n");
                        return;
                }
                // read data from input
                dprintf_full("DataChannel::run: Input is open\n");

                Frame *frm = NULL;
                //only continue if OPEN, PAUSED, MUTED, PREBUFFERING or STREAMEOF
                //STREAMEOF is not a problem, as long as eg. the Rtp Class still has buffered data!
                IO::State mystate = input->getState();

                while (! ( (mystate == IO::CLOSED) || (mystate == IO::CLOSING) ||
                                                         (mystate == IO::STREAMERR) ) //|| (input->getState() == IO::STREAMEOF) )
                                         && !tearDown && ((finalFrameNumber == -1) || (currentFrameNumber < finalFrameNumber)) ) {
                        lockInput.lock();
                        lockDataSink.lock();
                        if(mystate == IO::OPENING)

#ifndef WINCE
                                msleep(10);
#else
                                Sleep(5);
#endif
                        if(active.empty() && !localViewer) {
                                lockInput.release();
                                lockDataSink.release();
                                return;
                        }
                        lockDataSink.release();

                        //read frame
                        frm = input->getFrame();

                        if( frm == NULL ) {
                                dprintf_full("DataChannel::run  no frm got via getFrame\n");
                                if(mystate != IO::OPEN) {
                                        if(frm) {
                                                delete frm;
                                                frm=NULL;
                                        }
                                }
                                lockInput.release();
                        }       else {
                                dprintf_full("DataChannel::run  passing frame to adaptor\n");
                                // we got data
                                // adapt it with the global adaptor (chain)
                                lockAdaptor.lock();
                                list < Frame * >result;

                                //adapt (i.e. decode) frame
                                result = adaptor->adapt(frm);

                                lockAdaptor.release();
                                
                                if (!result.empty() ) {
                                        list < Frame * >::iterator rLi;
                                        // now append the received frame(s) to the sendqueue
                                        if(result.size()==1) {
                                                rLi=result.begin();
                                                if ( frm && (*rLi) != frm) {
                                                        // caching adaptors have to deep-copy the frame
                                                        dprintf_full("DataChannel::run  Frame was adapted and orig frame will be deleted now...\n");
                                                        delete frm; frm=NULL;
                                                }
                                                else {
                                                        // frame is in result
                                                        dprintf_full("DataChannel::run  frame is in result\n");
                                                        frm=NULL;
                                                }
                                        }
                                        //append results to sendQueue
                                        while (!result.empty()) {
                                                rLi = result.begin();

                                                if (!(*rLi)->isMarkedForDelete()) //may be dropped in ESSynchronizer 
                                                        sendQueue.push_back((*rLi));
                                                else {
                                                        dprintf_full("DataChannel::run adaptor result was marked for delete... dropping!\n");
                                                        delete (*rLi); (*rLi)=NULL;
                                                }
                                                result.pop_front();
                                        }
                                } else {
                                        dprintf_full("DataChannel::run  adaptor result was empty!!! dropped, so delete frm CTS %i...\n",frm->getAU()->cts);
                                        //ATTENTION: DC only deletes frames coming from inputIO, not eg. adaptor-decoded and then dropped ones!
                                        //so make sure, to markForDelete()
                                        delete frm; frm=NULL;
                                }
                                lockInput.release();
                                
                        }       // end frame != NULL    
                        
                        // result now empty
                        // check if we have data in the sendqueue
                        list < Frame * >::iterator rLi;
                        
                        if (sendQueue.empty() &&  (input->getState() == IO::STREAMEOF) )
                                break; //give up, no more data coming, nothing left to forward
                        
                        // must be an if not while -> avoids bursts                             
                        if (!sendQueue.empty() && !tearDown) {  
                                // distribute it to all active clients
                                lockDataSink.lock();
                                rLi = sendQueue.begin();

                                //set this frame to be deleted after all output-DataSinks, this may be changed by eg. buffering output-IOs
                                (*rLi)->markForDelete();

                                if(!active.empty()) {
                                        
                                        dprintf_full("DataChannel::run: READ FRAME %p (currentFrameNumber %li)\n",(*rLi),currentFrameNumber);
                                        currentFrameNumber++;
                                        
                                        list < DataSink * >::iterator li = active.begin();
                                        
                                        while (li != active.end()) {
                                                if(*rLi) {
                                                        dprintf_full("Sending one frame %p (TS %i size %i) to one client\n",(*rLi),
                                                                                                         (*rLi)->getAU()->cts,(*rLi)->getAU()->size);
                                                        
                                                        if (!(*li)->send((*rLi))) {  //finally, writeFrame of output class
                                                                // sth went wrong
                                                                dprintf_err("DataChannel::run: Sending of frame failed\n");
                                                                tearDown = true;
                                                        }
                                                }
                                                ++li;
                                        }
                                        dprintf_full("DataChannel::run: SENT one FRAME %p to all clients\n",(*rLi));
                                }
                                // when we have no client and not even a localViewer -> exit
                                else if(!localViewer) {
                                        lockDataSink.release();
                                        if((*rLi)) {
                                                delete (*rLi);
                                                (*rLi)=NULL;
                                        }
                                        return;
                                }
                                if(localViewer)
                                        localViewer->send( (*rLi) );
                                lockDataSink.release();

                                if ((*rLi)->isMarkedForDelete()) {
                                        dprintf_full("DataChannel::run  cleanup sendQueue with frame %p\n",(*rLi));                             
                                        if((*rLi))
                                                delete (*rLi);
                                } //else: ignore and let the buffering output-IO do the delete on it's own!

                                *rLi=NULL;
                                sendQueue.pop_front();

                        }       else {
                                dprintf_full("DataChannel::run  sendQueue was empty!!! teardown %i\n",tearDown);
                        }
                        
                        if(frm) {
                                dprintf_full("DataChannel::run  cleanup incoming frame %p\n",frm);
                                delete frm;
                                frm=NULL;
                        }

                        mystate = input->getState();
                }               // while() end reading
                
                //lockDataSink.lock();
                channelOpen = false;
                //lockDataSink.release();
                dprintf_full("DataChannel::run: Finished Reading (state %i) currentFrameNumber %li\n",
                                                                 input->getState(),currentFrameNumber);
                if(input->getState()==IO::STREAMEOF)
                        resetSendHeader=true;

                input->close(true);

                if (adaptor && (strcmp(adaptor->getName(),"Forwarder") != 0)) { //maybe there is some cached data?
                        list < Frame * >result;
                        dprintf_full("DataChannel::run: Closing Adaptor %s\n",adaptor->getName());
                        result = adaptor->close();
                        
                        list < Frame * >::iterator rLi;
                        if (result.size() > 0) {
                                dprintf_full("DataChannel::run: adaptor->close() returned %i frames\n",result.size());  
                                // now append the received frame(s) to the sendqueue
                                while (!result.empty()) {
                                        rLi = result.begin();
                                        sendQueue.push_back((*rLi));
                                        result.pop_front();
                                }
                                // now we have read all data, we have to ensure that everything is sent
                                dprintf_full("DataChannel::run: Sending %i cached frames\n",sendQueue.size());
                                while (!sendQueue.empty() && !tearDown) {
                                        // distribute it to all active clients
                                        rLi = sendQueue.begin();
                                //set this frame to be deleted after all output-DataSinks, this may be changed by eg. buffering output-IOs
                                (*rLi)->markForDelete();

                                        dprintf_full("DataChannel::run: READ cached FRAME\n");
                                        lockDataSink.lock();
                                        if(!active.empty()) {
                                                list < DataSink * >::iterator li = active.begin();                                      
                                                while (li != active.end() && !tearDown) {
                                                        dprintf_full("DataChannel::run: Sending one frame to one client\n");                                            
                                                        if (!(*li)->send((*rLi))) {
                                                                // sth went wrong
                                                                dprintf_err("DataChannel::run: Sending of frame failed\n");
                                                        }
                                                        ++li;
                                                }
                                        }
                                        if(localViewer)
                                                localViewer->send( (*rLi) );

                                        dprintf_full("DataChannel::run: SEND cached FRAME\n");
                                        // finished with sending one frame
                                        lockDataSink.release();

                                        if ((*rLi)->isMarkedForDelete()) {
                                                dprintf_full("DataChannel::run  cleanup sendQueue with frame %p\n",(*rLi));                             
                                                if((*rLi))
                                                        delete (*rLi);
                                        } //else: ignore and let the buffering output-IO do the delete on it's own!

                                        (*rLi)=NULL;
                                        sendQueue.pop_front();
                                }                       // end sending all frames 
                        }
                        // if a teardown was received while sending, empty sendqueue
                        while (!sendQueue.empty()) {
                                rLi = sendQueue.begin();
                                dprintf_full("DataChannel::run, emptying sendqueue because of teardown: frame %p\n",(*rLi));
                                delete(*rLi);
                                
                                (*rLi)=NULL;
                                sendQueue.pop_front();
                        }                       
                }// if(adaptor)

                // not necessary with BitField, but we better do it
                es->setCompleteState(true);
  }
        else {
                dprintf_small("DataChannel::run: Thread Started with no clients\n");
                input->close(true);
  }
        channelOpen=true;

        closeAllClients(false);

        
        dprintf_small("DataChannel::run: EXIT!\n");
        this->exit();
} //DataChannel::run


void DataChannel::setAdaptor(Adaptor * adapt)
{
    lockAdaptor.lock();
    delete adaptor;
    adaptor = adapt;
    if (!adaptor) {
                adaptor = new Forwarder();
    }
    lockAdaptor.release();
};

void DataChannel::setESInfo(ESInfo *new_es) {
        if(es && es->getUsageCounter())
                es->getUsageCounter()->decrease();

        es = new_es;
}

void DataChannel::setFinalFrameNumber(long frameNumber) {
        finalFrameNumber = frameNumber;
}

void DataChannel::initiateThreadStop(bool forceTearDown) {
        if (forceTearDown)
                tearDown=true;
        //if not forceTearDown, thread will stop when finalFrameNumber is reached
        while ( this->running() || !this->finished()) {
                msleep(40);
                dprintf_full("DataChannel::initiateThreadStop waiting on DataChannel::run finished... sendQueue (%i elems)\n",sendQueue.size());
        }
        if (forceTearDown)
                tearDown=false;
}

void DataChannel::setInput(IO *new_io) {
        dprintf_full("DataChannel::setInput\n");

        /*
                list <int> ids = getListOfActiveClientIds();
                for(list <int>::iterator id = ids.begin(); id != ids.end(); ++id) {
                        pause((*id));
                        getDataSink((*id))->getOutput()->setESInfo(new_es);
                }
                setESInfo(new_es);
        */


        lockInput.lock();

    assert(active.empty()); //no-one is playing
                deleteInput();
          input = new_io;
                currentFrameNumber=input->getCurrentFrameNumber();

                list < DataSink * >::iterator li = paused.begin();
                li = paused.begin();
                while (li != paused.end()) {
                        (*li)->getOutput()->setResendFrameHeader(true);
                        ++li;
                }
                lockInput.release();
}

int DataChannel::getNumberOfActiveDataSinks(){
        int result=0;
        lockDataSink.lock();
        result=active.size();
        lockDataSink.release();
        return result;
};

bool DataChannel::play(u32 clientId, double prefetch)
{
  dprintf_full("DataChannel::play(u32 clientId=%u prefetch %f)\r\n",clientId,prefetch);
        if(paused.empty()) {
                dprintf_err("DataChannel::play: (Empty) Failed to play non existing client %u\r\n",clientId);
                return false;
        }
        if(!input) {
                dprintf_err("bool DataChannel::play: No input is set!\r\n");
                return false;
        }

        //FIXME: unpausing will only work for one DC with ONE client
        input->play(prefetch); //open the io stream (if not yet open)

  lockDataSink.lock();
        
        //  if (channelOpen && !tearDown) {


        // localDataSinkAdaptors are not allowed to modify the ES directly
                        // FIXME: one DC -> one CLient
                        // Problem here is the global Adaptor and
                        // the local Adaptor, we could only clone an ES, when
                        // the global adaptor is initialized!
                        // the next three lines also break TC support
                        // client receives an invalid (old) ESheader
                        // through Rtp when the server adapts the stream
                        // for the client
        // ESInfo *es2 = es->clone(es->getContainerInfo());
        //c->getAdaptor()->setESInfo(es2);
        //c->getOutput()->setESInfo(es2);


                // remove the client from the paused list, 

                list<DataSink*>::iterator cLi=paused.begin();
                while(cLi!=paused.end()) {
                        if( (*cLi)->getDataSinkId()==clientId) {
                                active.push_front((*cLi));
                                (*cLi)->playOutput(prefetch);
                                paused.remove(*cLi);
                                lockDataSink.release();
                                dprintf_full("DataChannel::play: Started client %u\r\n",clientId);
                                return true;
                        }
                        cLi++;
                }
                lockDataSink.release();
                dprintf_err("DataChannel::play: Failed to play non existing paused client %u\r\n",clientId);
                return false;

                // } 
        dprintf_err("DataChannel::play: Failed to play client %u. channelOpen=%i, tearDown=%i\r\n",
                clientId,channelOpen,tearDown);
  lockDataSink.release();
  return false;
};

char *DataChannel::enableCaching(bool omitHeader, const char* dir, bool proxyMode, const Url* url)
{
        
    char *out = new char[1024];
                IO* output=NULL;
    out[0] = 0;
    lockDataSink.lock();
    if (channelOpen && es) {

                char intnum[32];
                out[0] = 0;
                intnum[0] = 0;
                sprintf(intnum, "%llu", es->getStreamId());
                if(!dir) {
                        output=new DevNull();
                }
                else {

                        strcpy(out,dir);
                        
                        if(!(out[strlen(out)-1]=='/' || out[strlen(out)-1]=='\\' )) {
                                int xy=strlen(out);
                                out[xy]=PATHSEPARATORCHAR;
                                out[xy+1]='\0';
                        }


                        // problem: recursive servers -> eliminate path just get filename
                        const char* tmp=es->getContainerInfo()->getLocalFile();
                        const char* t2=NULL;
                        char* dummy=NULL;
                        if(tmp!=NULL) {
                                if((t2=strrchr(tmp,PATHSEPARATORCHAR))!=NULL) {
                                        t2++;
                                }
                                else
                                        t2=tmp;
                        }
                        if(url) {
                                if(proxyMode) {
                                        dummy=Url::transformUrlToLocalFile(url);
                                        strcat(out,dummy);
                                        delete[] dummy;
                                }
                                else {
                                        dummy=Url::transformPathToLocalFile(url);
                                        strcat(out,dummy);
                                        delete[] dummy;
                                }
                        }
                        else {
                        
                                strcat(out,"localfile");
                                if(t2) {
                                        strcat(out,"_");
                                        strcat(out, t2);
                                }
                        }

                        if (es->isVisualStream())
                                strcat(out, ".video.");
                        else if (es->isAudioStream())
                                strcat(out, ".audio.");
                        else if(es->isSceneDescriptionStream())
                                strcat(out, ".bifs.");
                        else if(es->isODStream())
                                strcat(out, ".ODstream.");
                        else
                                strcat(out, ".generic.");

                        strcat(out, intnum);
                        // check if file exists!!!!
                        FILE* fp=NULL;
                        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, out is deep-copied, also in QFile!
                        output = new MPGStreamIO(es, out, true, omitHeader);
                        if (!output->open()) {
                                dprintf_err("DataChannel::enableCaching: Failed to open output channel\n");
                        }
                }
                
                char tmp2[12];
                strcpy(tmp2, "127.0.0.1");
                DataSink *c = new DataSink(0,output, tmp2, 0, 0);
                active.push_front(c);
        }
    lockDataSink.release();
        
    return out;
};

void DataChannel::visualizeCaching()
{
    lockDataSink.lock();
    if (channelOpen && es->isVisualStream()) {
                        if(!this->localViewer) {
                                char *tmp2 = new char[12];
                                
                                strcpy(tmp2, "127.0.0.1");
                                
                                AdaptorChain *chain =   new VideoAdaptorChain();
                                chain->setESInfo((ESInfo*)es);
#if defined (BUILD_ARCH_INTEL) || defined (WIN32) //assuming display with 24 or 32 bit (4byte) depth
                                // YV12 is super-fast with Visualizer::SDL!
                                UncompressedVideoFrame::ColorSpace cs = UncompressedVideoFrame::ColorSpaceYV12; 
                                //                              UncompressedVideoFrame::ColorSpace cs = UncompressedVideoFrame::ColorSpaceRGB32;
                                //                              UncompressedVideoFrame::ColorSpace cs = UncompressedVideoFrame::ColorSpaceGRAY8;
                                //        UncompressedVideoFrame::ColorSpace cs = UncompressedVideoFrame::ColorSpaceRGB565;
#elif BUILD_ARCH_ARM  //iPAQ only has 16 bit display!
                                UncompressedVideoFrame::ColorSpace cs = UncompressedVideoFrame::ColorSpaceRGB565;
#endif
                                chain->addAdaptor(new MP4Decoder((VideoESInfo*)es, false, cs, MP4Decoder::FFMPEG));
                                //chain->addAdaptor(new FrameRotate((VideoESInfo*)es, 90)); //add after decoder

                                //chain->addAdaptor(new Visualizer((VideoESInfo*)es, Visualizer::X11, cs));
                                chain->addAdaptor(new Visualizer((VideoESInfo*)es, Visualizer::SDL, cs));
                                this->localViewer = new DataSink(1,new DevNull(), tmp2, 0, chain);
                                delete tmp2;
                        }
    }
    lockDataSink.release();
};

bool DataChannel::mute(u32 clientId)
{
        dprintf_full("DataChannel::mute(u32 clientId=%u)\r\n",clientId);
        if(active.empty())
                return false;

  lockDataSink.lock();

        list<DataSink*>::iterator cLi=active.begin();
        while(cLi!=active.end()) {
                if( (*cLi)->getDataSinkId()==clientId) {
                        (*cLi)->muteOutput();
                        lockDataSink.release();
                        return true;
                }
                cLi++;
        }
  lockDataSink.release();
  return false;
};

bool DataChannel::unmute(u32 clientId)
{
        dprintf_full("DataChannel::unmute(u32 clientId=%u)\r\n",clientId);
        if(active.empty())
                return false;

  lockDataSink.lock();

        list<DataSink*>::iterator cLi=active.begin();
        while(cLi!=active.end()) {
                if( (*cLi)->getDataSinkId()==clientId) {
                        (*cLi)->playOutput(0.0);
                        lockDataSink.release();
                        return true;
                }
                cLi++;
        }
  lockDataSink.release();
  return false;
};

bool DataChannel::pause(u32 clientId)
{
        if(active.empty()) {
                dprintf_full("DataChannel::pause(u32 clientId=%u): NO ACTIVE CLIENTS!\r\n",clientId);
                return false;
        }

        //pause the input stream
        input->pause();


  lockDataSink.lock();
  // remove the client from the active list
        list<DataSink*>::iterator cLi=active.begin();
        while(cLi!=active.end()) {
                dprintf_full("DataChannel::pause   found client %p Id=%u, looking for Id %u)\n",
                                                                 (*cLi), (*cLi)->getDataSinkId(),clientId);
                if( (*cLi)->getDataSinkId()==clientId) {
                        paused.push_front((*cLi));
                        (*cLi)->pauseOutput();
                        active.remove(*cLi);
                        lockDataSink.release();
                        return true;
                }
                cLi++;
        }

  lockDataSink.release();
  return false;
};

bool DataChannel::insert(DataSink* c) 
{
        dprintf_full("DataChannel::insert(DataSink* c=%p,c->clientId=%u\r\n)\r\n",c, (c?c->getDataSinkId():0));
        lockDataSink.lock();
        paused.push_front(c);
        lockDataSink.release();
        return true;
};

DataSink* DataChannel::getDataSink(u32 clientId)
{
        DataSink* result=NULL;
        lockDataSink.lock();
        if(!paused.empty()) {
          list < DataSink * >::iterator li = paused.begin();
                while(li != paused.end()) {
                        if((*li)->getDataSinkId()==clientId) {
                                result=(*li);
                                li=paused.end();
                        
                        }
                        else
                                ++li;
                }
        }
        lockDataSink.release();
        return result;
};
const DataSink* DataChannel::getActiveDataSink(u32 clientId)
{
        DataSink* result=NULL;
        lockDataSink.lock();
        if(!active.empty()) {
          list < DataSink * >::iterator li = active.begin();
                while(li != active.end()) {
                        if((*li)->getDataSinkId()==clientId) {
                                result=(*li);
                                li=active.end();
                        
                        }
                        else
                                ++li;
                }
        }
        lockDataSink.release();
        return result;  
};

list <u32> DataChannel::getListOfActiveClientIds() {
        list<u32> idlist;
        lockDataSink.lock();
        if(!active.empty()) {
          list < DataSink * >::iterator li = active.begin();
                while(li != active.end()) {
                        idlist.push_back((*li)->getDataSinkId());
                        ++li;
                }
        }
        lockDataSink.release();
        return idlist;
}

list <u32> DataChannel::getListOfPausedClientIds() {
        list<u32> idlist;
        lockDataSink.lock();
        if(!paused.empty()) {
          list < DataSink * >::iterator li = paused.begin();
                while(li != paused.end()) {
                        idlist.push_back((*li)->getDataSinkId());
                        ++li;
                }
        }
        lockDataSink.release();
        return idlist;
}


void DataChannel::closeAllClients(bool immediateClose)
{
        bool resetSendHeader=false;
        // now close the clients (incl. their adaptors)
        lockDataSink.lock();
        tearDown=true;
        channelOpen=false;
        if(input->getState()==IO::STREAMEOF) 
                        resetSendHeader=true;
        if(!active.empty() || localViewer) {
                list < DataSink * >::iterator li = active.begin();
                while (li!=active.end()) {
                        
                        // check if we have reached the STREAMEOF at the input
                        // if yes, we have to enable resendFrameHeader
                        if(resetSendHeader) {
                                (*li)->getOutput()->setResendFrameHeader(true);
                                dprintf_full("DataChannel::closeAllClients resetting resendFrameHeader\r\n");
                        }
                        (*li)->close(immediateClose); // dataSink->close: this may take a while...
                        ++li;
                }
        }
        dprintf_full("DataChannel::closeAllClients: closed all client outs\n");
        if(localViewer)
                localViewer->close();
        lockDataSink.release();
};


bool DataChannel::teardown(u32 clientId,bool immediate)
{
        dprintf_full("DataChannel::teardown(u32 clientId=%u,bool immediate=%i)\r\n",clientId,immediate);
  lockDataSink.lock();
        immediateClose=immediate;
        if(active.empty()) {
                dprintf_full("DataChannel::teardown: Closing input\r\n");
                if (input)
                        input->close(immediate);
                tearDown=true;
                channelOpen=false;
        }
  // remove the client from the active list,
  list < DataSink * >::iterator li = active.begin();
  while (li != active.end()) {
                if (clientId == (*li)->getDataSinkId()) {
                        dprintf_full("DataChannel::teardown: Deleted client from active list\r\n");
                        delete *li;
                        *li=NULL;
                  active.erase(li);
                        if(active.empty()) {
                                dprintf_full("DataChannel::teardown: Closing input from active list \r\n");
                                if (input)
                                        input->close(immediate);
                                
                                tearDown=true;
                                channelOpen=false;
                        }
                        lockDataSink.release();
                        return true;
                }
                ++li;
  } //while
  // if not found on active, remove the client from the pause list,
  li = paused.begin();
  while (li != paused.end()) {
                if (clientId == (*li)->getDataSinkId()) {
                        dprintf_full("DataChannel::teardown: Deleted client from paused list\r\n");
                        delete *li;
                        *li=NULL;
                  paused.erase(li);
                        if(paused.empty()) {
                                dprintf_full("DataChannel::teardown: Closing input from paused list \r\n");
                                if (input)
                                        input->close(immediate);
                                
                                tearDown=true;
                                channelOpen=false;
                        }
                        lockDataSink.release();
                        return true;
                }
                ++li;
  } //while

        // client really not found
        dprintf_err("DataChannel::teardown DataSink not found\r\n");
        fflush(stdout);
  lockDataSink.release();
        
  return false;

};


void DataChannel::setOutputStatistics(Statistics *stats) {
        //just overwrite old stats, if exist (stats will be deleted elsewhere)
        outstats = stats;
}

Statistics *DataChannel::getOutputStatistics() {
        return outstats;
}

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