mirror of
https://github.com/FairRootGroup/FairMQ.git
synced 2025-10-13 16:46:47 +00:00
f1abb9ecdd
- Remove the compile time check of the transport implementation.
The transport (zeromq/nanomsg) can be chosen at run time with:
`device.SetTransport("zeromq"); // possible values are "zeromq" and "nanomsg"`.
For devices that use FairMQProgOptions, the transport can be configured via cmd option:
`--transport zeromq` or `--transport nanomsg`. Default values is "zeromq".
The device receives the configured value with:
`device.SetTransport(config.GetValue<std::string>("transport"));`
Old method of setting transport still works. But the NANOMSG constant is not defined.
- Remove old `fairmq/prototest` directory. It was only used as a test for protobuf.
The protobuf part of Tutorial3 does the same (with different values).
- Fix a bug in FairMQPollerNN, where the `revents` value was not initialized.
This caused the `poller->CheckOutput()` to trigger when it should not.
76 lines
1.8 KiB
C++
76 lines
1.8 KiB
C++
/*
|
|
* File: GenericMerger.h
|
|
* Author: winckler
|
|
*
|
|
* Created on April 9, 2015, 1:37 PM
|
|
*/
|
|
|
|
#ifndef GENERICMERGER_H
|
|
#define GENERICMERGER_H
|
|
|
|
|
|
#include <boost/thread.hpp>
|
|
#include <boost/bind.hpp>
|
|
|
|
#include "FairMQDevice.h"
|
|
#include "FairMQLogger.h"
|
|
#include "FairMQPoller.h"
|
|
|
|
|
|
template <typename MergerPolicy, typename InputPolicy, typename OutputPolicy>
|
|
class GenericMerger : public FairMQDevice, public MergerPolicy, public InputPolicy, public OutputPolicy
|
|
{
|
|
public:
|
|
GenericMerger()
|
|
: fBlockingTime(100)
|
|
{}
|
|
|
|
virtual ~GenericMerger()
|
|
{}
|
|
|
|
template <typename... Args>
|
|
void SetTransport(Args... args)
|
|
{
|
|
FairMQDevice::SetTransport(std::forward<Args>(args)...);
|
|
}
|
|
|
|
protected:
|
|
int fBlockingTime;
|
|
|
|
virtual void Run()
|
|
{
|
|
std::unique_ptr<FairMQPoller> poller(fTransportFactory->CreatePoller(fChannels["data-in"]));
|
|
|
|
int received = 0;
|
|
|
|
while (GetCurrentState() == RUNNING)
|
|
{
|
|
std::unique_ptr<FairMQMessage> msg(fTransportFactory->CreateMessage());
|
|
// MergerPolicy::
|
|
poller->Poll(fBlockingTime);
|
|
|
|
for (int i = 0; i < fChannels.at("data-in").size(); i++)
|
|
{
|
|
if (poller->CheckInput(i))
|
|
{
|
|
received = fChannels.at("data-in").at(i).Receive(msg)
|
|
if (received > 0)
|
|
{
|
|
MergerPolicy::Merge(InputPolicy::DeserializeMsg(msg));
|
|
}
|
|
}
|
|
|
|
OutputPolicy::SetMessage(msg);
|
|
|
|
if (received > 0 && MergerPolicy::ReadyToSend())
|
|
{
|
|
fChannels["data-out"].at(0).Send(OutputPolicy::SerializeMsg(MergerPolicy::GetOutputData()));
|
|
received = 0;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
};
|
|
|
|
#endif /* GENERICMERGER_H */
|