ChannelRef.cpp

Go to the documentation of this file.

/*
* This file is part of ArmarX.
*
* Copyright (C) 2011-2016, High Performance Humanoid Technologies (H2T), Karlsruhe Institute of Technology (KIT), all rights reserved.
*
* ArmarX is free software; you can redistribute it and/or modify
* it under the terms of the GNU General Public License version 2 as
* published by the Free Software Foundation.
*
* ArmarX 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 this program. If not, see <http://www.gnu.org/licenses/>.
*
* @package    ArmarX::Core
* @author     Kai Welke (welke@kit.edu)
* @date       2012 Kai Welke
* @copyright  http://www.gnu.org/licenses/gpl-2.0.txt
*             GNU General Public License
*/
 
#include <ArmarXCore/observers/variant/ChannelRef.h>
#include <ArmarXCore/observers/Observer.h>
 
#include <ArmarXCore/observers/exceptions/local/InvalidDataFieldException.h>
#include <ArmarXCore/core/exceptions/local/ProxyNotInitializedException.h>
#include <ArmarXCore/observers/AbstractObjectSerializer.h>
#include <ArmarXCore/statechart/Exception.h>
#include <ArmarXCore/observers/exceptions/user/InvalidDatafieldException.h>
 
#include <Ice/ObjectAdapter.h>
#include <Ice/ValueFactory.h>
 
#include <algorithm>
 
template class ::IceInternal::Handle<::armarx::ChannelRef>;
 
namespace armarx
{
    using namespace armarx::exceptions::local;
 
    // *******************************************************
    // Construction / destruction
    // *******************************************************
    ChannelRef::ChannelRef(Observer* observer, std::string channelName) :
        validationTimeout(4000),
        waitIntervallMs(10)
    {
        if (!observer)
        {
            ARMARX_FATAL_S << "observer must not be NULL";
            throw exceptions::local::eNullPointerException("observer must not be NULL");
        }
 
        this->observerProxy = ObserverInterfacePrx::uncheckedCast(observer->getProxy());
        this->communicator = observerProxy->ice_getCommunicator();
        this->observerName = observer->getName();
        this->channelName = channelName;
 
        // initialize data fields and initialized state
        initializeDataFields(observer);
    }
 
    ChannelRef::ChannelRef(ObserverInterfacePrx observerPrx, std::string channelName) :
        validationTimeout(4000),
        waitIntervallMs(10)
    {
        if (!observerPrx)
        {
            throw exceptions::local::eNullPointerException("observerPrx must not be NULL");
        }
 
        this->observerProxy = observerPrx;
        this->observerName = observerPrx->ice_getIdentity().name;
        this->channelName = channelName;
        this->communicator = observerProxy->ice_getCommunicator();
 
        // initialize data fields and initialized state
        initializeDataFields();
    }
 
    // *******************************************************
    // Datafield access
    // *******************************************************
    DataFieldIdentifierPtr ChannelRef::getDataFieldIdentifier(const std::string& datafieldName)
    {
        if (!assureProxy())
        {
            throw ProxyNotInitializedException(observerName);
        }
 
        if (std::find(datafieldNames.begin(), datafieldNames.end(), datafieldName) == datafieldNames.end())
        {
            refetchChannel();
 
            if (std::find(datafieldNames.begin(), datafieldNames.end(), datafieldName) == datafieldNames.end())
            {
                throw exceptions::user::InvalidDataFieldException(channelName, datafieldName);
            }
        }
 
        return new DataFieldIdentifier(observerName, channelName, datafieldName);
    }
 
    TimedVariantPtr ChannelRef::getDataField(const std::string& datafieldName)
    {
        if (!assureProxy())
        {
            throw ProxyNotInitializedException(observerName);
        }
        TimedVariantPtr result;
        try
        {
            result = TimedVariantPtr::dynamicCast(observerProxy->getDatafieldByName(getChannelName(), datafieldName));
        }
        catch (...)
        {
            if (communicator)
            {
                observerProxy = ObserverInterfacePrx::uncheckedCast(communicator->stringToProxy(observerName));
                result = TimedVariantPtr::dynamicCast(observerProxy->getDatafieldByName(getChannelName(), datafieldName));
            }
            else
            {
                throw;
            }
        }
        return result;
    }
 
    // *******************************************************
    // Properties
    // *******************************************************
    const ObserverInterfacePrx& ChannelRef::getObserverProxy()
    {
        if (!assureProxy())
        {
            throw ProxyNotInitializedException(observerName);
        }
 
        return observerProxy;
    }
 
    const std::string& ChannelRef::getObserverName() const
    {
        return observerName;
    }
 
    const std::string& ChannelRef::getChannelName() const
    {
        return channelName;
    }
 
    const Ice::StringSeq& ChannelRef::getDataFieldNames() const
    {
        return datafieldNames;
    }
 
