jaus++/2.0/src/jaus/extras/video/videosubscriber.cpp

Go to the documentation of this file.

#include "jaus/extras/video/videosubscriber.h"
#include "jaus/extras/video/querycameracount.h"
#include "jaus/extras/video/reportcameracount.h"
#include "jaus/extras/video/sharedimage.h"

using namespace JAUS;

const std::string VideoSubscriber::Name = "urn:jaus:jss:jpp:extras:VideoSubscriber";


VideoSubscriber::VideoSubscriber() : Management::Child(Service::ID(VideoSubscriber::Name),
                                                       Service::ID(Management::Name))
{
    
}


VideoSubscriber::~VideoSubscriber()
{
    Mutex::ScopedLock smLock(&mSharedImagesMutex);
    std::vector<SharedImage*>::iterator sm;
    for(sm = mSharedImages.begin();
        sm != mSharedImages.end();
        sm++)
    {
        (*sm)->CloseSharedImage();
        delete (*sm);
    }
    mSharedImages.clear();
}


void VideoSubscriber::Shutdown()
{
    Mutex::ScopedLock smLock(&mSharedImagesMutex);
    std::vector<SharedImage*>::iterator sm;
    for(sm = mSharedImages.begin();
        sm != mSharedImages.end();
        sm++)
    {
        (*sm)->CloseSharedImage();
        delete (*sm);
    }
    mSharedImages.clear();
    Management::Child::Shutdown();
}


bool VideoSubscriber::CreateVideoSubscription(const Address& id, 
                                              const Byte cameraID,
                                              const unsigned int waitTimeMs)
{
    std::vector<SharedImage*>::iterator sm;
    Mutex::ScopedLock smLock(&mSharedImagesMutex);
    for(sm = mSharedImages.begin();
        sm != mSharedImages.end();
        sm++)
    {
        if((*sm)->GetSourceID() == id && (*sm)->GetCameraID() == cameraID)
        {
            return true;
        }
    }
    
    // Try create a shared image connection first.
    SharedImage* simage = new SharedImage();
    if(simage->OpenSharedImage(id, cameraID))
    {
        simage->RegisterCallback(SharedImageCallback, this);
        mSharedImages.push_back(simage);
        return true;
    }
    delete simage;

    QueryImage query;
    query.SetFormat(Image::JPEG);
    query.SetCameraID(cameraID);
    return EventsService()->RequestEveryChangeEvent(id, &query, waitTimeMs);
}


bool VideoSubscriber::GetCameraCount(const Address& id, 
                                     unsigned int& count,
                                     const unsigned int waitTimeMs) const
{
    QueryCameraCount query(id, GetComponentID());
    ReportCameraCount report;
    count = 0;
    if(Send(&query, &report, waitTimeMs))
    {
        count = (unsigned int)report.GetCameraList()->size();
        return true;
    }

    return false;
}


bool VideoSubscriber::GetCameraList(const Address& id, 
                                    std::vector<Byte>& list,
                                    const unsigned int waitTimeMs) const
{
    QueryCameraCount query(id, GetComponentID());
    ReportCameraCount report;
    list.clear();
    if(Send(&query, &report, waitTimeMs))
    {
        list = *report.GetCameraList();
        return list.size() > 0 ? true : false;
    }

    return false;
}


bool VideoSubscriber::HaveVideoSubscription(const Address& id,
                                            const int cameraID) const
{
    bool result = false;

    std::vector<SharedImage*>::const_iterator sm;
    Mutex::ScopedLock smLock(&mSharedImagesMutex);
    for(sm = mSharedImages.begin();
        sm != mSharedImages.end();
        sm++)
    {
        if((*sm)->GetSourceID() == id && (*sm)->GetCameraID() == cameraID)
        {
            return true;
        }
    }

    Events::Subscription::List list = GetComponent()->EventsService()->GetSubscriptions(id, REPORT_IMAGE);
    Events::Subscription::List::iterator s;
    for(s = list.begin();
        s != list.end();
        s++)
    {
        if(cameraID < 0)
        {
            result = true;
            break;
        }
        else
        {
            QueryImage* query = dynamic_cast<QueryImage *>(s->mpQueryMessage);
            if(query && query->GetCameraID() == (Byte)cameraID)
            {
                result = true;
                break;
            }
        }
    }
    return result;
}


