Messenger.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 "Messenger.h"

namespace cmlabs {

Messenger::Messenger(const JString& name, MessageSender* sender, const JString& wakeupMessageType, const JString& phaseChangeType) {
        perfStat = NULL;
        signalHandler = NULL;
        messageSender = sender;
        this->wakeupMessageType = wakeupMessageType;
        this->phaseChangeType = phaseChangeType;
        parameters.noDelete();
        streams.noDelete();
        totalTriggerCount = 0;
        shouldContinue = true;
        this->name = name;
        currentMessage = NULL;
        currentPhaseSpec = NULL;
        startProcessTime.setInvalid();
        verbose = 0;
        priority = 0;
        inQueue.objectTable.sorting |= SORTREVERSE;
//      inQueue.objectTable.sorting |= SORTBYVALUE;
        inQueue.noDelete();
        recentMessages = NULL;
}

Messenger::~Messenger() {
        if (currentMessage != NULL)
                delete(currentMessage);
        if (phaseMutex.enterMutex(1000)) {
                if (currentPhaseSpec != NULL)
                        delete(currentPhaseSpec);
                currentPhaseSpec = NULL;
                phaseMutex.leaveMutex();
        }
}

// Crank Interface

JString Messenger::getName() {
        return name;
}

// Get name of trigger
JString Messenger::getTriggerAlias() {
        JString alias;
        if (phaseMutex.enterMutex(1000)) {
                PhaseSpec* spec = getPhaseSpec();
                if (spec != NULL)
                        alias = spec->lastTriggerName;
                phaseMutex.leaveMutex();
        }
        return alias;
}

Collection Messenger::getDestinations() {
        Collection destinations;
        
        if (messageSender == NULL)
                return destinations;

        if (!phaseMutex.enterMutex(500))
                return destinations;
        PhaseSpec* spec = getPhaseSpec();
        if (spec == NULL)
                return destinations;

        PostSpec* post;

        for (int n=0; n<spec->posts->getCount(); n++) {
                if ( (post = (PostSpec*) spec->posts->get(n)) != NULL) {
                        destinations.add(post->to);
                }
        }
        phaseMutex.leaveMutex();
        return destinations;
}

bool Messenger::isWakeupMessage(Message* msg) {
        if ((msg == NULL) || (msg->getObject() == NULL))
                return false;
        return msg->type.equalsIgnoreCase(wakeupMessageType);
}

int Messenger::getInputMessageCount() {
        return inQueue.getCount();
}

int Messenger::getOutputMessageCount() {
        return outputMessages.getCount();
}

int Messenger::getTriggerCount() {
        return totalTriggerCount;
}

bool Messenger::shouldContinueRunning() {
        if (!mutex.EnterMutex(1000))
                return true;
        bool res = shouldContinue;
        mutex.LeaveMutex();
        return res;
}


bool Messenger::processFirstMessage() {
        if (!shouldContinue)
                return false;

        Message* msg = NULL;
        if (inQueueMutex.EnterMutex(100)) {
                msg = (Message*) inQueue.getFirst();
                inQueueMutex.LeaveMutex();
        }

        if (msg == NULL)
                return false;

        PhaseSpec* newSpec = NULL;

        ObjectDictionary* dict = (ObjectDictionary*) msg->getObject();
        if (dict != NULL)
                newSpec = (PhaseSpec*) dict->get("PhaseSpec");

        if (newSpec == NULL)
                return false;

        checkNewPhaseSpec((PhaseSpec*)newSpec->clone(), msg->getTimestamp());
        return true;
}

Message* Messenger::waitForNewMessage(int timeout) {
        if (!shouldContinue)
                return NULL;

//      Message* msg = (Message*) inputMessages.waitForNewEntry(timeout);
        Message* msg = getNextMessage(timeout);

        if (currentMessage != NULL) {
                if ((perfStat != NULL) && (startProcessTime.isValid())) {
                        perfStat->msgProcessed++;
                        JTime endProcessTime;
                        perfStat->deltaRunSeconds = ((double)endProcessTime.microDifference(startProcessTime)) / 1000000.0;
                        perfStat->totalRunSeconds += perfStat->deltaRunSeconds;
                        perfStat->runCount++;
                }
        }

        if ( (perfStat != NULL) && (perfStat->lastUpdated.getAge() > 1000)) {
                JTime now;
                ThreadStat cur = aThread.getCallingThreadStatistics();
                perfStat->currentCPUUsage = cur.percentCPU / 100.0;
                perfStat->deltaCPUSeconds = ((double)cur.cpuUsage) / 1000000.0;
                perfStat->totalCPUSeconds += perfStat->deltaCPUSeconds;
                perfStat->deltaSeconds = now.microDifference(perfStat->lastUpdated) / 1000000.0;
                perfStat->lastUpdated = now;
                // Normalize to usage per second
                if (perfStat->deltaSeconds > 0)
                        perfStat->deltaCPUSeconds = perfStat->deltaCPUSeconds / perfStat->deltaSeconds;
                aThread.resetCallingThreadStatistics();
        }

        if (msg == NULL)
                return NULL;

        PhaseSpec* newSpec;
        if (msg->getType().equalsIgnoreCase(wakeupMessageType)) {
        //      if (verbose > 2)
        //              messageSender->log(1, "%s triggered by message from '%s' type '%s'...\n", (char*) compName, (char*) msg->getFrom(), (char*) msg->getType());
        //      else
        //              messageSender->log(3, "%s triggered by message from '%s' type '%s'...\n", (char*) compName, (char*) msg->getFrom(), (char*) msg->getType());

                ObjectDictionary* dict = (ObjectDictionary*) msg->getObject();
                if (dict == NULL) {
                        newSpec = NULL;
                }
                else {
                        newSpec = (PhaseSpec*) dict->get("PhaseSpec");
                        dict->removeNoDelete("PhaseSpec");
                }

                if (!checkNewPhaseSpec(newSpec, msg->getTimestamp())) {
                        // We should ignore this message, it is too old
                        delete(msg);
                        return NULL;
                }

                if (currentMessage != NULL) {
                        delete(currentMessage);
                        currentMessage = NULL;
                }
                currentMessage = msg;
                msg = getTriggerMessage();

                if ( (msg != NULL) && (recentMessages != NULL) && (!recentMessages->containsKey(msg)) ) {
                        recentMessages->add(msg->clone(), new ObjectDictionary());
                        if (recentMessages->getCount() > 10)
                                recentMessages->removeFirst();
                        if (dict != NULL) {
                                ObjectCollection* retrieves = new ObjectCollection();
                                dict->put("Retrieves", retrieves);
                                ObjectCollection* rets = this->getAllRetrievedMessages();
                                if (rets != NULL) {
                                        for (Message* mm = (Message*) rets->getFirst(); mm!=NULL; mm = (Message*) rets->getNext())
                                                retrieves->add(new Reference(mm));
                                }
                        }
                }

        }
        else {
        //      if (verbose > 2)
        //              messageSender->log(1, "%s triggered by direct message from '%s' type '%s'...\n", (char*) compName, (char*) msg->getFrom(), (char*) msg->getType());
        //      else
        //              messageSender->log(3, "%s triggered by direct message from '%s' type '%s'...\n", (char*) compName, (char*) msg->getFrom(), (char*) msg->getType());

                // Remove current message, but do not set this new one...
                if (currentMessage != NULL) {
                        delete(currentMessage);
                        currentMessage = NULL;
                }
        }

        if ((perfStat != NULL) && (!startProcessTime.isValid())) {
                aThread.resetCallingThreadStatistics();
        }
        startProcessTime = JTime();
        return msg;
}

bool Messenger::processPhaseSpec(PhaseSpec* newSpec, JTime t) {
        return checkNewPhaseSpec(newSpec, t);
}


bool Messenger::checkNewPhaseSpec(PhaseSpec* newSpec, JTime msgtime) {

        if (!phaseMutex.enterMutex(500))
                return false;

        if ( (newSpec != NULL) && (currentPhaseSpec != NULL) ) {
                if (newSpec->equals(currentPhaseSpec)) {
                        // Process message with current spec
                        delete(currentPhaseSpec);
                        currentPhaseSpec = newSpec;
                        // delete(newSpec);
                        phaseMutex.leaveMutex();
                        return true;
                }
                else {
                        // Is the new phase spec older than our current one
                        if (newSpec->isOlderThan(currentPhaseSpec)) {
                                // ignore this message all together
                                // printf("New Spec too old: %s", (char*) newSpec->name);
                        //      delete(newSpec);
                                phaseMutex.leaveMutex();
                                return false;
                        }
                        else {
                                // new spec is newer, use this
                                delete(currentPhaseSpec);
                                currentPhaseSpec = newSpec;
                                clearOldMessagesFromQueue(currentPhaseSpec->activationTime);
                                phaseMutex.leaveMutex();
                                return true;
                        }
                }
        }
        else if (newSpec != NULL) {
                // This is a new spec and we haven't got an old one
                currentPhaseSpec = newSpec;
                clearOldMessagesFromQueue(currentPhaseSpec->activationTime);
                phaseMutex.leaveMutex();
                return true;
        }
        else {
                // newSpec is NULL, use the current spec...
                delete(currentPhaseSpec);
                currentPhaseSpec = NULL;
                clearOldMessagesFromQueue(msgtime);
                phaseMutex.leaveMutex();
                return true;
        }
}


bool Messenger::log(int level, char *format, ...) {

    va_list args;
    va_start(args, format);

   char* str = new char[10000];
   #ifdef WIN32
                #ifdef CYGWIN
                        vsnprintf(str, 10000, format, args);
                #else
                        _vsnprintf(str, 10000, format, args);
                #endif
   #else
      vsnprintf(str, 10000, format, args);
   #endif // WIN32
   va_end(args);

        bool ret = messageSender->log(level, str);
   delete [] str;

   return ret;
}

bool Messenger::setStatus(const JString& name, const JString& value) {
        return messageSender->setStatus(name, value);
}

JString Messenger::getStatus(const JString& name) {
        return messageSender->getStatus(name);
}

Message* Messenger::getWakeupMessage() {
        if ((!shouldContinue) || (currentMessage == NULL))
                return NULL;
        if (!isWakeupMessage(currentMessage)) {
                return NULL;
        }
        return currentMessage;
}

Message* Messenger::getTriggerMessage() {
        if ((!shouldContinue) || (currentMessage == NULL))
                return NULL;

        if (!isWakeupMessage(currentMessage)) {
                return currentMessage;
        }

        ObjectDictionary* dict = (ObjectDictionary*) currentMessage->getObject();
        if (dict == NULL) {
                return NULL;
        }

        Message* hmsg;
        if ( (hmsg = (Message*) dict->get("TriggerMessage")) == NULL ) {
                return NULL;
        }
        return hmsg;
}

int Messenger::getRetrievedMessageCount() {
        ObjectCollection* retrievedMessages = getAllRetrievedMessages();
        if (retrievedMessages == NULL)
                return 0;
        else
                return retrievedMessages->getCount();
}

ObjectCollection* Messenger::getAllRetrievedMessages() {
        if ((!shouldContinue) || (currentMessage == NULL) || (!isWakeupMessage(currentMessage)))
                return NULL;

        ObjectDictionary* dict = (ObjectDictionary*) currentMessage->getObject();
        if (dict == NULL) {
                return NULL;
        }

        ObjectCollection* retrievedMessages = (ObjectCollection*) dict->get("RetrievedMessages");
        if (retrievedMessages == NULL) {
                return NULL;
        }
        return retrievedMessages;
}

Message* Messenger::getRetrievedMessage(int pos) {
        ObjectCollection* retrievedMessages = getAllRetrievedMessages();
        if (retrievedMessages != NULL) {
                return (Message*) retrievedMessages->get(pos);
        }
        else
                return NULL;
}

PhaseSpec* Messenger::getPhaseSpec() {
//      if ((!shouldContinue) || (currentMessage == NULL) || (!isWakeupMessage(currentMessage)))
//              return NULL;
        if (!shouldContinue)
                return NULL;

        return currentPhaseSpec;
/*      ObjectDictionary* dict = (ObjectDictionary*) currentMessage->getObject();
        if (dict == NULL) {
                return NULL;
        }

        PhaseSpec* spec = (PhaseSpec*) dict->get("PhaseSpec");
        if (spec == NULL) {
                return NULL;
        }
        return spec;*/
}

ObjectCollection* Messenger::retrieveMessages(const JString& retrieveFrom, const JString& retrieveSpecXML) {
        ObjectCollection* col = NULL;
        Message* msg;
        if (retrieveSpecXML.startsWithIgnoreCase("<retrieves"))
                msg = new Message(name, retrieveFrom, "AIR.Retrieves", retrieveSpecXML, "xml");
        else
                msg = new Message(name, retrieveFrom, "AIR.Retrieve", retrieveSpecXML, "xml");
        Message* rmsg = messageSender->sendReceiveMsgTo(retrieveFrom, msg);
        if ( (rmsg != NULL) && (rmsg->getType().containsIgnoreCase(".Success")) ) {
                col = (ObjectCollection*) rmsg->takeObject();
        }
        delete(rmsg);

        if (col == NULL) {
                this->messageSender->log(1, "retrieve from '%s' failed...\n", (char*) retrieveFrom);
        }
        else {
        //      if (verbose > 3)
                        this->messageSender->log(5, "retrieve from '%s' returned %d messages...\n", (char*) retrieveFrom, col->getCount());
        //      else
        //              this->messageSender->log(4, "retrieve from '%s' returned %d messages...\n", (char*) retrieveFrom, col->getCount());
        }
        return col;
}

ObjectCollection* Messenger::retrieveMessages(const JString& retrieveFrom, RetrieveSpec* retrieveSpec) {
        ObjectCollection* col = NULL;
        Message* msg = new Message(name, retrieveFrom, "AIR.Retrieve", retrieveSpec);
        Message* rmsg = messageSender->sendReceiveMsgTo(retrieveFrom, msg);
        if ( (rmsg != NULL) && (rmsg->getType().equalsIgnoreCase("AIR.Retrieve.Success")) ) {
                col = (ObjectCollection*) rmsg->takeObject();
        }
        delete(rmsg);
        if (col == NULL) {
                this->messageSender->log(1, "retrieve from '%s' failed...\n", (char*) retrieveFrom);
        }
        else {
                // if (verbose > 3)
                        this->messageSender->log(5, "retrieve from '%s' returned %d messages...\n", (char*) retrieveFrom, col->getCount());
                // else
                //      this->messageSender->log(4, "retrieve from '%s' returned %d messages...\n", (char*) retrieveFrom, col->getCount());
        }
        return col;
}

ObjectCollection* Messenger::retrieveMessages(const JString& retrieveFrom, const ObjectCollection& retrieveSpecs) {
        ObjectCollection* col = NULL;
        Message* msg = new Message(name, retrieveFrom, "AIR.Retrieves", retrieveSpecs.clone());
        Message* rmsg = messageSender->sendReceiveMsgTo(retrieveFrom, msg);
        if ( (rmsg != NULL) && (rmsg->getType().equalsIgnoreCase("AIR.Retrieves.Success")) ) {
                col = (ObjectCollection*) rmsg->takeObject();
        }
        delete(rmsg);
        if (col == NULL) {
                this->messageSender->log(1, "retrieve from '%s' failed...\n", (char*) retrieveFrom);
        }
        else {
                // if (verbose > 3)
                        this->messageSender->log(5, "retrieve from '%s' returned %d messages...\n", (char*) retrieveFrom, col->getCount());
                // else
                //      this->messageSender->log(4, "retrieve from '%s' returned %d messages...\n", (char*) retrieveFrom, col->getCount());
        }
        return col;
}


Collection Messenger::getParameterNames() {
        return parameters.getAllKeys();
}

Collection Messenger::getStreamNames() {
        return streams.getAllKeys();
}

MediaConnection* Messenger::getStreamConnection(const JString& name) {
        return (MediaConnection*) streams.get(name);
}

Parameter* Messenger::getParameter(const JString& name) {
        return (Parameter*) parameters.get(name);
}






bool Messenger::addOutputMessage(Message* msg) {
        return outputMessages.add(msg);
}

bool Messenger::addOutputMessage(Message* msg, JString destination) {
        
        if (!phaseMutex.enterMutex(500))
                return false;
        
        PhaseSpec* spec = getPhaseSpec();
        if (spec == NULL) {
                phaseMutex.leaveMutex();
                return outputMessages.add(msg);
        }

        PostSpec* post;
        for (int n=0; n<spec->posts->getCount(); n++) {
                if ( (post = (PostSpec*) spec->posts->get(n)) != NULL) {
                        if (post->alias.equalsIgnoreCase(destination)) {
                                msg->setType(post->type);
                                phaseMutex.leaveMutex();
                                return outputMessages.add(msg);
                        }
                }
        }

        msg->setType(destination);
        phaseMutex.leaveMutex();
        return outputMessages.add(msg);
}

bool Messenger::sendOutputMessages() {
        if (messageSender == NULL)
                return false;

        if (!phaseMutex.enterMutex(500))
                return false;

        PhaseSpec* spec = getPhaseSpec();
        if (spec == NULL) {
                phaseMutex.leaveMutex();
                return false;
        }

        Message* triggerMsg = getTriggerMessage();
//      if (triggerMsg == NULL)
//              return false;

        Message* msg;
        Message* msgCopy;
        ObjectCollection destinations;
        destinations.noDelete();
        PostSpec* post;
        Reference* ref;
        ObjectDictionary* dict;
        ObjectCollection* col;
        JString str;
        JString type, posttype, phaseChange;
        int n;

        while ( (msg = (Message*) outputMessages.getFirst()) != NULL) {
                outputMessages.removeNoDelete(0);
                destinations.removeAll();
                if (msg->getTo().length() > 0) {
                        msg->setFrom(getName());
                        msg->setPostedTime(JTime());
                        if (msg->priority == 0)
                                msg->priority = priority;
                        if (perfStat != NULL) {
                                perfStat->msgPosted++;
                                if (msg->stored.length() == 0)
                                        msg->stored = msg->to;
                                perfStat->setLastMessagePost(msg);
                        }
                        // Send message
                        str = JString::format("message from '%s' to '%s' type '%s'", (char*) msg->getFrom(), (char*) msg->getTo(), (char*) msg->getType());
                        if (!messageSender->sendMsgTo(msg->getTo(), msg)) {
                                messageSender->log(1, "Posting of %s failed.\n", (char*) str);
                        }
                        else {
                        //      if (verbose > 1)
                                        messageSender->log(2, "Posted %s...\n", (char*) str);
                        //      else
                        //              messageSender->log(2, "Posted %s...\n", (char*) str);
                        }
                }
                else {
                        type = msg->getType();
                        for (n=0; n<spec->posts->getCount(); n++) {
                                if ( (post = (PostSpec*) spec->posts->get(n)) != NULL) {
                                        if ( (type.length() > 0) &&
                                                (post->type.print().length() > 0) &&
                                                (!post->type.isPartOf(type)) && (!type.equalsIgnoreCase(post->alias)) ) {
                                                // Ignore...
                                        }
                                        else {
                                                destinations.add(post);
                                                if ( (post->phasechange.length() > 0) && (phaseChange.length() == 0) )
                                                        phaseChange = post->to;
                                                if (type.equalsIgnoreCase(post->alias))
                                                        msg->setType(post->type.print());
                                        }
                                }
                        }
                        if (destinations.getCount() == 0) {
                                for (n=0; n<spec->posts->getCount(); n++) {
                                        if ( (post = (PostSpec*) spec->posts->get(n)) != NULL) {
                                                posttype = post->type.print();
                                                if ( (type.length() > 0) && posttype.containsIgnoreCase(type) ) {
                                                        destinations.add(post);
                                                        if ( (post->phasechange.length() > 0) && (phaseChange.length() == 0) )
                                                                phaseChange = post->to;
                                                        msg->setType(posttype);
                                                        if (msg->priority == 0)
                                                                msg->priority = post->priority;
                                                        break;
                                                }
                                        }
                                }
                        }
                        if (destinations.getCount() == 0) {
                                delete(msg);
                        }
                        else {
                                if (phaseChange.length() > 0)
                                        sendPhaseChange(phaseChange);
                                for (int k=0; k < destinations.getCount(); k++) {
                                        if ( (post = (PostSpec*) destinations.get(k)) != NULL ) {
                                                if (k < destinations.getCount() - 1)
                                                        msgCopy = (Message*) msg->clone();
                                                else
                                                        msgCopy = msg;
                                                // if (shouldChangePhase) {
                                                        // msgCopy->setPhaseChangeMsg(true);
                                                        // shouldChangePhase = false;
                                                // }
                                                // If from is hardcoded, overwrite
                                                if (post->from.length() > 0)
                                                        msgCopy->setFrom(post->from);
                                                else
                                                        msgCopy->setFrom(getName());
                                                if (msgCopy->getType().length() == 0)
                                                        msgCopy->setType(post->type);
                                                // If content hardcoded, overwrite
                                                if (post->object != NULL)
                                                        msgCopy->setObject(post->object->clone());
                                                else if (post->content.length() > 0)
                                                        msgCopy->setContentNoParse(post->content, post->language);
                                                msgCopy->setTo(post->to);
                                                if (post->cc.length() > 0)
                                                        msgCopy->setCC(post->cc);
                                                msgCopy->setPostedTime(JTime());
                                                if (msg->priority == 0)
                                                        msg->priority = post->priority;

                                                if (msgCopy->inReplyTo == NULL) {
                                                        if ( (triggerMsg != NULL) && (post->useInReplyTo) && (triggerMsg->inReplyTo != NULL))
                                                                msgCopy->setInReplyTo((Reference*)triggerMsg->inReplyTo->clone());
                                                        else if ( (triggerMsg != NULL) && (post->useInReplyTo) && (triggerMsg->inReplyTo == NULL))
                                                                msgCopy->setInReplyTo(new Reference(triggerMsg));
                                                }

                                                if (perfStat != NULL) {
                                                        perfStat->msgPosted++;
                                                        if (msgCopy->stored.length() == 0)
                                                                msgCopy->stored = msgCopy->to;
                                                        perfStat->setLastMessagePost(msg);
                                                }
                                                // Send message
                                                str = JString::format("message from '%s' to '%s' type '%s'", (char*) msg->getFrom(), (char*) msgCopy->getTo(), (char*) msgCopy->getType());
                                                ref = new Reference(msgCopy);
                                                if (post->noreply)
                                                        msgCopy->noreply = "Yes";
                                                if (!messageSender->sendMsgTo(msgCopy->getTo(), msgCopy)) {
                                                        delete(ref);
                                                        messageSender->log(1, "Posting of %s failed.\n", (char*) str);
                                                }
                                                else {
                                                        messageSender->log(2, "Posted %s...\n", (char*) str);
                                                        if (recentMessages != NULL) {
                                                                if ( (dict = (ObjectDictionary*) recentMessages->getLast()) == NULL)
                                                                        recentMessages->add(new Message(), (dict = new ObjectDictionary()));
                                                                if ( (col = (ObjectCollection*) dict->get("Posted")) == NULL )
                                                                        dict->put("Posted", (col = new ObjectCollection()));
                                                                col->add(ref);
                                                                if (col->getCount() > 10)
                                                                        col->removeFirst();
                                                        }
                                                        else {
                                                                delete(ref);
                                                        }
                                                }
                                        }
                                }
                        }
                }
        }
        phaseMutex.leaveMutex();
        return true;
}

bool Messenger::sendPhaseChange(const JString& dest) {
//      printf("[%s] PHASECHANGE: %s!\n", (char*) getName(), (char*) dest);
        if (messageSender == NULL)
                return false;

        if (!phaseMutex.enterMutex(500))
                return false;

        PhaseSpec* spec = getPhaseSpec();
        if (spec == NULL) {
                phaseMutex.leaveMutex();
                return false;
        }

//      printf("[%s] PHASECHANGE 2!\n", (char*) getName());
        PostSpec* post;
        if (dest.length() > 0) {
//              printf("[%s] PHASECHANGE 3!\n", (char*) getName());
                for (post = (PostSpec*) spec->posts->getFirst(); post != NULL; post = (PostSpec*) spec->posts->getNext()) {
                        if (post->to.equalsIgnoreCase(dest))
                                break;
                }
        }
        else {
//              printf("[%s] PHASECHANGE 4!\n", (char*) getName());
                for (post = (PostSpec*) spec->posts->getFirst(); post != NULL; post = (PostSpec*) spec->posts->getNext()) {
                        if (post->phasechange.equalsIgnoreCase("yes"))
                                break;
                }

                if (post == NULL)
                        post = (PostSpec*) spec->posts->getFirst();
        }

        if (post == NULL) {
                phaseMutex.leaveMutex();
                return false;
        }

//      printf("[%s] PHASECHANGE 5!\n", (char*) getName());
        JString to = post->to;

        Message* msg = new Message(getName(), to, phaseChangeType);
        msg->setPostedTime(JTime());

        if (perfStat != NULL) {
                perfStat->msgPosted++;
                if (msg->stored.length() == 0)
                        msg->stored = to;
                perfStat->setLastMessagePost(msg);
        }
        JString str = JString::format("message from '%s' to '%s' type '%s'", (char*) msg->getFrom(), (char*) to, (char*) msg->getType());

        Reference* ref = new Reference(msg);
        ObjectDictionary* dict;
        ObjectCollection* col;

        if (!messageSender->sendMsgTo(to, msg)) {
                messageSender->log(1, "Posting of %s failed.\n", (char*) str);
//              printf("[%s] PHASECHANGE 6!\n", (char*) getName());
                phaseMutex.leaveMutex();
                delete(ref);
                return false;
        }
        else {
                messageSender->log(2, "Posted %s...\n", (char*) str);
                if (recentMessages != NULL) {
                        if ( (dict = (ObjectDictionary*) recentMessages->getLast()) == NULL)
                                recentMessages->add(new Message(), (dict = new ObjectDictionary()));
                        if ( (col = (ObjectCollection*) dict->get("Posted")) == NULL )
                                dict->put("Posted", (col = new ObjectCollection()));
                        col->add(ref);
                        if (col->getCount() > 10)
                                col->removeFirst();
                }
                else
                        delete(ref);
        }

//      printf("[%s] PHASECHANGE End!\n", (char*) getName());
        phaseMutex.leaveMutex();
        return true;
}

JTime Messenger::getCurrentContextStart() {
        return getContextStart(getCurrentContextName());
}

JTime Messenger::getContextStart(const JString& context) {
        JTime t;
        t.setInvalid();
        PhaseSpec* spec = getPhaseSpec();
        if (spec == NULL)
                return t;
        if (spec->currentContexts == NULL)
                return t;
        JTime* tp = (JTime*) spec->currentContexts->get(context);
        if (tp != NULL)
                return *tp;
        else
                return t;
}

JTime Messenger::getCurrentPhaseStart() {
        JTime t;
        t.setInvalid();
        PhaseSpec* spec = getPhaseSpec();
        if (spec == NULL)
                return t;
        return spec->activationTime;
}

// return current Context name
JString Messenger::getCurrentContextName() {
        PhaseSpec* spec = getPhaseSpec();
        if (spec == NULL)
                return "";
        if (spec->currentContexts == NULL)
                return spec->context;
        JString root = spec->context.split(".").getFirst();
        if (!root.equals(spec->context))
                root += ".";
        JString key;
        for (key = spec->currentContexts->getFirstKey(); key.length() > 0; key = spec->currentContexts->getNextKey()) {
                if (key.startsWithIgnoreCase(root))
                        return key;
        }
        return spec->context;
}

// return current Phase name
JString Messenger::getCurrentPhaseName() {
        PhaseSpec* spec = getPhaseSpec();
        if (spec == NULL)
                return "";
        return spec->name;
}

bool Messenger::clearOldMessagesFromQueue(JTime stamp) {
        if (!stamp.isValid())
                return false;
        JString id;
        ObjectDictionary* dict;
        PhaseSpec* spec;
        Message* hmsg = NULL;
        if (inQueue.getCount() > 0) {
                if (inQueueMutex.EnterMutex(100)) {
                        for (hmsg = (Message*) inQueue.getFirst(); hmsg != NULL; hmsg = (Message*) inQueue.getNext()) {
                                dict = (ObjectDictionary*) hmsg->getObject();
                                if (dict != NULL) {
                                        spec = (PhaseSpec*) dict->get("PhaseSpec");
                                        if (spec != NULL) {
                                                if ( (!spec->equals(currentPhaseSpec)) && (spec->activationTime.isValid()) && (spec->activationTime < stamp) ) {
                                                        inQueue.remove(hmsg);
                                                        delete(hmsg);
                                                }
                                        }
                                }
                        }
                        inQueueMutex.LeaveMutex();
                }
                else
                        return false;
        }
        return true;
}

Message* Messenger::getNextMessage(long ms) {
        JString id;
        Message* hmsg = NULL;
        if (inQueue.getCount() == 0) {
                JTime t;
                if (inQueueSemaphore.wait(ms)) {
                        if (inQueueMutex.EnterMutex(ms)) {
                                if ( (hmsg = (Message*) inQueue.getFirst()) != NULL)
                                        inQueue.removeFirst();
                                inQueueMutex.LeaveMutex();
                        }
                }
        //      if ( (hmsg == NULL) && (t.getAge() < ms) ) {
        //              printf("\n\n   ######## Semaphore too early %.3f #########\n\n", t.getMicroAge());
        //              fflush(stdout);
        //              wait(2000);
        //      }
                return hmsg;
        }
        else {
                // Get signal waiting
                inQueueSemaphore.wait(0);
                if (inQueueMutex.EnterMutex(ms)) {
                        if ( (hmsg = (Message*) inQueue.getFirst()) != NULL)
                                inQueue.removeFirst();
                        inQueueMutex.LeaveMutex();
                }
                return hmsg;
        }
}


// Module Interface
bool Messenger::addInputMessage(Message* msg) {

        if (inQueueMutex.EnterMutex(3000)) {
                inQueue.add(msg);
                inQueueMutex.LeaveMutex();
                inQueueSemaphore.post();
                return true;
        }
        else {
                return false;
        }
}

bool Messenger::addParameter(const JString& name, Parameter* parameter) {
        return parameters.put(name, parameter);
}

bool Messenger::addStream(const JString& name, MediaConnection* media) {
        return streams.put(name, media);
}

bool Messenger::clearStreams() {
        streams.removeAllNoDelete();
//      MediaConnection* media;
//      while ( (media = (MediaConnection*) streams.getFirst()) != NULL) {
//              media->stopContinuousBackgroundReceive();
//              streams.removeNoDelete(0);
//      }
        return true;
}

bool Messenger::terminate() {
        if (!mutex.EnterMutex(1000))
                return false;
        shouldContinue = false;
        clearStreams();
        mutex.LeaveMutex();
        return true;
}



// Connect to a media server by name
MediaConnection* Messenger::connectToMediaServer(JString mediaName) {
        TCPLocation loc = messageSender->resolve(mediaName);
        if (!loc.isValid()) {
        //      printf("\n  Resolving stream '%s' failed...\n", (char*) mediaName);
                return NULL;
        }

        MediaConnection* mediacon = new MediaConnection(getName(), loc);
        if (!mediacon->init()) {
        //      printf("\n  Connection init to stream '%s' failed...\n", (char*) mediaName);
                delete(mediacon);
                return NULL;
        }

        return mediacon;
}

// Recomment to a mediaserver by name
bool Messenger::reconnectToMediaServer(JString mediaName, MediaConnection* mediacon) {
        if (mediacon == NULL)
                return false;

        TCPLocation loc = messageSender->resolve(mediaName);
        if (!loc.isValid())
                return false;

        if (!mediacon->reinit(getName(), loc)) {
                return false;
        }

        return true;
}

// Get the server session ID
JString Messenger::getServerID() {
        JString id;
        Message* send = new Message(getName(), "AIRServer", "GET_ID");
        Message* recv = messageSender->sendReceiveMsgTo("CNS", send);
        if ( (recv != NULL) && (recv->type.equals("ID")) ) {
                id = recv->getContent();
        }
        return id;
}

// Get the server name
JString Messenger::getServerName() {
        JString name;
        Message* send = new Message(getName(), "AIRServer", "GET_NAME");
        Message* recv = messageSender->sendReceiveMsgTo("CNS", send);
        if ( (recv != NULL) && (recv->type.equals("NAME")) ) {
                name = recv->getContent();
        }
        return name;
}





bool Messenger::postOutputMessage(Message* msg) {
        addOutputMessage(msg);
        return sendOutputMessages();
}

bool Messenger::postMessageToAllDestinations(Message* msg) {
        if (msg == NULL)
                return false;

    if (msg->priority == 0)
      msg->priority = priority;
        Message* hmsg;
        Collection destinations = getDestinations();
        int count = destinations.getCount();
        for (int n=0; n<count; n++) {
                if (n == count - 1)
                        hmsg = msg;
                else
                        hmsg = (Message*) msg->clone();
                hmsg->setTo(destinations.get(n));
                messageSender->sendMsgTo(destinations.get(n), hmsg);
        }
        return true;
}


JString Messenger::getParameterString(JString module, JString parameter) {
        JString str = getParameter(module, parameter);
        if (str.startsWith("PARAM_ERROR"))
                return "";
        else
                return str;
}

int Messenger::getParameterInteger(JString module, JString parameter) {
        JString str = getParameter(module, parameter);
        if (str.startsWith("PARAM_ERROR"))
                return PARAM_ERROR;
        else
                return str.toInt();
}

double Messenger::getParameterDouble(JString module, JString parameter) {
        JString str = getParameter(module, parameter);
        if (str.startsWith("PARAM_ERROR"))
                return PARAM_ERROR;
        else
                return str.toDouble();
}

bool Messenger::setParameterString(JString module, JString parameter, JString value) {
        return setParameter(module, parameter, value);
}

bool Messenger::setParameterInteger(JString module, JString parameter, int value) {
        return setParameter(module, parameter, JString(value));
}

bool Messenger::setParameterDouble(JString module, JString parameter, double value) {
        return setParameter(module, parameter, JString(value));
}


ObjectDictionary* Messenger::getParameterSpecs(JString module) {
        Message* msg = messageSender->sendReceiveMsgTo(module, new Message(getName(), module, "PARAM_GET_ALL_SPECS"));
        if ((msg != NULL) && (msg->getType().equals("PARAM_GET_ALL_SPECS_SUCCESS"))) {
                ObjectDictionary* content = (ObjectDictionary*) msg->takeObject();
                delete(msg);
                return content;
        }
        else {
                delete(msg);
                return NULL;
        }
}

Parameter* Messenger::getParameterSpec(JString module, JString parameter) {
        Message* msg = messageSender->sendReceiveMsgTo(module, new Message(getName(), module, "PARAM_GET_SPEC", parameter, ""));
        if ((msg != NULL) && (msg->getType().equals("PARAM_GET_SPEC_SUCCESS"))) {
                Parameter* content = (Parameter*) msg->takeObject();
                delete(msg);
                return content;
        }
        else {
                delete(msg);
                return NULL;
        }
}

JString Messenger::getParameter(JString module, JString parameter) {
        Message* msg = messageSender->sendReceiveMsgTo(module, new Message(getName(), module, "PARAM_GET", parameter, ""));
        if ((msg != NULL) && (msg->getType().equals("PARAM_GET_SUCCESS"))) {
                JString content = msg->getContent();
                delete(msg);
                return content;
        }
        else if (msg != NULL) {
                delete(msg);
                return JString::format("PARAM_ERROR: %s", (char*) msg->getType());
        }
        else {
                return "PARAM_ERROR: Unable to connect to parameter";
        }
}

bool Messenger::setParameter(JString module, JString parameter, JString value) {
//      Message* notifyMsg = new Message(getName(), "", "AIR.Param.Set", JString::format("Module: %s  Parameter: %s  Value: %s", (char*)module, (char*)parameter, (char*)value), "");
//      postMessageToAllDestinations(notifyMsg);
        Message* msg = messageSender->sendReceiveMsgTo(module, new Message(getName(), module, "PARAM_SET", JString::format("%s=%s", (char*)parameter, (char*)value), ""));
        if ((msg != NULL) && (msg->getType().equals("PARAM_SET_SUCCESS"))) {
                value = msg->getContent();
                // AIR.Param.  Tweak.Up / Tweak.Down / Set / Reset                              content: N
                // AIR.Param.  Tweak.Up / Tweak.Down / Set / Reset  .Success            content: Resulting value
//              notifyMsg = new Message(getName(), "", "AIR.Param.Set.Complete", JString::format("Module: %s  Parameter: %s  Value: %s", (char*)module, (char*)parameter, (char*)value), "");
//              notifyMsg->addReferenceToHistory(msg, getName());
                delete(msg);
//              postMessageToAllDestinations(notifyMsg);
                return true;
        }
        else {
                if (msg == NULL) {
                        value = "No Response";
                }
                else {
                        value = msg->getType();
                }
                delete(msg);
//              notifyMsg = new Message(getName(), "", "AIR.Param.Set.Failed", JString::format("Module: %s  Parameter: %s  Reason: %s", (char*)module, (char*)parameter, (char*)value), "");
//              postMessageToAllDestinations(notifyMsg);
                return false;
        }
}

bool Messenger::increaseParameter(JString module, JString parameter, int steps) {
        JString value = JString(steps);
//      Message* notifyMsg = new Message(getName(), "", "AIR.Param.Tweak.Up", JString::format("Module: %s  Parameter: %s  Value: %s", (char*)module, (char*)parameter, (char*)value), "");
//      postMessageToAllDestinations(notifyMsg);
        Message* msg = messageSender->sendReceiveMsgTo(module, new Message(getName(), module, "PARAM_INCREMENT", JString::format("%s=%s", (char*)parameter, (char*)value), ""));
        if ((msg != NULL) && (msg->getType().equals("PARAM_INCREMENT_SUCCESS"))) {
                value = msg->getContent();
//              notifyMsg = new Message(getName(), "", "AIR.Param.Tweak.Up.Complete", JString::format("Module: %s  Parameter: %s  Value: %s", (char*)module, (char*)parameter, (char*)value), "");
//              notifyMsg->addReferenceToHistory(msg, getName());
                delete(msg);
//              postMessageToAllDestinations(notifyMsg);
                return true;
        }
        else {
                if (msg == NULL) {
                        value = "No Response";
                }
                else {
                        value = msg->getType();
                }
                delete(msg);
//              notifyMsg = new Message(getName(), "", "AIR.Param.Tweak.Up.Failed", JString::format("Module: %s  Parameter: %s  Reason: %s", (char*)module, (char*)parameter, (char*)value), "");
//              postMessageToAllDestinations(notifyMsg);
                return false;
        }
}

bool Messenger::decreaseParameter(JString module, JString parameter, int steps) {
        JString value = JString(steps);
//      Message* notifyMsg = new Message(getName(), "", "AIR.Param.Tweak.Down", JString::format("Module: %s  Parameter: %s  Value: %s", (char*)module, (char*)parameter, (char*)value), "");
//      postMessageToAllDestinations(notifyMsg);
        Message* msg = messageSender->sendReceiveMsgTo(module, new Message(getName(), module, "PARAM_DECREMENT", JString::format("%s=%s", (char*)parameter, (char*)value), ""));
        if ((msg != NULL) && (msg->getType().equals("PARAM_DECREMENT_SUCCESS"))) {
                value = msg->getContent();
//              notifyMsg = new Message(getName(), "", "AIR.Param.Tweak.Down.Complete", JString::format("Module: %s  Parameter: %s  Value: %s", (char*)module, (char*)parameter, (char*)value), "");
//              notifyMsg->addReferenceToHistory(msg, getName());
                delete(msg);
//              postMessageToAllDestinations(notifyMsg);
                return true;
        }
        else {
                if (msg == NULL) {
                        value = "No Response";
                }
                else {
                        value = msg->getType();
                }
                delete(msg);
//              notifyMsg = new Message(getName(), "", "AIR.Param.Tweak.Down.Failed", JString::format("Module: %s  Parameter: %s  Reason: %s", (char*)module, (char*)parameter, (char*)value), "");
//              postMessageToAllDestinations(notifyMsg);
                return false;
        }
}

bool Messenger::addParameterItem(JString module, JString parameter, JString value) {
//      Message* notifyMsg = new Message(getName(), "", "AIR.Param.AddItem", JString::format("Module: %s  Parameter: %s  Value: %s", (char*)module, (char*)parameter, (char*)value), "");
//      postMessageToAllDestinations(notifyMsg);
        Message* msg = messageSender->sendReceiveMsgTo(module, new Message(getName(), module, "PARAM_ADD_ITEM", JString::format("%s=%s", (char*)parameter, (char*)value), ""));
        if ((msg != NULL) && (msg->getType().equals("PARAM_ADD_ITEM_SUCCESS"))) {
                value = msg->getContent();
                // AIR.Param.  Tweak.Up / Tweak.Down / Set / Reset                              content: N
                // AIR.Param.  Tweak.Up / Tweak.Down / Set / Reset  .Success            content: Resulting value
//              notifyMsg = new Message(getName(), "", "AIR.Param.AddItem.Complete", JString::format("Module: %s  Parameter: %s  Value: %s", (char*)module, (char*)parameter, (char*)value), "");
//              notifyMsg->addReferenceToHistory(msg, getName());
                delete(msg);
//              postMessageToAllDestinations(notifyMsg);
                return true;
        }
        else {
                if (msg == NULL) {
                        value = "No Response";
                }
                else {
                        value = msg->getType();
                }
                delete(msg);
//              notifyMsg = new Message(getName(), "", "AIR.Param.AddItem.Failed", JString::format("Module: %s  Parameter: %s  Reason: %s", (char*)module, (char*)parameter, (char*)value), "");
//              postMessageToAllDestinations(notifyMsg);
                return false;
        }
}

bool Messenger::removeParameterItem(JString module, JString parameter, JString value) {
//      Message* notifyMsg = new Message(getName(), "", "AIR.Param.RemoveItem", JString::format("Module: %s  Parameter: %s  Value: %s", (char*)module, (char*)parameter, (char*)value), "");
//      postMessageToAllDestinations(notifyMsg);
        Message* msg = messageSender->sendReceiveMsgTo(module, new Message(getName(), module, "PARAM_REMOVE_ITEM", JString::format("%s=%s", (char*)parameter, (char*)value), ""));
        if ((msg != NULL) && (msg->getType().equals("PARAM_REMOVE_ITEM_SUCCESS"))) {
                value = msg->getContent();
                // AIR.Param.  Tweak.Up / Tweak.Down / Set / Reset                              content: N
                // AIR.Param.  Tweak.Up / Tweak.Down / Set / Reset  .Success            content: Resulting value
//              notifyMsg = new Message(getName(), "", "AIR.Param.RemoveItem.Complete", JString::format("Module: %s  Parameter: %s  Value: %s", (char*)module, (char*)parameter, (char*)value), "");
//              notifyMsg->addReferenceToHistory(msg, getName());
                delete(msg);
//              postMessageToAllDestinations(notifyMsg);
                return true;
        }
        else {
                if (msg == NULL) {
                        value = "No Response";
                }
                else {
                        value = msg->getType();
                }
                delete(msg);
//              notifyMsg = new Message(getName(), "", "AIR.Param.RemoveItem.Failed", JString::format("Module: %s  Parameter: %s  Reason: %s", (char*)module, (char*)parameter, (char*)value), "");
//              postMessageToAllDestinations(notifyMsg);
                return false;
        }
}

bool Messenger::resetParameter(JString module, JString parameter) {
        JString value;
//      Message* notifyMsg = new Message(getName(), "", "AIR.Param.Reset", JString::format("Module: %s  Parameter: %s", (char*)module, (char*)parameter), "");
//      postMessageToAllDestinations(notifyMsg);
        Message* msg = messageSender->sendReceiveMsgTo(module, new Message(getName(), module, "PARAM_RESET", parameter, ""));
        if ((msg != NULL) && (msg->getType().equals("PARAM_RESET_SUCCESS"))) {
                value = msg->getContent();
//              notifyMsg = new Message(getName(), "", "AIR.Param.Reset.Complete", JString::format("Module: %s  Parameter: %s  Value: %s", (char*)module, (char*)parameter, (char*)value), "");
//              notifyMsg->addReferenceToHistory(msg, getName());
                delete(msg);
//              postMessageToAllDestinations(notifyMsg);
                return true;
        }
        else {
                if (msg == NULL) {
                        value = "No Response";
                }
                else {
                        value = msg->getType();
                }
                delete(msg);
//              notifyMsg = new Message(getName(), "", "AIR.Param.Reset.Failed", JString::format("Module: %s  Parameter: %s  Reason: %s", (char*)module, (char*)parameter, (char*)value), "");
//              postMessageToAllDestinations(notifyMsg);
                return false;
        }
}

bool Messenger::ping(JString module) {
        Message* msg = messageSender->sendReceiveMsgTo(module, new Message(getName(), module, "PING"));
        JString type = msg->getType();
        delete(msg);
        if (type.equals("PING_SUCCESS"))
                return true;
        else
                return false;
}

TCPLocation Messenger::resolve(const JString& name, bool forceCNSLookup) {
        return messageSender->resolve(name, forceCNSLookup);
}

bool Messenger::addParameter(JString paramname, JString xml) {
        Parameter* parameter = new Parameter(xml);
        if ((parameter->name.length() == 0) || (!parameter->name.equals(paramname))) {
                delete(parameter);
                return false; 
        }
        return parameters.put(parameter->name, parameter);
}

bool Messenger::hasParameter(JString name) {
        return parameters.contains(name);
}

JString Messenger::getParameterString(JString param) {
        Parameter* parameter = getParameter(param);
        if (parameter == NULL)
                return "";
        return parameter->getString();
}

int Messenger::getParameterInteger(JString param) {
        Parameter* parameter = getParameter(param);
        if (parameter == NULL)
                return PARAM_ERROR;
        return parameter->getInteger();
}

double Messenger::getParameterDouble(JString param) {
        Parameter* parameter = getParameter(param);
        if (parameter == NULL)
                return PARAM_ERROR;
        return parameter->getDouble();
}

bool Messenger::setParameterString(JString param, JString value) {
        Parameter* parameter = getParameter(param);
        if (parameter == NULL)
                return false;
        return parameter->setString(value);
}

bool Messenger::setParameterInteger(JString param, int value) {
        Parameter* parameter = getParameter(param);
        if (parameter == NULL)
                return false;
        return parameter->setInteger(value);
}

bool Messenger::setParameterDouble(JString param, double value) {
        Parameter* parameter = getParameter(param);
        if (parameter == NULL)
                return false;
        return parameter->setDouble(value);
}


bool Messenger::addParameterItem(JString param, JString value) {
        Parameter* parameter = getParameter(param);
        if (parameter == NULL)
                return false;
        return parameter->addItem(value);
}

bool Messenger::removeParameterItem(JString param, JString value) {
        Parameter* parameter = getParameter(param);
        if (parameter == NULL)
                return false;
        return parameter->removeItem(value);
}

Collection Messenger::getParameterItems(JString param) {
        Parameter* parameter = getParameter(param);
        if (parameter == NULL)
                return Collection();
        return parameter->values;
}

InfoItem* Messenger::queryCatalog(const JString& catalog, Message* query) {

        if (query == NULL)
                return NULL;

        if (query->from.length() == 0)
                query->from = getName();
        query->setTo(catalog);
        Message* rmsg = messageSender->sendReceiveMsgTo(catalog, query);
        InfoItem* answer = NULL;
        if (rmsg != NULL) { // && (rmsg->getType().equalsIgnoreCase("CATALOG_ANSWER")) ) {
                answer = (InfoItem*) rmsg->takeObject();
        }
        delete(rmsg);
        return answer;
}

JString Messenger::getConfigXML(const JString& module) {

        // if module is empty, use own name
        JString entry = module;
        if (entry.length() == 0) entry = name;

        Message* send = new Message(name, "Config", "GET_CONFIGXML", entry, "");
        Message* recv = messageSender->sendReceiveMsgTo("Config", send);

        JString type = recv->getType();
        JString content = recv->getContent();
        delete(recv);
        if (type.equals("CONFIGXML"))
                return content;
        else
                return "";
}

bool Messenger::setSignalHandler(bool (*handler)(const JString& module, const JString& signal, const JString& content)) {
        signalHandler = handler;
        return true;
}

bool Messenger::handleSignal(const JString& module, const JString& signal, const JString& content) {
        if (signalHandler == NULL)
                return false;
        return signalHandler(module, signal, content);
}

} // namespace cmlabs

---