consumerImpl.h

/*
 * Orca-Robotics Project: Components for robotics 
 *               http://orca-robotics.sf.net/
 * Copyright (c) 2004-2009 Alex Brooks, Alexei Makarenko, Tobias Kaupp
 *
 * This copy of Orca is licensed to you under the terms described in
 * the LICENSE file included in this distribution.
 *
 */
 
#ifndef ORCAIFACEIMPL_CONSUMER_IMPL_H
#define ORCAIFACEIMPL_CONSUMER_IMPL_H

#include <orcaice/context.h>
#include <orcaice/multiconnectutils.h>
#include <orcaice/iceutils.h>
#include <orcaice/icestormutils.h>
#include <orcaice/icegridutils.h>
#include <gbxsickacfr/gbxiceutilacfr/store.h>
// we only need the definition of Stoppable and checkedSleep() function.
// (we don't need the actual Thread class).
#include <gbxsickacfr/gbxiceutilacfr/threadutils.h>
#include <gbxsickacfr/gbxiceutilacfr/timer.h>

namespace orcaifaceimpl
{

class ConsumerSubscriber
{
public:
    ConsumerSubscriber( const orcaice::Context& context );

    virtual ~ConsumerSubscriber() {};

    virtual void subscribeWithString( const std::string& proxyString )=0;

    void subscribeWithTag( const std::string& interfaceTag );

    virtual void unsubscribe()=0;

    virtual bool subscribeWithString( const std::string& proxyString,  
                          gbxutilacfr::Stoppable* activity, const std::string& subsysName="", 
                          int retryIntervalSec=2, int retryNumber=-1 )=0;

    bool subscribeWithTag( const std::string& interfaceTag, 
                          gbxutilacfr::Stoppable* activity, const std::string& subsysName="", 
                          int retryIntervalSec=2, int retryNumber=-1 );

    bool hasBeenSubscribed()
        { return !topic_.isEmpty(); }

protected:

    // This could be used for re-subscribing.
//     gbxiceutilacfr::Store<std::string> proxyString_;

    //
    // Can connect to Admin interfaces to subscribe their consumers (Status and Tracer)
    //
    gbxiceutilacfr::Store<IceStorm::TopicPrx> topic_;

    orcaice::Context context_;
};

// have to define it as a separate abstract class
// because ConsumerTypeI only needs to know about dataEvent()
// (and does not know about the 2 extra template parameters in
// ConsumerImpl)
template<class ObjectType>
class AbstractConsumer
{
public:
    virtual ~AbstractConsumer() {}

    // Implement this callback in the derived class. 
    virtual void dataEvent( const ObjectType& data )=0;
};

// implements Slice consumer interface
// the end user does not need to know about it.
// redirects incoming data to ConsumerImpl's derivatives.
template<class ConsumerType, class ObjectType>
class ConsumerTypeI : virtual public ConsumerType
{
public:
    ConsumerTypeI( AbstractConsumer<ObjectType> &impl ) :
        impl_(impl)
        {}
    virtual ~ConsumerTypeI() {}

