cdi_daq.cxx

//    cdi_daq.cxx
//
//    Test application for TDAQ conditions data collection
//
//    Luis Pedro,  November 2003
//
//    description:
//          Implements an interce between the TDAQ IS service
//          and the ATLAS Conditions Database
//      

// changed after the release of the new IS API
//#include <is/isinfo.h>
//#include <is/isinfoany.h>
#include <is/info.h>
#include <is/infoany.h>

#include <iostream>
#include <owl/time.h>
#include <ipc/core.h>
#include <ipc/partition.h>
#include <cmdl/cmdargs.h>
#include <signal.h>
#include <list>
#include <unistd.h>

#include "cdi/cdiHandle.h"
#include "cdi/cdiDataHandle.h"
#include "cdi/cdiCOOLDataHandle.h"
#include <exception>

#define LISBON 0
#define COOL   1

//enum apis {LISBON=0,COOL=1};

int verbose;

ISInfoReceiver *    ISReceiver;

struct CallbackParams
{
  std::string           partition_name;
  //  cdiDataHandle *       data_handle;
  cdiHandle*                    data_handle;
  ISInfoReceiver *      receiver;
  std::vector<std::string>  is_names;
  cdiAux            mout;
  std::string                   conf_obj; 
};

void subscribe_to_IS( CallbackParams & params );
void unsubscribe_from_IS( CallbackParams & params );

void signal_exit( int )
{
   if ( ISReceiver )
     ISReceiver->stop();        
}

void is_callback_function( ISCallbackInfo * isc )
{

  CallbackParams * params = static_cast<CallbackParams*>(isc->parameter());
  params->mout.info( std::string("Callback::" + (std::string)isc->name()) );
  
  // NBARROS: This might lead to problems
  // currently the value is hardcoded as  "RunParams.Conditions";
  
  // here we have the issue that each time that someone changes the Conditions
  // object all data is stored all over again
  
  if( std::string(isc->name()) == params->conf_obj) // std::string("RunParams.Conditions")) //cdiDataHandle::configuration_name )
    {
      params->mout.debug(std::string( "Processing configuration callback"));
      unsubscribe_from_IS( *params );
      params->data_handle->getConfig( params->is_names );
      subscribe_to_IS( *params );  
    } 
  else
    {
      params->mout.debug( std::string( "Processing normal callback (not configuration)"));
    }
  params->data_handle->process_data( std::string( isc->name()), isc );
}

void subscribe_to_IS( CallbackParams & params )
{
  params.mout.verbose( std::string("::cdi_daq::unsubscribe_to_IS -> Entering..."));
  
  ISInfo::Status status;
    
  for( size_t i = 0; i < params.is_names.size(); i++ )
  {
    if ( (status = params.receiver->subscribe(  params.is_names[i].c_str(), 
                        is_callback_function, 
                        &params ) ) != ISInfo::Success ) {
    if ( status == ISInfo::AlreadyExist)
      params.mout.error( std::string( "Subscribe for IS info " + params.is_names[i] + " FAILED: AlreadyExist!") );
      else if ( status == ISInfo::CommFailure)
    params.mout.error( std::string( "Subscribe for IS info " + params.is_names[i] + " FAILED: CommFailure !") );
      else
    params.mout.error( std::string( "Subscribe for IS info " + params.is_names[i] + " FAILED: Unknown !") );
    }
    else
    {
      params.mout.info( std::string( "Subscribe IS server " + params.is_names[i] + " SUCCESS.") );
    }
  }

  params.mout.info( std::string("Ready to data aquisition!") );
}


void unsubscribe_from_IS( CallbackParams & params )
{  
  ISInfo::Status status;
  
  for( size_t i = 0; i < params.is_names.size(); i++ )
  {
    if ( (status = params.receiver->unsubscribe( params.is_names[i].c_str() ) ) != ISInfo::Success )
      params.mout.error( std::string( "Unsubscribe from IS info " + params.is_names[i] + " FAILED:" ) );
    else
      params.mout.info ( std::string( "Unsubscribe from IS info " + params.is_names[i] + " SUCCESS.") );
  }
}


