OSubject.cc

Go to the documentation of this file.

/*
 * This file is part of openSDK.
 *
 * Copyright (C) 2006-2007 openSDK team
 *
 * openSDK is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; version 2.
 *
 * openSDK 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 General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with openSDK; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
 *
 *     $Id: OSubject.cc,v 1.22 2007/03/25 12:58:16 nuno-lopes Exp $
 */

#include <iostream>
#include <OPENR/OSubject.h>
#include <opensdkAPI.h>
#include <ModuleData.h>

using namespace std;

#define FETCH_ITERATOR(ret) \
        ObserverConstIterator it = findObserver(id); \
        if (it == end()) { return ret; }


ObserverIterator OSubject::findObserver(const ObserverID& id)
{
        for (ObserverIterator it = observers_.begin(); it != observers_.end(); ++it) {
                if (it->GetObserverID() == id) {
                        return it;
                }
        }

        return observers_.end(); // invalid iterator
}

ObserverConstIterator OSubject::findObserver(const ObserverID& id) const
{
        for (ObserverConstIterator it = begin(); it != end(); ++it) {
                if (it->GetObserverID() == id) {
                        return it;
                }
        }

        return end(); // invalid iterator
}

OSubject::OSubject(void) : maxBufferSize_(DEFAULT_MAX_BUFFER_SIZE), notifyUnitSize_(1)
{
        //TODO
}

OSubject::~OSubject()
{
        for (ObserverIterator it = observers_.begin(); it != observers_.end(); ++it) {
                it->DestroyBuffer();
                it->Delete();
                observers_.erase(it);
        }
}

OStatus OSubject::SetReadyEntry(const OServiceEntry& entry)
{
        myID_.Set(entry.GetOID(), entry.GetSelector());

        for (std::list< std::pair<int,int> >::iterator it = MOD_DATA(ConnectHandlers).begin(); it != MOD_DATA(ConnectHandlers).end(); ++it) {
                if (it->second == 0) {
                        it->second = entry.GetSelector();
                        return oSUCCESS;
                }
        }
        return oFAIL;

}

OStatus OSubject::SetBufferSize(size_t size)
{
        maxBufferSize_ = size;
        return oSUCCESS;
}

OStatus OSubject::SetNotifyUnitSize(size_t size)
{
        if (size > GetBufferSize() || size > MAX_NUM_DATA) {
                return oFAIL;
        }

        notifyUnitSize_ = size;
        return oSUCCESS;
}

OStatus OSubject::SetData(const OObserverInfo& info, RCRegion* r)
{
        r->AddReference();

        // queue is full, overwrite the oldest
        if (info.impl_->d_skip) {
                info.impl_->queue[info.impl_->d_min]->RemoveReference();
                info.impl_->d_min = (info.impl_->d_min + 1) % info.impl_->q_size;
                --(info.impl_->n_data);
        }

        info.impl_->queue[info.impl_->d_max] = r;

        info.impl_->d_max = (info.impl_->d_max + 1) % info.impl_->q_size;
        if (info.impl_->d_max == info.impl_->d_min) {
                info.impl_->d_skip = 1; // queue became full
        }

        ++(info.impl_->n_data);

        return oSUCCESS;
}

OStatus OSubject::SetData(const ObserverID& id, RCRegion* r)
{
        FETCH_ITERATOR(oFAIL)
        return SetData(*it, r);
}

OStatus OSubject::SetData(RCRegion* r)
{
        for (ObserverConstIterator it = begin(); it != end(); ++it) {
                OStatus tmp = SetData(*it, r);
                if (tmp != oSUCCESS) {
                        return tmp;
                }
        }

        return oSUCCESS;
}

OStatus OSubject::SetData(const ObserverInfo& info, const void* buf, size_t size)
{
        RCRegion* rgn = new RCRegion(size);
        memcpy(rgn->Base(), buf, size);
        OStatus tmp = SetData(info, rgn);

        // remove our reference since each observer will increment the reference counter
        rgn->RemoveReference();

        return tmp;
}

