DataStreamTool.cpp

Go to the documentation of this file.

// Include files

// from Gaudi
#include "GaudiKernel/DataStreamTool.h"
#include "GaudiKernel/EventSelectorDataStream.h"
#include "GaudiKernel/IAddressCreator.h"
#include "GaudiKernel/ISvcLocator.h"
#include "GaudiKernel/ISvcManager.h"
#include "GaudiKernel/Incident.h"
#include "GaudiKernel/MsgStream.h"
#include "GaudiKernel/MsgStream.h"
#include "GaudiKernel/PropertyHolder.h"
#include "GaudiKernel/Service.h"
#include "GaudiKernel/SmartIF.h"

//-----------------------------------------------------------------------------
// Implementation file for class : DataStreamTool
//
// 2006-09-21 : Andres Felipe Osorio Oliveros
//-----------------------------------------------------------------------------

//=============================================================================
// Standard constructor, initializes variables
//=============================================================================
DataStreamTool::DataStreamTool( const std::string& type, const std::string& name, const IInterface* parent )
    : base_class( type, name, parent )
{
  // declareInterface<IDataStreamTool>(this);
}
//=============================================================================
StatusCode DataStreamTool::initialize()
{

  StatusCode status = AlgTool::initialize();
  if ( !status.isSuccess() ) {
    fatal() << "Error. Cannot initialize base class." << endmsg;
    return status;
  }

  // Get the references to the services that are needed by the ApplicationMgr itself
  m_incidentSvc = serviceLocator()->service( "IncidentSvc" );
  if ( !m_incidentSvc ) {
    fatal() << "Error retrieving IncidentSvc." << endmsg;
    return StatusCode::FAILURE;
  }

  return StatusCode::SUCCESS;
}

StatusCode DataStreamTool::addStream( const std::string& input )
{

  if ( getStream( input ) ) {
    warning() << "Input stream " << input << "already in use" << endmsg;
  }

  m_streamSpecs.push_back( input );

  auto strname               = name() + '_' + std::to_string( ++m_streamCount );
  EventSelectorDataStream* s = nullptr;

  StatusCode status = createStream( strname, input, s );

  if ( status.isSuccess() && s ) {
    s->addRef();
    m_streams.push_back( s );
    status = StatusCode::SUCCESS;
  } else {
    if ( s ) {
      s->release();
      error() << "Error connecting/creating Stream: " << s << endmsg;
    }
    error() << "Error connecting/creating Stream: " << input << endmsg;
    status = StatusCode::FAILURE;
  }
  return status;
}

StatusCode DataStreamTool::addStreams( const StreamSpecs& inputs )
{

  StatusCode status = StatusCode::SUCCESS;
  for ( auto& i : inputs ) {
    status = addStream( i );
    if ( !status.isSuccess() ) break;
  }
  return status;
}

StatusCode DataStreamTool::finalize()
{
  clear().ignore();
  m_incidentSvc.reset();
  return AlgTool::finalize();
}

StatusCode DataStreamTool::initializeStream( EventSelectorDataStream* s )
{
  IEvtSelector* sel = nullptr;
  StatusCode status = s->initialize();
  if ( status.isSuccess() ) {
    status = createSelector( s->name(), s->selectorType(), sel );
    if ( status.isSuccess() ) {
      SmartIF<IProperty> prop( sel ); // Att: IProperty, IService used to point to EventSelector
      SmartIF<IService> isvc( sel );
      s->setSelector( sel );
      sel->release(); // No need for this interface anymore, it is passed to the stream
      if ( prop && isvc ) {
        for ( const auto& i : s->properties() ) prop->setProperty( i ).ignore();
        prop->setProperty( Gaudi::Property<int>( "OutputLevel", msgLevel() ) ).ignore();
        // FIXME: (MCl) Why do we have to initialize the selector here?
        return isvc->sysInitialize();
      }
    }
  }
  return StatusCode::FAILURE;
}

// Create (sub-) Event selector service
StatusCode DataStreamTool::createSelector( const std::string& nam, const std::string& typ, IEvtSelector*& sel )
{
  auto isvc = make_SmartIF( Service::Factory::create( typ, nam, serviceLocator() ) );
  if ( isvc ) {
    auto isel = isvc.as<IEvtSelector>();
    if ( isel ) {
      sel = isel.get();
      sel->addRef(); // make sure that sel is not left dangling once isel and isvc go out of scope...
      return StatusCode::SUCCESS;
    }
  }
  sel = nullptr;
  error() << "Failed to create IEvtSelector " << typ << "/" << nam << endmsg;
  return StatusCode::FAILURE;
}

