FairMQExParamsClient.cxx

Go to the documentation of this file.

/********************************************************************************
 *    Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH    *
 *                                                                              *
 *              This software is distributed under the terms of the             *
 *              GNU Lesser General Public Licence (LGPL) version 3,             *
 *                  copied verbatim in the file "LICENSE"                       *
 ********************************************************************************/
#include "FairMQExParamsClient.h"

#include "FairMQExParamsParOne.h"
#include "RootSerializer.h"

#include <FairMQLogger.h>
#include <chrono>
#include <thread>   // this_thread::sleep_for

using namespace std;

FairMQExParamsClient::FairMQExParamsClient()
    : fRunId(0)
    , fParameterName()
    , fMaxIterations(0)
    , fNumIterations(0)
{}

FairMQExParamsClient::~FairMQExParamsClient() {}

void FairMQExParamsClient::InitTask()
{
    fParameterName = fConfig->GetValue<string>("parameter-name");
    fMaxIterations = fConfig->GetValue<uint64_t>("max-iterations");
    fRunId = 2000;
}

bool FairMQExParamsClient::ConditionalRun()
{
    LOG(info) << "Requesting parameter \"" << fParameterName << "\" for Run ID " << fRunId << ".";

    // 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
    FairMQMessagePtr req(NewSimpleMessage(fParameterName + "," + to_string(fRunId)));
    FairMQMessagePtr rep(NewMessage());

    if (Send(req, "data") > 0) {
        if (Receive(rep, "data") >= 0) {
            if (rep->GetSize() != 0) {
                FairMQExParamsParOne* par = nullptr;
                Deserialize<RootSerializer>(*rep, par);
                LOG(info) << "Received parameter from the server:";
                par->print();
                delete par;
            } else {
                LOG(error) << "Received empty reply. Parameter not available";
            }
        }
    }

    fRunId == 2099 ? fRunId = 2000 : fRunId++;

    if (fMaxIterations > 0 && ++fNumIterations >= fMaxIterations) {
        LOG(info) << "Configured maximum number of iterations reached. Leaving RUNNING state.";
        return false;
    }

    this_thread::sleep_for(chrono::seconds(1));

    return true;
}