mirror of
https://github.com/FairRootGroup/FairMQ.git
synced 2025-10-13 16:46:47 +00:00
16fd63cd5b
- OnData() channel data handler. - ConditionalRun() for devices without incoming data. - Header file with common main(), to be extended with getDevice/addCustomOptions. - Update examples (MQ/Tutorial3) to use the new API and config. - NewSimpleMessage() for simpler creation of small messages (additional copy). - Replace SetProperty/GetProperty with fConfig access. - Runtime configurable channel names for common devices. - Configurable logging interval per channel. - FairMQMultiplier for distributing same data to multiple outputs. - Cleanup state machine messages. - Cmd option to toggle signal handling. - Simpler API for send/receive timeouts. - Enable --log-to-file. - Fix coverity issues, warnings. - Various code cleanup and minor tweaks.
132 lines
3.9 KiB
C++
132 lines
3.9 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" *
|
|
********************************************************************************/
|
|
/**
|
|
* runTransferTimeoutTester.cxx
|
|
*
|
|
* @since 2015-09-05
|
|
* @author A. Rybalchenko
|
|
*/
|
|
|
|
#include "FairMQLogger.h"
|
|
#include "FairMQDevice.h"
|
|
|
|
class TransferTimeoutTester : public FairMQDevice
|
|
{
|
|
public:
|
|
TransferTimeoutTester() {}
|
|
virtual ~TransferTimeoutTester() {}
|
|
|
|
protected:
|
|
virtual void Run()
|
|
{
|
|
bool sendCanceling = false;
|
|
bool receiveCanceling = false;
|
|
|
|
FairMQMessagePtr msg1(NewMessage());
|
|
FairMQMessagePtr msg2(NewMessage());
|
|
|
|
if (Send(msg1, "data-out", 0, 1000) == -2)
|
|
{
|
|
LOG(INFO) << "send canceled";
|
|
sendCanceling = true;
|
|
}
|
|
else
|
|
{
|
|
LOG(ERROR) << "send did not cancel";
|
|
}
|
|
|
|
if (Receive(msg2, "data-in", 0, 1000) == -2)
|
|
{
|
|
LOG(INFO) << "receive canceled";
|
|
receiveCanceling = true;
|
|
}
|
|
else
|
|
{
|
|
LOG(ERROR) << "receive did not cancel";
|
|
}
|
|
|
|
if (sendCanceling && receiveCanceling)
|
|
{
|
|
LOG(INFO) << "Transfer timeout test successfull";
|
|
}
|
|
}
|
|
};
|
|
|
|
int main(int argc, char** argv)
|
|
{
|
|
TransferTimeoutTester timeoutTester;
|
|
timeoutTester.CatchSignals();
|
|
|
|
std::string transport;
|
|
if (argc != 2)
|
|
{
|
|
LOG(ERROR) << "Transport for the test not specified!";
|
|
return 1;
|
|
}
|
|
transport = argv[1];
|
|
|
|
if (transport == "zeromq" || transport == "nanomsg")
|
|
{
|
|
timeoutTester.SetTransport(transport);
|
|
}
|
|
else
|
|
{
|
|
LOG(ERROR) << "Incorrect transport requested! Expected 'zeromq' or 'nanomsg', found: " << transport;
|
|
return 1;
|
|
}
|
|
|
|
reinit_logger(false);
|
|
|
|
timeoutTester.SetProperty(TransferTimeoutTester::Id, "timeoutTester");
|
|
|
|
FairMQChannel dataOutChannel;
|
|
dataOutChannel.UpdateType("push");
|
|
dataOutChannel.UpdateMethod("bind");
|
|
dataOutChannel.UpdateAddress("tcp://127.0.0.1:5559");
|
|
if (transport == "nanomsg")
|
|
{
|
|
dataOutChannel.UpdateAddress("tcp://127.0.0.1:5759");
|
|
}
|
|
dataOutChannel.UpdateSndBufSize(1000);
|
|
dataOutChannel.UpdateRcvBufSize(1000);
|
|
dataOutChannel.UpdateRateLogging(0);
|
|
timeoutTester.fChannels["data-out"].push_back(dataOutChannel);
|
|
|
|
FairMQChannel dataInChannel;
|
|
dataInChannel.UpdateType("pull");
|
|
dataInChannel.UpdateMethod("bind");
|
|
dataInChannel.UpdateAddress("tcp://127.0.0.1:5560");
|
|
if (transport == "nanomsg")
|
|
{
|
|
dataInChannel.UpdateAddress("tcp://127.0.0.1:5760");
|
|
}
|
|
dataInChannel.UpdateSndBufSize(1000);
|
|
dataInChannel.UpdateRcvBufSize(1000);
|
|
dataInChannel.UpdateRateLogging(0);
|
|
timeoutTester.fChannels["data-in"].push_back(dataInChannel);
|
|
|
|
timeoutTester.ChangeState(TransferTimeoutTester::INIT_DEVICE);
|
|
timeoutTester.WaitForEndOfState(TransferTimeoutTester::INIT_DEVICE);
|
|
|
|
timeoutTester.ChangeState(TransferTimeoutTester::INIT_TASK);
|
|
timeoutTester.WaitForEndOfState(TransferTimeoutTester::INIT_TASK);
|
|
|
|
timeoutTester.ChangeState(TransferTimeoutTester::RUN);
|
|
timeoutTester.WaitForEndOfState(TransferTimeoutTester::RUN);
|
|
|
|
timeoutTester.ChangeState(TransferTimeoutTester::RESET_TASK);
|
|
timeoutTester.WaitForEndOfState(TransferTimeoutTester::RESET_TASK);
|
|
|
|
timeoutTester.ChangeState(TransferTimeoutTester::RESET_DEVICE);
|
|
timeoutTester.WaitForEndOfState(TransferTimeoutTester::RESET_DEVICE);
|
|
|
|
timeoutTester.ChangeState(TransferTimeoutTester::END);
|
|
|
|
return 0;
|
|
}
|