CbmDeviceMcbmEventSink.cxx

Go to the documentation of this file.

/* Copyright (C) 2020-2021 Facility for Antiproton and Ion Research in Europe, Darmstadt
   SPDX-License-Identifier: GPL-3.0-only
   Authors: Pierre-Alain Loizeau [committer] */
 
#include "CbmDeviceMcbmEventSink.h"
 
 
#include "CbmEvent.h"
#include "CbmFlesCanvasTools.h"
#include "CbmMQDefs.h"
 
#include "TimesliceMetaData.h"
 
#include "FairMQLogger.h"
#include "FairMQProgOptions.h"  // device->fConfig
#include "FairParGenericSet.h"
#include "FairRootFileSink.h"
#include "FairRootManager.h"
#include "FairRunOnline.h"
 
#include "BoostSerializer.h"
 
#include "RootSerializer.h"
 
#include "TCanvas.h"
#include "TFile.h"
#include "TH1.h"
#include "TList.h"
#include "TNamed.h"
 
#include <boost/archive/binary_iarchive.hpp>
#include <boost/serialization/utility.hpp>
 
#include <thread>  // this_thread::sleep_for
 
#include <array>
#include <iomanip>
#include <stdexcept>
#include <string>
struct InitTaskError : std::runtime_error {
  using std::runtime_error::runtime_error;
};
 
using namespace std;
 
//Bool_t bMcbm2018MonitorTaskBmonResetHistos = kFALSE;
 
CbmDeviceMcbmEventSink::CbmDeviceMcbmEventSink() {}
 
