jaus++/2.0/src/jaus/core/events/events.cpp

Go to the documentation of this file.

#include "jaus/core/events/events.h"
#include "jaus/core/events/cancelevent.h"
#include "jaus/core/events/confirmeventrequest.h"
#include "jaus/core/events/createevent.h"
#include "jaus/core/events/event.h"
#include "jaus/core/events/queryevents.h"
#include "jaus/core/events/rejecteventrequest.h"
#include "jaus/core/events/reportevents.h"
#include "jaus/core/events/updateevent.h"
#include "jaus/core/transport/transport.h"
#include "jaus/core/component.h"
#include <cxutils/math/cxmath.h>
#include <cstring>
#include <iomanip>
#include <iostream>


using namespace JAUS;

const std::string Events::Name = "urn:jaus:jss:core:Events";

Events::Subscription::Subscription()
{
    mID = mSequenceNumber = 0;
    mPeriodicRate = 0.0;
    mpQueryMessage = NULL;
    mType = Events::Periodic;
    mUpdateTimeMs = 0;
    mpEventService = NULL;
    mTriggerTimeSeconds = 0;
}


Events::Subscription::Subscription(const Events::Subscription& data)
{
    mID = mSequenceNumber = 0;
    mPeriodicRate = 0.0;
    mpQueryMessage = NULL;
    mType = Events::Periodic;
    mUpdateTimeMs = 0;
    mpEventService = NULL;
    mTriggerTimeSeconds = 0;
    *this = data;
}


Events::Subscription::~Subscription()
{
    if(mpQueryMessage)
    {
        delete mpQueryMessage;
    }
}


Events::Subscription& Events::Subscription::operator =(const Events::Subscription& data)
{
    if(this != &data)
    {
        mID = data.mID;
        mSequenceNumber = data.mSequenceNumber;
        mPeriodicRate = data.mPeriodicRate;
        mType = data.mType;
        mUpdateTimeMs = data.mUpdateTimeMs;
        mProducer = data.mProducer;
        mClients = data.mClients;
        if(mpQueryMessage)
        {
            delete mpQueryMessage;
            mpQueryMessage = NULL;
        }
        if(data.mpQueryMessage)
        {
            mpQueryMessage = data.mpQueryMessage->Clone();
        }
        mpEventService = data.mpEventService;
        mTriggerTimeSeconds = data.mTriggerTimeSeconds;
    }
    return *this;
}


Events::Child::Child(const ID& serviceIdentifier, 
                     const ID& parentServiceIdentifier) : Service(serviceIdentifier,
                                                                  parentServiceIdentifier)
{
}


Events::Child::~Child()
{
}


void Events::Child::SignalEvent(const UShort reportMessageCode)
{
    Component* component = GetComponent();
    if(component)
    {
        component->EventsService()->SignalEvent(reportMessageCode);
    }
}


void Events::Child::SignalEvent(const Subscription& info)
{
    Component* component = GetComponent();
    if(component)
    {
        component->EventsService()->SignalEvent(info);
    }
}


Events* Events::Child::EventsService()
{
    return this->GetComponent()->EventsService();
}


const Events* Events::Child::EventsService() const
{
    return this->GetComponent()->EventsService();
}



void Events::Child::SendEvent(const Events::Subscription& info,
                              const Message* payload) const
{
    JAUS::Event eventMessage;
    eventMessage.SetSourceID(GetComponentID());
    eventMessage.SetSequenceNumber(info.mSequenceNumber);
    eventMessage.SetEventID(info.mID); 
    eventMessage.SetReportMessage(payload);
    Address::Set::const_iterator dest;
    for(dest = info.mClients.begin();
        dest != info.mClients.end();
        dest++)
    {
        eventMessage.SetDestinationID(*dest);
        Send(&eventMessage);
    }
}


Events::Events() : Service(Service::ID(Events::Name, 1.0), Service::ID(Transport::Name, 1.0))
{
    mCheckEventsTimeMs = Time::GetUtcTimeMs();
    mCheckEventDelayMs = 5000;
    mPeriodicTimer.RegisterTimerEvent(Events::PeriodicEvent, this);
    mShutdownFlag = false;
}


Events::~Events()
{
}


void Events::Shutdown()
{
    mShutdownFlag = true;
    
    if(mDebugMessagesFlag)
    {
        Mutex::ScopedLock lock(&mDebugMessagesMutex);
        std::cout << "[" << GetServiceID().ToString() << "-" << mComponentID.ToString() << "] - Stopping Events Timer\n";
    }
    mPeriodicTimer.Stop();
    if(mDebugMessagesFlag)
    {
        Mutex::ScopedLock lock(&mDebugMessagesMutex);
        std::cout << "[" << GetServiceID().ToString() << "-" << mComponentID.ToString() << "] - Events Timer Stopped\n";
    }

    Subscription::List copy;
    Subscription::List removed;

    // Make a copy of the current subscriptions list so 
    // that we don't block any other thread.

    mEventsMutex.Lock();
    mEvents.clear();

    Subscription::List::iterator sub;

    if(mDebugMessagesFlag)
    {
        Mutex::ScopedLock lock(&mDebugMessagesMutex);
        std::cout << "[" << GetServiceID().ToString() << "-" << mComponentID.ToString() << "] - Canceling Event Subscriptions.\n";
    }    

    // Remove events we successfully canceled.
    sub = mSubscriptions.begin();
    while(sub != mSubscriptions.end())
    {
        CancelSubscription(*sub, Service::DefaultWaitMs);
        sub = mSubscriptions.erase(sub);
    }    
    
    mSubscriptions.clear();
    mEventsMutex.Unlock();   
    
    if(mDebugMessagesFlag)
    {
        Mutex::ScopedLock lock(&mDebugMessagesMutex);
        std::cout << "[" << GetServiceID().ToString() << "-" << mComponentID.ToString() << "] - Finished Canceling Event Subscriptions.\n";
    }

    mShutdownFlag = false;
}


