bulkDataSender.h

Go to the documentation of this file.

#ifndef _BULKDATA_SENDER_H
#define _BULKDATA_SENDER_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 <iostream>

#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 "bulkDataSenderDefaultCb.h"
#include "bulkDataFlowProducer.h"

#include "bulkDataC.h"

// #include "ace/High_Res_Timer.h"

// #include <acsQoS.h>

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

        void createSingleFlow();

        void createMultipleFlows(const char *fepsConfig);

        void connectToPeer(bulkdata::BulkDataReceiverConfig *recvConfig_p);

        void getFlowProtocol(ACE_CString &flowname, TAO_AV_Protocol_Object *&currentProtocol_p);

        void startSend(CORBA::ULong flownumber, ACE_Message_Block *param = 0);

        void startSend(CORBA::ULong flownumber, const char *param, size_t len);

        void sendData(CORBA::ULong flownumber, ACE_Message_Block *buffer);
           
        void sendData(CORBA::ULong flownumber, const char *buffer, size_t len);

        void stopSend(CORBA::ULong flownumber);

        void disconnectPeer();
 
        TAO_StreamCtrl * getStreamCtrl() 
            {
                return streamctrl_p;
            }

        const char *getFlowSpec(const ACE_CString & flowName);

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

        // checks if the handler referenced by handle is registered inside the ACE Reactor
        ACE_HANDLE findHandle(ACE_CString &flowname);

        /* THE FOLLOWING METHODS ARE UNDER TESTING - PLEASE DO NOT USE THEM */
        /********************************************************************/

        void startStream(CORBA::ULong flownumber);
        
        void sendStream(CORBA::ULong flownumber, ACE_Message_Block *buffer);

        void stopStream(CORBA::ULong flownumber);

        /********************************************************************/

      private:

        typedef ACE_Hash_Map_Manager<ACE_CString, BulkDataFlowProducer<TSenderCallback> *, ACE_Null_Mutex> FepObjects;
        typedef ACE_Hash_Map_Iterator<ACE_CString, BulkDataFlowProducer<TSenderCallback> *, 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 initPartA();

        AVStreams::StreamEndPoint_A_ptr createSepA();

        AVStreams::FlowProducer_ptr createFepProducerA(ACE_CString &flowname,
                                                       AVStreams::protocolSpec protocols,
                                                       ACE_CString &format,
                                                       TAO_StreamCtrl *strctrl_p);


        void addFepToSep(AVStreams::StreamEndPoint_A_ptr locSepA_p, AVStreams::FlowProducer_ptr locFepA_p);


        TAO_StreamCtrl *createStreamCtrl();     


        //      AVStreams::streamQoS * create_QoS();


        const char * createFwdFlowSpec(ACE_CString &flowname,
                                       ACE_CString &direction,
                                       ACE_CString &formatName,
                                       ACE_CString &flowProtocol,
                                       ACE_CString &carrierProtocol,
                                       ACE_CString &localAddress,
                                       ACE_CString &remoteAddress);


        void setReceiverConfig(bulkdata::BulkDataReceiverConfig *recvConfig_p);

                    

        AVStreams::StreamEndPoint_A_ptr getStreamEndPointA();


        void deleteStreamCtrl();



        void deleteFepsA();

        void deleteSepA();

        void deleteConnector();

        void deleteHandler();

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

        void mergeFlowSpecs();

        TAO_AV_Endpoint_Reactive_Strategy_A<TAO_StreamEndPoint_A,TAO_VDev,AV_Null_MediaCtrl> endpointStrategy_m;

        AVStreams::StreamEndPoint_A_var sepA_p;

        AVStreams::StreamEndPoint_B_var sepB_p;


        //AVStreams::streamQoS_var the_qos;
        struct FepsCfgA
        {
            ACE_CString fepsFlowname;
            ACE_CString fepsFormat;
            ACE_CString fepsProtocol;
        };

        FepObjects fepMap_m;

        HandleMap handleMap_m;

        AVStreams::flowSpec_var recvFeps_p;

        AVStreams::flowSpec senderFeps_m;

        TAO_StreamEndPoint_A * sepRefCount_p;

        CORBA::Boolean disconnectPeerFlag;
    
        AVStreams::flowSpec flowSpec_m;

        TAO_StreamCtrl *streamctrl_p;

        //void operator=(const BulkDataSender&);

//      public:
// this part should be private
// deprecated and back-incompatible
// needed for the distributer but will be removed as soon as possible
//
//      AVStreams::flowSpec flowSpec_m;
//      TAO_StreamCtrl *streamctrl_p;
    };
}

#include "bulkDataSender.i"

#endif /* _BULKDATA_SENDER_H */