void CbmDeviceMcbmEventSink::InitTask()
try {
  LOG(info) << "Init options for CbmDeviceMcbmEventSink.";
 
  fsOutputFileName = fConfig->GetValue<std::string>("OutFileName");
 
  fsChannelNameDataInput = fConfig->GetValue<std::string>("EvtNameIn");
  fsAllowedChannels[0]   = fsChannelNameDataInput;
 
  fbFillHistos              = fConfig->GetValue<bool>("FillHistos");
  fsChannelNameHistosInput  = fConfig->GetValue<std::string>("ChNameIn");
  fsChannelNameHistosConfig = fConfig->GetValue<std::string>("ChNameHistCfg");
  fsChannelNameCanvasConfig = fConfig->GetValue<std::string>("ChNameCanvCfg");
  fuPublishFreqTs           = fConfig->GetValue<uint32_t>("PubFreqTs");
  fdMinPublishTime          = fConfig->GetValue<double_t>("PubTimeMin");
  fdMaxPublishTime          = fConfig->GetValue<double_t>("PubTimeMax");
 
  OnData(fsChannelNameMissedTs, &CbmDeviceMcbmEventSink::HandleMissTsData);
 
  OnData(fsChannelNameCommands, &CbmDeviceMcbmEventSink::HandleCommand);
 
  // Get the information about created channels from the device
  // Check if the defined channels from the topology (by name)
  // are in the list of channels which are possible/allowed
  // for the device
  // The idea is to check at initilization if the devices are
  // properly connected. For the time beeing this is done with a
  // nameing convention. It is not avoided that someone sends other
  // data on this channel.
  //logger::SetLogLevel("INFO");
  int noChannel = fChannels.size();
  LOG(info) << "Number of defined channels: " << noChannel;
  for (auto const& entry : fChannels) {
    LOG(info) << "Channel name: " << entry.first;
    if (std::string::npos != entry.first.find(fsChannelNameDataInput)) {
      if (!IsChannelNameAllowed(entry.first)) throw InitTaskError("Channel name does not match.");
      OnData(entry.first, &CbmDeviceMcbmEventSink::HandleData);
    }  // if( entry.first.find( "ts" )
  }    // for( auto const &entry : fChannels )
 
  //   InitContainers();
 
  fvDigiBmon = new std::vector<CbmTofDigi>();
  fvDigiSts  = new std::vector<CbmStsDigi>();
  fvDigiMuch = new std::vector<CbmMuchBeamTimeDigi>();
  fvDigiTrd  = new std::vector<CbmTrdDigi>();
  fvDigiTof  = new std::vector<CbmTofDigi>();
  fvDigiRich = new std::vector<CbmRichDigi>();
  fvDigiPsd  = new std::vector<CbmPsdDigi>();
 
  fTimeSliceMetaDataArray = new TClonesArray("TimesliceMetaData", 1);
  if (NULL == fTimeSliceMetaDataArray) {
    throw InitTaskError("Failed creating the TS meta data TClonesarray ");
  }  // if( NULL == fTimeSliceMetaDataArray )
  fEventsArray = new TClonesArray("CbmEvent", 500);
  if (NULL == fEventsArray) {
    throw InitTaskError("Failed creating the Events TClonesarray ");
  }  // if( NULL == fEventsArray )
 
  if ("" != fsOutputFileName) {
    fpRun         = new FairRunOnline();
    fpFairRootMgr = FairRootManager::Instance();
    fpFairRootMgr->SetSink(new FairRootFileSink(fsOutputFileName));
    if (nullptr == fpFairRootMgr->GetOutFile()) {
      throw InitTaskError("Could not open root file");
    }  // if( nullptr == fpFairRootMgr->GetOutFile() )
  }    // if( "" != fsOutputFileName )
  else {
    throw InitTaskError("Empty output filename!");
  }  // else of if( "" != fsOutputFileName )
 
  LOG(info) << "Init Root Output to " << fsOutputFileName;
 
  fpFairRootMgr->InitSink();
  //      fEvtHeader = new FairEventHeader();
  //      fEvtHeader->SetRunId(iRunId);
  //      rootMgr->Register("EventHeader.", "Event", fEvtHeader, kTRUE);
  //      rootMgr->FillEventHeader(fEvtHeader);
 
  fpFairRootMgr->Register("TimesliceMetaData", "TS Meta Data", fTimeSliceMetaDataArray, kTRUE);
  fpFairRootMgr->RegisterAny("BmonDigi", fvDigiBmon, kTRUE);
  fpFairRootMgr->RegisterAny("StsDigi", fvDigiSts, kTRUE);
  fpFairRootMgr->RegisterAny("MuchBeamTimeDigi", fvDigiMuch, kTRUE);
  fpFairRootMgr->RegisterAny("TrdDigi", fvDigiTrd, kTRUE);
  fpFairRootMgr->RegisterAny("TofDigi", fvDigiTof, kTRUE);
  fpFairRootMgr->RegisterAny("RichDigi", fvDigiRich, kTRUE);
  fpFairRootMgr->RegisterAny("PsdDigi", fvDigiPsd, kTRUE);
  fpFairRootMgr->Register("CbmEvent", "Cbm Event", fEventsArray, kTRUE);
  /*
   TTree* outTree =new TTree(FairRootManager::GetTreeName(), "/cbmout", 99);
   LOG(info) << "define Tree " << outTree->GetName();
 
   fpFairRootMgr->GetSink()->SetOutTree(outTree);
*/
  fpFairRootMgr->WriteFolder();
 
  LOG(info) << "Initialized outTree with rootMgr at " << fpFairRootMgr;
 
  if (kTRUE == fbFillHistos) {
    /*
      std::vector< std::pair< TNamed *, std::string > > vHistos = fpAlgo->GetHistoVector();
      std::vector< std::pair< TCanvas *, std::string > > vCanvases = fpAlgo->GetCanvasVector();
 
      for( UInt_t uHisto = 0; uHisto < vHistos.size(); ++uHisto )
      {
//         LOG(info) << "Registering  " << vHistos[ uHisto ].first->GetName()
//                   << " in " << vHistos[ uHisto ].second.data()
//                   ;
         fArrayHisto.Add( vHistos[ uHisto ].first );
         std::pair< std::string, std::string > psHistoConfig( vHistos[ uHisto ].first->GetName(),
                                                              vHistos[ uHisto ].second );
         fvpsHistosFolder.push_back( psHistoConfig );
 
         FairMQMessagePtr messageHist( NewMessage() );
//         Serialize< BoostSerializer < std::pair< std::string, std::string > > >( *messageHist, psHistoConfig );
         BoostSerializer < std::pair< std::string, std::string > >.Serialize( *messageHist, psHistoConfig );
 
         if( Send( messageHist, fsChannelNameHistosConfig ) < 0 )
         {
            throw InitTaskError( "Problem sending histo config" );
         } // if( Send( messageHist, fsChannelNameHistosConfig ) < 0 )
 
         LOG(info) << "Config of hist  " << psHistoConfig.first.data()
                   << " in folder " << psHistoConfig.second.data() ;
      } // for( UInt_t uHisto = 0; uHisto < vHistos.size(); ++uHisto )
 
      for( UInt_t uCanv = 0; uCanv < vCanvases.size(); ++uCanv )
      {
//         LOG(info) << "Registering  " << vCanvases[ uCanv ].first->GetName()
//                   << " in " << vCanvases[ uCanv ].second.data();
         std::string sCanvName = (vCanvases[ uCanv ].first)->GetName();
         std::string sCanvConf = GenerateCanvasConfigString( vCanvases[ uCanv ].first );
 
         std::pair< std::string, std::string > psCanvConfig( sCanvName, sCanvConf );
 
         fvpsCanvasConfig.push_back( psCanvConfig );
 
         FairMQMessagePtr messageCan( NewMessage() );
//         Serialize< BoostSerializer < std::pair< std::string, std::string > > >( *messageCan, psCanvConfig );
         BoostSerializer < std::pair< std::string, std::string > >.Serialize( *messageCan, psCanvConfig );
 
         if( Send( messageCan, fsChannelNameCanvasConfig ) < 0 )
         {
            throw InitTaskError( "Problem sending canvas config" );
         } // if( Send( messageCan, fsChannelNameCanvasConfig ) < 0 )
 
         LOG(info) << "Config string of Canvas  " << psCanvConfig.first.data()
                   << " is " << psCanvConfig.second.data() ;
      } //  for( UInt_t uCanv = 0; uCanv < vCanvases.size(); ++uCanv )
*/
  }  // if( kTRUE == fbFillHistos )
}
catch (InitTaskError& e) {
  LOG(error) << e.what();
  // Wrapper defined in CbmMQDefs.h to support different FairMQ versions
  cbm::mq::ChangeState(this, cbm::mq::Transition::ErrorFound);
}
 