bool Events::RequestEveryChangeEvent(const Address& provider,
                                     const Message* query,
                                     const unsigned int waitTimeMs)
{
    bool result = false;

    CreateEvent request(provider, GetComponentID());

    request.SetType(Events::EveryChange);
    request.SetQueryMessage(query);
    
    // Generate a local request ID, then add to
    // list of pending responses.
    Byte requestID = 0;
    std::set<Byte>::iterator riter;
    mRequestMutex.Lock();
    riter = mRequestSet.find(requestID);
    while(riter != mRequestSet.end() )
    {
        requestID++;
        riter = mRequestSet.find(requestID);
    }
    mRequestMutex.Unlock();
    
    request.SetRequestID(requestID);
    
    // Setup list of possible responses to command.
    Message::List responses;
    ConfirmEventRequest confirm;
    RejectEventRequest reject;
    responses.push_back(&confirm);
    responses.push_back(&reject);
    
    if(Send(&request, responses, waitTimeMs))
    {
        if(confirm.GetSourceID().IsValid()) // Confirmation.
        {
            Subscription sub;             
            Subscription::List::iterator siter;

            sub.mID = confirm.GetEventID();
            sub.mSequenceNumber = 0;
            sub.mpQueryMessage = query->Clone();
            sub.mType = EveryChange;
            sub.mPeriodicRate = confirm.GetConfirmedPeriodicRate();
            sub.mUpdateTimeMs = Time::GetUtcTimeMs();
            sub.mProducer = provider;
            sub.mClients.insert(GetComponentID());

            Mutex::ScopedLock lock(&mEventsMutex);

            // Don't add duplicate events.
            bool exists = false;
            for(siter = mSubscriptions.begin();
                siter != mSubscriptions.end();
                siter++)
            {
                if(siter->mID == sub.mID &&
                   siter->mProducer == provider &&
                   siter->mpQueryMessage->GetMessageCode() == query->GetMessageCode())
                {
                    *siter = sub;
                    exists = true;
                    break;
                }
            }
            if(!exists)
            {
                mSubscriptions.push_back(sub);
            }

            result = true;
        }
        else if(mDebugMessagesFlag)
        {
            Mutex::ScopedLock lock(&mDebugMessagesMutex);
            std::cout << "[" << GetServiceID().ToString() << "-" << mComponentID.ToString() << "] - Could not create subscription to " << provider.ToString() << " for Query: " << query->GetMessageName() << "\n";
        }
    }
    
    // Remove local request ID from queue.
    mRequestMutex.Lock();
    riter = mRequestSet.find(requestID);
    if(riter != mRequestSet.end() )
    {
        mRequestSet.erase(riter);
    }
    mRequestMutex.Unlock();

    return result;
}


bool Events::RequestPeriodicEvent(const Address &provider, 
                                  const Message *query, 
                                  const double desiredPeriodicRate, 
                                  const double minimumPeriodicRate, 
                                  const unsigned int waitTimeMs)
{
    bool result = false;

    CreateEvent request(provider, GetComponentID());

    request.SetRequestedPeriodicRate(desiredPeriodicRate);
    request.SetType(Events::Periodic);
    request.SetQueryMessage(query);
    
    // Generate a local request ID, then add to
    // list of pending responses.
    Byte requestID = 0;
    std::set<Byte>::iterator riter;
    mRequestMutex.Lock();
    riter = mRequestSet.find(requestID);
    while(riter != mRequestSet.end() )
    {
        requestID++;
        riter = mRequestSet.find(requestID);
    }
    mRequestMutex.Unlock();
    
    request.SetRequestID(requestID);
    
    // Setup list of possible responses to command.
    Message::List responses;
    ConfirmEventRequest confirm;
    RejectEventRequest reject;
    responses.push_back(&confirm);
    responses.push_back(&reject);
    
    if(Send(&request, responses, waitTimeMs))
    {
        if(confirm.GetSourceID().IsValid()) // Confirmation.
        {
            if(confirm.GetConfirmedPeriodicRate() < minimumPeriodicRate)
            {
                // Cancel requested event since it does not 
                // meet minimum requirements.
                CancelEvent cancel(confirm.GetSourceID(), GetComponentID());
                cancel.SetEventID(confirm.GetEventID());
                cancel.SetRequestID(requestID);
                ConfirmEventRequest confirmResponse;
                // Cancel the requested event.
                Send(&cancel, &confirmResponse, waitTimeMs);
            }
            else
            {
                Subscription sub;
                Subscription::List::iterator siter;

                sub.mID = confirm.GetEventID();
                sub.mSequenceNumber = 0;
                sub.mpQueryMessage = query->Clone();
                sub.mType = Periodic;
                sub.mPeriodicRate = confirm.GetConfirmedPeriodicRate();
                sub.mUpdateTimeMs = Time::GetUtcTimeMs();
                sub.mProducer = provider;
                sub.mClients.insert(GetComponentID());

                Mutex::ScopedLock lock(&mEventsMutex);

                // Don't add duplicate events.
                bool exists = false;
                for(siter = mSubscriptions.begin();
                    siter != mSubscriptions.end();
                    siter++)
                {
                    if(siter->mProducer == provider &&
                       siter->mpQueryMessage->GetMessageCode() == query->GetMessageCode() &&
                       siter->mpQueryMessage->GetPresenceVector() == query->GetPresenceVector())
                    {
                        *siter = sub;
                        exists = true;
                        break;
                    }
                }
                if(!exists)
                {
                    mSubscriptions.push_back(sub);
                }

                result = true;
            }
        }
        else if(mDebugMessagesFlag)
        {
            Mutex::ScopedLock lock(&mDebugMessagesMutex);
            std::cout << "[" << GetServiceID().ToString() << "-" << mComponentID.ToString() << "] - Could not create subscription to " << provider.ToString() << " for Query: " << query->GetMessageName() << "\n";
        }
    }
    
    // Remove local request ID from queue.
    mRequestMutex.Lock();
    riter = mRequestSet.find(requestID);
    if(riter != mRequestSet.end() )
    {
        mRequestSet.erase(riter);
    }
    mRequestMutex.Unlock();

    return result;
}


