AthenaOutputStreamTool.cxx

Go to the documentation of this file.

/*
  Copyright (C) 2002-2025 CERN for the benefit of the ATLAS collaboration
*/

 
#include "AthenaOutputStreamTool.h"
 
// Gaudi
#include "GaudiKernel/IOpaqueAddress.h"
#include "GaudiKernel/INamedInterface.h"
#include "GaudiKernel/IClassIDSvc.h"
#include "GaudiKernel/ThreadLocalContext.h"
 
// Athena
#include "AthenaKernel/IDecisionSvc.h"
#include "AthenaBaseComps/AthCnvSvc.h"
#include "StoreGate/StoreGateSvc.h"
#include "SGTools/DataProxy.h"
#include "SGTools/SGIFolder.h"
#include "PersistentDataModel/AthenaAttributeList.h"
#include "PersistentDataModel/DataHeader.h"
#include "PersistentDataModel/TokenAddress.h"
 
 
namespace {

bool hasInputAlias (const SG::DataProxy& dp)
{
  std::string inputName = dp.name() + "_Input";
  return dp.hasAlias (inputName);
}
 
 
} // anonymous namespace

AthenaOutputStreamTool::AthenaOutputStreamTool(const std::string& type,
        const std::string& name,
        const IInterface* parent) : base_class(type, name, parent),
    m_clidSvc("ClassIDSvc", name),
    m_decSvc("DecisionSvc/DecisionSvc", name) {
}
//__________________________________________________________________________
StatusCode AthenaOutputStreamTool::initialize() {
   ATH_MSG_INFO("Initializing " << name());
 
   ATH_CHECK( m_clidSvc.retrieve() );
   ATH_CHECK( m_conversionSvc.retrieve() );
 
   // Autoconfigure
   if (m_dataHeaderKey.empty()) {
      m_dataHeaderKey.setValue(name());
      // Remove "ToolSvc." from m_dataHeaderKey.
      if (m_dataHeaderKey.value().starts_with( "ToolSvc.")) {
         m_dataHeaderKey.setValue(m_dataHeaderKey.value().substr(8));
         // Remove "Tool" from m_dataHeaderKey.
         if (m_dataHeaderKey.value().find("Tool") == m_dataHeaderKey.size() - 4) {
            m_dataHeaderKey.setValue(m_dataHeaderKey.value().substr(0, m_dataHeaderKey.size() - 4));
         }
      } else {
         const INamedInterface* parentAlg = dynamic_cast<const INamedInterface*>(parent());
         if (parentAlg != 0) {
            m_dataHeaderKey.setValue(parentAlg->name());
         }
      }
   }
   if (m_processTag.empty()) {
      m_processTag.setValue(m_dataHeaderKey);
   }
 
   { // handle the AttrKey overwrite
     const std::string keyword = "[AttributeListKey=";
     std::string::size_type pos = m_outputName.value().find(keyword);
     if( (pos != std::string::npos) ) {
       ATH_MSG_INFO("The AttrListKey will be overwritten/set by the value from the OutputName: " << m_outputName);
       const std::string attrListKey = m_outputName.value().substr(pos + keyword.size(),
                                   m_outputName.value().find(']', pos + keyword.size()) - pos - keyword.size());
       m_attrListKey = attrListKey;
     }
   }
   if ( ! m_attrListKey.key().empty() ) {
     ATH_CHECK(m_attrListKey.initialize());
     m_attrListWrite = m_attrListKey.key() + "Decisions";
     //ATH_CHECK(m_attrListWrite.initialize());
   }
 
   if (!m_outputCollection.value().empty()) {
      m_outputAttributes += "[OutputCollection=" + m_outputCollection.value() + "]";
   }
   if (!m_containerPrefix.value().empty()) {
      m_outputAttributes += "[PoolContainerPrefix=" + m_containerPrefix.value() + "]";
   }
   if (m_containerNameHint.value() != "0") {
      m_outputAttributes += "[TopLevelContainerName=" + m_containerNameHint.value() + "]";
   }
   if (m_branchNameHint.value() != "0") {
      m_outputAttributes += "[SubLevelBranchName=" + m_branchNameHint.value() + "]";
   }
   if (!m_metaDataOutputCollection.value().empty()) {
      m_metaDataOutputAttributes += "[OutputCollection=" + m_metaDataOutputCollection.value() + "]";
   }
   if (!m_metaDataContainerPrefix.value().empty()) {
      m_metaDataOutputAttributes += "[PoolContainerPrefix=" + m_metaDataContainerPrefix.value() + "]";
   }
 
   if (m_extend) ATH_CHECK(m_decSvc.retrieve());
   return(StatusCode::SUCCESS);
}
//__________________________________________________________________________
StatusCode AthenaOutputStreamTool::connectServices(const std::string& dataStore,
    const std::string& cnvSvc,
    bool extendProvenenceRecord) {
   // Release old data store
   if (m_store.isValid()) {
      if (m_store.release().isFailure()) {
         ATH_MSG_ERROR("Could not release " << m_store.typeAndName() << " store");
      }
   }
   m_store = ServiceHandle<StoreGateSvc>(dataStore, this->name());
   if (cnvSvc != m_conversionSvc.type() && cnvSvc != "EventPersistencySvc") {
      if (m_conversionSvc.release().isFailure()) {
         ATH_MSG_ERROR("Could not release " << m_conversionSvc.type());
      }
      m_conversionSvc = ServiceHandle<IConversionSvc>(cnvSvc, this->name());
      if (m_conversionSvc.retrieve().isFailure() || m_conversionSvc == 0) {
         ATH_MSG_ERROR("Could not locate " << m_conversionSvc.type());
         return(StatusCode::FAILURE);
      }
   }
   m_extendProvenanceRecord = extendProvenenceRecord;
   auto pprop = dynamic_cast<const IProperty*>(parent());
   if (not pprop){
     ATH_MSG_ERROR("'parent' could not be cast to IProperty");
     return(StatusCode::FAILURE);
   }
   auto keep = dynamic_cast<const StringProperty&>( pprop->getProperty("KeepProvenanceTagsRegEx") );
   m_keepProvenancesStr = keep.value();
   // create RegEx pattern from the property value, specify extended grammar
   m_keepProvenancesRE = std::regex(m_keepProvenancesStr, std::regex::extended);
 
   return(connectServices());
}
//__________________________________________________________________________
StatusCode AthenaOutputStreamTool::connectServices() {
   // Find the data store
   if (m_store.retrieve().isFailure() || m_store == 0) {
      ATH_MSG_ERROR("Could not locate " << m_store.typeAndName() << " store");
      return(StatusCode::FAILURE);
   }
   return(StatusCode::SUCCESS);
}
//__________________________________________________________________________
StatusCode AthenaOutputStreamTool::connectOutput(const std::string& outputName) {
   ATH_MSG_DEBUG("In connectOutput " << outputName);
 
   // Use arg if not empty, save the output name
   if (!outputName.empty()) {
      m_outputName.setValue(outputName);
   }
   if (m_outputName.value().empty()) {
      ATH_MSG_ERROR("No OutputName provided");
      return(StatusCode::FAILURE);
   }
   // Connect services if not already available
   if (m_store == 0 || m_conversionSvc == 0) {
     ATH_CHECK( connectServices() );
   }
   // Connect the output file to the service
   if (m_conversionSvc->connectOutput(m_outputName.value()).isFailure()) {
      ATH_MSG_ERROR("Unable to connect output " << m_outputName.value());
      return(StatusCode::FAILURE);
   } else {
      ATH_MSG_DEBUG("Connected to " << m_outputName.value());
   }
 
   // Remove DataHeader with same key if it exists
   if (m_store->contains<DataHeader>(m_dataHeaderKey)) {
      const DataHeader* preDh = nullptr;
      if (m_store->retrieve(preDh, m_dataHeaderKey).isSuccess()) {
         if (m_store->removeDataAndProxy(preDh).isFailure()) {
            ATH_MSG_ERROR("Unable to get proxy for the DataHeader with key " << m_dataHeaderKey);
            return(StatusCode::FAILURE);
         }
         ATH_MSG_DEBUG("Released DataHeader with key " << m_dataHeaderKey);
      }
   }
 
   // Create new DataHeader
   m_dataHeader = new DataHeader();
   m_dataHeader->setProcessTag(m_processTag);
 
   // Retrieve all existing DataHeaders from StoreGate
   const DataHeader* dh = nullptr;
   std::vector<std::string> dhKeys;
   m_store->keys<DataHeader>(dhKeys);
   //construct string out of loop
   const std::string boolTypeStr{"bool"};
   for (const std::string& dhKey : dhKeys) {
      bool primaryDH = false;
      if (!m_store->transientContains<DataHeader>(dhKey)) {
         if (dhKey == "EventSelector") primaryDH = true;
         ATH_MSG_DEBUG("No transientContains DataHeader with key " << dhKey);
      }
      if (m_store->retrieve(dh, dhKey).isFailure()) {
         ATH_MSG_DEBUG("Unable to retrieve the DataHeader with key " << dhKey);
      }
      if (dh->isInput() || hasInputAlias(*m_store->proxy(dh)) || primaryDH) {
         propagateProvenance( *dh );
      }
   }
 
   // Attach the attribute list to the DataHeader if requested
   if (!m_attrListKey.key().empty() && m_store->storeID() == StoreID::EVENT_STORE) {
      auto attrListHandle = SG::makeHandle(m_attrListKey);
      if (!attrListHandle.isValid()) {
         ATH_MSG_WARNING("Unable to retrieve AttributeList with key " << m_attrListKey);
      } else {
         m_dataHeader->setAttributeList(attrListHandle.cptr());
         if (m_extend) {  // Add streaming decisions
            ATH_MSG_DEBUG("Adding stream decisions to " << m_attrListWrite);
            // Look for attribute list created for mini-EventInfo
            const AthenaAttributeList* attlist(attrListHandle.cptr());
 
            // Build new attribute list for modification
            AthenaAttributeList* newone = new AthenaAttributeList(attlist->specification());
            newone->copyData(*attlist);
 
            // Now loop over stream definitions and add decisions
            auto streams = m_decSvc->getStreams();
            for (auto it  = streams.begin();
                      it != streams.end(); ++it) {
               newone->extend(*it,boolTypeStr);
               (*newone)[*it].data<bool>() = m_decSvc->isEventAccepted(*it,Gaudi::Hive::currentContext());
               ATH_MSG_DEBUG("Added stream decision for " << *it << " to " << m_attrListKey);
            }
            // record new attribute list with old key + suffix
            const AthenaAttributeList* attrList2 = nullptr;
            if (!m_store->contains<AthenaAttributeList>(m_attrListWrite)) {
               if (m_store->record(newone,m_attrListWrite).isFailure()) {
                  ATH_MSG_ERROR("Unable to record att list " << m_attrListWrite);
               }
            } else {
               ATH_MSG_DEBUG("Decisions already added by a different stream");
            }
            if (m_store->retrieve(attrList2,m_attrListWrite).isFailure()) {
               ATH_MSG_ERROR("Unable to record att list " << m_attrListWrite);
            } else {
               m_dataHeader->setAttributeList(attrList2);
            }
         }    // list extend check
      }       // list retrieve check
   }          // list property check
 
   // Record DataHeader in StoreGate
   SG::WriteHandle<DataHeader> wh(m_dataHeaderKey, m_store->name());
   if (wh.record(std::unique_ptr<DataHeader>(m_dataHeader)).isFailure()) {
      ATH_MSG_ERROR("Unable to record DataHeader with key " << m_dataHeaderKey);
      return(StatusCode::FAILURE);
   } else {
      ATH_MSG_DEBUG("Recorded DataHeader with key " << m_dataHeaderKey);
   }
   m_dataHeader->setStatus(DataHeader::Output);
   // Set flag that connection is open
   m_connectionOpen = true;
   return(StatusCode::SUCCESS);
}
 