bool CbmDeviceMcbmEventSink::IsChannelNameAllowed(std::string channelName)
{
  for (auto const& entry : fsAllowedChannels) {
    std::size_t pos1 = channelName.find(entry);
    if (pos1 != std::string::npos) {
      const vector<std::string>::const_iterator pos =
        std::find(fsAllowedChannels.begin(), fsAllowedChannels.end(), entry);
      const vector<std::string>::size_type idx = pos - fsAllowedChannels.begin();
      LOG(info) << "Found " << entry << " in " << channelName;
      LOG(info) << "Channel name " << channelName << " found in list of allowed channel names at position " << idx;
      return true;
    }  // if (pos1!=std::string::npos)
  }    // for(auto const &entry : fsAllowedChannels)
  LOG(info) << "Channel name " << channelName << " not found in list of allowed channel names.";
  LOG(error) << "Stop device.";
  return false;
}
/*
Bool_t CbmDeviceMcbmEventSink::InitContainers()
{
   LOG(info) << "Init parameter containers for CbmDeviceMcbmEventSink.";
 
   if( kFALSE == InitParameters( fpAlgo ->GetParList() ) )
      return kFALSE;
 
   fpAlgo ->SetIgnoreOverlapMs( fbIgnoreOverlapMs );
 
   Bool_t initOK = fpAlgo->InitContainers();
 
//   Bool_t initOK = fMonitorAlgo->ReInitContainers();
 
  return initOK;
}
 
Bool_t CbmDeviceMcbmEventSink::InitParameters( TList* fParCList )
{
   for( int iparC = 0; iparC < fParCList->GetEntries(); iparC++ )
   {
      FairParGenericSet* tempObj = (FairParGenericSet*)( fParCList->At( iparC ) );
      fParCList->Remove( tempObj );
      std::string paramName{ tempObj->GetName() };
      // NewSimpleMessage creates a copy of the data and takes care of its destruction (after the transfer takes place).
      // Should only be used for small data because of the cost of an additional copy
 
      // Her must come the proper Runid
      std::string message = paramName + ",111";
      LOG(info) << "Requesting parameter container " << paramName << ", sending message: " << message;
 
      FairMQMessagePtr req( NewSimpleMessage(message) );
      FairMQMessagePtr rep( NewMessage() );
 
      FairParGenericSet* newObj = nullptr;
 
      if( Send(req, "parameters") > 0 )
      {
         if( Receive( rep, "parameters" ) >= 0)
         {
            if( 0 !=  rep->GetSize() )
            {
               CbmMqTMessage tmsg( rep->GetData(), rep->GetSize() );
               newObj = static_cast< FairParGenericSet* >( tmsg.ReadObject( tmsg.GetClass() ) );
               LOG( info ) << "Received unpack parameter from the server:";
               newObj->print();
            } // if( 0 !=  rep->GetSize() )
               else
               {
                  LOG( error ) << "Received empty reply. Parameter not available";
                  return kFALSE;
               } // else of if( 0 !=  rep->GetSize() )
         } // if( Receive( rep, "parameters" ) >= 0)
      } // if( Send(req, "parameters") > 0 )
      fParCList->AddAt( newObj, iparC );
      delete tempObj;
   } // for( int iparC = 0; iparC < fParCList->GetEntries(); iparC++ )
 
   return kTRUE;
}
*/
//--------------------------------------------------------------------//
// handler is called whenever a message arrives on fsChannelNameMissedTs, with a reference to the message and a sub-channel index (here 0)
bool CbmDeviceMcbmEventSink::HandleMissTsData(FairMQMessagePtr& msg, int /*index*/)
{
  std::vector<uint64_t> vIndices;
  std::string msgStrMissTs(static_cast<char*>(msg->GetData()), msg->GetSize());
  std::istringstream issMissTs(msgStrMissTs);
  boost::archive::binary_iarchive inputArchiveMissTs(issMissTs);
  inputArchiveMissTs >> vIndices;
 
  fvulMissedTsIndices.insert(fvulMissedTsIndices.end(), vIndices.begin(), vIndices.end());
 
  CheckTsQueues();
 
  return true;
}
//--------------------------------------------------------------------//
// handler is called whenever a message arrives on "data", with a reference to the message and a sub-channel index (here 0)
bool CbmDeviceMcbmEventSink::HandleData(FairMQParts& parts, int /*index*/)
{
  fulNumMessages++;
  LOG(debug) << "Received message number " << fulNumMessages << " with " << parts.Size() << " parts"
             << ", size0: " << parts.At(0)->GetSize();
 
  if (0 == fulNumMessages % 10000) LOG(info) << "Received " << fulNumMessages << " messages";
 
  uint32_t uPartIdx = 0;
  /*
  std::string msgStrTsMeta( static_cast< char * >( parts.At( uPartIdx )->GetData() ),
                            ( parts.At( uPartIdx ) )->GetSize() );
  std::istringstream issTsMeta(msgStrTsMeta);
  boost::archive::binary_iarchive inputArchiveTsMeta(issTsMeta);
  inputArchiveTsMeta >> (*fTsMetaData);
  ++uPartIdx;
*/
  //  Deserialize<RootSerializer>(*parts.At(uPartIdx), fTsMetaData);
  RootSerializer().Deserialize(*parts.At(uPartIdx), fTsMetaData);
  LOG(debug) << "TS metadata extracted";
 
  if (fuPrevTsIndex + 1 == fTsMetaData->GetIndex()
      || (0 == fuPrevTsIndex && 0 == fulTsCounter && 0 == fTsMetaData->GetIndex())) {
    LOG(debug) << "TS direct to dump";
    PrepareTreeEntry(parts);
    DumpTreeEntry();
    fuPrevTsIndex = fTsMetaData->GetIndex();
    fulTsCounter++;
  }  // if( fuPrevTsIndex + 1 == fTsMetaData->GetIndex() ||  ( 0 == fuPrevTsIndex && 0 == fulTsCounter ) )
  else {
    LOG(debug) << "TS direct to storage";
    fmFullTsStorage.emplace_hint(fmFullTsStorage.end(), std::pair<uint64_t, CbmUnpackedTimeslice>(
                                                          fTsMetaData->GetIndex(), CbmUnpackedTimeslice(parts)));
  }  // else of if( fuPrevTsIndex + 1 == fTsMetaData->GetIndex() ||  ( 0 == fuPrevTsIndex && 0 == fulTsCounter && 0 == fTsMetaData->GetIndex() )
  LOG(debug) << "TS metadata checked";
 
  //   delete fTsMetaData;
 
  CheckTsQueues();
  LOG(debug) << "TS queues checked";
 
  if (kTRUE == fbFillHistos) {
    std::chrono::system_clock::time_point currentTime = std::chrono::system_clock::now();
    std::chrono::duration<double_t> elapsedSeconds    = currentTime - fLastPublishTime;
    if ((fdMaxPublishTime < elapsedSeconds.count())
        || (0 == fulNumMessages % fuPublishFreqTs && fdMinPublishTime < elapsedSeconds.count())) {
      SendHistograms();
      fLastPublishTime = std::chrono::system_clock::now();
    }  // if( ( fdMaxPublishTime < elapsedSeconds.count() ) || ( 0 == fulNumMessages % fuPublishFreqTs && fdMinPublishTime < elapsedSeconds.count() ) )
  }    // if( kTRUE == fbFillHistos )
 
  return true;
}
//--------------------------------------------------------------------//
bool CbmDeviceMcbmEventSink::HandleCommand(FairMQMessagePtr& msg, int /*index*/)
{
  /*
   std::string sCommand( static_cast< char * >( msg->GetData() ),
                          msg->GetSize() );
*/
  std::string sCommand;
  std::string msgStrCmd(static_cast<char*>(msg->GetData()), msg->GetSize());
  std::istringstream issCmd(msgStrCmd);
  boost::archive::binary_iarchive inputArchiveCmd(issCmd);
  inputArchiveCmd >> sCommand;
 
  std::string sCmdTag = sCommand;
  size_t charPosDel   = sCommand.find(' ');
  if (std::string::npos != charPosDel) {
    sCmdTag = sCommand.substr(0, charPosDel);
  }  // if( std::string::npos != charPosDel )
 
  if ("EOF" == sCmdTag) {
    fbReceivedEof = true;
 
    if (std::string::npos == charPosDel) {
      LOG(fatal) << "CbmDeviceMcbmEventSink::HandleCommand => "
                 << "Incomplete EOF command received: " << sCommand;
      return false;
    }  // if( std::string::npos == charPosDel )
    charPosDel++;
    std::string sNext = sCommand.substr(charPosDel);
    charPosDel        = sNext.find(' ');
 
    if (std::string::npos == charPosDel) {
      LOG(fatal) << "CbmDeviceMcbmEventSink::HandleCommand => "
                 << "Incomplete EOF command received: " << sCommand;
      return false;
    }  // if( std::string::npos == charPosDel )
    fuLastTsIndex = std::stoul(sNext.substr(0, charPosDel));
    charPosDel++;
    fuTotalTsCount = std::stoul(sNext.substr(charPosDel));
 
    LOG(info) << "CbmDeviceMcbmEventSink::HandleCommand => "
              << "Received EOF command with final TS index " << fuLastTsIndex << " and total nb TS " << fuTotalTsCount;
    if (fuPrevTsIndex == fuLastTsIndex && fulTsCounter == fuTotalTsCount) {
      LOG(info) << "CbmDeviceMcbmEventSink::HandleCommand => "
                << "Found final TS index " << fuLastTsIndex << " and total nb TS " << fuTotalTsCount;
      Finish();
    }  // if( fuPrevTsIndex == fuLastTsIndex && fulTsCounter == fuTotalTsCount )
  }    // if( "EOF" == sCmdTag )
  else if ("STOP" == sCmdTag) {
    Finish();
  }  // else if( "STOP" == sCmdTag )
  else {
    LOG(warning) << "Unknown command received: " << sCmdTag << " => will be ignored!";
  }  // else if command not recognized
 
  return true;
}
//--------------------------------------------------------------------//
void CbmDeviceMcbmEventSink::CheckTsQueues()
{
  bool bHoleFoundInBothQueues = false;
 
  std::map<uint64_t, CbmUnpackedTimeslice>::iterator itFullTs = fmFullTsStorage.begin();
  std::vector<uint64_t>::iterator itMissTs                    = fvulMissedTsIndices.begin();
 
  while (!bHoleFoundInBothQueues) {
    if (fmFullTsStorage.end() != itFullTs && fuPrevTsIndex + 1 == (*itFullTs).first) {
      PrepareTreeEntry((*itFullTs).second);
      DumpTreeEntry();
 
      fuPrevTsIndex = (*itFullTs).first;
      fulTsCounter++;
 
      ++itFullTs;
      continue;
    }  // if( fmFullTsStorage.end() != itFullTs && fuPrevTsIndex + 1 == (*itFullTs).first() )
    if (fvulMissedTsIndices.end() != itMissTs && fuPrevTsIndex + 1 == (*itMissTs)) {
      new ((*fTimeSliceMetaDataArray)[fTimeSliceMetaDataArray->GetEntriesFast()])
        TimesliceMetaData(0, 0, 0, (*itMissTs));
 
      DumpTreeEntry();
 
      fuPrevTsIndex = (*itMissTs);
      fulMissedTsCounter++;
 
      ++itMissTs;
      continue;
    }  // if( fvulMissedTsIndices.end() != itMissTs && fuPrevTsIndex + 1 == (*itMissTs ) )
 
    bHoleFoundInBothQueues = true;
  }  // while( !bHoleFoundInBothQueues )
 
  fmFullTsStorage.erase(fmFullTsStorage.begin(), itFullTs);
  fvulMissedTsIndices.erase(fvulMissedTsIndices.begin(), itMissTs);
 
  if (fbReceivedEof && fuPrevTsIndex == fuLastTsIndex && fulTsCounter == fuTotalTsCount) {
    LOG(info) << "CbmDeviceMcbmEventSink::CheckTsQueues => "
              << "Found final TS index " << fuLastTsIndex << " and total nb TS " << fuTotalTsCount;
    Finish();
  }  // if( fbReceivedEof && fuPrevTsIndex == fuLastTsIndex && fulTsCounter == fuTotalTsCount )
}
//--------------------------------------------------------------------//
void CbmDeviceMcbmEventSink::PrepareTreeEntry(CbmUnpackedTimeslice unpTs)
{
 
  new ((*fTimeSliceMetaDataArray)[fTimeSliceMetaDataArray->GetEntriesFast()])
    TimesliceMetaData(std::move(unpTs.fTsMetaData));
 
  /*
   fvDigiBmon->insert( fvDigiBmon->end(), unpTs.fvDigiBmon.begin(), unpTs.fvDigiBmon.end() );
   fvDigiSts->insert( fvDigiSts->end(), unpTs.fvDigiSts.begin(), unpTs.fvDigiSts.end() );
   fvDigiMuch->insert( fvDigiMuch->end(), unpTs.fvDigiMuch.begin(), unpTs.fvDigiMuch.end() );
   fvDigiTrd->insert( fvDigiTrd->end(), unpTs.fvDigiTrd.begin(), unpTs.fvDigiTrd.end() );
   fvDigiTof->insert( fvDigiTof->end(), unpTs.fvDigiTof.begin(), unpTs.fvDigiTof.end() );
   fvDigiRich->insert( fvDigiRich->end(), unpTs.fvDigiRich.begin(), unpTs.fvDigiRich.end() );
   fvDigiPsd->insert( fvDigiPsd->end(), unpTs.fvDigiPsd.begin(), unpTs.fvDigiPsd.end() );
*/
  (*fvDigiBmon) = std::move(unpTs.fvDigiBmon);
  (*fvDigiSts) = std::move(unpTs.fvDigiSts);
  (*fvDigiMuch) = std::move(unpTs.fvDigiMuch);
  (*fvDigiTrd) = std::move(unpTs.fvDigiTrd);
  (*fvDigiTof) = std::move(unpTs.fvDigiTof);
  (*fvDigiRich) = std::move(unpTs.fvDigiRich);
  (*fvDigiPsd) = std::move(unpTs.fvDigiPsd);
 
  fEventsArray->AbsorbObjects(&(unpTs.fEventsArray));
}
void CbmDeviceMcbmEventSink::DumpTreeEntry()
{
  // Unpacked digis + CbmEvent output to root file
  /*
 * NH style
//      fpFairRootMgr->FillEventHeader(fEvtHeader);
//      LOG(info) << "Fill WriteOutBuffer with FairRootManager at " << fpFairRootMgr;
//      fpOutRootFile->cd();
      fpFairRootMgr->Fill();
      fpFairRootMgr->StoreWriteoutBufferData( fpFairRootMgr->GetEventTime() );
      //fpFairRootMgr->StoreAllWriteoutBufferData();
      fpFairRootMgr->DeleteOldWriteoutBufferData();
*/
  fpFairRootMgr->StoreWriteoutBufferData(fpFairRootMgr->GetEventTime());
  fpFairRootMgr->Fill();
  fpFairRootMgr->DeleteOldWriteoutBufferData();
 
  fTimeSliceMetaDataArray->Clear();
 
  fvDigiBmon->clear();
  fvDigiSts->clear();
  fvDigiMuch->clear();
  fvDigiTrd->clear();
  fvDigiTof->clear();
  fvDigiRich->clear();
  fvDigiPsd->clear();
 
  //   fEventsArray->Delete();
  fEventsArray->Clear("C");
  //   fEventsArray->Clear();
}
 