void Events::SignalEvent(const UShort reportMessageCode,
                         const bool changeOnly)
{
    mEventsMutex.Lock();
    Subscription::Map toGenerate = mEvents;
    mEventsMutex.Unlock();

    Subscription::Map::iterator sub;
    for(sub = toGenerate.begin();
        sub != toGenerate.end();
        sub++)
    {
        if(reportMessageCode == (UShort)(sub->first & 0xFFFF) &&
            (sub->second.mType == EveryChange || changeOnly == false) &&
            sub->second.mpEventService != NULL)
        {
            ((Child *)(sub->second.mpEventService))->GenerateEvent(sub->second);
            // Update the status of the event generated.
            mEventsMutex.Lock();
            Subscription::Map::iterator e = mEvents.find(sub->first);
            if(e != mEvents.end())
            {
                e->second.mSequenceNumber++;
                e->second.mTriggerTimeSeconds = CxUtils::Timer::GetTimeSeconds();
                e->second.mUpdateTimeMs = Time::GetUtcTimeMs();
            }
            mEventsMutex.Unlock();
        }
    }
}


void Events::SignalEvent(const Subscription& info,
                         const bool changeOnly)
{
    mEventsMutex.Lock();
    Subscription::Map toGenerate = mEvents;
    mEventsMutex.Unlock();

    UInt key = info.mID;
    key <<= USHORT_BITS;
    key |= info.mpQueryMessage->GetMessageCodeOfResponse();

    Subscription::Map::iterator sub;
    for(sub = toGenerate.begin();
        sub != toGenerate.end();
        sub++)
    {
        if(sub->first == key)
        {
            ((Child *)(sub->second.mpEventService))->GenerateEvent(sub->second);
            // Update the status of the event generated.
            mEventsMutex.Lock();
            Subscription::Map::iterator e = mEvents.find(sub->first);
            if(e != mEvents.end())
            {
                e->second.mSequenceNumber++;
                e->second.mTriggerTimeSeconds = CxUtils::Timer::GetTimeSeconds();
                e->second.mUpdateTimeMs = Time::GetUtcTimeMs();
            }
            mEventsMutex.Unlock();   
            break;
        }
    }
}


void Events::PrintStatus() const
{

    std::cout << "[" << GetServiceID().ToString() << "] - " << GetComponentID().ToString() << "\n";
    mEventsMutex.Lock();
    unsigned int supporting = (unsigned int)mEvents.size();
    unsigned int subscribing = (unsigned int)mSubscriptions.size();
    mEventsMutex.Unlock();
    std::cout << "Supporting " << supporting << " and Requesting " << subscribing << " Events.\n";
}


bool Events::CancelSubscription(const Address& id,
                                const UShort reportMessageCode,
                                const unsigned int waitTimeMs)
{
    bool result = true;
    
    Subscription::List copy;
    Subscription::List removed;

    // Make a copy of the current subscriptions list so 
    // that we don't block any other thread.

    mEventsMutex.Lock();
    copy = mSubscriptions;
    mEventsMutex.Unlock();

    Subscription::List::iterator sub;

    for(sub = copy.begin();
        sub != copy.end();
        sub++)
    {
        if((sub->mProducer == id || id == Address())&&
            (reportMessageCode == 0 || sub->mpQueryMessage->GetMessageCodeOfResponse() == reportMessageCode))
        {
            if(CancelSubscription(*sub, waitTimeMs))
            {
                removed.push_back(*sub);
            }
            else
            {
                result = false;
            }
        }
    }

    // Remove events we successfully canceled.
    mEventsMutex.Lock();
    for(sub = removed.begin();
        sub != removed.end();
        sub++)
    {
        Subscription::List::iterator rem;
        for(rem = mSubscriptions.begin();
            rem != mSubscriptions.end();
            rem++)
        {
            if(sub->mProducer == rem->mProducer &&
                sub->mID == rem->mID &&
                sub->mpQueryMessage->GetMessageCode() == rem->mpQueryMessage->GetMessageCode())
            {
                mSubscriptions.erase(rem);
                break;
            }
        }
    }
    mEventsMutex.Unlock();

    return result;
}