int main(int argc, char ** argv){
  
  IPCCore::init(argc, argv);
  
  CmdArgStr partition_name('p', "partition", "partition-name", "partition to work in.", CmdArg::isREQ);
  CmdArgInt backend('B', "backend", "cdi-backend", "Choose the backend to be used (0:Lisbon_CONDDB, 1:COOL).", CmdArg::isREQ);
  CmdArgInt verb_level('v', "verbose", "verbosity-level", "0:Silent, 1:Info, 2:Debug, 3:Verbose (defaults to 0).");
  CmdArgStr    conf_obj('O',"config-list","configuration-object","Name of the ISobject containing the list of objects to be stored (defaults to \"RunParams.Conditions\")");
  
  CmdLine  cmd(*argv, &partition_name, &backend ,&verb_level, &conf_obj,NULL);
  CmdArgvIter  arg_iter(--argc, ++argv);
  
  cmd.description("This program implements the functionality to store data published in subscribed\n"
          "IS Servers into the ATLAS Conditions Database.\n"
          "\t It provides the functionality to interface the TDAQ system to the Conditions Database.\n"
          "The user must choose which backend myust be used (COOL or Lisbon CondDB)");
  
  
  partition_name = 0;
  verb_level = 0;
  conf_obj = "RunParams.Conditions";

  cmd.parse(arg_iter);
  
  std::cout << OWLTime() << " :: cdi_daq is initializing with " << argc << " command line parameters" << std::endl;
  
  
  //Initialization
  //  cdiDataHandle cdi_handle(verb_level, (std::string)partition_name);
   
  cdiHandle *cdi_handle;
  
  if ( backend == COOL ) {
    cdi_handle = new cdiCOOLDataHandle(verb_level,(std::string)partition_name,(std::string)conf_obj);
  } else
    if ( backend == LISBON ) {
      cdi_handle = new cdiDataHandle(verb_level,(std::string)partition_name);
    }
    else {
      std::cerr << OWLTime() << " :: cdi_daq failed because the backend [" << backend 
        << "] is not valid!" << std::endl;
      exit(1);
    }

  // initialize signal handlers
  signal(SIGINT,  signal_exit);
  signal(SIGTERM, signal_exit);
 
  IPCPartition partition(partition_name);  
  
  ISInfoReceiver receiver(partition);
 
  ISReceiver = &receiver;

  std::vector<std::string> is_names;
  
  //  cdi_handle.getConfig( is_names );

  // this fills the ISobjects subscribed using the Conditions object and the mandatory elements
  cdi_handle->getConfig( is_names );
  
  if (verb_level >= 3)
    std::cout << OWLTime() << "::cdi_daq -> Creating CallbackParams..." << std::endl;
  
  CallbackParams params;
  
  if (verb_level >= 3)
    std::cout << OWLTime() << "::cdi_daq -> Defining partition as : [" << partition_name << "]..." << std::endl;
  params.partition_name = partition_name;
  
  if (verb_level >= 3)
    std::cout << OWLTime() << "::cdi_daq -> Configuration object to be used : [" << conf_obj << "]..." << std::endl;
  params.conf_obj = conf_obj;
  

  if (verb_level >= 3)
    std::cout << OWLTime() << "::cdi_daq -> Uploading the cdiHandle..." << std::endl;
  params.data_handle = cdi_handle;//&cdi_handle;
  params.receiver = &receiver;
  params.is_names = is_names;
  params.mout = cdiAux( verb_level );
  params.conf_obj = conf_obj;

  if (verb_level >= 3)
    std::cout << OWLTime() << "::cdi_daq -> Subscribing to IS..." << std::endl;
  subscribe_to_IS( params );
  
  std::cout << OWLTime() << " :: cdi_daq has been started in the \"" 
        << partition.name() << "\" partition" << std::endl;


  receiver.run();
  
  /*
    try {
    receiver.run();
    }
    catch (std::exception &e)
    {
    std::cerr << "Standard exception caught [" << e.message() << "]" << std::endl;
    std::cerr << OWLTime() << " :: cdi_daq is going to quit!" << std::endl;
    
    }
  */
  ISReceiver = 0;
  
  unsubscribe_from_IS( params );
  
  return 0;
}

---