bulkDataReceiver.h

Go to the documentation of this file.

#ifndef _BULKDATA_RECEIVER_H
#define _BULKDATA_RECEIVER_H
/*******************************************************************************
 *    ALMA - Atacama Large Millimiter Array
 *    (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
 *
 *
 * "@(#)"
 *
 * who       when      what
 * --------  --------  ----------------------------------------------
 * oat       27/01/05  created 
 */

#include <vector>

#include "orbsvcs/AV/AVStreams_i.h"
#include "orbsvcs/AV/Endpoint_Strategy.h"
#include "orbsvcs/AV/Protocol_Factory.h"
#include "orbsvcs/AV/Flows_T.h"
#include "orbsvcs/AV/Transport.h"
#include "orbsvcs/AV/Policy.h"

#include "ACSBulkDataError.h"
#include "bulkDataFlowConsumer.h"

#include "bulkDataC.h"

namespace AcsBulkdata
{  
    template<class TReceiverCallback>
    class BulkDataReceiver
    {
      public:
      
        BulkDataReceiver();
      
        virtual ~BulkDataReceiver();
      
        void initialize();

        void createSingleFlow();

        void createMultipleFlows(const char *fepsConfig);

        bulkdata::BulkDataReceiverConfig * getReceiverConfig();

        void getFlowCallback(ACE_CString &flowName, TReceiverCallback *&cb_p);

        void getFlowCallback(CORBA::ULong flowNumber, TReceiverCallback *&cb_p);

        void closeReceiver();

        std::vector<std::string> getFlowNames();

        void setReceiverName(ACE_CString recvName);

        void subscribeNotification(ACS::CBvoid_ptr notifCb);
     /*
     *  @throw ACSBulkDataError::AVNotificationMechanismErrorExImpl
     */
        void notifySender(const ACSErr::Completion& comp);


        // method called by the distributor object to set the reactor handle
        // on its internal receiver
        void addHandle(ACE_CString flowName, ACE_HANDLE handle)
            {
                handleMap_m.rebind(flowName,handle);    
            }

        void setCbTimeout(const char * cbTimeout);


      private:

        typedef ACE_Hash_Map_Manager<ACE_CString, BulkDataFlowConsumer<TReceiverCallback> *, ACE_Null_Mutex> FepObjects;
        typedef ACE_Hash_Map_Iterator<ACE_CString, BulkDataFlowConsumer<TReceiverCallback> *, ACE_Null_Mutex>  FepObjectsIterator;

        typedef ACE_Hash_Map_Manager<ACE_CString, ACE_HANDLE, ACE_Null_Mutex> HandleMap;
        typedef ACE_Hash_Map_Iterator<ACE_CString, ACE_HANDLE, ACE_Null_Mutex>  HandleMapIterator;

        void initPartB();

        AVStreams::StreamEndPoint_B_ptr createSepB();

        AVStreams::FlowConsumer_ptr createFepConsumerB(ACE_CString &flowName, AVStreams::protocolSpec protocols, ACE_CString &format);


        void addFepToSep(AVStreams::StreamEndPoint_B_ptr locSepB_p,AVStreams::FlowConsumer_ptr locFepB_p);

        AVStreams::StreamEndPoint_B_ptr getStreamEndPointB();

        AVStreams::flowSpec * getFepsConfig();

        void deleteFepsB();


        void deleteSepB();

        void deleteAcceptor();

        void deleteHandler();

        void closeSocket();

        const char * createFlowSpec(ACE_CString &flowName,
                                    ACE_CString &fepProtocol);


      public:
        bulkdata::Connection checkFlowCallbacks();

        bulkdata::Connection getSenderConnectionState()
            {
                return recvConfig_p->connectionState;
            }

      private:

        FepObjects fepMap_m;

        HandleMap handleMap_m;

        TAO_AV_Endpoint_Reactive_Strategy_B <TAO_StreamEndPoint_B,TAO_VDev,AV_Null_MediaCtrl> reactiveStrategy_m;    

        AVStreams::StreamEndPoint_B_var sepB_p;

        struct FepsCfgB
        {
            ACE_CString fepsFlowname;
            ACE_CString fepsFormat;
            ACE_CString fepsProtocol;
        };

        AVStreams::flowSpec fepsData;

        //bulkdata::BulkDataReceiverConfig_var recvConfig_p; 
        bulkdata::BulkDataReceiverConfig * recvConfig_p; 

        TAO_StreamEndPoint_B *sepRefCount_p;

        CORBA::Boolean closeReceiverFlag;

        ACS::CBvoid_ptr locNotifCb_p;

        ACE_Time_Value cbTimeout_m;

        //void operator=(const BulkDataReceiver&);
    };
}

#include "bulkDataReceiver.i"

#endif /* _BULKDATA_RECEIVER_H */