bool Events::CancelSubscription(const Address& id,
                                const UShort reportMessageCode,
                                const Byte eventID,
                                const unsigned int waitTimeMs)
{
    bool result = true;
    
    Subscription::List copy;
    Subscription::List removed;

    // Make a copy of the current subscriptions list so 
    // that we don't block any other thread.

    mEventsMutex.Lock();
    copy = mSubscriptions;
    mEventsMutex.Unlock();

    Subscription::List::iterator sub;

    for(sub = copy.begin();
        sub != copy.end();
        sub++)
    {
        if(sub->mProducer == id &&
            sub->mID == eventID && 
            (reportMessageCode == 0 || sub->mpQueryMessage->GetMessageCodeOfResponse() == reportMessageCode))
        {
            if(CancelSubscription(*sub, waitTimeMs))
            {
                removed.push_back(*sub);
            }
            else
            {
                result = false;
            }
        }
    }

    // Remove events we successfully canceled.
    mEventsMutex.Lock();
    for(sub = removed.begin();
        sub != removed.end();
        sub++)
    {
        Subscription::List::iterator rem;
        for(rem = mSubscriptions.begin();
            rem != mSubscriptions.end();
            rem++)
        {
            if(sub->mProducer == rem->mProducer &&
                sub->mID == rem->mID &&
                sub->mpQueryMessage->GetMessageCode() == rem->mpQueryMessage->GetMessageCode())
            {
                mSubscriptions.erase(rem);
                break;
            }
        }
    }
    mEventsMutex.Unlock();

    return result;
}


bool Events::HaveSubscription(const UShort reportMessageCode, 
                              const Address &id, 
                              const bool verifyWithQuery, 
                              const unsigned int waitTimeMs) const
{
    bool result = false;
    if(verifyWithQuery && id.IsValid())
    {
        QueryEvents query(id, GetComponentID());
        query.SetQueryType(QueryEvents::MessageID);
        query.SetQueryFilter(reportMessageCode);

        ReportEvents report;

        if(Send(&query, &report, waitTimeMs))
        {
            if(report.GetRecords()->size() > 0)
            {
                result = true;
            }
        }
    }
    else
    {
        Mutex::ScopedLock lock(&mEventsMutex);
        Subscription::List::const_iterator sub;
        for(sub = mSubscriptions.begin();
            sub != mSubscriptions.end() && false == result;
            sub++)
        {
            if((id == Address() || sub->mProducer == id) && 
                sub->mpQueryMessage->GetMessageCodeOfResponse() == reportMessageCode)
            {
                result = true;
                break;
            }
        }
    }

    return result;
}


bool Events::HaveSubscribers(const UShort reportMessageCode) const
{
    bool result = false;

    Mutex::ScopedLock lock(&mEventsMutex);
    Subscription::Map::const_iterator sub;
    for(sub = mEvents.begin();
        sub != mEvents.end() && false == result;
        sub++)
    {
        if(sub->second.mpQueryMessage->GetMessageCodeOfResponse() == reportMessageCode)
        {
            result = true;
            break;
        }
    }

    return result;
}


