---

OutputStream.cpp

Go to the documentation of this file.

//      ====================================================================
//  OutputStream.cpp
//      --------------------------------------------------------------------
//
//      Package   : GaudiSvc/PersistencySvc
//
//      Author    : Markus Frank
//
//      ====================================================================
#define GAUDISVC_PERSISTENCYSVC_OUTPUTSTREAM_CPP

// STL include files
#include <iostream>
// Framework include files
#include "GaudiKernel/Tokenizer.h"
#include "GaudiKernel/AlgFactory.h"
#include "GaudiKernel/IAlgManager.h"
#include "GaudiKernel/ISvcLocator.h"
#include "GaudiKernel/IConversionSvc.h"
#include "GaudiKernel/IDataDirectory.h"
#include "GaudiKernel/IDataManagerSvc.h"
#include "GaudiKernel/IDataProviderSvc.h"
#include "GaudiKernel/IPersistencySvc.h"

#include "GaudiKernel/MsgStream.h"
#include "GaudiKernel/DataStoreItem.h"
#include "OutputStream.h"
#include "OutputStreamAgent.h"

// Define the algorithm factory for the standard output data writer
static const AlgFactory<OutputStream> s_factory;
const IAlgFactory& OutputStreamFactory = s_factory;

// Standard Constructor
OutputStream::OutputStream(const std::string& name, ISvcLocator* pSvcLocator)
 : Algorithm(name, pSvcLocator)
{
  m_doPreLoad      = true;
  m_output         = "";
  m_outputName     = "";
  m_storeName      = "EventDataSvc";
  m_persName       = "EventPersistencySvc";
  m_agent          = new OutputStreamAgent(this);
  m_pDataManager   = 0;
  m_pDataProvider  = 0;
  m_pConversionSvc = 0;
  m_acceptAlgs     = new std::vector<Algorithm*>( );
  m_requireAlgs    = new std::vector<Algorithm*>( );
  m_vetoAlgs       = new std::vector<Algorithm*>( );
        declareProperty("ItemList",         m_itemNames);
        declareProperty("Output",           m_output);
        declareProperty("OutputFile",       m_outputName);
        declareProperty("EvtDataSvc",       m_storeName);
        declareProperty("EvtConversionSvc", m_persName);
        declareProperty("AcceptAlgs",       m_acceptNames);
        declareProperty("RequireAlgs",      m_requireNames);
        declareProperty("VetoAlgs",         m_vetoNames);

    // Associate action handlers with the AcceptAlgs, RequireAlgs & VetoAlgs properties
  m_acceptNames.declareUpdateHandler ( &OutputStream::acceptAlgsHandler , this );
    m_requireNames.declareUpdateHandler( &OutputStream::requireAlgsHandler, this );
    m_vetoNames.declareUpdateHandler   ( &OutputStream::vetoAlgsHandler   , this );
}

// Standard Destructor
OutputStream::~OutputStream()   {
  delete m_agent;
  delete m_acceptAlgs;
  delete m_requireAlgs;
  delete m_vetoAlgs;
}