OStatus OSubject::SetData(const ObserverID& id, const void* buf, size_t size)
{
        FETCH_ITERATOR(oFAIL)
        return SetData(*it, buf, size);
}

OStatus OSubject::SetData(const void* buf, size_t size)
{
        OStatus tmp = oSUCCESS;
        RCRegion* rgn = new RCRegion(size);
        memcpy(rgn->Base(), buf, size);

        for (ObserverConstIterator it = begin(); it != end(); ++it) {
                tmp = SetData(*it, rgn);
                if (tmp != oSUCCESS) {
                        break;
                }
        }

        // remove our reference since each observer will increment the reference counter
        rgn->RemoveReference();

        return tmp;
}

OStatus OSubject::NotifyObserver(const OObserverInfo& info)
{
        if (IsReady(info)) { //ASSERT-READY state
                ONotifyMessage *msg = new ONotifyMessage();
                msg->subjectID      = GetID();
                msg->numOfData      = 0;
                msg->numOfNotify    = ++(info.impl_->n_notify);

                for (unsigned int i = 0; info.impl_->n_data; ++i) {
                        msg->data[i].srcRCR = info.impl_->queue[info.impl_->d_min];
                        info.impl_->d_min = (info.impl_->d_min + 1) % info.impl_->q_size;
                        --(info.impl_->n_data);
                        msg->numOfData++;
                }

                info.impl_->d_skip = 0;

                return _sendMessage(info.GetObserverID(), msg);

        } else if (ReadyStatus(info) == -1) { // DEASSERT-READY state
                ClearBuffer(info);
        }

        return oSUCCESS;
}

OStatus OSubject::NotifyObserver(const ObserverID& id)
{
        FETCH_ITERATOR(oFAIL)
        return NotifyObserver(*it);
}

OStatus OSubject::NotifyObservers(void)
{
        for (ObserverConstIterator it = begin(); it != end(); ++it) {
                OStatus tmp = NotifyObserver(*it);
                if (tmp != oSUCCESS) {
                        return tmp;
                }
        }

        return oSUCCESS;
}

size_t OSubject::RemainBuffer(const OObserverInfo& info) const
{
        return (info.impl_->q_size - info.impl_->n_data);
}

size_t OSubject::RemainBuffer(const ObserverID& id) const
{
        FETCH_ITERATOR(0)
        return RemainBuffer(*it);
}

size_t OSubject::RemainBuffer(void) const
{
        size_t min = DEFAULT_MAX_BUFFER_SIZE;
        for (ObserverConstIterator it = begin(); it != end(); ++it) {
                size_t tmp = RemainBuffer(*it);
                if (tmp < min) {
                        min = tmp;
                }
        }

        return min;
}


OStatus OSubject::ClearBuffer(const OObserverInfo& info)
{
        while (info.impl_->n_data) {
                info.impl_->queue[info.impl_->d_min]->RemoveReference();
                info.impl_->d_min = (info.impl_->d_min + 1) % info.impl_->q_size;
                --(info.impl_->n_data);
        }

        info.impl_->d_min  = 0;
        info.impl_->d_max  = 0;
        info.impl_->d_skip = 0;
        return oSUCCESS;
}

OStatus OSubject::ClearBuffer(const ObserverID& id)
{
        FETCH_ITERATOR(oFAIL)
        return ClearBuffer(*it);
}

OStatus OSubject::ClearBuffer()
{
        for (ObserverConstIterator it = begin(); it != end(); ++it) {
                OStatus tmp = ClearBuffer(*it);
                if (tmp != oSUCCESS) {
                        return tmp;
                }
        }

        return oSUCCESS;
}