void Events::Receive(const Message* message)
{
    if(mShutdownServiceFlag)
    {
        return;
    }
    switch(message->GetMessageCode())
    {
    case CANCEL_EVENT:
        {
            const CancelEvent* input = dynamic_cast<const CancelEvent*>(message);
            double maxFrequencyHz = 0.5;
            bool canceled = false;
            {
                Mutex::ScopedLock lock(&mEventsMutex);
                Subscription::Map::iterator sub;
                
                for(sub = mEvents.begin();
                    sub != mEvents.end() && false == canceled;
                    sub++)
                {
                    if(input->GetEventID() == sub->second.mID)
                    {
                        Address::Set::iterator client = sub->second.mClients.find(input->GetSourceID());
                        if(client != sub->second.mClients.end())
                        {
                            sub->second.mClients.erase(client);
                            canceled = true;
                        }
                        // If no more clients, erase the event alltogether.
                        if(sub->second.mClients.size() == 0)
                        {
                            mEvents.erase(sub);
                        }
                        if(canceled)
                        {
                            break;
                        }
                    }
                }
              
                // Look up how what the highest update frequency is
                // for events, so if needed, we can slow down the
                // periodic timer.
                for(sub = mEvents.begin();
                    sub != mEvents.end() && true == canceled;
                    sub++)
                {
                    if(sub->second.mPeriodicRate > maxFrequencyHz)
                    {
                        maxFrequencyHz = sub->second.mPeriodicRate;
                    }
                }                
            }

            // Send confirmation or rejection of cancelation
            if(canceled)
            {
                if(mDebugMessagesFlag)
                {
                    Mutex::ScopedLock lock(&mDebugMessagesMutex);
                    std::cout << "[" << GetServiceID().ToString() << "-" << mComponentID.ToString() << "] - Event for " << input->GetSourceID().ToString() << " was Canceled.\n";
                }
                ConfirmEventRequest response(input->GetSourceID(), GetComponentID());
                response.SetRequestID(input->GetRequestID());
                response.SetEventID(input->GetEventID());
                Send(&response);   
                
                // See if we need to change the periodic timer update frequency.  Throttle 
                // down as needed.
                if(mPeriodicTimer.IsActive() && mPeriodicTimer.GetFrequency() > maxFrequencyHz)
                {
                    mPeriodicTimer.ChangeFrequency(maxFrequencyHz);
                }   
            }
            else
            {
                RejectEventRequest response(input->GetSourceID(), GetComponentID());
                response.SetRequestID(input->GetRequestID());
                response.SetResponseCode(RejectEventRequest::InvalidEventID);
                Send(&response);
            }
        }
        break;
    case UPDATE_EVENT:
    case CREATE_EVENT:
        {
            const CreateEvent* create = dynamic_cast<const CreateEvent*>(message);
            const UpdateEvent* update = dynamic_cast<const UpdateEvent*>(message);
            Service::Map children = GetChildServices();
            Service::Map::iterator citer;
            Events::Child* child = NULL;
            Message* queryMessage = NULL;
            double confirmedPeriodicRate = 0.0;
            std::string errorMessage;
            UShort signalEventCode = 0;
        
            bool sentResponse = false;

            for(citer = children.begin();
                citer != children.end();
                citer++)
            {
                child = dynamic_cast<Events::Child*>(citer->second);
                if(child && (queryMessage = child->CreateMessage(create->GetQueryMessageCode())) != NULL)
                {
                    
                    if(queryMessage->ReadMessageBody(*create->GetQueryMessage()) < 0)
                    {
                        // Failed to read message data.
                        return;
                    }
                    if((create && child->IsEventSupported(create->GetType(),
                                                          create->GetRequestedPeriodicRate(),
                                                          queryMessage,
                                                          confirmedPeriodicRate,
                                                          errorMessage)) ||
                       (update && child->IsEventSupported(update->GetType(),
                                                          update->GetRequestedPeriodicRate(),
                                                          queryMessage,
                                                          confirmedPeriodicRate,
                                                          errorMessage)) )
                    {                        
                        Mutex::ScopedLock lock(&mEventsMutex);
                        Subscription::Map::iterator siter;
                        Subscription subscription;

                        bool exists = false;
                        for(siter = mEvents.begin();
                            siter != mEvents.end();
                            siter ++)
                        {
                            // Keep trying to generate a "unique" subscription ID that is
                            // not already used for this type of data.
                            if(siter->second.mpQueryMessage->GetMessageCode() == queryMessage->GetMessageCode())
                            {
                                subscription.mID = siter->second.mID + 1;
                            }
                            // See if this event already exists, and if so, add the
                            // subscriber to it.
                            Packet queryBody1, queryBody2;
                            siter->second.mpQueryMessage->WriteMessageBody(queryBody1);
                            queryMessage->WriteMessageBody(queryBody2);
                            if(siter->second.mpQueryMessage->GetMessageCode() == queryMessage->GetMessageCode() &&
                                siter->second.mpQueryMessage->GetPresenceVector() == queryMessage->GetPresenceVector() &&
                                siter->second.mType == create->GetType() &&
                                queryBody1.Length() == queryBody2.Length() &&
                                memcmp(queryBody1.Ptr(), queryBody2.Ptr(), queryBody1.Length()) == 0)
                            {
                                exists = true;
                                Type type = create ? create->GetType() : update->GetType();
                                if(type == Periodic)
                                {
                                    if(confirmedPeriodicRate > siter->second.mPeriodicRate)
                                    {
                                        siter->second.mPeriodicRate = confirmedPeriodicRate;
                                    }
                                }
                                siter->second.mClients.insert(message->GetSourceID());
                                subscription = siter->second;
                                break;
                            }
                        }
                        // If this is an Update Event message, and
                        // the event doesn't exist, return error.
                        if(update && (exists == false || subscription.mType != update->GetType()))
                        {
                            RejectEventRequest reject(update->GetSourceID(), GetComponentID());
                            if(subscription.mType != update->GetType())
                            {
                                reject.SetResponseCode(RejectEventRequest::InvalidSetup);
                                errorMessage = "Change of Event Type (Periodic/Every Change) Not Supported";
                            }
                            else
                            {
                                reject.SetResponseCode(RejectEventRequest::InvalidEventID);
                            }
                            reject.SetRequestID(update->GetRequestID());
                            if(errorMessage.size())
                            {
                                reject.SetErrorMessage(errorMessage);
                            }
                            Send(&reject);
                            return;
                        }                        
                        // Create new event if it doesn't exist.
                        else if(create && exists == false)
                        {
                            subscription.mPeriodicRate = confirmedPeriodicRate;
                            subscription.mpQueryMessage = queryMessage; queryMessage = NULL;
                            subscription.mType = create->GetType();
                            subscription.mUpdateTimeMs = Time::GetUtcTimeMs();
                            subscription.mProducer = GetComponentID();
                            subscription.mClients.insert(create->GetSourceID());
                            subscription.mpEventService = child;
                            UInt key = subscription.mID;
                            key <<= USHORT_BITS;
                            key |= subscription.mpQueryMessage->GetMessageCodeOfResponse();
                            mEvents[key] = subscription;
                        }
                        
                        ConfirmEventRequest confirm(message->GetSourceID(), GetComponentID());
                        confirm.SetRequestID((create ? create->GetRequestID() : update->GetRequestID()));
                        confirm.SetConfirmedPeriodicRate(confirmedPeriodicRate);
                        confirm.SetEventID(subscription.mID);
                        if(true == (sentResponse = Send(&confirm)))
                        {
                            signalEventCode = subscription.mpQueryMessage->GetMessageCodeOfResponse();
                            //SignalEvent(subscription.mpQueryMessage->GetMessageCodeOfResponse(), false);
                        }
                        // Start periodic timer to either generate periodic
                        // events, or to verify events are still valid (or re-create subscriptions).
                        if(mPeriodicTimer.IsActive() == false)
                        {
                            std::stringstream tname;
                            tname << "JEVENTS - " << GetComponentID().ToString();
                            mPeriodicTimer.SetName(tname.str());
                            if(subscription.mType == Events::Periodic)
                            {
                                mPeriodicTimer.Start(subscription.mPeriodicRate);
                            }
                            else
                            {
                                mPeriodicTimer.Start(0.5);
                            }
                        }
                        else if(subscription.mPeriodicRate > mPeriodicTimer.GetFrequency())
                        {
                            mPeriodicTimer.ChangeFrequency(subscription.mPeriodicRate);
                        }
                    }
                    
                    // If the child can create the query message
                    // then it is the Service that we need to check.
                    // Make sure to delete any memory we created.
                    if(queryMessage)
                    {
                        delete queryMessage;
                        queryMessage = NULL;
                        break;
                    }   
                    
                }
                if(signalEventCode != 0)
                {
                    SignalEvent(signalEventCode, false);
                }
            }
            if(sentResponse == false)
            {
                RejectEventRequest reject(message->GetSourceID(), GetComponentID());
                reject.SetResponseCode(RejectEventRequest::MessageNotSupported);
                reject.SetRequestID((create ? create->GetRequestID() : update->GetRequestID()));
                if(errorMessage.size())
                {
                    reject.SetErrorMessage(errorMessage);
                }
                Send(&reject);
            }
        }
        break;
    case EVENT:
        {
            const Event* input = dynamic_cast<const Event*>(message);
            Transport* transport = dynamic_cast<Transport *>(GetParentService());
            Message* reportMessage = NULL;
            if(transport)
            {
                // Read the Report Data, first create a message to read it.
                reportMessage = transport->CreateMessage(input->GetReportMessageCode());
                if(reportMessage)
                {
                    reportMessage->CopyHeaderData(input);
                    if(reportMessage->ReadMessageBody(*input->GetReportMessage()) >= 0)
                    {
                        // Push the message to child services who will process.
                        GetParentService()->Receive(reportMessage);
                    }
                }
                else if(mDebugMessagesFlag)
                {
                    Mutex::ScopedLock lock(&mDebugMessagesMutex);
                    std::cout << "[" << GetServiceID().ToString() << "-" << mComponentID.ToString() << "] - Received unsupported Report [" << std::setbase(16) << input->GetReportMessageCode() << std::setbase(10) << "] within Event Message\n";
                }
                
                
            }
            // Update event subscription stats.
            Mutex::ScopedLock lock(&mEventsMutex);
            Subscription::List::iterator subscription;
            for(subscription = mSubscriptions.begin();
                subscription != mSubscriptions.end();
                subscription++)
            {
                if(subscription->mID == input->GetEventID() &&
                    subscription->mProducer == input->GetSourceID())
                {
                    subscription->mSequenceNumber = input->GetSequenceNumber();
                    subscription->mUpdateTimeMs = Time::GetUtcTimeMs();

                    Callback::Set::iterator cb;
                    mEventCallbackMutex.Lock();
                    for(cb = mEventCallbacks.begin();
                        cb != mEventCallbacks.end() && reportMessage;
                        cb++)
                    {
                        Events::Callback* myCB = dynamic_cast<Events::Callback*>(*cb);
                        if(myCB)
                        {
                            myCB->ProcessEvent(subscription->mType,
                                               subscription->mID,
                                               subscription->mSequenceNumber,
                                               reportMessage);
                        }
                    }
                    mEventCallbackMutex.Unlock();
                    break;
                }
            }
            // Delete message when finished.
            if(reportMessage)
            {
                delete reportMessage;
            }
        }
        break;
    case QUERY_EVENTS:
        {
            const QueryEvents* input = dynamic_cast<const QueryEvents*>(message);
            ReportEvents report(input->GetSourceID(), GetComponentID());
            Mutex::ScopedLock lock(&mEventsMutex);
            Events::Subscription::Map::iterator sub;
            ReportEvents::Record record;
            // Add events to report and send.
            for(sub = mEvents.begin();
                sub != mEvents.end();
                sub++)
            {
                // Only report events associated with the component
                // requesting.
                if(sub->second.mClients.find(message->GetSourceID()) != sub->second.mClients.end())
                {
                    record.mEventID = sub->second.mID;
                    record.mEventType = sub->second.mType;
                    record.mQueryCode = sub->second.mpQueryMessage->GetMessageCode();
                    record.mQueryMessage.Clear();
                    sub->second.mpQueryMessage->WriteMessageBody(record.mQueryMessage);
                    if(input->GetQueryType() == QueryEvents::AllEvents)
                    {
                        report.GetRecords()->push_back(record);
                    }
                    else if(input->GetQueryType() == QueryEvents::EventID && 
                            record.mEventID == (Byte)input->GetQueryFilter())
                    {
                        report.GetRecords()->push_back(record);
                    }
                    else if(input->GetQueryType() == QueryEvents::EventType &&
                            input->GetQueryFilter() == (UShort)record.mEventType)
                    {
                        report.GetRecords()->push_back(record);
                    }
                    else if(input->GetQueryType() == QueryEvents::MessageID &&
                            (record.mQueryCode == input->GetQueryFilter() ||
                            sub->second.mpQueryMessage->GetMessageCodeOfResponse() == input->GetQueryFilter()))
                    {
                        report.GetRecords()->push_back(record);
                    }
                }
            }
            Send(&report);
        }
        break;
    case CONFIRM_EVENT_REQUEST:
    case REJECT_EVENT_REQUEST:
        break;
    case REPORT_EVENTS:
        {
            const ReportEvents* input = dynamic_cast<const ReportEvents*>(message);
            Subscription::List subscriptions;
            Subscription::List::iterator sub;
            
            mEventsMutex.Lock();
            subscriptions = mSubscriptions;
            mEventsMutex.Unlock();

            // Go through the list of subscriptions we have
            // and verify the subscription is still there.
            for(sub = subscriptions.begin();
                sub != subscriptions.end();
                sub++)
            {
                if(sub->mProducer == input->GetSourceID())
                {
                    ReportEvents::Record::List::const_iterator record;
                    bool exists = false;
                    for(record = input->GetRecords()->begin();
                        record != input->GetRecords()->end();
                        record++)
                    {
                        if(record->mEventType == sub->mType &&
                           record->mQueryCode == sub->mpQueryMessage->GetMessageCode())
                        {
                            exists = true;
                            break;
                        }
                    }
                    if(exists == false)
                    {
                        // Re-create the event.
                        if(sub->mType == Periodic)
                        {
                            RequestPeriodicEvent(sub->mProducer,
                                                 sub->mpQueryMessage,
                                                 sub->mPeriodicRate,
                                                 sub->mPeriodicRate - 0.1);
                        }
                        else
                        {
                            RequestEveryChangeEvent(sub->mProducer,
                                                    sub->mpQueryMessage);
                        }
                    }
                }
            }
        }
        break;
    default:
        break;
    };
}


