acsncConsumer.h

Go to the documentation of this file.

#ifndef CONSUMER_H
#define CONSUMER_H

/* @(#) $Id: acsncConsumer.h,v 1.72 2011/11/24 14:18:52 rtobar Exp $
*
*    Consumer Abstract base class for notification channel push structured event
*    consumers.
*    ALMA - Atacama Large Millimiter Array
*
*    (c) Associated Universities Inc., 2002   
*    (c) European Southern Observatory, 2002
*    Copyright by ESO (in the framework of the ALMA collaboration)
*    and Cosylab 2002, All rights reserved
*
*    This library is free software; you can redistribute it and/or
*    modify it under the terms of the GNU Lesser General Public
*    License as published by the Free Software Foundation; either
*    version 2.1 of the License, or (at your option) any later version.
*
*    This library 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
*    Lesser General Public License for more details.
*
*    You should have received a copy of the GNU Lesser General Public
*    License along with this library; if not, write to the Free Software
*    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307  USA
*/

#include "acsncHelper.h"
#include <acstimeProfiler.h>
#include "acsncCDBProperties.h"

#include <list>
namespace nc {

class Consumer :
    protected Helper,
    public POA_CosNotifyComm::StructuredPushConsumer,
    protected virtual PortableServer::RefCountServantBase
{
  public:
    Consumer(const char* channelName, const char* acsNCDomainName = 0);

    Consumer(const char* channelName, 
             CORBA::ORB_ptr orb_mp,
                 const char* acsNCDomainName = 0);
    
    Consumer(const char* channelName, 
             int argc, 
             char *argv[],
                 const char* acsNCDomainName = 0);

    virtual void 
    disconnect();
    
    virtual void 
    push_structured_event(const CosNotification::StructuredEvent &publishedEvent)= 0;
    
    virtual void 
    offer_change(const CosNotification::EventTypeSeq &added,
                 const CosNotification::EventTypeSeq &removed);
    
    virtual void 
    disconnect_structured_push_consumer();

    void
    consumerReady();

    void
    resume();
    
    void
    suspend();

    template <class T> void
    addSubscription()
        {
            //Create a tempory any
            CORBA::Any any;
            //Create a temporary instance of the ALMA event.
            T junk;
            
            //Store the ALMA event in the any to extract the name
            //Probably a nicer way of doing this but it's the best that
            //I can come up with at the moment.
            any <<= junk;

            //Use the version of addSubscription which really subscribes
            //to events from the channel
           
            if (any.type()->kind()!=CORBA::tk_sequence)
                {
                addSubscription(any.type()->name());
                }
            else
                {
                std::string etName= acsnc::SEQUENCE_EVENT_TYPE_PREFIX; //_SequenceOf_
                CORBA::Any a;
                a._tao_set_typecode(any.type()->content_type());
                etName+=a.type()->name();
                addSubscription(etName.c_str());
                }
        }

    template <class T> void
    removeSubscription()
        {
            //Create a tempory any
            CORBA::Any any;
            //Create a temporary instance of the ALMA event.
            T junk;
            
            //Store the ALMA event in the any to extract the name
            //Probably a nicer way of doing this but it's the best that
            //I can come up with at the moment.
            any <<= junk;
            
            //Use the version of removeSubscription which really unsubscribes
            //from an event type.
            if (any.type()->kind()!=CORBA::tk_sequence)
                {
                removeSubscription(any.type()->name());
                }
            else
                {
                std::string etName= acsnc::SEQUENCE_EVENT_TYPE_PREFIX; //_SequenceOf_"
                CORBA::Any a;
                a._tao_set_typecode(any.type()->content_type());
                etName+=a.type()->name();
                removeSubscription(etName.c_str());
                }
        }
    
    int
    addFilter(const char* type_name,
              const char* filter);
    
    bool
    removeFilter(int filter_id);

    virtual void reconnect(::NotifyMonitoringExt::EventChannelFactory *ecf);
    
    void setAntennaName(std::string antennaName);

    void setAutoreconnect(bool autoreconnect);

    bool setEventReceptionTimeout(int eventReceptionTimeout); 

    bool setConnectionCheckerFreq(int connectionCheckerFreq);


  protected:
    void
    init();

    virtual ~Consumer(){}

    void 
    createConsumer();
        


    CosNotifyChannelAdmin::ConsumerAdmin_var consumerAdmin_m;

    CosNotifyChannelAdmin::StructuredProxyPushSupplier_var proxySupplier_m;

    unsigned long long numEvents_m;
    
    CosNotifyComm::StructuredPushConsumer_var reference_m;

    virtual const char* 
    getFilterLanguage();

    CDBProperties::EventHandlerTimeoutMap handlerTimeoutMap_m;

    static double DEFAULT_MAX_PROCESS_TIME;

    Profiler *profiler_mp;

    std::string antennaName;

    bool autoreconnect_m;

  private:

    void
    init(CORBA::ORB_ptr orb);

    void reinit();

    bool shouldReconnect();

    static void* ncChecker(void* arg);

    void checkNotifyChannel();

    void createCheckerThread();

    void destroyCheckerThread();

    CORBA::ORB_ptr orb_mp;

    void operator=(const Consumer&);

    Consumer(const Consumer&);

    void
    addSubscription(const char* type_name);

    void
    removeSubscription(const char* type_name);

    CosNotifyChannelAdmin::AdminID adminid;
    CosNotifyChannelAdmin::ProxyID proxySupplierID;

    static const bool DEFAULT_AUTORECONNECT;
    static const int DEFAULT_EVENT_RECEPTION_TIMEOUT;
    static const int DEFAULT_CONNECTION_CHECKER_FREQ;
    bool stopNCCheckerThread;
    pthread_t ncCheckerThread;
    ACE_Thread_Mutex checkerThMutex_m;
    int eventReceptionTimeout_m;
    int connectionCheckerFreq_m;
};
 }; 

#endif /* CONSUMER_H */