StatusCode DataStreamTool::finalizeStream( EventSelectorDataStream* s )
{
  if ( s ) {
    IEvtSelector* sel = s->selector();
    if ( sel ) {
      SmartIF<IService> isvc( sel );
      if ( isvc ) {
        isvc->finalize().ignore();
        s->finalize().ignore();
        // Fire EndStream "Incident"
        m_incidentSvc->fireIncident( Incident( name(), IncidentType::EndStream ) );
        return StatusCode::SUCCESS;
      }
      // Failed to get service interface of sub-event selector
      return StatusCode::FAILURE;
    }
    // No selector (yet) attached - no need to finalize it!
    return StatusCode::SUCCESS;
  }
  return StatusCode::FAILURE;
}

StatusCode DataStreamTool::eraseStream( const std::string& info )
{

  auto i = getStreamIterator( info );
  if ( i != m_streams.end() ) {
    ( *i )->release();
    m_streams.erase( i );
    return StatusCode::SUCCESS;
  }
  return StatusCode::FAILURE;
}

StatusCode DataStreamTool::createStream( const std::string& nam, const std::string& info,
                                         EventSelectorDataStream*& stream )
{
  stream = new EventSelectorDataStream( nam, info, serviceLocator() );
  return StatusCode::SUCCESS;
}

EventSelectorDataStream* DataStreamTool::getStream( const std::string& info )
{
  auto i = getStreamIterator( info );
  return i != m_streams.end() ? *i : nullptr;
}

DataStreamTool::Streams::iterator DataStreamTool::getStreamIterator( const std::string& info )
{
  return std::find_if( std::begin( m_streams ), std::end( m_streams ),
                       [&]( const EventSelectorDataStream* i ) { return i->definition() == info; } );
}

EventSelectorDataStream* DataStreamTool::getStream( size_type pos )
{
  // pos has to point inside the vector
  return ( ( pos >= 0 ) && ( (size_t)pos < m_streams.size() ) ) ? m_streams[pos] : nullptr;
}

EventSelectorDataStream* DataStreamTool::lastStream() { return m_streams.back(); }

StatusCode DataStreamTool::clear()
{

  StatusCode iret, status = StatusCode::SUCCESS;
  iret.ignore();

  // disconnect the streams
  for ( auto& il : m_streamSpecs ) {
    EventSelectorDataStream* s = getStream( il );
    if ( s ) {
      if ( s->isInitialized() ) {
        iret = finalizeStream( s );
        if ( !iret.isSuccess() ) {
          error() << "Error finalizing Stream" << il << endmsg;
          status = iret;
        }
      }
      iret = eraseStream( il );
      if ( !iret.isSuccess() ) {
        error() << "Error diconnecting Stream" << il << endmsg;
        status = iret;
      }
    }
  }

  m_streamSpecs.clear();

  return status;
}

StatusCode DataStreamTool::connectStream( EventSelectorDataStream* s )
{

  if ( !s ) return StatusCode::FAILURE;
  s->addRef();
  m_streams.push_back( s );
  return StatusCode::SUCCESS;
}

StatusCode DataStreamTool::connectStream( const std::string& info )
{
  if ( getStream( info ) ) {
    warning() << "Input stream " << info << "already in use" << endmsg;
  }
  auto nam                   = name() + '_' + std::to_string( ++m_streamCount );
  EventSelectorDataStream* s = nullptr;
  StatusCode status          = createStream( nam, info, s );
  if ( status.isSuccess() ) return connectStream( s );
  s->release();
  return status;
}

/*

  Taking control over Streams and return them to EventSelector

*/

StatusCode DataStreamTool::getNextStream( const EventSelectorDataStream*& esds, size_type& dsid )
{

  EventSelectorDataStream* nextStream = getStream( dsid );
  if ( !nextStream ) return StatusCode::FAILURE; //<-end of streams reached

  esds = nextStream;
  ++m_streamID;

  return StatusCode::SUCCESS;
}

StatusCode DataStreamTool::getPreviousStream( const EventSelectorDataStream*& esds, size_type& dsid )
{

  EventSelectorDataStream* previousStream = getStream( dsid );
  if ( !previousStream ) return StatusCode::FAILURE; //<-begin of streams reached

  esds = previousStream;
  --m_streamID;

  return StatusCode::SUCCESS;
}