Message* Events::CreateMessage(const UShort messageCode) const
{
    Message* message = NULL;

    switch(messageCode)
    {
    case CANCEL_EVENT:
        message = new CancelEvent();
        break;
    case CONFIRM_EVENT_REQUEST:
        message = new ConfirmEventRequest();
        break;
    case CREATE_EVENT:
        message = new CreateEvent();
        break;
    case EVENT:
        message = new Event();
        break;
    case QUERY_EVENTS:
        message = new QueryEvents();
        break;
    case REJECT_EVENT_REQUEST:
        message = new RejectEventRequest();
        break;
    case REPORT_EVENTS:
        message = new ReportEvents();
        break;
    case UPDATE_EVENT:
        message = new UpdateEvent();
        break;
    default:
        message = NULL;
        break;
    };

    return message;
}


void Events::CheckServiceStatus(const unsigned int timeSinceLastCheckMs)
{
    // Check to see if we need to re-create any event
    // subscriptions.
    if(mShutdownFlag == false && Time::GetUtcTimeMs() - mCheckEventsTimeMs >= mCheckEventDelayMs)
    {
        // Query status of events and re-create as needed.
        Subscription::List subscriptions;
        Subscription::List::iterator sub;
        mEventsMutex.Lock();
        subscriptions = mSubscriptions;
        mEventsMutex.Unlock();

        QueryEvents query;
        query.SetSourceID(GetComponentID());
        // Go through each subscription and verify with
        // the producer of the event.
        for(sub = subscriptions.begin();
            sub != subscriptions.end() && mShutdownFlag == false;
            sub++)
        {
            query.SetDestinationID(sub->mProducer);            
            query.SetQueryType(QueryEvents::EventID);
            query.SetQueryFilter(sub->mID);
            Send(&query);
        }
        // Update check time.
        mCheckEventsTimeMs = Time::GetUtcTimeMs();
    }
}