OStatus OSubject::ControlHandler(const OControlMessage& msg, OStatus status)
{
        if (status != oSUCCESS) {
                return status;
        }

        if (msg.msgType != omsgCONTROL && msg.msgType != omsgCONTROL_V21) {
                OSYSLOG1((osyslogERROR, "Unknown message type received in ControlHandler: %d (Entry: %d/%d)", msg.msgType, (size_t)myID_.GetOID().GetAddress(), (int)myID_.GetSelector()));
                return oFAIL;
        }

        switch (msg.command) {
                case objcommCMD_ADDOBSERVER:
                case objcommCMD_ADDOBSERVER_SELF:
                case objcommCMD_ADDOBSERVER_V22:
                        {
                        OObserverInfo info;
                        info.impl_ = new _OObserverInfo(msg.observerID, 0, 0, 0, objcommVERSION22);
                        info.CreateBuffer(GetBufferSize());
                        observers_.push_back(info);
                        }
                        break;

                case objcommCMD_REMOVEOBSERVER:
                case objcommCMD_REMOVEOBSERVER_SELF:
                case objcommCMD_REMOVEOBSERVER_V22:
                        for (ObserverIterator it = observers_.begin(); it != observers_.end(); ++it) {
                                if (it->GetObserverID().GetOID() == msg.observerID.GetOID() &&
                                it->GetObserverID().GetSelector() == msg.observerID.GetSelector()) {
                                        it->DestroyBuffer();
                                        it->Delete();
                                        observers_.erase(it);
                                        break;
                                }
                        }
                        break;

                default:
                        OSYSLOG1((osyslogERROR, "Unknown message command received in ControlHandler: %d", msg.command));
                        return oFAIL;
        }

        return oSUCCESS;
}

OStatus OSubject::ReadyHandler(const OReadyMessage& msg)
{
        if (msg.msgType != omsgREADY && msg.msgType != omsgREADY_V21 && msg.msgType != omsgREADY_V22) {
                OSYSLOG1((osyslogERROR, "Unknown message type received in ReadyHandler: %d (Entry: %d/%d)", msg.msgType, (size_t)myID_.GetOID().GetAddress(), (int)myID_.GetSelector()));
                return oFAIL;
        }

        switch (msg.command) {
                case objcommCMD_ASSERT_READY:
                        for (ObserverIterator it = observers_.begin(); it != observers_.end(); ++it) {
                                if (it->GetObserverID().GetOID() == msg.observerID.GetOID() &&
                                it->GetObserverID().GetSelector() == msg.observerID.GetSelector()) {
                                        it->impl_->ready_flag = 1;
                                        break;
                                }
                        }
                        break;

                case objcommCMD_DEASSERT_READY:
                        for (ObserverIterator it = observers_.begin(); it != observers_.end(); ++it) {
                                if (it->GetObserverID().GetOID() == msg.observerID.GetOID() &&
                                it->GetObserverID().GetSelector() == msg.observerID.GetSelector()) {
                                        it->impl_->ready_flag = -1;
                                        break;
                                }
                        }
                        break;

                default:
                        OSYSLOG1((osyslogERROR, "Unknown message command received in ReadyHandler: %d", msg.command));
                        return oFAIL;
        }

        return oSUCCESS;
}

int OSubject::NumberOfObservers(void) const
{
        return observers_.size();
}

int OSubject::ReadyStatus(const ObserverID& id) const
{
        FETCH_ITERATOR(0)
        return ReadyStatus(*it);
}

int OSubject::IsAllReady(void) const
{
        for (ObserverConstIterator it = begin(); it != end(); ++it) {
                if (ReadyStatus(*it) == 0) { // sent state
                        return 0;
                }
        }

        return 1;
}

int OSubject::IsAnyReady(void) const
{
        for (ObserverConstIterator it = begin(); it != end(); ++it) {
                if (IsReady(*it)) { // assert-ready
                        return 1;
                }
        }

        return 0;
}

int OSubject::IsReady(const ObserverID& id) const
{
        FETCH_ITERATOR(0)
        return IsReady(*it);
}

---