//--------------------------------------------------------------------//
bool CbmDeviceMcbmEventSink::SendHistograms()
{
  FairMQMessagePtr message(NewMessage());
  //  Serialize<RootSerializer>(*message, &fArrayHisto);
  RootSerializer().Serialize(*message, &fArrayHisto);
 
  if (Send(message, fsChannelNameHistosInput) < 0) {
    LOG(error) << "Problem sending data";
    return false;
  }  // if( Send( message, fsChannelNameHistosInput ) < 0 )
 
  //   fpAlgo->ResetHistograms( kFALSE );
 
  return true;
}
 
//--------------------------------------------------------------------//
CbmDeviceMcbmEventSink::~CbmDeviceMcbmEventSink()
{
 
  if (!fbFinishDone) Finish();
 
  fTimeSliceMetaDataArray->Clear();
  delete fTimeSliceMetaDataArray;
  delete fTsMetaData;
 
  fvDigiBmon->clear();
  fvDigiSts->clear();
  fvDigiMuch->clear();
  fvDigiTrd->clear();
  fvDigiTof->clear();
  fvDigiRich->clear();
  fvDigiPsd->clear();
 
  fEventsArray->Clear();
  delete fEventsArray;
 
  delete fpRun;
}
 
void CbmDeviceMcbmEventSink::Finish()
{
  // Clean closure of output to root file
  fpFairRootMgr->Write();
  //   fpFairRootMgr->GetSource()->Close();
  fpFairRootMgr->CloseSink();
  LOG(info) << "File closed after saving " << (fulTsCounter + fulMissedTsCounter) << " TS (" << fulTsCounter
            << " full ones and " << fulMissedTsCounter << " missed/empty ones)";
 
  if (kTRUE == fbFillHistos) {
    SendHistograms();
    fLastPublishTime = std::chrono::system_clock::now();
  }  // if( kTRUE == fbFillHistos )
 
  ChangeState(fair::mq::Transition::Stop);
  std::this_thread::sleep_for(std::chrono::milliseconds(3000));
  ChangeState(fair::mq::Transition::End);
 
  fbFinishDone = kTRUE;
}
 