//__________________________________________________________________________
void AthenaOutputStreamTool::propagateProvenance( const DataHeader& src_dh )
{
   // keep track of provenance entries inserted into the new DataHeader
   std::set<std::string> insertedTags{};
   // Add DataHeader token to the new DataHeader
   if (m_extendProvenanceRecord) {
      std::string pTag;
      std::unique_ptr<SG::TransientAddress> dhTransAddr;
      for (const DataHeaderElement& dhe : src_dh) {
         if (dhe.getPrimaryClassID() == ClassID_traits<DataHeader>::ID()) {
            pTag = dhe.getKey();
            dhTransAddr.reset( dhe.getAddress( m_conversionSvc->repSvcType() ) );
         }
      }
      // Update dhTransAddr to handle fast merged files.
      if( auto dhProxy=m_store->proxy(&src_dh); dhProxy && dhProxy->address() ) {
         DataHeaderElement dhe(dhProxy, dhProxy->address(), pTag);
         m_dataHeader->insertProvenance(dhe);
         insertedTags.insert(std::move(pTag));
      }
      else if( dhTransAddr ) {
         DataHeaderElement dhe(dhTransAddr.get(), dhTransAddr->address(), pTag);
         m_dataHeader->insertProvenance(dhe);
         insertedTags.insert(std::move(pTag));
      }
   }
 
   // empty regexpr means do not keep any provenance
   if( !m_keepProvenancesStr.empty() ) {
      // Each stream tag is written only once in the provenance record
      // In files where there are multiple entries per stream tag
      // the record is in reverse, i.e., the latest appears first.
      // Therefore, only keep the first entry if there are multiple
      // matches so that we retain the latest one.
      for(auto iter=src_dh.beginProvenance(), iEnd=src_dh.endProvenance(); iter != iEnd; ++iter) {
         const auto & currentKey = (*iter).getKey();
         if( insertedTags.insert(currentKey).second ) {
            // first prov with that tag. Now check if we want to keep that tag
            bool keep = false;
            auto it =  m_keepProvenanceMatch.find( currentKey );
            if( it != m_keepProvenanceMatch.end() ) {
               keep = it->second;
            } else {
               keep = std::regex_search(currentKey, m_keepProvenancesRE);
               m_keepProvenanceMatch[currentKey] = keep;
            }
            if( keep ) {
               m_dataHeader->insertProvenance(*iter);
            }
         }
      }
   }
}
   
