ObjectQueue.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

************************************************************************/
// ObjectQueue.cpp: implementation of the ObjectQueue class.
//
//////////////////////////////////////////////////////////////////////

#include "ObjectQueue.h"

namespace cmlabs {

//////////////////////////////////////////////////////////////////////
// Construction/Destruction
//////////////////////////////////////////////////////////////////////

ObjectQueue::ObjectQueue()
{
        dict = new ObjectDictionary();
//      dict->noDelete();
}

ObjectQueue::~ObjectQueue()
{
        delete(dict);
}

//! Get total size of payload
unsigned long ObjectQueue::getPayloadSize() const {
        return dict->getPayloadSize();
}

bool ObjectQueue::isEmpty() {
        bool result;
        if (mutex.EnterMutex(1000)) {
                result = (dict->getCount() == 0);
                mutex.LeaveMutex();
                return result;
        }
        else
                return true;
}

int ObjectQueue::getCount() {
        int result;
        if (!mutex.EnterMutex(1000))
                return 0;
        result = dict->getCount();
        mutex.LeaveMutex();
        return result;
}

bool ObjectQueue::removeAll() {
        Object* obj;
        while ( (obj = this->retrieveEntry(0)) != NULL)
                delete(obj);
        return true;
}

ObjectCollection* ObjectQueue::retrieveAll() {
        if (!mutex.EnterMutex(1000))
                return NULL;

        if (dict->getCount() == 0) {
                mutex.LeaveMutex();
                return NULL;
        }

        ObjectCollection* col = new ObjectCollection();
        col->addAll(dict);
        dict->removeAllNoDelete();
        mutex.LeaveMutex();
        return col;
}



JString ObjectQueue::add(Object* obj) {
        JString id = createUniqueID("queue");
        if (!mutex.EnterMutex(1000))
                return "";
        dict->put(id, obj);
        mutex.LeaveMutex();
        sem.post();
        return id;
}

JString ObjectQueue::add(JString id, Object* obj) {
        if (!mutex.EnterMutex(1000))
                return "";
        dict->put(id, obj);
        mutex.LeaveMutex();
        sem.post();
        return id;
}



JString ObjectQueue::waitForNewEntryID(long ms) {
        JString id;

        if (!mutex.EnterMutex(ms))
                return "";
        
        if (dict->getCount() > 0) {
                sem.tryWait();
                id = dict->getFirstKey();
                mutex.LeaveMutex();
        }
        else if (ms > 0) {
                mutex.LeaveMutex();
                if (sem.wait(ms)) {
                        if (!mutex.EnterMutex(1000))
                                return "";
                        id = dict->getFirstKey();
                        mutex.LeaveMutex();
                }
        }
        else {
                mutex.LeaveMutex();
        }
        
        return id;
}

Object* ObjectQueue::waitForNewEntry(long ms) {

        if (!mutex.EnterMutex(ms))
                return NULL;

        Object* obj = NULL;
        if (dict->getCount() > 0) {
                sem.tryWait();
                obj = dict->get(0);
                dict->removeNoDelete(0);
                mutex.LeaveMutex();
        }
        else if (ms > 0) {
                mutex.LeaveMutex();
                if (sem.wait(ms)) {
                        if (!mutex.EnterMutex(1000))
                                return NULL;
                        obj = dict->get(0);
                        dict->removeNoDelete(0);
                        mutex.LeaveMutex();
                }
        }
        else {
                mutex.LeaveMutex();
        }
        
        return obj;
}

bool ObjectQueue::waitForNewEntryToAppear(long ms) {

        if (!mutex.EnterMutex(ms))
                return false;

        Object* obj = NULL;
        if (dict->getCount() > 0) {
                sem.tryWait();
                obj = dict->get(0);
        //      dict->removeNoDelete(0);
                mutex.LeaveMutex();
        }
        else if (ms > 0) {
                mutex.LeaveMutex();
                if (sem.wait(ms)) {
                        if (!mutex.EnterMutex(ms))
                                return false;
                        obj = dict->get(0);
        //              dict->removeNoDelete(0);
                        mutex.LeaveMutex();
                }
        }
        else {
                mutex.LeaveMutex();
        }
        
        return (obj != NULL);
}

Object* ObjectQueue::waitForNewEntry(JString id, long ms) {
        
        Object* entry;
        long waitedSoFar = 0;
        bool done = false;

        // Check all already in there, only wait if not
        while (! done) {
                if (!mutex.EnterMutex(ms))
                        return NULL;
                done = ! ((entry = dict->get(id)) == NULL);
                mutex.LeaveMutex();

                if (! done) {
                        if (!sem.wait(ms - waitedSoFar))
                                waitedSoFar += (ms - waitedSoFar);

                                if (waitedSoFar >= ms) {
                                break;
                        }
                }
        }

        if (done) {
                if (!mutex.EnterMutex(ms))
                        return NULL;
                dict->removeNoDelete(id);
                mutex.LeaveMutex();
        }
        
        return entry;
}


Object* ObjectQueue::retrieveEntry(JString id) {
        if (!mutex.EnterMutex(1000))
                return NULL;
        Object* obj = dict->get(id);
        dict->removeNoDelete(id);
        mutex.LeaveMutex();
        return obj;
}

Object* ObjectQueue::retrieveEntry(int pos) {
        if (!mutex.EnterMutex(1000))
                return NULL;
        Object* obj = dict->get(pos);
        dict->removeNoDelete(pos);
        mutex.LeaveMutex();
        return obj;
}

Object* ObjectQueue::viewEntry(JString id) {
        return dict->get(id);
}

Object* ObjectQueue::viewEntry(int pos) {
        return dict->get(pos);
}

// ***************************************************************
// Binary Functions
// ***************************************************************


//! Get the size of any binary data
long ObjectQueue::getBinarySize(int chunk) {
        if (dict == NULL)
                return 0;
        return dict->getBinarySize(chunk);
}

//! Get the number of data chunks
int ObjectQueue::getBinaryChunkCount() {
        if (dict == NULL)
                return 0;
        return dict->getBinaryChunkCount();
}

//! Get a pointer to the binary data at chunk number 'chunk'
long ObjectQueue::toBinaryBuffer(int chunk, char* buffer, int maxlen) {
        if (dict == NULL)
                return false;
        return dict->toBinaryBuffer(chunk, buffer, maxlen);
}

//! Set the data after separate reception
bool ObjectQueue::fromBinaryBuffer(int chunk, char* buffer, long len) {
        if (dict == NULL)
                return false;
        return dict->fromBinaryBuffer(chunk, buffer, len);
}


bool ObjectQueue::unitTest() {
        Queue testQ;
        return (testQ.unitTest());
}





} // namespace cmlabs

---