// initialize data writer
StatusCode OutputStream::initialize() {
  StatusCode status = StatusCode::SUCCESS;
        MsgStream log(msgSvc(), name());
        
        // Reset the number of events written
        m_events = 0;
        
        // Use the Job options service to set the Algorithm's parameters
        setProperties();
  // Get access to the DataManagerSvc
  status = serviceLocator()->getService(m_storeName, IID_IDataManagerSvc, (IInterface*&)m_pDataManager );
  if( !status.isSuccess() )   {
    log << MSG::FATAL << "Unable to locate IDataManagerSvc interface" << endreq;
    return status;
  }
  m_pDataManager->addRef();
  // Get access to the assigned data service
  status = serviceLocator()->getService(m_storeName, IID_IDataProviderSvc, (IInterface*&)m_pDataProvider );
  if( !status.isSuccess() )   {
    log << MSG::FATAL << "Unable to locate IDataProviderSvc interface of " << m_storeName << endreq;
    return status;
  }
  m_pDataProvider->addRef();
  status = connectConversionSvc();
  if( !status.isSuccess() )   {
    log << MSG::FATAL << "Unable to connect to conversion service." << endreq;
    return status;
  }
  m_pConversionSvc->addRef();

  // Clear the item list
  clearItems();
  // Take the new item list from the properties.
  for(ItemNames::iterator i = m_itemNames.begin(); i != m_itemNames.end(); i++)   {
    addItem( *i );
  }
  // Take the item list to the data service preload list.
  if ( m_doPreLoad )    {
    for(Items::iterator j = m_itemList.begin(); j != m_itemList.end(); j++)   {
      m_pDataProvider->addPreLoadItem( *(*j) );
    }
    // Not working: bad reference counting! pdataSvc->release();
  }
  log << MSG::INFO << "Data source: " << m_storeName  << endreq;
  log << MSG::INFO << "Data output: " << m_output << endreq;

  // Decode the accept, required and veto Algorithms. The logic is the following:
  //  a. The event is accepted if all lists are empty.
  //  b. The event is provisionally accepted if any Algorithm in the accept list
  //     has been executed and has indicated that its filter is passed. This
  //     provisional acceptance can be overridden by the other lists.
  //  c. The event is rejected unless all Algorithms in the required list have
  //     been executed and have indicated that their filter passed.
  //  d. The event is rejected if any Algorithm in the veto list has been
  //     executed and has indicated that its filter has passed.
  decodeAcceptAlgs ( );
  decodeRequireAlgs( );
  decodeVetoAlgs   ( );
  return status;
}

// terminate data writer
StatusCode OutputStream::finalize() {
  MsgStream log(msgSvc(), name());
  log << MSG::INFO << "Events output: " << m_events << endreq;
  if ( m_pDataProvider ) m_pDataProvider->release();
  m_pDataProvider = 0;
  if ( m_pDataManager ) m_pDataManager->release();
  m_pDataManager = 0;
  if ( m_pConversionSvc ) m_pConversionSvc->release();
  m_pConversionSvc = 0;
  clearItems();
        return StatusCode::SUCCESS;
}

// Work entry point
StatusCode OutputStream::execute() {
  StatusCode status = StatusCode::SUCCESS;
  // Clear any previously existing item list
  clearSelection();

  // Test whether this event should be output
  if ( isEventAccepted( ) ) {
    // Connect the output file to the service
    status = m_pConversionSvc->connectOutput(m_outputName);
    if ( status.isSuccess() )   {
      // Traverse the tree and collect the requested objects
      for ( Items::iterator i = m_itemList.begin(); i != m_itemList.end(); i++ )    {
        m_currentItem = (*i);
        status = m_pDataManager->traverseSubTree(m_currentItem->path(), m_agent);
      }
      if ( status.isSuccess() )   {
        // Now pass the collection to the persistency service
        status = m_pConversionSvc->createReps( selectedObjects() );
      }
      clearSelection();
      m_events++;
    }
  }
  return status;
}

// Place holder to create configurable data store agent
bool OutputStream::collect(IDataDirectory* dir, int level)    {
  if ( level < m_currentItem->depth() )   {
    if ( dir->object() != 0 )   {
      /*
      std::cout << "Analysing (" 
                << dir->name() 
                << ") Object:" 
                << ((dir->object()==0) ? "UNLOADED" : "LOADED") 
                << std::endl;
      */
      m_objects.push_back(dir->object());
      return true;
    }
  }
  return false;
}

// Clear collected object list
void OutputStream::clearSelection()     {
  m_objects.erase(m_objects.begin(), m_objects.end());
}

// Remove all items from the output streamer list;
void OutputStream::clearItems()     {
  for ( Items::iterator i = m_itemList.begin(); i != m_itemList.end(); i++ )    {  
    delete (*i);
  }
  m_itemList.erase(m_itemList.begin(), m_itemList.end());
}