bool VideoSubscriber::CancelVideoSubscription(const Address& id,
                                              const int cameraID)
{
    bool result = false;

    std::vector<SharedImage*>::iterator sm;
    Mutex::ScopedLock smLock(&mSharedImagesMutex);
    sm = mSharedImages.begin();
    while(sm != mSharedImages.end())
    {
        if(id.IsValid() == false ||
           ((*sm)->GetSourceID() == id && (*sm)->GetCameraID() == cameraID) ||
           ((*sm)->GetSourceID() == id && cameraID < 0))
        {
            (*sm)->CloseSharedImage();
            delete (*sm);
            mSharedImages.erase(sm);
            sm = mSharedImages.begin();
            result = true;
        }
        else
        {
            sm++;
        }
    }

    Events::Subscription::List list = GetComponent()->EventsService()->GetSubscriptions(id, REPORT_IMAGE);
    Events::Subscription::List::iterator s;
    for(s = list.begin();
        s != list.end();
        s++)
    {
        if(cameraID < 0)
        {
            if(GetComponent()->EventsService()->CancelSubscription(s->mProducer,
                                                                   REPORT_IMAGE,
                                                                   s->mID,
                                                                   Service::DefaultWaitMs))
            {
                result = true;
            }
        }
        else
        {
            QueryImage* query = dynamic_cast<QueryImage *>(s->mpQueryMessage);
            if(query && query->GetCameraID() == (Byte)cameraID)
            {
                if(GetComponent()->EventsService()->CancelSubscription(s->mProducer,
                                                                       REPORT_IMAGE,
                                                                       s->mID,
                                                                       Service::DefaultWaitMs))
                {
                    result = true;
                }
                break;
            }
        }
    }
    return result;
}


void VideoSubscriber::RegisterCallback(Callback* callback, const bool rawImage, const bool add)
{
    Mutex::ScopedLock lock(&mVideoCallbacksMutex);
    if(add)
    {
        if(rawImage)
        {
            mRawCallbacks.insert(callback);
        }
        else
        {
            mCompressedCallbacks.insert(callback);
        }
    }
    else
    {
        Callback::Set::iterator cb;
        if(rawImage)
        {
            cb = mRawCallbacks.find(callback);
            if(cb != mRawCallbacks.end())
            {
                mRawCallbacks.erase(cb);
            }
        }
        else
        {
            cb = mCompressedCallbacks.find(callback);
            if(cb != mCompressedCallbacks.end())
            {
                mCompressedCallbacks.erase(cb);
            }
        }

    }
}


void VideoSubscriber::Receive(const Message* message)
{
    switch(message->GetMessageCode())
    {
    case REPORT_IMAGE:
        {
            const ReportImage* report = dynamic_cast<const ReportImage*>(message);
            if(report)
            {
                Mutex::ScopedLock lock(&mVideoCallbacksMutex);

                Callback::Set::iterator cb;
                for(cb = mCompressedCallbacks.begin();
                    cb != mCompressedCallbacks.end();
                    cb++)
                {
                    (*cb)->ProcessCompressedVideo(report->GetSourceID(),
                                                  report->GetCameraID(),
                                                  report->GetFormat(),
                                                  report->GetImage()->Ptr(),
                                                  report->GetImage()->Length(),
                                                  report->GetFrameNumber());
                }
                // Don't decompress the image data if there are
                // no callbacks to use it.
                if(mRawCallbacks.size() > 0)
                {
                    Image raw;
                    // Decompress and trigger callbacks.
                    if(raw.Decompress(report->GetImage()->Ptr(),
                                      report->GetImage()->Length(),
                                      report->GetFormat()))
                    {
                        for(cb = mRawCallbacks.begin();
                            cb != mRawCallbacks.end();
                            cb++)
                        {
                            (*cb)->ProcessRawImage(report->GetSourceID(),
                                                   report->GetCameraID(),
                                                   raw,
                                                   report->GetFrameNumber());
                        }
                    }
                }
            }
        }
        break;
    default:
        break;
    }
}


Message* VideoSubscriber::CreateMessage(const UShort messageCode) const
{
    Message* message;
    switch(messageCode)
    {
    case QUERY_IMAGE:
        message = new QueryImage();
        break;
    case QUERY_CAMERA_COUNT:
        message = new JAUS::QueryCameraCount();
        break;
    case REPORT_IMAGE:
        message = new ReportImage();
        break;
    case REPORT_CAMERA_COUNT:
        message = new JAUS::ReportCameraCount();
        break;
    default:
        message = NULL;
        break;
    }
    return message;
}


void VideoSubscriber::SharedImageCallback(const Address& source,
                                          const Byte cameraID,
                                          const Image& img,
                                          const unsigned int frameNumber,
                                          void* fargs)
{
    VideoSubscriber* subscriber = (VideoSubscriber*)fargs;
    Mutex::ScopedLock lock(&subscriber->mVideoCallbacksMutex);

    Callback::Set::iterator cb;
    for(cb = subscriber->mCompressedCallbacks.begin();
        cb != subscriber->mCompressedCallbacks.end();
        cb++)
    {
        (*cb)->ProcessCompressedVideo(source,
                                      cameraID,
                                      Image::RAW,
                                      img.mpImage,
                                      img.mDataSize,
                                      frameNumber);
    }
    // Don't decompress the image data if there are
    // no callbacks to use it.
    if(subscriber->mRawCallbacks.size() > 0)
    {
        for(cb = subscriber->mRawCallbacks.begin();
            cb != subscriber->mRawCallbacks.end();
            cb++)
        {
            (*cb)->ProcessRawImage(source,
                                   cameraID,
                                   img,
                                   frameNumber);
        }
    }
}


/* End of File */