//__________________________________________________________________________
StatusCode AthenaOutputStreamTool::commitOutput(bool doCommit) {
   ATH_MSG_DEBUG("In commitOutput");
   // Connect the output file to the service
   if (m_conversionSvc->commitOutput(m_outputName.value(), doCommit).isFailure()) {
      ATH_MSG_ERROR("Unable to commit output " << m_outputName.value());
      return(StatusCode::FAILURE);
   }
   // Set flag that connection is closed
   m_connectionOpen = false;
   return(StatusCode::SUCCESS);
}
//__________________________________________________________________________
StatusCode AthenaOutputStreamTool::finalizeOutput() {
   AthCnvSvc* athConversionSvc = dynamic_cast<AthCnvSvc*>(m_conversionSvc.get());
   if (athConversionSvc != 0) {
      if (athConversionSvc->disconnectOutput(m_outputName.value()).isFailure()) {
         ATH_MSG_ERROR("Unable to finalize output " << m_outputName.value());
         return(StatusCode::FAILURE);
      }
   }
   return(StatusCode::SUCCESS);
}
//__________________________________________________________________________
StatusCode AthenaOutputStreamTool::streamObjects(const TypeKeyPairs& typeKeys, const std::string& outputName) {
   ATH_MSG_DEBUG("In streamObjects");
   // Check that a connection has been opened
   if (!m_connectionOpen) {
      ATH_MSG_ERROR("Connection NOT open. Please open a connection before streaming out objects.");
      return(StatusCode::FAILURE);
   }
   // Use arg if not empty, save the output name
   if (!outputName.empty()) {
      m_outputName.setValue(outputName);
   }
   if (m_outputName.value().empty()) {
      ATH_MSG_ERROR("No OutputName provided");
      return(StatusCode::FAILURE);
   }
   // Now iterate over the type/key pairs and stream out each object
   std::vector<DataObject*> dataObjects;
   for (TypeKeyPairs::const_iterator first = typeKeys.begin(), last = typeKeys.end();
       first != last; ++first) {
      const std::string& type = (*first).first;
      const std::string& key  = (*first).second;
      // Find the clid for type name from the CLIDSvc
      CLID clid;
      if (m_clidSvc->getIDOfTypeName(type, clid).isFailure()) {
         ATH_MSG_ERROR("Could not get clid for typeName " << type);
         return(StatusCode::FAILURE);
      }
      DataObject* dObj = 0;
      // Two options: no key or explicit key
      if (key.empty()) {
         ATH_MSG_DEBUG("Get data object with no key");
         // Get DataObject without key
         dObj = m_store->accessData(clid);
      } else {
         ATH_MSG_DEBUG("Get data object with key");
         // Get DataObjects with key
         dObj = m_store->accessData(clid, key);
      }
      if (dObj == 0) {
         // No object - print warning and return
         ATH_MSG_DEBUG("No object found for type " << type << " key " << key);
         return(StatusCode::SUCCESS);
      } else {
         ATH_MSG_DEBUG("Found object for type " << type << " key " << key);
      }
      // Save the dObj
      dataObjects.push_back(dObj);
   }
   // Stream out objects
   if (dataObjects.size() == 0) {
      ATH_MSG_DEBUG("No data objects found");
      return(StatusCode::SUCCESS);
   }
   StatusCode status = streamObjects(dataObjects, m_outputName.value());
   if (!status.isSuccess()) {
      ATH_MSG_ERROR("Could not stream out objects");
      return(status);
   }
   return(StatusCode::SUCCESS);
}
//__________________________________________________________________________
StatusCode AthenaOutputStreamTool::streamObjects(const DataObjectVec& dataObjects, const std::string& outputName) {
   // Check that a connection has been opened
   if (!m_connectionOpen) {
      ATH_MSG_ERROR("Connection NOT open. Please open a connection before streaming out objects.");
      return(StatusCode::FAILURE);
   }
   // Connect the output file to the service
   std::string outputConnectionString = outputName;
   const std::string defaultMetaDataString = "[OutputCollection=MetaDataHdr][PoolContainerPrefix=MetaData]";
   if (std::string::size_type mpos = outputConnectionString.find(defaultMetaDataString); mpos!=std::string::npos) {
     // If we're in here we're writing MetaData
     // Now let's see if we should be overwriting the MetaData attributes
     // For the time-being this happens when we're writing MetaData in the augmentation mode
     if (!m_metaDataOutputAttributes.empty()) {
       // Note: This won't work quite right if only one attribute is set though!
       outputConnectionString.replace(mpos, defaultMetaDataString.length(), m_metaDataOutputAttributes);
     }
   }
   for (std::string::size_type pos = m_outputAttributes.find('['); pos != std::string::npos; pos = m_outputAttributes.find('[', ++pos)) {
      if (outputConnectionString.find(m_outputAttributes.substr(pos, m_outputAttributes.find('=', pos) + 1 - pos)) == std::string::npos) {
         outputConnectionString += m_outputAttributes.substr(pos, m_outputAttributes.find(']', pos) + 1 - pos);
      }
   }
 
   // Check that the DataHeader is still valid
   DataObject* dataHeaderObj = m_store->accessData(ClassID_traits<DataHeader>::ID(), m_dataHeaderKey);
   std::map<DataObject*, IOpaqueAddress*> written;
   for (DataObject* dobj : dataObjects) {
      // Do not write the DataHeader via the explicit list
      if (dobj->clID() == ClassID_traits<DataHeader>::ID()) {
         ATH_MSG_DEBUG("Explicit request to write DataHeader: " << dobj->name() << " - skipping it.");
      // Do not stream out same object twice
      } else if (written.find(dobj) != written.end()) {
         // Print warning and skip
         ATH_MSG_DEBUG("Trying to write DataObject twice (clid/key): " << dobj->clID() << " " << dobj->name());
         ATH_MSG_DEBUG("    Skipping this one.");
      } else {
         // Write object
         IOpaqueAddress* addr = new TokenAddress(0, dobj->clID(), outputConnectionString);
         addr->addRef();
         if (m_conversionSvc->createRep(dobj, addr).isSuccess()) {
            written.insert(std::pair<DataObject*, IOpaqueAddress*>(dobj, addr));
         } else {
            ATH_MSG_ERROR("Could not create Rep for DataObject (clid/key):" << dobj->clID() << " " << dobj->name());
            return(StatusCode::FAILURE);
         }
      }
   }
   // End of loop over DataObjects, write DataHeader
   if ((m_conversionSvc.type() == "AthenaPoolCnvSvc" || m_conversionSvc.type() == "AthenaPoolSharedIOCnvSvc") && dataHeaderObj != nullptr) {
      IOpaqueAddress* addr = new TokenAddress(0, dataHeaderObj->clID(), outputConnectionString);
      addr->addRef();
      if (m_conversionSvc->createRep(dataHeaderObj, addr).isSuccess()) {
         written.insert(std::pair<DataObject*, IOpaqueAddress*>(dataHeaderObj, addr));
      } else {
         ATH_MSG_ERROR("Could not create Rep for DataHeader");
         return(StatusCode::FAILURE);
      }
   }
   for (DataObject* dobj : dataObjects) {
      // call fillRepRefs of persistency service
      SG::DataProxy* proxy = dynamic_cast<SG::DataProxy*>(dobj->registry());
      if (proxy != nullptr && written.find(dobj) != written.end()) {
         IOpaqueAddress* addr(written.find(dobj)->second);
         if ((m_conversionSvc->fillRepRefs(addr, dobj)).isSuccess()) {
            if (dobj->clID() != 1 || addr->par()[0] != "\n") {
               if (dobj->clID() != ClassID_traits<DataHeader>::ID()) {
                  m_dataHeader->insert(proxy, addr);
               } else {
                  m_dataHeader->insert(proxy, addr, m_processTag);
               }
               if (proxy->address() == nullptr) {
                  proxy->setAddress(addr);
               }
               addr->release();
            }
         } else {
            ATH_MSG_ERROR("Could not fill Object Refs for DataObject (clid/key):" << dobj->clID() << " " << dobj->name());
            return(StatusCode::FAILURE);
         }
      } else {
         ATH_MSG_WARNING("Could cast DataObject " << dobj->clID() << " " << dobj->name());
      }
   }
   m_dataHeader->addHash(&*m_store);
   if ((m_conversionSvc.type() == "AthenaPoolCnvSvc" || m_conversionSvc.type() == "AthenaPoolSharedIOCnvSvc") && dataHeaderObj != nullptr) {
      // End of DataObjects, fill refs for DataHeader
      SG::DataProxy* proxy = dynamic_cast<SG::DataProxy*>(dataHeaderObj->registry());
      if (proxy != nullptr && written.find(dataHeaderObj) != written.end()) {
         IOpaqueAddress* addr(written.find(dataHeaderObj)->second);
         if ((m_conversionSvc->fillRepRefs(addr, dataHeaderObj)).isSuccess()) {
            if (dataHeaderObj->clID() != 1 || addr->par()[0] != "\n") {
               if (dataHeaderObj->clID() != ClassID_traits<DataHeader>::ID()) {
                  m_dataHeader->insert(proxy, addr);
               } else {
                  m_dataHeader->insert(proxy, addr, m_processTag);
               }
               addr->release();
            }
         } else {
            ATH_MSG_ERROR("Could not fill Object Refs for DataHeader");
            return(StatusCode::FAILURE);
         }
      } else {
         ATH_MSG_ERROR("Could not cast DataHeader");
         return(StatusCode::FAILURE);
      }
   }
   return(StatusCode::SUCCESS);
}
//__________________________________________________________________________
StatusCode AthenaOutputStreamTool::getInputItemList(SG::IFolder* p2BWrittenFromTool) {
   const std::string hltKey = "HLTAutoKey";
   SG::ConstIterator<DataHeader> beg;
   SG::ConstIterator<DataHeader> ending;
   if (m_store->retrieve(beg, ending).isFailure() || beg == ending) {
      ATH_MSG_DEBUG("No DataHeaders present in StoreGate");
   } else {
      for ( ; beg != ending; ++beg) {
         if (m_store->transientContains<DataHeader>(beg.key()) && beg->isInput()) {
            for (std::vector<DataHeaderElement>::const_iterator it = beg->begin(), itLast = beg->end();
                it != itLast; ++it) {
               // Only insert the primary clid, not the ones for the symlinks!
               CLID clid = it->getPrimaryClassID();
               if (clid != ClassID_traits<DataHeader>::ID()) {
                  //check the typename is known ... we make an exception if the key contains 'Aux.' ... aux containers may not have their keys known yet in some cases
                  //see https://its.cern.ch/jira/browse/ATLASG-59 for the solution
                  std::string typeName;
                  if (m_clidSvc->getTypeNameOfID(clid, typeName).isFailure() && it->getKey().find("Aux.") == std::string::npos) {
                     if (m_skippedItems.find(it->getKey()) == m_skippedItems.end()) {
                        ATH_MSG_WARNING("Skipping " << it->getKey() << " with unknown clid " << clid << " . Further warnings for this item are suppressed" );
                        m_skippedItems.insert(it->getKey());
                     }
                     continue;
                  }
                  ATH_MSG_DEBUG("Adding " << typeName << "#" << it->getKey() << " (clid " << clid << ") to itemlist");
                  const std::string keyName = it->getKey();
                  if (keyName.size() > 10 && keyName.compare(0, 10,hltKey)==0) {
                     p2BWrittenFromTool->add(clid, hltKey + "*").ignore();
                  } else if (keyName.size() > 10 && keyName.compare(keyName.size() - 10, 10, hltKey)==0) {
                     p2BWrittenFromTool->add(clid, "*" + hltKey).ignore();
                  } else {
                     p2BWrittenFromTool->add(clid, keyName).ignore();
                  }
               }
            }
         }
      }
   }
   ATH_MSG_DEBUG("Adding DataHeader for stream " << name());
   return(StatusCode::SUCCESS);
}