    bool ChannelRef::hasDatafield(const std::string& datafieldName) const
    {
        for (unsigned int i = 0; i < datafieldNames.size(); ++i)
        {
            if (datafieldName == datafieldNames.at(i))
            {
                return true;
            }
        }
 
        return false;
    }
 
    bool ChannelRef::getInitialized()
    {
        // once all datafields are initialized -> channel initialized
        if (initialized)
        {
            return true;
        }
 
        // if proxy not there -> not initialized
        if (!assureProxy())
        {
            return false;
        }
 
        ARMARX_INFO_S << "ChannelRef:" << getObserverName() << "." << getChannelName() << ": Initialized as retrieved from proxy : " << initialized;
 
        return initialized;
    }
 
    void ChannelRef::refetchChannel()
    {
        initializeDataFields();
    }
 
 
    // *******************************************************
    // Inherited from VariantDataClass
    // *******************************************************
    VariantDataClassPtr ChannelRef::clone(const Ice::Current& c) const
    {
        ChannelRefPtr cln = new ChannelRef(*this);
 
        cln->initialized = this->initialized;
 
        cln->observerName = this->observerName;
        cln->channelName = this->channelName;
        cln->datafieldNames = this->datafieldNames;
 
        cln->observerProxy = this->observerProxy;
 
        return cln;
    }
 
    std::string ChannelRef::output(const Ice::Current& c) const
    {
        std::stringstream s;
        s << "Reference to channel " << channelName << " on observer " << observerName;
 
        return s.str();
    }
 
    VariantTypeId ChannelRef::getType(const Ice::Current& c) const
    {
        return VariantType::ChannelRef;
    }
 
    bool ChannelRef::validate(const Ice::Current& c)
    {
        int validateWaitIntervallMs = waitIntervallMs;
        IceUtil::Time waitStartTime = IceUtil::Time::now();
 
        IceUtil::Time timeout = IceUtil::Time::milliSeconds(validationTimeout);
 
        IceUtil::Time lastOutput = IceUtil::Time::milliSeconds(0);
 
        while (IceUtil::Time(waitStartTime + timeout - IceUtil::Time::now()).toMilliSeconds() > 0)
        {
            if (getInitialized())
            {
                return true;
            }
 
            if (IceUtil::Time(IceUtil::Time::now() - lastOutput).toSeconds() >= 1) // output only every second
            {
                ARMARX_VERBOSE_S << "Waiting for ChannelRef:  " << getObserverName() << "." << getChannelName() << flush;
                lastOutput = IceUtil::Time::now();
            }
 
            usleep(validateWaitIntervallMs * 1000);
 
            if (validateWaitIntervallMs < validationTimeout * 0.5f) // increase wait time for lower cpu usage
            {
                validateWaitIntervallMs *= 2;
            }
        }
 
        ARMARX_WARNING_S << "Could not validate ChannelRef:  " << getObserverName() << "." << getChannelName() << flush;
        return false;
 
    }
 
    // private
    bool ChannelRef::assureProxy()
    {
        if (!observerProxy)
        {
            try
            {
                observerProxy = ObserverInterfacePrx::checkedCast(communicator->stringToProxy(observerName));
            }
            catch (...)
            {
            }
        }
 
        if (observerProxy && !initialized)
        {
            initializeDataFields();
        }
 
        return (observerProxy != 0);
    }
 
    void ChannelRef::initializeDataFields(Observer* obs)
    {
        ChannelRegistryEntry channel = obs ?
                                       obs->getChannel(getChannelName()) :
                                       observerProxy->getChannel(getChannelName());
 
        DataFieldRegistry::const_iterator iterData = channel.dataFields.begin();
        datafieldNames.clear();
 
        while (iterData != channel.dataFields.end())
        {
            datafieldNames.push_back(iterData->second.identifier->datafieldName);
            iterData++;
        }
 
        initialized = channel.initialized;
    }
 
    void ChannelRef::ice_postUnmarshal()
    {
        if (observerProxy)
        {
            communicator = observerProxy->ice_getCommunicator();
        }
    }
 
    void ChannelRef::serialize(const ObjectSerializerBasePtr& serializer, const ::Ice::Current&) const
    {
        AbstractObjectSerializerPtr obj = AbstractObjectSerializerPtr::dynamicCast(serializer);
 
        obj->setString("observerName", observerName);
        obj->setString("channelName", channelName);
    }
 
    void ChannelRef::deserialize(const ObjectSerializerBasePtr& serializer, const ::Ice::Current& c)
    {
        AbstractObjectSerializerPtr obj = AbstractObjectSerializerPtr::dynamicCast(serializer);
 
        observerName = obj->getString("observerName");
        channelName = obj->getString("channelName");
 
        if (!c.adapter)
        {
            throw LocalException("The adapter in the Ice::current object must be set to load a ChannelRef from XML");
        }
 
        communicator = c.adapter->getCommunicator();
    }
}