Events::Subscription::List Events::GetSubscriptions(const Address& source,
                                            const UShort reportType) const
{
    Mutex::ScopedLock lock(&mEventsMutex);
    Subscription::List result;
    Subscription::List::const_iterator s;
    for(s = mSubscriptions.begin();
        s != mSubscriptions.end();
        s++)
    {
        if(source.IsValid() == false ||
           s->mProducer == source)
        {
            if(reportType == 0 || 
               reportType == (int)s->mpQueryMessage->GetMessageCodeOfResponse())
            {
                result.push_back(*s);
            }
        }
    }
    return result;
}


Events::Subscription::List Events::GetSubscriptions(const UShort reportType) const
{
    Mutex::ScopedLock lock(&mEventsMutex);
    Subscription::List result;
    Subscription::List::const_iterator s;
    for(s = mSubscriptions.begin();
        s != mSubscriptions.end();
        s++)
    {
        if(reportType == 0 || 
           reportType == (int)s->mpQueryMessage->GetMessageCodeOfResponse())
        {
            result.push_back(*s);
        }
    }
    return result;
}


Events::Subscription::List Events::GetProducedEvents(const UShort reportType) const
{
    Mutex::ScopedLock lock(&mEventsMutex);
    Subscription::List result;
    Subscription::Map::const_iterator s;
    for(s = mEvents.begin();
        s != mEvents.end();
        s++)
    {
        if(((UShort)(s->first & reportType)) == reportType)
        {
            result.push_back(s->second);
        }
    }
    return result;
}


