MediaStream.cpp

Go to the documentation of this file.

/***************************** License **********************************

 Copyright (C) 2008 by Communicative Machines
 http://www.cmlabs.com   All rights reserved.

 This library is free software; you can redistribute it and/or
 modify it under the terms of the GNU Lesser General Public
 License as published by the Free Software Foundation; either
 version 2.1 of the License, or (at your option) any later version.

 This library 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
 Lesser General Public License for more details.

 You should have received a copy of the GNU Lesser General Public
 License along with this library; if not, write to the Free Software
 Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA

************************************************************************/
#include "MediaStream.h"

namespace cmlabs {

#define STREAMMAXCOUNTDEFAULT 1000
#define BACKUPMINCOUNTDEFAULT 20
#define BACKUPMAXCOUNTDEFAULT 100

MediaStream::MediaStream(const JString& streamname, long hardMaxSize, long softMaxSize, long maxCount, long maxBackupCount) {
        outputData.sorting = SORTBYKEY;
        dataSamples.sorting = SORTBYKEY;
        allowMultipleWriters = false;

        name = streamname;
        if (name.length() == 0)
                name = createUniqueID("stream");
        
        if ( hardMaxSize < 0 )
                hardMax = 30*1024*1024;
        else
                hardMax = hardMaxSize;

        if ( softMaxSize < 0 )
                softMax = (long)(hardMaxSize*0.9);
        else
                softMax = softMaxSize;

        this->maxCount = maxCount;
        if (maxBackupCount <= 0)
                this->maxBackupCount = BACKUPMAXCOUNTDEFAULT;
        else
                this->maxBackupCount = maxBackupCount;
        bytesEverHandled = 0;
        countEverHandled = 0;
        currentSize = 0;
}

MediaStream::~MediaStream() {
        dataSamples.removeAll();
        currentSize = 0;
}

Object* MediaStream::clone() const {
        return NULL;
}

JString MediaStream::print() {

        JString str;
        DataSample* sample;

        for (int n=0; n < dataSamples.getCount(); n++) {
                sample = (DataSample*) dataSamples.get(n);
                if (sample != NULL)
                        str += JString::format("%3d %s\n", n, (char*) sample->print());
        }
        return str;
}

bool MediaStream::handleMultipleWriters(bool allow) {
        allowMultipleWriters = allow;
        return true;
}

InfoItem* MediaStream::getInfo() {
        InfoItem* info = new InfoItem();

        long val;
        info->setEntry("CurrentCount", JString(getCount()));
        info->setEntry("TotalCount", JString(this->getTotalCountEverHandled()));
        //val = getTotalDataMemUse();
        val = currentSize;
        info->setEntry("CurrentSize", JString(val));
        info->setEntry("CurrentSizeString", JString::bytifySize(val));
        double dval = this->getTotalDataEverHandled(); 
        info->setEntry("TotalSize", JString(dval));
        info->setEntry("TotalSizeString", JString::bytifySize(dval));
        double valin = getInputDataRate();
        double valout = getOutputDataRate();
        info->setEntry("CurrentRateIn", JString(valin));
        info->setEntry("CurrentRateOut", JString(valout));
        info->setEntry("CurrentRateInString", JString::bytifyRate(valin));
        info->setEntry("CurrentRateOutString", JString::bytifyRate(valout));
        info->setEntry("CurrentRatesString", JString::bytifyRates(valin, valout));

        DataSample* sample = getNewestSample();
        if (sample == NULL) {
                info->setEntry("PackageSize", JString("0"));
                info->setEntry("PackageSizeString", JString::bytifySize((long)0));
        }
        else {
                info->setEntry("PackageSize", JString(sample->size));
                info->setEntry("PackageSizeString", JString::bytifySize(sample->size));
        }

        info->setTime("OldestTime", getOldestSampleTime());
        info->setTime("NewestTime", getNewestSampleTime());

        return info;
}


bool MediaStream::addSample(DataSample* sample) {

        if (sample == NULL)
                return false;

        bool res = false;

        // JTime* t;
        if (allowMultipleWriters)
                sample->timestamp.reset();
        //}
        //else
        //      t = (JTime*) sample->getTimestamp().clone();

        JTime* t = (JTime*) sample->timestamp.clone();
        t->perfCountReset();
        // sample->timestamp.perfCountReset();

        if (accessMutex.EnterMutex(1000)) {
                while (dataSamples.containsKey(t))
                        t->atomicOffset++;
                sample->timestamp.atomicOffset = t->atomicOffset;

                if (!(res = dataSamples.add(t, sample))) {
                        delete(sample);
                        delete(t);
                }
                else {
                        if (channels.getCount() > 0) {
                                addChannelSample(sample);
                        }
                        countEverHandled++;
                        bytesEverHandled += sample->getDataSize();
                        currentSize += sample->getDataSize();
                }
        }
        accessMutex.LeaveMutex();

        doMaintenance();
        newDataSem.post();

        return res;
}

bool MediaStream::addSamples(ObjectCollection* samples, bool deleteCol) {

        if ((samples == NULL) || (samples->getCount() == 0)) {
                if (deleteCol)
                        delete(samples);
                return false;
        }

        JTime* t;
        DataSample* sample;
        // JTime t;
        for (int n=0; n<samples->getCount(); n++) {
                if ( (sample = (DataSample*) samples->get(n)) != NULL ) {
                        if (allowMultipleWriters)
                                sample->timestamp.reset();
                        //      t = new JTime();
                        //      sample->setTimestamp(*t);
                        //}
                        //else
                        t = (JTime*) sample->getTimestamp().clone();
                        t->perfCountReset();
                        while (dataSamples.containsKey(t))
                                t->atomicOffset++;
                        sample->timestamp.atomicOffset = t->atomicOffset;
                        if (accessMutex.EnterMutex(1000)) {
                                if (!dataSamples.add(t, sample)) {
                                        samples->remove(n);
                                        delete(t);
                                        n--;
                                }
                                else {
                                        if (channels.getCount() > 0)
                                                addChannelSample(sample);
                                        countEverHandled++;
                                        bytesEverHandled += sample->getDataSize();
                                        currentSize += sample->getDataSize();
                                }
                        }
                        accessMutex.LeaveMutex();
                }
        }

        doMaintenance();
        newDataSem.post();

        if (deleteCol) {
                samples->removeAllNoDelete();
                delete(samples);
        }

        return true;
}

bool MediaStream::deleteSample(DataSample* sample) {
        if (sample == NULL)
                return false;
        bool ret = false;
        if (accessMutex.EnterMutex(1000)) {
                currentSize -= sample->getDataSize();
        //      JTime t = sample->timestamp;
        //      t.perfCountReset();
                dataSamples.removeNoDelete(&sample->timestamp);
                sample->markForDeletion();
                // If sample size is large, lower the number of samples to backup
        //      if (deletedSamples.getCount() == 0) {
        //              if (sample->size > 1024)
        //                      maxBackupCount = BACKUPMINCOUNTDEFAULT;
        //      }
                deletedSamples.add(sample);
                accessMutex.LeaveMutex();
        }
        return true;
}

bool MediaStream::deleteSample(int pos) {
        bool ret = false;
        if (accessMutex.EnterMutex(1000)) {
                ret = deleteSample((DataSample*)dataSamples.get(pos));
                accessMutex.LeaveMutex();
        }
        return ret;
}


double MediaStream::getTotalDataEverHandled() {
        return bytesEverHandled;
}

long MediaStream::getTotalCountEverHandled() {
        return countEverHandled;
}

double MediaStream::getInputDataRate() {

        if (dataSamples.getCount() == 0)
                return 0;

        int max = 10, c;
        
        DataSample* sample;
        JTime now, *oldest; //, *oldest2;
        long ms = 1;
        double size = 0;

        if (accessMutex.EnterMutex(1000)) {
                c = dataSamples.getCount();
                if (c < max)
                        max = c;
                oldest = (JTime*) dataSamples.getKey(c - max);
        //      oldest = dataSamples.getTime(dataSamples.getCount() - 1 - max - 1);
        //      sample = (DataSample*) dataSamples.get(dataSamples.getCount() - 1 - max - 1);
        //      oldest2 = sample->getTimestamp();

        //      JFile::writeAFileASCII("out.txt", JString::format("\n\n%s\n\n", (char*) dataSamples.print()));

                ms = now - *oldest;
                size = 0;
                for (int n=0; n<max; n++) {
                        if ( (sample = (DataSample*) dataSamples.get(c-1-n)) != NULL) {
                                size += sample->getDataSize();
                        }
                        else
                                ms = ms;
                }
        }
        accessMutex.LeaveMutex();

//      printf("InputRate: %.2f in %d ms   %s   %s\n", size, ms, (char*) oldest.print(), (char*) sample->getTimestamp().print() );
        double res = (size/((double)ms/1000));
        if (res < 0)
                res = 0;
        return res;
}

double MediaStream::getOutputDataRate() {
        if (outputData.getCount() == 0)
                return 0;
        JTime now, *oldest;
        long ms = 1;
        double size = 0;
        
        if (accessMutex.EnterMutex(1000)) {
                oldest = (JTime*) outputData.getFirstKey();
                if (oldest == NULL) {
                        accessMutex.LeaveMutex();
                        return 0;
                }
                ms = now - *oldest;
                if (ms > 100000)
                        ms = ms;

                for (int n=0; n<outputData.getCount(); n++)
                        size += ((JString*)outputData.get(n))->toDouble();
        }
        accessMutex.LeaveMutex();
//      printf("OutputRate: %.2f in %d ms  %s\n", size, ms, (char*) oldest.print());
        double res = (size/((double)ms/1000));
        if (res < 0)
                res = 0;
        return res;
}


bool MediaStream::deleteSampleAt(const JTime& time) {
        bool ret = false;
        if (accessMutex.EnterMutex(1000)) {
        //      ret = dataSamples.remove(&time);
        //      JTime t = time;
        //      t.perfCountReset();
                DataSample* sample = (DataSample*) dataSamples.get(&time);
                if (sample != NULL) {
                        currentSize -= sample->getDataSize();
                        dataSamples.removeNoDelete(&time);
                        sample->markForDeletion();
                        // If sample size is large, lower the number of samples to backup
                //      if (deletedSamples.getCount() == 0) {
                //              if (sample->size > 1024)
                //                      maxBackupCount = BACKUPMINCOUNTDEFAULT;
                //      }
                        deletedSamples.add(sample);
                }
                accessMutex.LeaveMutex();
        }
        return ret;
}

bool MediaStream::deleteSamplesOlderThan(const JTime& time) {
        int ret = 0;
        DataSample* sample;
        if (accessMutex.EnterMutex(1000)) {
        //      ret = dataSamples.removeAllBefore(&time);
                while ( ( (sample = (DataSample*) dataSamples.getFirst()) != NULL ) &&
                  (sample->timestamp - time < 0) ) {
                        if (!deleteSample(sample))
                                break;
                        ret++;
                }
        }
        accessMutex.LeaveMutex();
        return (ret > 0);
}

bool MediaStream::deleteSamplesNewerThan(const JTime& time) {
        int ret = 0;
        DataSample* sample;
        if (accessMutex.EnterMutex(1000)) {
        //      ret = dataSamples.removeAllAfter(&time);
                while ( ( (sample = (DataSample*) dataSamples.getLast()) != NULL ) &&
                  (sample->timestamp - time > 0) ) {
                        if (!deleteSample(sample))
                                break;
                        ret++;
                }
        }
        accessMutex.LeaveMutex();
        return (ret > 0);
}

bool MediaStream::deleteSamplesBetween(const JTime& fromTime, const JTime& toTime) {
        int ret = 0;
        DataSample* sample;
        if (accessMutex.EnterMutex(1000)) {
        //      ret = dataSamples.removeAllEntriesBetweenKeys(&fromTime, &toTime);
                ObjectCollection* coll = dataSamples.getAllBetweenKeys(&fromTime, &toTime);
                if (coll != NULL) {
                        while ( ( sample = (DataSample*) coll->getFirst()) != NULL ) {
                                coll->removeNoDelete(0);
                                if (!deleteSample(sample))
                                        break;
                                ret++;
                        }
                        coll->removeAllNoDelete();
                        delete(coll);
                }
        }
        accessMutex.LeaveMutex();
        return (ret > 0);
}

long MediaStream::getTotalDataSize() {

        return currentSize;

        //long size = 0;
        //DataSample* sample;

        //if (accessMutex.EnterMutex(1000)) {
        //      for (int n=0; n<dataSamples.getCount(); n++) {
        //              sample = (DataSample*) dataSamples.get(n);
        //              if (sample != NULL) {
        //                      size += sample->getDataSize();
        //              }
        //      }
        //}
        //accessMutex.LeaveMutex();
        //return size;
}

/*
long MediaStream::getTotalDataMemUseNoBlock() {
        long size = 0;
        DataSample* sample;

        for (int n=0; n<dataSamples.getCount(); n++) {
                sample = (DataSample*) dataSamples.get(n);
                if (sample != NULL) {
                        size += sample->getDataMemUse();
                }
        }
        return size;
}
*/

long MediaStream::getTotalDataMemUse() {

        long size = 0;
        DataSample* sample;

        if (accessMutex.EnterMutex(1000)) {
                for (int n=0; n<dataSamples.getCount(); n++) {
                        sample = (DataSample*) dataSamples.get(n);
                        if (sample != NULL) {
                                size += sample->getDataMemUse();
                        }
                }
        }
        accessMutex.LeaveMutex();
        return size;
}

int MediaStream::getCount() {
        return dataSamples.getCount();
}

ObjectCollection* MediaStream::getDataSampleList() {
        ObjectCollection* col = new ObjectCollection();
        DataSample* sample;

        if (accessMutex.EnterMutex(1000)) {
                for (int n=0; n<dataSamples.getCount(); n++) {
                        sample = (DataSample*) dataSamples.get(n);
                        if (sample != NULL) {
                                col->add(sample->cloneWithoutData());
                        }
                }
        }
        accessMutex.LeaveMutex();
        return col;
}

ObjectCollection* MediaStream::getTimestampList() {
        ObjectCollection* col = new ObjectCollection();
        JTime* t;

        if (accessMutex.EnterMutex(1000)) {
                for (int n=0; n<dataSamples.getCount(); n++) {
                        t = (JTime*) dataSamples.getKey(n);
                        if (t != NULL) {
                                col->add(t->clone());
                        }
                }
        }
        accessMutex.LeaveMutex();
        return col;
}

JTime MediaStream::getOldestSampleTime() {
        JTime *t = NULL;
        // t.setInvalid();
        if (accessMutex.EnterMutex(1000)) {
                t = (JTime*) dataSamples.getFirstKey(); 
        }
        accessMutex.LeaveMutex();
        if (t != NULL)
                return *t;
        else {
                JTime tt;
                tt.setInvalid();
                return tt;
        }
}

JTime MediaStream::getNewestSampleTime() {
        JTime* t = NULL;
        // t.setInvalid();
        if (accessMutex.EnterMutex(1000)) {
                t = (JTime*) dataSamples.getLastKey();
        }
        accessMutex.LeaveMutex();
        if (t != NULL)
                return *t;
        else {
                JTime tt;
                tt.setInvalid();
                return tt;
        }
}

DataSample* MediaStream::getOldestSample() {
        DataSample* sample = NULL;
        if (accessMutex.EnterMutex(1000)) {
                sample = (DataSample*) dataSamples.getFirst();
                if (sample != NULL) {
                        // printf("Putting TimeSeries: \n%s\n\n", (char*) outputData.print());
                        outputData.add(new JTime(), new JString(sample->getDataSize()));
                        // printf("Putting TimeSeries: \n%s\n\n", (char*) outputData.print());
                        while (outputData.getCount() > 10)
                                outputData.removeFirst();
                }
        }
        accessMutex.LeaveMutex();
        return sample;
}

DataSample* MediaStream::getNewestSample() {
        DataSample* sample = NULL;
        
        if (accessMutex.EnterMutex(1000)) {
                sample = (DataSample*) dataSamples.getLast();
                if (sample != NULL) {
                        // printf("Putting TimeSeries: \n%s\n\n", (char*) outputData.print());
                        outputData.add(new JTime(), new JString(sample->getDataSize()));
                        // printf("Putting TimeSeries: \n%s\n\n", (char*) outputData.print());
                        while (outputData.getCount() > 10)
                                outputData.removeFirst();
                }
        }
        accessMutex.LeaveMutex();
        return sample;
}

DataSample* MediaStream::getSample(const JString& id) {
        DataSample* sample = NULL;
        
        if (accessMutex.EnterMutex(1000)) {
                for (int n=getCount()-1; n>=0; n--) {
                        sample = (DataSample*) dataSamples.get(n);
                        if (sample != NULL) {
                                if (sample->id.equals(id)) {
                                        break;
                                }
                        }
                }
                if (sample != NULL) {
                        outputData.add(new JTime(), new JString(sample->getDataSize()));
                        // printf("Putting TimeSeries: \n%s\n\n", (char*) outputData.print());
                        while (outputData.getCount() > 10)
                                outputData.removeFirst();
                }
        }
        accessMutex.LeaveMutex();
        return sample;
}

DataSample* MediaStream::getSampleAt(const JTime& time) {
        DataSample* sample = NULL;
        
        if (accessMutex.EnterMutex(1000)) {
        //      JTime t = time;
        //      t.perfCountReset();
                sample = (DataSample*) dataSamples.get(&time);
                if (sample != NULL) {
                        // printf("Putting TimeSeries: \n%s\n\n", (char*) outputData.print());
                        outputData.add(new JTime(), new JString(sample->getDataSize()));
                        // printf("Putting TimeSeries: \n%s\n\n", (char*) outputData.print());
                        while (outputData.getCount() > 10)
                                outputData.removeFirst();
                }
        }
        accessMutex.LeaveMutex();
        return sample;
}

DataSample* MediaStream::getFirstSampleAfter(const JString& id) {
        DataSample* sample = NULL;
        
        if (accessMutex.EnterMutex(1000)) {
                for (int n=getCount()-1; n>=0; n--) {
                        sample = (DataSample*) dataSamples.get(n);
                        if (sample != NULL) {
                                if (sample->id.equals(id)) {
                                        sample = (DataSample*) dataSamples.getNext();
                                        break;
                                }
                        }
                }
                if (sample != NULL) {
                        outputData.add(new JTime(), new JString(sample->getDataSize()));
                        // printf("Putting TimeSeries: \n%s\n\n", (char*) outputData.print());
                        while (outputData.getCount() > 10)
                                outputData.removeFirst();
                }
        }
        accessMutex.LeaveMutex();
        return sample;
}

DataSample* MediaStream::getFirstSampleAfter(const JTime& time) {
        DataSample* sample = NULL;
        
        if (accessMutex.EnterMutex(1000)) {
                sample = (DataSample*) dataSamples.getFirstAfterKey(&time);
                if (sample != NULL) {
                        // printf("Putting TimeSeries: \n%s\n\n", (char*) outputData.print());
                        outputData.add(new JTime(), new JString(sample->getDataSize()));
                        // printf("Putting TimeSeries: \n%s\n\n", (char*) outputData.print());
                        while (outputData.getCount() > 10)
                                outputData.removeFirst();
                }
        }
        accessMutex.LeaveMutex();
        return sample;
}

DataSample* MediaStream::getLastSampleBefore(const JTime& time) {
        DataSample* sample = NULL;
        
        if (accessMutex.EnterMutex(1000)) {
                sample = (DataSample*) dataSamples.getLastBeforeKey(&time);
                if (sample != NULL) {
                        // printf("Putting TimeSeries: \n%s\n\n", (char*) outputData.print());
                        outputData.add(new JTime(), new JString(sample->getDataSize()));
                        // printf("Putting TimeSeries: \n%s\n\n", (char*) outputData.print());
                        while (outputData.getCount() > 10)
                                outputData.removeFirst();
                }
        }
        accessMutex.LeaveMutex();
        return sample;
}

ObjectCollection* MediaStream::getSamplesBetween(const JTime& fromTime, const JTime& toTime) {
        ObjectCollection* samples = NULL;

        if (accessMutex.EnterMutex(1000)) {
                samples = dataSamples.getAllBetweenKeys(&fromTime, &toTime);
                if (samples != NULL) {
                        samples->noDelete();
                        long size = 0;
                        for (DataSample* sample = (DataSample*) samples->getFirst(); sample != NULL; sample = (DataSample*) samples->getNext()) {
                                size += sample->getDataSize();
                        }
                        outputData.add(new JTime(), new JString(size));
                        while (outputData.getCount() > 10)
                                outputData.removeFirst();
//                      for (int n=0; n<samples->getCount(); n++) {
//                              DataSample* sample = (DataSample*) samples->get(n);
//                              if (sample != NULL) {
////                                    printf("Putting TimeSeries: \n%s\n\n", (char*) outputData.print());
//                                      outputData.add(new JTime(), new JString(sample->getDataSize()));
////                                    printf("Putting TimeSeries: \n%s\n\n", (char*) outputData.print());
//                                      while (outputData.getCount() > 10)
//                                              outputData.removeFirst();
//                              }
//                      }
                }
        }
        accessMutex.LeaveMutex();
        return samples;
}




DataSample* MediaStream::waitForFirstSampleAfter(const JTime& time, long ms) {
//      DataSample* sample = getFirstSampleAfter(time+1);
        DataSample* sample = getFirstSampleAfter(time);
        if (sample != NULL)
                return sample;

        JTime start;
        // JTime now;
        long dif = start.getAge();
        
        while ( dif < ms ) {
                newDataSem.wait(ms - dif);
//              if ( (sample = getFirstSampleAfter(time+1)) != NULL)
                if ( (sample = getFirstSampleAfter(time)) != NULL)
                        return sample;
                // now = JTime();
                dif = start.getAge();
        }
        return NULL;
}

DataSample* MediaStream::waitForLastSampleAfter(const JTime& time, long ms) {
        JTime newest = getNewestSampleTime();
//      if ( (newest.isValid()) && (newest - time > 0) )
        if ( (newest.isValid()) && (newest.compare(&time) > 0) )
                return this->getSampleAt(newest);
        
/*      DataSample* sample = getNewestSample();
        if (sample != NULL) {
                else
                        outputData.removeNewest();
        }
*/
        JTime start;
        // JTime now;
        long dif = start.getAge();
        
//      JTime t1, t2;
        while ( dif < ms ) {
        //      t1 = JTime();
                newDataSem.wait(ms - dif);
        //      t2 = JTime();
        //      printf("[%d/%d]", t2.microDifference(t1), (ms - dif)); fflush(stdout);

                newest = getNewestSampleTime();
        //      if ( (newest.isValid()) && (newest - time > 0) ) {
                if ( (newest.isValid()) && (newest.compare(&time) > 0) ) {
                        return this->getSampleAt(newest);
                }
                // now = JTime();
                dif = start.getAge();
        }
        return NULL;
}

DataSample* MediaStream::waitForFirstSampleAfter(const JString& id, long ms) {
        DataSample* sample = getFirstSampleAfter(id);
        if (sample != NULL)
                return sample;

        JTime start;
        // JTime now;
        long dif = start.getAge();
        
        while ( dif < ms ) {
                newDataSem.wait(ms - dif);
                if ( (sample = getFirstSampleAfter(id)) != NULL)
                        return sample;
                // now = JTime();
                dif = start.getAge();
        }
        return NULL;
}

DataSample* MediaStream::waitForLastSampleAfter(const JString& id, long ms) {
        DataSample* sample = getNewestSample();
        if ((sample != NULL) && (!sample->id.equals(id)))
                return sample;

        JTime start;
        // JTime now;
        long dif = start.getAge();
        
        while ( dif < ms ) {
                newDataSem.wait(ms - dif);
                sample = getNewestSample();
                if ((sample != NULL) && (!sample->id.equals(id)))
                        return sample;
                // now = JTime();
                dif = start.getAge();
        }
        return NULL;
}



bool MediaStream::doMaintenance() {

        int maximumCount = maxCount;
        if (maxCount <= 0)
                maximumCount = STREAMMAXCOUNTDEFAULT;

        DataSample* sample;
        if (this->getCount() > maximumCount) {
                if (accessMutex.EnterMutex(1000)) {
                        while (this->getCount() > maximumCount) {
                                if (!deleteSample(0)) {
                                        accessMutex.LeaveMutex();
                                        return true;
                                }
                                //sample = (DataSample*) dataSamples.getFirst();
                                //if (sample == NULL) {
                                //      accessMutex.LeaveMutex();
                                //      return true;
                                //}
                                //dataSamples.removeNoDelete(0);
                                //sample->markForDeletion();
                                //deletedSamples.add(sample);
                        }
                        accessMutex.LeaveMutex();
                }
        }

        while (deletedSamples.getCount() > maxBackupCount)
                deletedSamples.removeFirst();

        if (currentSize < softMax)
                return true;

        if (accessMutex.EnterMutex(1000)) {

                while ((hardMax > 0) && (currentSize > hardMax)) {
                        if (!deleteSample(0)) {
                                accessMutex.LeaveMutex();
                                return true;
                        }
                        //sample = (DataSample*) dataSamples.getFirst();
                        //dataSamples.removeNoDelete(0);
                        //size -= sample->getDataMemUse();
                        //sample->markForDeletion();
                        //deletedSamples.add(sample);
                }

                while ((softMax > 0) && (currentSize > softMax)) {
                        if ( (sample = (DataSample*) dataSamples.getFirst()) != NULL) {
                                if (sample->isInUse()) {
                                        sample->lockBy(name);
                                        break;
                                }
                                else {
                                        if (!deleteSample(0)) {
                                                accessMutex.LeaveMutex();
                                                return true;
                                        }
                                        //size -= sample->getDataMemUse();
                                        //dataSamples.removeNoDelete(0);
                                        //sample->markForDeletion();
                                        //deletedSamples.add(sample);
                                }
                        }
                }
        }
        accessMutex.LeaveMutex();

        return true;
}



bool MediaStream::createChannel(const JString& name, const JString& fieldname) {
        if (channels.contains(name))
                return false;
        channelNames.put(name, fieldname);
        SortedObjectCollection* channel = new SortedObjectCollection();
        channels.put(name, channel);
        return true;
}

bool MediaStream::destroyChannel(const JString& name) {
        channelNames.remove(name);
        SortedObjectCollection* channel = (SortedObjectCollection*) channels.get(name);
        channels.removeNoDelete(name);
        delete(channel);
        return true;
}

ObjectCollection* MediaStream::searchChannel(const JString& name, double val1, double val2) {
        SortedObjectCollection* channel = (SortedObjectCollection*) channels.get(name);
        if (channel == NULL)
                return NULL;
        SortedEntry e1 = SortedEntry(val1, NULL, NULL);
        SortedEntry e2 = SortedEntry(val2, NULL, NULL);
        ObjectCollection* samples = channel->objectTable.getAllBetween(&e1, &e2);
        if (samples != NULL) {
                ObjectCollection* coll = new ObjectCollection();
                coll->noDelete();
                long size = 0;
                for (SortedEntry* entry = (SortedEntry*) samples->getFirst(); entry != NULL; entry = (SortedEntry*) samples->getNext()) {
                        coll->addLast(entry->object);
                        size += ((DataSample*)entry->object)->getDataSize();
                }
                outputData.add(new JTime(), new JString(size));
                while (outputData.getCount() > 10)
                        outputData.removeFirst();
                delete(samples);
                return coll;
        }
        return NULL;
}

bool MediaStream::addChannelSample(DataSample* sample) {
        if (sample == NULL)
                return false;
        JString fieldname;
        SortedObjectCollection* channel;
        SortedEntry* entry;
        for (int n=0; n<channelNames.getCount(); n++) {
                name = channelNames.getKey(n);
                if ( (channel = (SortedObjectCollection*) channels.get(name)) != NULL) {
                        fieldname = channelNames.get(n);
                        if (sample->params.contains(fieldname)) {
                                entry = new SortedEntry(sample->getParamFloat(fieldname), sample, &channel->objectTable);
                                channel->add(entry);
                                if (sample->channels == NULL)
                                        sample->channels = new ObjectDictionary;
                                sample->channels->put(name, entry);
                                // printf("[%s] Adding sample to channel %s/%s (%d)\n", (char*) sample->timestamp.printTimeMS(), (char*) name, (char*) fieldname, (int)entry->value); fflush(stdout);
                        }
                }
        }
        return true;
}

bool MediaStream::removeChannelSample(DataSample* sample) {
        if (sample == NULL)
                return true;;
        SortedEntry* entry;
        while ( (entry = (SortedEntry*) sample->channels->getFirst()) != NULL) {
                // Remove the linked entry from the channel
                if (entry->table != NULL)
                        entry->table->remove(entry);
                sample->channels->removeFirst();
        }
        return true;
}









} // namespace cmlabs

---