consumerImpl.h

/*
 * Orca-Robotics Project: Components for robotics 
 *               http://orca-robotics.sf.net/
 * Copyright (c) 2004-2008 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>
#include <gbxsickacfr/gbxiceutilacfr/thread.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,  
                          gbxiceutilacfr::Thread*  thread, const std::string& subsysName="", 
                          int retryInterval=2, int retryNumber=-1 )=0;

    bool subscribeWithTag( const std::string& interfaceTag, 
                          gbxiceutilacfr::Thread*  thread, const std::string& subsysName="", 
                          int retryInterval=2, int retryNumber=-1 );
protected:

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

    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:
    // 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 ProviderPrxType, class ConsumerType, class ConsumerPrxType, class ObjectType>
class ConsumerImpl : public ConsumerSubscriber, 
                     public AbstractConsumer<ObjectType>,
                     public IceUtil::Shared
{

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 ConsumerTypeI<ConsumerType,ObjectType>( *this );
        // this function does not throw, because it never talks to the Registry
        consumerPrx_ = orcaice::createConsumerInterface<ConsumerPrxType>(context_,consumerPtr_);
    }

    virtual ~ConsumerImpl() 
    {
//         context_.tracer().debug( "ConsumerImpl::~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 destroy our consumer object
        if ( !consumerPrx_ )
            return;

        orcaice::tryRemoveInterfaceWithIdentity( context_, consumerPrx_->ice_getIdentity() );
    }

    ConsumerPrxType consumerPrx() const { return consumerPrx_; }

    // This is tricky! Can't leave it pure virtual because we unsubscribe and detsroy
    // in ConsumerImpl destructor. By that time, the derived class (e.g. StoringConsumer)
    // is already destroyed and we'll get "pure virtual method called".
    virtual void dataEvent( const ObjectType& data )
    {
    }

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

    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 " << proxyString;
        context_.tracer().debug( ss.str() );
    }

    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() );
        }
    }

    virtual bool subscribeWithString( const std::string& proxyString, 
                          gbxiceutilacfr::Thread*  thread, const std::string& subsysName="", 
                          int retryInterval=2, int retryNumber=-1 )
    {
        ProviderPrxType providerPrx;
        // multi-try
        orcaice::connectToInterfaceWithString( context_,
                                               providerPrx,
                                               proxyString,
                                               thread,
                                               subsysName,
                                               retryInterval,
                                               retryNumber );

        int count = 0;
        while ( !thread->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() );
                return true;
            }
            catch ( const orca::OrcaException &e )
            {
                std::stringstream ss;
                ss << "Failed to subscribe: " << e.what << std::endl
                   <<"Will retry in "<<retryInterval<<"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 "<<retryInterval<<"s.";
                context_.tracer().warning( ss.str() );
            }
            catch ( ... )
            {
                std::stringstream ss;
                ss << "Failed to subscribe for unknown reason. "
                   <<"Will retry in "<<retryInterval<<"s.";
                context_.tracer().warning( ss.str() );
            }
            ++count;
            if ( !subsysName.empty() )
                context_.status().heartbeat( subsysName );
            IceUtil::ThreadControl::sleep(IceUtil::Time::seconds(retryInterval));
            if ( !subsysName.empty() )
                context_.status().heartbeat( subsysName );
        }

        return false;
    }
};

} // namespace

#endif