    // implementation of remote call defined in all *Consumer interfaces
    // this implementation redirects to the Impl class
    virtual void setData( const ObjectType& data, const Ice::Current& )
    {
//         context_.tracer().debug( "Received data from provider", 8 );
        impl_.dataEvent( data );
    }

private:
    AbstractConsumer<ObjectType>& impl_;
};

template<class ProviderType, 
         class ConsumerType, 
         class ObjectType,
         class ConsumerTypeIType=ConsumerTypeI<ConsumerType,ObjectType> >
class ConsumerImpl : public ConsumerSubscriber, 
                     public AbstractConsumer<ObjectType>,
                     public IceUtil::Shared
{
typedef typename ProviderType::ProxyType ProviderPrxType;
typedef typename ConsumerType::ProxyType ConsumerPrxType;

protected:
    // these are protected so that it's possible to re-implement initConsumer()

    ConsumerPrxType consumerPrx_;

    Ice::ObjectPtr consumerPtr_;

public:
    ConsumerImpl( const orcaice::Context &context ) :
        ConsumerSubscriber(context)
    {
        consumerPtr_ = new ConsumerTypeIType( *this );
        // this function does not throw, because it never talks to the Registry
        // we do NOT currently convert to a one-way proxy, but maybe we should for efficiency.
        consumerPrx_ = orcaice::createConsumerInterface<ConsumerPrxType>(context_,consumerPtr_);
    }

    virtual ~ConsumerImpl() 
    {
            // unsubscribe from the info provider
            try {
                unsubscribe();
            }
            catch ( const std::exception& e ) {
                std::stringstream ss;
                ss << "failed to unsubscribe in destructor: " << e.what();
                context_.tracer().warning( ss.str() );
            }
            catch ( ... ) {
                context_.tracer().warning( "failed to unsubscribe in destructor." );
            }

            // now, remove our consumer object from Adapter
            if ( consumerPrx_ )
            {
                orcaice::tryRemoveInterfaceWithIdentity( context_, consumerPrx_->ice_getIdentity() );
            }

            // Make sure that the adapter has let go of its smart point -- if not, it means
            // that a 'setData()' call is still executing
            gbxiceutilacfr::Timer timer;
            while ( !context_.adapter()->isDeactivated() &&
                    consumerPtr_->__getRef() > 1 )
            {
                const double maxSec = 10.0; // This is a bit arbitrary...
                if ( timer.elapsedSec() > maxSec )
                {
                    std::cout<<"TRACE(consumerImpl.h): been waiting " << timer.elapsedSec() << "s for setData() call to finish..." << std::endl;                    
                }
                usleep(100000);
            }
        }

    ConsumerPrxType consumerPrx() const { return consumerPrx_; }

    orcaice::Context& context() { return context_; };

    // This is tricky! Can't leave it pure virtual because we unsubscribe and destroy
    // in ConsumerImpl destructor. By that time, the derived class (e.g. StoringConsumer)
    // is already destroyed and we'll get "pure virtual method called".
    //
    // It's tempting to try to make this function private and declare the class which
    // calls it a friend. But this is complicated with templates.
    virtual void dataEvent( const ObjectType& data )
    {
    }

    // no doxytags, these functions are already documented above.

    //
    // Can connect to Admin interfaces to subscribe their consumers (Status and Tracer)
    //
    virtual void subscribeWithString( const std::string& proxyString )
    {
        ProviderPrxType providerPrx;

        std::string staticId = ConsumerType::ice_staticId();
        // Home does not have a consumer
        if ( staticId == "::orca::StatusConsumer" || staticId == "::orca::TracerConsumer" ) {
            orca::FQInterfaceName fqIfaceName = orcaice::toInterfaceName( proxyString );
            orca::FQComponentName fqCompName;
            fqCompName.platform = fqIfaceName.platform;
            fqCompName.component = fqIfaceName.component;
            orcaice::connectToAdminInterface<ProviderType,ProviderPrxType>( context_, providerPrx, fqCompName );
        }
        else {
            orcaice::connectToInterfaceWithString( context_, providerPrx, proxyString );
        }

        IceStorm::TopicPrx topicPrx = providerPrx->subscribe( consumerPrx_ );
        topic_.set( topicPrx );

        std::stringstream ss;
        ss << "Subscribed to topic=" << topicPrx->ice_toString() << " consumer=" << proxyString;
        context_.tracer().debug( ss.str(),6 );
    }

    virtual void unsubscribe()
    {
        if ( !topic_.isEmpty() )
        {
            IceStorm::TopicPrx topicPrx;
            topic_.get( topicPrx );

            topicPrx->unsubscribe( consumerPrx_ );
            std::stringstream ss;
            ss << "Unsubscribed from " << topicPrx->ice_toString();
            context_.tracer().debug( ss.str(),6 );
        }
    }

    virtual bool subscribeWithString( const std::string& proxyString, 
                          gbxutilacfr::Stoppable* activity, const std::string& subsysName="", 
                          int retryIntervalSec=2, int retryNumber=-1 )
    {
        ProviderPrxType providerPrx;
        // multi-try
        orcaice::connectToInterfaceWithString( context_,
                                               providerPrx,
                                               proxyString,
                                               activity,
                                               subsysName,
                                               retryIntervalSec,
                                               retryNumber );

        int count = 0;
        while ( !activity->isStopping() && ( retryNumber<0 || count<retryNumber) )
        {
            try {
                IceStorm::TopicPrx topicPrx = providerPrx->subscribe( consumerPrx_ );
                topic_.set( topicPrx );

                std::stringstream ss;
                ss << "Subscribed to " << proxyString;
                context_.tracer().debug( ss.str(),6 );
                return true;
            }
            catch ( const orca::OrcaException &e )
            {
                std::stringstream ss;
                ss << "Failed to subscribe: " << e.what << std::endl
                   <<"Will retry in "<<retryIntervalSec<<"s.";
                context_.tracer().warning( ss.str() );                
            }
            catch ( const std::exception &e )
            {
                std::stringstream ss;
                ss << "Failed to subscribe: " << e.what() << std::endl
                   <<"Will retry in "<<retryIntervalSec<<"s.";
                context_.tracer().warning( ss.str() );
            }
            catch ( ... )
            {
                std::stringstream ss;
                ss << "Failed to subscribe for unknown reason. "
                   <<"Will retry in "<<retryIntervalSec<<"s.";
                context_.tracer().warning( ss.str() );
            }
            ++count;
            if ( !subsysName.empty() )
                context_.status().heartbeat( subsysName );
            gbxiceutilacfr::checkedSleep( activity, retryIntervalSec*1000 );
            if ( !subsysName.empty() )
                context_.status().heartbeat( subsysName );
        }

        return false;
    }
};

} // namespace

#endif