FairMQ/fairmq/examples/2-sampler-processor-sink/FairMQExample2Processor.cxx
Alexey Rybalchenko 1302e77a16 Several FairMQ fixes and improvements:
- FairMQ: add possibility to poll on multiple channels.
- FairMQ: include command channel when polling on blocking calls (for unblocking without termination).
- FairMQ: move signal handler inside of FairMQDevice class (call FairMQDevice::CatchSignals() in the main function).
- FairMQ: add 'bool CheckCurrentState(statename)' (instead of 'GetCurrentState() == statename' that cannot be thread safe).
- FairMQDevice: add 'InteractiveStateLoop()' method that can be used to change states from the command line.
- FairMQDevice: add automatic transition to IDLE state if Run() exits without an external event.
- FairMQDevice: implement device reset.
- FairMQDevice: use unordered_map for device channels.
- FairMQChannel: improve address validation for channels.
- FairMQChannel: add ExpectsAnotherPart() method to check if another msg part is expected (old approach still works).
- FairMQ: remove invalid transition from the run files.
- FairMQFileSink: disable ROOT termination signal handler.
- Tutorial3: spawn xterm windows from start scripts without overlapping for better visibility.
- FairMQ Examples: update protobuf test and move its files to a common directory.
- FairMQStateMachine: improve feedback on invalid transitions (more readable).
2015-09-28 12:17:22 +02:00

53 lines
1.6 KiB
C++

/********************************************************************************
* Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence version 3 (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
/**
* FairMQExample2Processor.cpp
*
* @since 2014-10-10
* @author A. Rybalchenko
*/
#include <boost/thread.hpp>
#include <boost/bind.hpp>
#include "FairMQExample2Processor.h"
#include "FairMQLogger.h"
FairMQExample2Processor::FairMQExample2Processor()
: fText()
{
}
void FairMQExample2Processor::CustomCleanup(void *data, void *object)
{
delete (std::string*)object;
}
void FairMQExample2Processor::Run()
{
while (GetCurrentState() == RUNNING)
{
FairMQMessage* input = fTransportFactory->CreateMessage();
fChannels.at("data-in").at(0).Receive(input);
LOG(INFO) << "Received data, processing...";
std::string* text = new std::string(static_cast<char*>(input->GetData()), input->GetSize());
*text += " (modified by " + fId + ")";
delete input;
FairMQMessage* msg = fTransportFactory->CreateMessage(const_cast<char*>(text->c_str()), text->length(), CustomCleanup, text);
fChannels.at("data-out").at(0).Send(msg);
}
}
FairMQExample2Processor::~FairMQExample2Processor()
{
}