BmnOnlineSource.cxx

Go to the documentation of this file.

#include "BmnOnlineSource.h"
 
#include <FairRootManager.h>
#include <TBranchElement.h>
 
BmnOnlineSource::BmnOnlineSource(TString inputTransportAddress)
    : fEndpoint(inputTransportAddress)
    , fZmqContext(nullptr)
    , fZmqSubSocket(nullptr)
    , fAreBranchesRegistered(kFALSE)
{}
 
BmnOnlineSource::~BmnOnlineSource() {}
 
Bool_t BmnOnlineSource::Init()
{
    LOG(info) << "BmnOnlineSource::Init()";
 
    fZmqContext.reset(zmq_ctx_new(), zmq_ctx_term);
    fZmqSubSocket.reset(zmq_socket(fZmqContext.get(), ZMQ_SUB), zmq_close);
 
    if (zmq_connect(fZmqSubSocket.get(), fEndpoint.Data()) != 0) {
        LOG(info) << "BmnOnlineSource::Init() | An error occurred while "
                     "connecting the socket to the address '"
                  << fEndpoint << "': " << zmq_strerror(errno);
        return kFALSE;
    }
 
    if (zmq_setsockopt(fZmqSubSocket.get(), ZMQ_SUBSCRIBE, "", 0) != 0) {
        LOG(error) << "BmnOnlineSource::Init() | An error occurred while "
                      "subscribing the socket to the address '"
                   << fEndpoint << "': " << zmq_strerror(errno);
        return kFALSE;
    }
 
    return kTRUE;
}
 
Int_t BmnOnlineSource::ReadEvent(UInt_t)
{
    zmq_msg_t _message;
    zmq_msg_init(&_message);
 
    const auto _messageSize = zmq_msg_recv(&_message, fZmqSubSocket.get(), 0);
 
    if (_messageSize == -1) {
        LOG(warn) << "BmnOnlineSource::ReadEvent() | An error occurred while "
                     "receiving a message: "
                  << zmq_strerror(errno);
        zmq_msg_close(&_message);
        return kFALSE;
    } else {
        LOG(info) << "BmnOnlineSource::ReadEvent() | Message received: " << _messageSize << " B";
    }
 
    std::unique_ptr<TTree> tree;
 
    DeserializeZmqMessage(&_message, tree);
    RegisterBranches(tree);
    UploadDataToRegisteredBranches(tree);
 
    zmq_msg_close(&_message);
 
    return 0;
}
 
void BmnOnlineSource::Close()
{
    fZmqSubSocket.reset();
    fZmqContext.reset();
 
    fRegisteredBranches.clear();
}
 
void BmnOnlineSource::RegisterBranches(std::unique_ptr<TTree>& tree)
{
    if (fAreBranchesRegistered) {
        return;
    }
 
    const auto _branches = tree->GetListOfBranches();
    for (Long64_t branchIndex = 0; branchIndex < _branches->GetEntriesFast(); branchIndex++) {
        const auto _branch = static_cast<TBranchElement*>(_branches->At(branchIndex));
        const auto _branchName = _branch->GetName();
 
        if (fRegisteredBranches.find(_branchName) == fRegisteredBranches.end()
            && strcmp(_branch->GetClassName(), "TClonesArray") == 0)
        {
            auto _branch_data = std::make_unique<TClonesArray>(_branch->GetClonesName());
 
            FairRootManager::Instance()->RegisterInputObject(_branchName, _branch_data.get());
 
            fRegisteredBranches[_branchName] = std::move(_branch_data);
 
            LOG(info) << "BmnOnlineSource::RegisterBranches() | Register " << _branchName;
        }
    }
 
    fAreBranchesRegistered = kTRUE;
}
 
void BmnOnlineSource::UploadDataToRegisteredBranches(std::unique_ptr<TTree>& tree)
{
    for (auto& _branch : fRegisteredBranches) {
        _branch.second->Delete();
        auto tmp = _branch.second.get();
        if (tree->SetBranchAddress(_branch.first.c_str(), &tmp)) {
            LOG(error) << "BmnOnlineSource::UploadData(): Branch '" << _branch.first << "' not found!";
        }
    }
 
    tree->GetEntry(0);
}