FairMQ/fairmq/devices/GenericFileSink.h
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

92 lines
2.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" *
********************************************************************************/
/*
* File: GenericFileSink.h
* Author: winckler
*
* Created on October 7, 2014, 6:06 PM
*/
#ifndef GENERICFILESINK_H
#define GENERICFILESINK_H
#include "FairMQDevice.h"
#include <boost/thread.hpp>
#include <boost/bind.hpp>
#include "FairMQLogger.h"
/*********************************************************************
* -------------- NOTES -----------------------
* All policies must have a default constructor
* Function to define in (parent) policy classes :
*
* -------- INPUT POLICY --------
* InputPolicy::InitContainer(...)
* CONTAINER_TYPE InputPolicy::DeSerializeMsg(FairMQMessage* msg)
*
*
* -------- OUTPUT POLICY --------
* OutputPolicy::AddToFile(CONTAINER_TYPE);
* OutputPolicy::InitOutputFile()
**********************************************************************/
#include "FairMQDevice.h"
template <typename InputPolicy, typename OutputPolicy>
class GenericFileSink : public FairMQDevice, public InputPolicy, public OutputPolicy
{
public:
GenericFileSink()
: InputPolicy()
, OutputPolicy()
{}
virtual ~GenericFileSink()
{}
void SetTransport(FairMQTransportFactory* transport)
{
FairMQDevice::SetTransport(transport);
}
template <typename... Args>
void InitInputContainer(Args... args)
{
InputPolicy::InitContainer(std::forward<Args>(args)...);
}
protected:
virtual void InitTask()
{
OutputPolicy::InitOutputFile();
}
virtual void Run()
{
int receivedMsg = 0;
while (GetCurrentState() == RUNNING)
{
FairMQMessage* msg = fTransportFactory->CreateMessage();
if (fChannels["data-in"].at(0).Receive(msg) > 0)
{
OutputPolicy::AddToFile(InputPolicy::DeSerializeMsg(msg));
receivedMsg++;
}
delete msg;
}
MQLOG(INFO) << "Received " << receivedMsg << " messages!";
}
};
#endif /* GENERICFILESINK_H */