// Add item to output streamer list
void OutputStream::addItem(const std::string& descriptor)   {
        MsgStream log(msgSvc(), name());
  int sep = descriptor.rfind("#");
  int level = 0;
  std::string obj_path (descriptor,0,sep);
  std::string slevel   (descriptor,sep+1,descriptor.length());
  if ( slevel == "*" )  {
    level = 9999999;
  }
  else   {
    level = atoi(slevel.c_str());
  }
  DataStoreItem* item = new DataStoreItem(obj_path, level);
  log << MSG::INFO << "Adding OutputStream item " << item->path()
      << " with " << item->depth() 
      << " level(s)." << endreq;
  m_itemList.push_back( item );
}

// Connect to proper conversion service
StatusCode OutputStream::connectConversionSvc()   {
  StatusCode status = StatusCode::FAILURE;
        MsgStream log(msgSvc(), name());
  // Get output file from input
  std::string dbType, svc;
  Tokenizer tok;
  tok.analyse(m_output, " ", "", "", "=", "'", "'");
  for ( Tokenizer::Items::iterator i = tok.items().begin(); i != tok.items().end(); i++ )   {
    const std::string& tag = (*i).tag();
    const std::string& val = (*i).value();
    switch( ::toupper(tag[0]) )    {
    case 'D':
      m_outputName = val;
      break;
    case 'T':
      dbType = val;
      break;
    case 'S':
      svc = val;
      break;
    default:
      break;
    }
  }
  // Get access to the default Persistency service
  // The default service is the same for input as for output.
  // If this is not desired, then a specialized OutputStream must overwrite 
  // this value.
  if ( dbType.length() > 0 && svc.length()==0 )   {
    IPersistencySvc* ipers = 0;
    status = serviceLocator()->getService(m_persName, IID_IPersistencySvc, (IInterface*&)ipers );
    if( !status.isSuccess() )   {
      log << MSG::FATAL << "Unable to locate IPersistencySvc interface of " << m_persName << endreq;
      return status;
    }
    status = ipers->getService(dbType, m_pConversionSvc);
    if( !status.isSuccess() )   {
      log << MSG::FATAL << "Unable to locate IConversionSvc interface of database type " << dbType << endreq;
      return status;
    }
  }
  else if ( svc.length() > 0 )    {
    status = serviceLocator()->getService(svc, IID_IConversionSvc, (IInterface*&)m_pConversionSvc );
    if( !status.isSuccess() )   {
      log << MSG::FATAL << "Unable to locate IConversionSvc interface of " << svc << endreq;
      return status;
    }
  }
  else    {
    if ( m_output.length() == 0 )  m_output = m_outputName + " SVC='"+m_persName+"'";
    if ( m_outputName.length() == 0 )  m_outputName = m_output;
    status = serviceLocator()->getService(m_persName, IID_IConversionSvc, (IInterface*&)m_pConversionSvc );
    if( !status.isSuccess() )   {
      log << MSG::FATAL << "Unable to locate IConversionSvc interface of " << m_persName << endreq;
      return status;
    }
  }
  return status;
}

StatusCode
OutputStream::decodeAcceptAlgs( )
{
  StatusCode result = decodeAlgorithms( m_acceptNames, m_acceptAlgs );
  return result;
}

void
OutputStream::acceptAlgsHandler( Property& /* theProp */ )
{
  decodeAlgorithms( m_acceptNames, m_acceptAlgs );
}

StatusCode
OutputStream::decodeRequireAlgs( )
{
  StatusCode result = decodeAlgorithms( m_requireNames, m_requireAlgs );
  return result;
}

void
OutputStream::requireAlgsHandler( Property& /* theProp */ )
{
  decodeAlgorithms( m_requireNames, m_requireAlgs );
}

StatusCode
OutputStream::decodeVetoAlgs( )
{
  StatusCode result = decodeAlgorithms( m_vetoNames, m_vetoAlgs );
  return result;
}

void
OutputStream::vetoAlgsHandler( Property& /* theProp */ )
{
  decodeAlgorithms( m_vetoNames, m_vetoAlgs );
}