void Events::RegisterCallback(Events::Callback* callback)
{
    Mutex::ScopedLock lock(&mEventCallbackMutex);
    mEventCallbacks.insert(callback);
}


bool Events::CancelSubscription(Subscription& sub, const unsigned int waitTimeMs)
{
    bool result = false;
   
    // Generate a local request ID, then add to
    // list of pending responses.
    Byte requestID = 0;
    std::set<Byte>::iterator riter;
    mRequestMutex.Lock();
    riter = mRequestSet.find(requestID);
    while(riter != mRequestSet.end() )
    {
        requestID++;
        riter = mRequestSet.find(requestID);
    }
    mRequestMutex.Unlock();
    
    CancelEvent request(sub.mProducer, GetComponentID());
    request.SetEventID(sub.mID);
    request.SetRequestID(requestID);
    
    // Setup list of possible responses to command.
    Message::List responses;
    ConfirmEventRequest confirm;
    RejectEventRequest reject;
    responses.push_back(&confirm);
    responses.push_back(&reject);
    
    if(Send(&request, responses, waitTimeMs))
    {
        if(confirm.GetSourceID().IsValid()) // Received confirmation.
        {                   
            result = true;
        }
        else if(mDebugMessagesFlag)
        {
            Mutex::ScopedLock lock(&mDebugMessagesMutex);
            std::cout << "[" << GetServiceID().ToString() << "-" << mComponentID.ToString() << "] - Could not cancel subscription to " << sub.mProducer.ToString() << "\n";
        }
    }
    
    // Remove local request ID from queue.
    mRequestMutex.Lock();
    riter = mRequestSet.find(requestID);
    if(riter != mRequestSet.end() )
    {
        mRequestSet.erase(riter);
    }
    mRequestMutex.Unlock();

    return result;
}


void Events::PeriodicEvent(void *args)
{
    Events* service = (Events *)args;
    Events::Subscription::Map::iterator local;
    
    if(service->mShutdownFlag)
    {
        return;
    }
    
    Subscription::Map eventsCopy;
    // Make a copy of events so that we do not call other services
    // code within our events mutex.  This will avoid deadlock 
    // situations
    service->mEventsMutex.Lock();
    eventsCopy = service->mEvents;
    service->mEventsMutex.Unlock();

    for(local = eventsCopy.begin();
        local != eventsCopy.end() && service->mShutdownFlag == false && service->mPeriodicTimer.IsShuttingDown() == false;
        local++)
    {
        Events::Child* child = dynamic_cast<Events::Child*>(local->second.mpEventService);
        if(local->second.mType == Events::Periodic && child)
        {
            double delay = 1.0/(local->second.mPeriodicRate + CxUtils::CX_EPSILON);
            if(CxUtils::Timer::GetTimeSeconds() - local->second.mTriggerTimeSeconds >= delay)
            {
                // Trigger periodic event. 
                child->GenerateEvent(local->second);
                // Update states in actual map of data within the mutex.
                service->mEventsMutex.Lock();
                Events::Subscription::Map::iterator actual = service->mEvents.find(local->first);
                if(actual != service->mEvents.end() &&
                   actual->second.mSequenceNumber == local->second.mSequenceNumber &&
                   actual->second.mType == local->second.mType)
                {
                    actual->second.mUpdateTimeMs = Time::GetUtcTimeMs();
                    actual->second.mTriggerTimeSeconds = CxUtils::Timer::GetTimeSeconds();
                    actual->second.mSequenceNumber++;
                }
                service->mEventsMutex.Unlock();
            }
        }
    }

    /*
    service->mEventsMutex.Lock();
    
    for(myEvent = service->mEvents.begin();
        myEvent != service->mEvents.end() && service->mShutdownFlag == false && service->mPeriodicTimer.IsShuttingDown() == false;
        myEvent++)
    {
        Events::Child* child = dynamic_cast<Events::Child*>(myEvent->second.mpEventService);
        if(myEvent->second.mType == Events::Periodic && child)
        {
            double delay = 1.0/(myEvent->second.mPeriodicRate + CxUtils::CX_EPSILON);
            if(CxUtils::Timer::GetTimeSeconds() - myEvent->second.mTriggerTimeSeconds >= delay)
            {
                // Trigger periodic event.
                child->GenerateEvent(myEvent->second);
                myEvent->second.mUpdateTimeMs = Time::GetUtcTimeMs();
                myEvent->second.mTriggerTimeSeconds = CxUtils::Timer::GetTimeSeconds();
                myEvent->second.mSequenceNumber++;
            }
        }
    }

    service->mEventsMutex.Unlock();
    */
}




/*  End of File */