CbmUnpackedTimeslice::CbmUnpackedTimeslice(FairMQParts& parts) : fEventsArray("CbmEvent", 500)
{
  uint32_t uPartIdx = 0;
  /*
  std::string msgStrTsMeta( static_cast< char * >( parts.At( uPartIdx )->GetData() ),
                            ( parts.At( uPartIdx ) )->GetSize() );
  std::istringstream issTsMeta(msgStrTsMeta);
  boost::archive::binary_iarchive inputArchiveTsMeta(issTsMeta);
  inputArchiveTsMeta >> (*fTsMetaData);
  ++uPartIdx;
*/
  TObject* tempObjectMeta = nullptr;
  RootSerializer().Deserialize(*parts.At(uPartIdx), tempObjectMeta);
  ++uPartIdx;
 
  if (TString(tempObjectMeta->ClassName()).EqualTo("TimesliceMetaData")) {
    fTsMetaData = *(static_cast<TimesliceMetaData*>(tempObjectMeta));
  }  // if( TString( tempObject->ClassName() ).EqualTo( "TClonesArray") )
 
  std::string msgStrBmon(static_cast<char*>(parts.At(uPartIdx)->GetData()), (parts.At(uPartIdx))->GetSize());
  std::istringstream issBmon(msgStrBmon);
  boost::archive::binary_iarchive inputArchiveBmon(issBmon);
  inputArchiveBmon >> fvDigiBmon;
  ++uPartIdx;
 
  std::string msgStrSts(static_cast<char*>(parts.At(uPartIdx)->GetData()), (parts.At(uPartIdx))->GetSize());
  std::istringstream issSts(msgStrSts);
  boost::archive::binary_iarchive inputArchiveSts(issSts);
  inputArchiveSts >> fvDigiSts;
  ++uPartIdx;
 
  std::string msgStrMuch(static_cast<char*>(parts.At(uPartIdx)->GetData()), (parts.At(uPartIdx))->GetSize());
  std::istringstream issMuch(msgStrMuch);
  boost::archive::binary_iarchive inputArchiveMuch(issMuch);
  inputArchiveMuch >> fvDigiMuch;
  ++uPartIdx;
 
  std::string msgStrTrd(static_cast<char*>(parts.At(uPartIdx)->GetData()), (parts.At(uPartIdx))->GetSize());
  std::istringstream issTrd(msgStrTrd);
  boost::archive::binary_iarchive inputArchiveTrd(issTrd);
  inputArchiveTrd >> fvDigiTrd;
  ++uPartIdx;
 
  std::string msgStrTof(static_cast<char*>(parts.At(uPartIdx)->GetData()), (parts.At(uPartIdx))->GetSize());
  std::istringstream issTof(msgStrTof);
  boost::archive::binary_iarchive inputArchiveTof(issTof);
  inputArchiveTof >> fvDigiTof;
  ++uPartIdx;
 
  std::string msgStrRich(static_cast<char*>(parts.At(uPartIdx)->GetData()), (parts.At(uPartIdx))->GetSize());
  std::istringstream issRich(msgStrRich);
  boost::archive::binary_iarchive inputArchiveRich(issRich);
  inputArchiveRich >> fvDigiRich;
  ++uPartIdx;
 
  std::string msgStrPsd(static_cast<char*>(parts.At(uPartIdx)->GetData()), (parts.At(uPartIdx))->GetSize());
  std::istringstream issPsd(msgStrPsd);
  boost::archive::binary_iarchive inputArchivePsd(issPsd);
  inputArchivePsd >> fvDigiPsd;
  ++uPartIdx;
 
  TObject* tempObject = nullptr;
  RootSerializer().Deserialize(*parts.At(uPartIdx), tempObject);
  ++uPartIdx;
 
  if (TString(tempObject->ClassName()).EqualTo("TClonesArray")) {
    TClonesArray* arrayEventsIn = static_cast<TClonesArray*>(tempObject);
 
    fEventsArray.AbsorbObjects(arrayEventsIn);
  }  // if( TString( tempObject->ClassName() ).EqualTo( "TClonesArray") )
}
 
CbmUnpackedTimeslice::~CbmUnpackedTimeslice()
{
  fvDigiBmon.clear();
  fvDigiSts.clear();
  fvDigiMuch.clear();
  fvDigiTrd.clear();
  fvDigiTof.clear();
  fvDigiRich.clear();
  fvDigiPsd.clear();
  //  fEventsArray.Clear("C");
  fEventsArray.Delete();
}