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

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

#include "Queue.h"

namespace cmlabs {

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

Queue::Queue()
{
        dict = new Dictionary();
}

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

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

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

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

bool Queue::removeAll() {
        return dict->removeAll();
}


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

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


JString Queue::waitForNewEntry(long ms) {

        if (!mutex.EnterMutex(1000))
                return "";

        JString entry;
        if (dict->getCount() > 0) {
                sem.tryWait();
                entry = dict->get(0);
                dict->remove(0);
                mutex.LeaveMutex();
        }
        else if (ms > 0) {
                mutex.LeaveMutex();
                if (sem.wait(ms)) {
                        if (!mutex.EnterMutex(1000)) {
                                printf(" NO GOT MUTEX!\n");
                                return "";
                        }
                        entry = dict->get(0);
                        dict->remove(0);
                        mutex.LeaveMutex();
                }
        }
        else {
                mutex.LeaveMutex();
        }
        
        return entry;
}


JString Queue::waitForNewEntryID(long ms) {
        JString id;
        bool res;

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

        return id;
}


JString Queue::waitForNewEntry(JString id, long ms) {
        
        JString entry;

        long waitedSoFar = 0;

        bool done = false;

        // Check all already in there, only wait if not
        while (! done) {
                if (!mutex.EnterMutex(1000)) {
                        printf(" NO GOT MUTEX!\n");
                        return "";
                }
                done = ! ((entry = dict->get(id)).length() == 0);
                mutex.LeaveMutex();

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

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

/*      mutex.EnterMutex();
        
        // Check all already in there, only wait if not
        while ((entry = dict->get(id)).length() == 0) {
                if (!sem.wait(ms - waitedSoFar))
                        waitedSoFar += (ms - waitedSoFar);

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

        mutex.LeaveMutex();*/
        
//      printf("waitForNewEntry took %d ms\n", JTime() - t);

        if (done) {
                if (!mutex.EnterMutex(1000)) {
                        printf(" NO GOT MUTEX!\n");
                        return "";
                }
//              printf("--- Left in API:  %d\n", dict->getCount());
                dict->remove(id);
//              printf("*** Left in API:  %d\n", dict->getCount());
                mutex.LeaveMutex();
        }

        return entry;
}
/*
JString Queue::retrieveEntry(JString id) {

        if (id.length() == 0) return "";
        
        JString value;
        
        if (!mutex.EnterMutex(1000))
                return "";

        value = dict->get(id);
        dict->remove(id);

        mutex.LeaveMutex();

        return value;
}
*/
JString Queue::retrieveEntry(JString id) {
        if (!mutex.EnterMutex(1000))
                return "";
        JString str = dict->get(id);
        dict->remove(id);
        mutex.LeaveMutex();
        return str;
}

JString Queue::retrieveEntry(int pos) {
        if (!mutex.EnterMutex(1000))
                return "";
        JString str = dict->get(pos);
        dict->remove(pos);
        mutex.LeaveMutex();
        return str;
}

JString Queue::viewEntry(JString id) {
        return dict->get(id);
}

JString Queue::viewEntry(int pos) {
        return dict->get(pos);
}

bool Queue::unitTest() {

        Queue* q = new Queue();

        // We have added none, this should return ""
        if (q->waitForNewEntryID(500).length() > 0)
                return false;
        
        JString id = q->add("Test");

        // Now id should be in there...
        if (!id.equals(q->waitForNewEntryID(500)))
                return false;

        if (!q->retrieveEntry(id).equals("Test"))
                return false;

        delete(q);

        return true;
}





} // namespace cmlabs

---