StatusCode
OutputStream::decodeAlgorithms( StringArrayProperty& theNames,
                                std::vector<Algorithm*>* theAlgs )
{

  // Reset the list of Algorithms
  theAlgs->clear( );

  MsgStream log( msgSvc( ), name( ) );

  IAlgManager* theAlgMgr;
  StatusCode result = serviceLocator( )->getService( "ApplicationMgr",
                                           IID_IAlgManager,
                                           (IInterface*&)theAlgMgr );
  if ( result.isSuccess( ) ) {

    // Build the list of Algorithms from the names list
    const std::vector<std::string> nameList = theNames.value( );
    std::vector<std::string>::const_iterator it;
    std::vector<std::string>::const_iterator itend = nameList.end( );
    for (it = nameList.begin(); it != itend; it++) {

      // Check whether the suppied name corresponds to an existing
      // Algorithm object.
      std::string theName = (*it);
      IAlgorithm* theIAlg;
      Algorithm*  theAlgorithm;
      result = theAlgMgr->getAlgorithm( theName, theIAlg );
      if ( result.isSuccess( ) ) {
            try{
                  theAlgorithm = dynamic_cast<Algorithm*>(theIAlg);
            } catch(...){
                  result = StatusCode::FAILURE;
            }
          }
      if ( result.isSuccess( ) ) {

        // Check that the specified algorithm doesn't already exist in the list
        std::vector<Algorithm*>::iterator ita;
        std::vector<Algorithm*>::iterator itaend = theAlgs->end( );
        for (ita = theAlgs->begin(); ita != itaend; ita++) {
          Algorithm* existAlgorithm = (*ita);
          if ( theAlgorithm == existAlgorithm ) {
            result = StatusCode::FAILURE;
            break;
          }
        }
        if ( result.isSuccess( ) ) {
          theAlgs->push_back( theAlgorithm );
        }
      } else {
        log << MSG::INFO << theName << " doesn't exist - ignored" << endreq;
      }
    }
    result = StatusCode::SUCCESS;
  } else {
    log << MSG::FATAL << "Can't locate ApplicationMgr!!!" << endreq;
  }
  return result;
}

bool
OutputStream::isEventAccepted( ) const
{
  bool result = true;

  // Loop over all Algorithms in the accept list to see
  // whether any have been executed and have their filter
  // passed flag set. Any match causes the event to be
  // provisionally accepted.
  if ( ! m_acceptAlgs->empty( ) ) {
    result = false;
    std::vector<Algorithm*>::iterator it;
    std::vector<Algorithm*>::iterator itend = m_acceptAlgs->end( );
    for (it = m_acceptAlgs->begin(); it != itend; it++) {
      Algorithm* theAlgorithm = (*it);
      if ( theAlgorithm->isExecuted( ) && theAlgorithm->filterPassed( ) ) {
        result = true;
        break;
      }
    }
  }

  // Loop over all Algorithms in the required list to see
  // whether all have been executed and have their filter
  // passed flag set. Any mismatch causes the event to be
  // rejected.
  if ( result && ! m_requireAlgs->empty( ) ) {
    std::vector<Algorithm*>::iterator it;
    std::vector<Algorithm*>::iterator itend = m_requireAlgs->end( );
    for (it = m_requireAlgs->begin(); it != itend; it++) {
      Algorithm* theAlgorithm = (*it);
      if ( ! theAlgorithm->isExecuted( ) || ! theAlgorithm->filterPassed( ) ) {
        result = false;
        break;
      }
    }
  }

  // Loop over all Algorithms in the veto list to see
  // whether any have been executed and have their filter
  // passed flag set. Any match causes the event to be
  // rejected.
  if ( result && ! m_vetoAlgs->empty( ) ) {
    std::vector<Algorithm*>::iterator it;
    std::vector<Algorithm*>::iterator itend = m_vetoAlgs->end( );
    for (it = m_vetoAlgs->begin(); it != itend; it++) {
      Algorithm* theAlgorithm = (*it);
      if ( theAlgorithm->isExecuted( ) && theAlgorithm->filterPassed( ) ) {
        result = false;
        break;
      }
    }
  }
  return result;
}

---