Mirror/FairMQ
1
0
Fork 0
mirror of https://github.com/FairRootGroup/FairMQ.git synced 2025-10-13 08:41:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

Don't use to-be-deprecated names

Browse Source
This commit is contained in:
Alexey Rybalchenko 2022-01-14 00:48:34 +01:00 committed by Dennis Klein
parent f15f669853
commit bfd08bb33f
90 changed files with 411 additions and 391 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
examples/1-1/sampler.cxx
View File

@ -30,7 +30,7 @@ struct Sampler : fair::mq::Device
// create message object with a pointer to the data buffer, its size, // create message object with a pointer to the data buffer, its size,
// custom deletion function (called when transfer is done), // custom deletion function (called when transfer is done),
// and pointer to the object managing the data buffer // and pointer to the object managing the data buffer
FairMQMessagePtr msg(NewMessage( fair::mq::MessagePtr msg(NewMessage(
const_cast<char*>(text->c_str()), const_cast<char*>(text->c_str()),
text->length(), text->length(),
[](void* /*data*/, void* object) { delete static_cast<std::string*>(object); }, [](void* /*data*/, void* object) { delete static_cast<std::string*>(object); },

2
examples/1-1/sink.cxx
View File

@ -27,7 +27,7 @@ struct Sink : fair::mq::Device
fMaxIterations = fConfig->GetProperty<uint64_t>("max-iterations"); fMaxIterations = fConfig->GetProperty<uint64_t>("max-iterations");
} }
bool HandleData(FairMQMessagePtr& msg, int) bool HandleData(fair::mq::MessagePtr& msg, int)
{ {
LOG(info) << "Received: \"" << std::string(static_cast<char*>(msg->GetData()), msg->GetSize()) << "\""; LOG(info) << "Received: \"" << std::string(static_cast<char*>(msg->GetData()), msg->GetSize()) << "\"";

4
examples/1-n-1/processor.cxx
View File

@ -20,7 +20,7 @@ struct Processor : fair::mq::Device
OnData("data1", &Processor::HandleData); OnData("data1", &Processor::HandleData);
} }
bool HandleData(FairMQMessagePtr& msg, int) bool HandleData(fair::mq::MessagePtr& msg, int)
{ {
LOG(info) << "Received data, processing..."; LOG(info) << "Received data, processing...";
@ -32,7 +32,7 @@ struct Processor : fair::mq::Device
// its size, // its size,
// custom deletion function (called when transfer is done), // custom deletion function (called when transfer is done),
// and pointer to the object managing the data buffer // and pointer to the object managing the data buffer
FairMQMessagePtr msg2(NewMessage(const_cast<char*>(text->c_str()), fair::mq::MessagePtr msg2(NewMessage(const_cast<char*>(text->c_str()),
text->length(), text->length(),
[](void* /*data*/, void* object) { delete static_cast<std::string*>(object); }, [](void* /*data*/, void* object) { delete static_cast<std::string*>(object); },
text)); text));

2
examples/1-n-1/sampler.cxx
View File

@ -28,7 +28,7 @@ struct Sampler : fair::mq::Device
{ {
// Initializing message with NewStaticMessage will avoid copy // Initializing message with NewStaticMessage will avoid copy
// but won't delete the data after the sending is completed. // but won't delete the data after the sending is completed.
FairMQMessagePtr msg(NewStaticMessage(fText)); fair::mq::MessagePtr msg(NewStaticMessage(fText));
LOG(info) << "Sending \"" << fText << "\""; LOG(info) << "Sending \"" << fText << "\"";

2
examples/1-n-1/sink.cxx
View File

@ -27,7 +27,7 @@ struct Sink : fair::mq::Device
fMaxIterations = fConfig->GetProperty<uint64_t>("max-iterations"); fMaxIterations = fConfig->GetProperty<uint64_t>("max-iterations");
} }
bool HandleData(FairMQMessagePtr& msg, int) bool HandleData(fair::mq::MessagePtr& msg, int)
{ {
LOG(info) << "Received: \"" << std::string(static_cast<char*>(msg->GetData()), msg->GetSize()) << "\""; LOG(info) << "Received: \"" << std::string(static_cast<char*>(msg->GetData()), msg->GetSize()) << "\"";

4
examples/copypush/sampler.cxx
View File

@ -27,10 +27,10 @@ struct Sampler : fair::mq::Device
{ {
// NewSimpleMessage creates a copy of the data and takes care of its destruction (after the transfer takes place). // 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 // Should only be used for small data because of the cost of an additional copy
FairMQMessagePtr msg(NewSimpleMessage(fCounter++)); fair::mq::MessagePtr msg(NewSimpleMessage(fCounter++));
for (int i = 0; i < fNumDataChannels - 1; ++i) { for (int i = 0; i < fNumDataChannels - 1; ++i) {
FairMQMessagePtr msgCopy(NewMessage()); fair::mq::MessagePtr msgCopy(NewMessage());
msgCopy->Copy(*msg); msgCopy->Copy(*msg);
Send(msgCopy, "data", i); Send(msgCopy, "data", i);
} }

2
examples/copypush/sink.cxx
View File

@ -27,7 +27,7 @@ struct Sink : fair::mq::Device
fMaxIterations = fConfig->GetProperty<uint64_t>("max-iterations"); fMaxIterations = fConfig->GetProperty<uint64_t>("max-iterations");
} }
bool HandleData(FairMQMessagePtr& msg, int) bool HandleData(fair::mq::MessagePtr& msg, int)
{ {
LOG(info) << "Received message: \"" << *(static_cast<uint64_t*>(msg->GetData())) << "\""; LOG(info) << "Received message: \"" << *(static_cast<uint64_t*>(msg->GetData())) << "\"";

4
examples/dds/processor.cxx
View File

@ -20,7 +20,7 @@ struct Processor : fair::mq::Device
OnData("data1", &Processor::HandleData); OnData("data1", &Processor::HandleData);
} }
bool HandleData(FairMQMessagePtr& msg, int) bool HandleData(fair::mq::MessagePtr& msg, int)
{ {
LOG(info) << "Received data, processing..."; LOG(info) << "Received data, processing...";
@ -32,7 +32,7 @@ struct Processor : fair::mq::Device
// its size, // its size,
// custom deletion function (called when transfer is done), // custom deletion function (called when transfer is done),
// and pointer to the object managing the data buffer // and pointer to the object managing the data buffer
FairMQMessagePtr msg2(NewMessage(const_cast<char*>(text->c_str()), fair::mq::MessagePtr msg2(NewMessage(const_cast<char*>(text->c_str()),
text->length(), text->length(),
[](void* /*data*/, void* object) { delete static_cast<std::string*>(object); }, [](void* /*data*/, void* object) { delete static_cast<std::string*>(object); },
text)); text));

2
examples/dds/sampler.cxx
View File

@ -24,7 +24,7 @@ struct Sampler : fair::mq::Device
{ {
// NewSimpleMessage creates a copy of the data and takes care of its destruction (after the transfer takes place). // 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 // Should only be used for small data because of the cost of an additional copy
FairMQMessagePtr msg(NewSimpleMessage("Data")); fair::mq::MessagePtr msg(NewSimpleMessage("Data"));
LOG(info) << "Sending \"Data\""; LOG(info) << "Sending \"Data\"";

2
examples/dds/sink.cxx
View File

@ -25,7 +25,7 @@ struct Sink : fair::mq::Device
fIterations = fConfig->GetValue<uint64_t>("iterations"); fIterations = fConfig->GetValue<uint64_t>("iterations");
} }
bool HandleData(FairMQMessagePtr& msg, int) bool HandleData(fair::mq::MessagePtr& msg, int)
{ {
LOG(info) << "Received: \"" << std::string(static_cast<char*>(msg->GetData()), msg->GetSize()) << "\""; LOG(info) << "Received: \"" << std::string(static_cast<char*>(msg->GetData()), msg->GetSize()) << "\"";

2
examples/multipart/README.md
View File

@ -5,7 +5,7 @@ A topology of two devices - Sampler and Sink, communicating with PUSH-PULL patte
The Sampler sends a multipart message to the Sink, consisting of two message parts - header and body. The Sampler sends a multipart message to the Sink, consisting of two message parts - header and body.
Each message part is a regular FairMQMessage. To combine them into a multi-part message use `FairMQParts`. Add messages to `FairMQParts` with `AddPart` method. Each message part is a regular fair::mq::Message. To combine them into a multi-part message use `fair::mq::Parts`. Add messages to `fair::mq::Parts` with `AddPart` method.
All parts are guaranteed to be delivered together. The Receive call in the sink will recive the entire parts structure. All parts are guaranteed to be delivered together. The Receive call in the sink will recive the entire parts structure.

6
examples/multipart/sampler.cxx
View File

@ -35,15 +35,15 @@ struct Sampler : fair::mq::Device
} }
LOG(info) << "Sending header with stopFlag: " << header.stopFlag; LOG(info) << "Sending header with stopFlag: " << header.stopFlag;
FairMQParts parts; fair::mq::Parts parts;
// NewSimpleMessage creates a copy of the data and takes care of its destruction (after the transfer takes place). // 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 // Should only be used for small data because of the cost of an additional copy
parts.AddPart(NewSimpleMessage(header)); parts.AddPart(NewSimpleMessage(header));
parts.AddPart(NewMessage(1000)); parts.AddPart(NewMessage(1000));
// create more data parts, testing the FairMQParts in-place constructor // create more data parts, testing the fair::mq::Parts in-place constructor
FairMQParts auxData{ NewMessage(500), NewMessage(600), NewMessage(700) }; fair::mq::Parts auxData{ NewMessage(500), NewMessage(600), NewMessage(700) };
assert(auxData.Size() == 3); assert(auxData.Size() == 3);
parts.AddPart(std::move(auxData)); parts.AddPart(std::move(auxData));
assert(auxData.Size() == 0); assert(auxData.Size() == 0);

2
examples/multipart/sink.cxx
View File

@ -20,7 +20,7 @@ struct Sink : fair::mq::Device
OnData("data", &Sink::HandleData); OnData("data", &Sink::HandleData);
} }
bool HandleData(FairMQParts& parts, int) bool HandleData(fair::mq::Parts& parts, int)
{ {
LOG(info) << "Received message with " << parts.Size() << " parts"; LOG(info) << "Received message with " << parts.Size() << " parts";

2
examples/multiple-channels/README.md
View File

@ -5,4 +5,4 @@ This example demonstrates how to work with multiple channels and multiplex betwe
A topology of three devices - **Sampler**, **Sink** and **Broadcaster**. The Sampler sends data to the Sink via the PUSH-PULL pattern. The Broadcaster device sends a message to both Sampler and Sink containing a string "OK" every second. The Broadcaster sends the message via PUB pattern. Both Sampler and Sink, besides doing their PUSH-PULL job, listen via SUB to the Broadcaster. A topology of three devices - **Sampler**, **Sink** and **Broadcaster**. The Sampler sends data to the Sink via the PUSH-PULL pattern. The Broadcaster device sends a message to both Sampler and Sink containing a string "OK" every second. The Broadcaster sends the message via PUB pattern. Both Sampler and Sink, besides doing their PUSH-PULL job, listen via SUB to the Broadcaster.
The multiplexing between their data channels and the broadcast channels happens with `FairMQPoller`. The multiplexing between their data channels and the broadcast channels happens with `fair::mq::Poller`.

2
examples/multiple-channels/broadcaster.cxx
View File

@ -22,7 +22,7 @@ struct Broadcaster : fair::mq::Device
// NewSimpleMessage creates a copy of the data and takes care of its destruction (after the transfer takes place). // 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 // Should only be used for small data because of the cost of an additional copy
FairMQMessagePtr msg(NewSimpleMessage("OK")); fair::mq::MessagePtr msg(NewSimpleMessage("OK"));
LOG(info) << "Sending OK"; LOG(info) << "Sending OK";

8
examples/multiple-channels/sampler.cxx
View File

@ -7,7 +7,7 @@
********************************************************************************/ ********************************************************************************/
#include <fairmq/Device.h> #include <fairmq/Device.h>
#include <FairMQPoller.h> #include <fairmq/Poller.h>
#include <fairmq/runDevice.h> #include <fairmq/runDevice.h>
#include <chrono> #include <chrono>
@ -26,13 +26,13 @@ struct Sampler : fair::mq::Device
void Run() override void Run() override
{ {
FairMQPollerPtr poller(NewPoller("data", "broadcast")); fair::mq::PollerPtr poller(NewPoller("data", "broadcast"));
while (!NewStatePending()) { while (!NewStatePending()) {
poller->Poll(100); poller->Poll(100);
if (poller->CheckInput("broadcast", 0)) { if (poller->CheckInput("broadcast", 0)) {
FairMQMessagePtr msg(NewMessage()); fair::mq::MessagePtr msg(NewMessage());
if (Receive(msg, "broadcast") > 0) { if (Receive(msg, "broadcast") > 0) {
LOG(info) << "Received broadcast: \"" << std::string(static_cast<char*>(msg->GetData()), msg->GetSize()) << "\""; LOG(info) << "Received broadcast: \"" << std::string(static_cast<char*>(msg->GetData()), msg->GetSize()) << "\"";
@ -42,7 +42,7 @@ struct Sampler : fair::mq::Device
if (poller->CheckOutput("data", 0)) { if (poller->CheckOutput("data", 0)) {
std::this_thread::sleep_for(std::chrono::seconds(1)); std::this_thread::sleep_for(std::chrono::seconds(1));
FairMQMessagePtr msg(NewSimpleMessage(fText)); fair::mq::MessagePtr msg(NewSimpleMessage(fText));
if (Send(msg, "data") > 0) { if (Send(msg, "data") > 0) {
LOG(info) << "Sent \"" << fText << "\""; LOG(info) << "Sent \"" << fText << "\"";

4
examples/multiple-channels/sink.cxx
View File

@ -27,7 +27,7 @@ struct Sink : fair::mq::Device
fMaxIterations = fConfig->GetProperty<uint64_t>("max-iterations"); fMaxIterations = fConfig->GetProperty<uint64_t>("max-iterations");
} }
bool HandleBroadcast(FairMQMessagePtr& msg, int /*index*/) bool HandleBroadcast(fair::mq::MessagePtr& msg, int /*index*/)
{ {
LOG(info) << "Received broadcast: \"" << std::string(static_cast<char*>(msg->GetData()), msg->GetSize()) << "\""; LOG(info) << "Received broadcast: \"" << std::string(static_cast<char*>(msg->GetData()), msg->GetSize()) << "\"";
fReceivedBroadcast = true; fReceivedBroadcast = true;
@ -35,7 +35,7 @@ struct Sink : fair::mq::Device
return CheckIterations(); return CheckIterations();
} }
bool HandleData(FairMQMessagePtr& msg, int /*index*/) bool HandleData(fair::mq::MessagePtr& msg, int /*index*/)
{ {
LOG(info) << "Received message: \"" << std::string(static_cast<char*>(msg->GetData()), msg->GetSize()) << "\""; LOG(info) << "Received message: \"" << std::string(static_cast<char*>(msg->GetData()), msg->GetSize()) << "\"";
fReceivedData = true; fReceivedData = true;

4
examples/multiple-transports/sampler1.cxx
View File

@ -28,7 +28,7 @@ struct Sampler1 : fair::mq::Device
bool ConditionalRun() override bool ConditionalRun() override
{ {
// Creates a message using the transport of channel data1 // Creates a message using the transport of channel data1
FairMQMessagePtr msg(NewMessageFor("data1", 0, 1000000)); fair::mq::MessagePtr msg(NewMessageFor("data1", 0, 1000000));
// in case of error or transfer interruption, return false to go to IDLE state // in case of error or transfer interruption, return false to go to IDLE state
// successfull transfer will return number of bytes transfered (can be 0 if sending an empty message). // successfull transfer will return number of bytes transfered (can be 0 if sending an empty message).
@ -54,7 +54,7 @@ struct Sampler1 : fair::mq::Device
uint64_t numAcks = 0; uint64_t numAcks = 0;
while (!NewStatePending()) { while (!NewStatePending()) {
FairMQMessagePtr ack(NewMessageFor("ack", 0)); fair::mq::MessagePtr ack(NewMessageFor("ack", 0));
if (Receive(ack, "ack") < 0) { if (Receive(ack, "ack") < 0) {
break; break;
} }

2
examples/multiple-transports/sampler2.cxx
View File

@ -22,7 +22,7 @@ struct Sampler2 : fair::mq::Device
bool ConditionalRun() override bool ConditionalRun() override
{ {
FairMQMessagePtr msg(NewMessage(1000)); fair::mq::MessagePtr msg(NewMessage(1000));
// in case of error or transfer interruption, return false to go to IDLE state // in case of error or transfer interruption, return false to go to IDLE state
// successfull transfer will return number of bytes transfered (can be 0 if sending an empty message). // successfull transfer will return number of bytes transfered (can be 0 if sending an empty message).

6
examples/multiple-transports/sink.cxx
View File

@ -26,11 +26,11 @@ struct Sink : fair::mq::Device
} }
// handler is called whenever a message arrives on "data", with a reference to the message and a sub-channel index (here 0) // handler is called whenever a message arrives on "data", with a reference to the message and a sub-channel index (here 0)
bool HandleData1(FairMQMessagePtr& /*msg*/, int /*index*/) bool HandleData1(fair::mq::MessagePtr& /*msg*/, int /*index*/)
{ {
fNumIterations1++; fNumIterations1++;
// Creates a message using the transport of channel ack // Creates a message using the transport of channel ack
FairMQMessagePtr ack(NewMessageFor("ack", 0)); fair::mq::MessagePtr ack(NewMessageFor("ack", 0));
if (Send(ack, "ack") < 0) { if (Send(ack, "ack") < 0) {
return false; return false;
} }
@ -40,7 +40,7 @@ struct Sink : fair::mq::Device
} }
// handler is called whenever a message arrives on "data", with a reference to the message and a sub-channel index (here 0) // handler is called whenever a message arrives on "data", with a reference to the message and a sub-channel index (here 0)
bool HandleData2(FairMQMessagePtr& /*msg*/, int /*index*/) bool HandleData2(fair::mq::MessagePtr& /*msg*/, int /*index*/)
{ {
fNumIterations2++; fNumIterations2++;
// return true if want to be called again (otherwise go to IDLE state) // return true if want to be called again (otherwise go to IDLE state)

6
examples/n-m/receiver.cxx
View File

@ -24,7 +24,7 @@ namespace bpo = boost::program_options;
struct TFBuffer struct TFBuffer
{ {
FairMQParts parts; fair::mq::Parts parts;
chrono::steady_clock::time_point start; chrono::steady_clock::time_point start;
chrono::steady_clock::time_point end; chrono::steady_clock::time_point end;
}; };
@ -43,7 +43,7 @@ struct Receiver : fair::mq::Device
fMaxTimeframes = GetConfig()->GetValue<int>("max-timeframes"); fMaxTimeframes = GetConfig()->GetValue<int>("max-timeframes");
} }
bool HandleData(FairMQParts& parts, int /* index */) bool HandleData(fair::mq::Parts& parts, int /* index */)
{ {
Header& h = *(static_cast<Header*>(parts.At(0)->GetData())); Header& h = *(static_cast<Header*>(parts.At(0)->GetData()));
// LOG(info) << "Received sub-time frame #" << h.id << " from Sender" << h.senderIndex; // LOG(info) << "Received sub-time frame #" << h.id << " from Sender" << h.senderIndex;
@ -107,7 +107,7 @@ void addCustomOptions(bpo::options_description& options)
("max-timeframes", bpo::value<int>()->default_value(0), "Maximum number of timeframes to receive (0 - unlimited)"); ("max-timeframes", bpo::value<int>()->default_value(0), "Maximum number of timeframes to receive (0 - unlimited)");
} }
std::unique_ptr<fair::mq::Device> getDevice(FairMQProgOptions& /* config */) std::unique_ptr<fair::mq::Device> getDevice(fair::mq::ProgOptions& /* config */)
{ {
return std::make_unique<Receiver>(); return std::make_unique<Receiver>();
} }

8
examples/n-m/sender.cxx
View File

@ -28,11 +28,11 @@ struct Sender : fair::mq::Device
void Run() override void Run() override
{ {
FairMQChannel& dataInChannel = GetChannel("sync", 0); fair::mq::Channel& dataInChannel = GetChannel("sync", 0);
while (!NewStatePending()) { while (!NewStatePending()) {
Header h; Header h;
FairMQMessagePtr id(NewMessage()); fair::mq::MessagePtr id(NewMessage());
if (dataInChannel.Receive(id) > 0) { if (dataInChannel.Receive(id) > 0) {
h.id = *(static_cast<uint16_t*>(id->GetData())); h.id = *(static_cast<uint16_t*>(id->GetData()));
h.senderIndex = fIndex; h.senderIndex = fIndex;
@ -40,7 +40,7 @@ struct Sender : fair::mq::Device
continue; continue;
} }
FairMQParts parts; fair::mq::Parts parts;
parts.AddPart(NewSimpleMessage(h)); parts.AddPart(NewSimpleMessage(h));
parts.AddPart(NewMessage(fSubtimeframeSize)); parts.AddPart(NewMessage(fSubtimeframeSize));
@ -66,7 +66,7 @@ void addCustomOptions(bpo::options_description& options)
("subtimeframe-size", bpo::value<int>()->default_value(1000), "Subtimeframe size in bytes") ("subtimeframe-size", bpo::value<int>()->default_value(1000), "Subtimeframe size in bytes")
("num-receivers", bpo::value<int>()->required(), "Number of EPNs"); ("num-receivers", bpo::value<int>()->required(), "Number of EPNs");
} }
std::unique_ptr<fair::mq::Device> getDevice(FairMQProgOptions& /* config */) std::unique_ptr<fair::mq::Device> getDevice(fair::mq::ProgOptions& /* config */)
{ {
return std::make_unique<Sender>(); return std::make_unique<Sender>();
} }

4
examples/n-m/synchronizer.cxx
View File

@ -19,7 +19,7 @@ struct Synchronizer : fair::mq::Device
{ {
bool ConditionalRun() override bool ConditionalRun() override
{ {
FairMQMessagePtr msg(NewSimpleMessage(fTimeframeId)); fair::mq::MessagePtr msg(NewSimpleMessage(fTimeframeId));
if (Send(msg, "sync") > 0) { if (Send(msg, "sync") > 0) {
if (++fTimeframeId == UINT16_MAX - 1) { if (++fTimeframeId == UINT16_MAX - 1) {
@ -37,7 +37,7 @@ struct Synchronizer : fair::mq::Device
}; };
void addCustomOptions(bpo::options_description& /* options */) {} void addCustomOptions(bpo::options_description& /* options */) {}
std::unique_ptr<fair::mq::Device> getDevice(FairMQProgOptions& /* config */) std::unique_ptr<fair::mq::Device> getDevice(fair::mq::ProgOptions& /* config */)
{ {
return std::make_unique<Synchronizer>(); return std::make_unique<Synchronizer>();
} }

4
examples/qc/qCDispatcher.cxx
View File

@ -30,10 +30,10 @@ struct QCDispatcher : fair::mq::Device
}); });
} }
bool HandleData(FairMQMessagePtr& msg, int) bool HandleData(fair::mq::MessagePtr& msg, int)
{ {
if (fDoQC.load() == true) { if (fDoQC.load() == true) {
FairMQMessagePtr msgCopy(NewMessage()); fair::mq::MessagePtr msgCopy(NewMessage());
msgCopy->Copy(*msg); msgCopy->Copy(*msg);
if (Send(msg, "qc") < 0) { if (Send(msg, "qc") < 0) {
return false; return false;

4
examples/qc/qCTask.cxx
View File

@ -9,12 +9,12 @@
#include <fairmq/Device.h> #include <fairmq/Device.h>
#include <fairmq/runDevice.h> #include <fairmq/runDevice.h>
class QCTask : public FairMQDevice class QCTask : public fair::mq::Device
{ {
public: public:
QCTask() QCTask()
{ {
OnData("qc", [](FairMQMessagePtr& /*msg*/, int) { OnData("qc", [](fair::mq::MessagePtr& /*msg*/, int) {
LOG(info) << "received data"; LOG(info) << "received data";
return false; return false;
}); });

2
examples/qc/sampler.cxx
View File

@ -16,7 +16,7 @@ struct Sampler : fair::mq::Device
{ {
bool ConditionalRun() override bool ConditionalRun() override
{ {
FairMQMessagePtr msg(NewMessage(1000)); fair::mq::MessagePtr msg(NewMessage(1000));
if (Send(msg, "data1") < 0) { if (Send(msg, "data1") < 0) {
return false; return false;

2
examples/qc/sink.cxx
View File

@ -14,7 +14,7 @@
struct Sink : fair::mq::Device struct Sink : fair::mq::Device
{ {
Sink() { OnData("data2", &Sink::HandleData); } Sink() { OnData("data2", &Sink::HandleData); }
bool HandleData(FairMQMessagePtr& /*msg*/, int /*index*/) { return true; } bool HandleData(fair::mq::MessagePtr& /*msg*/, int /*index*/) { return true; }
}; };
namespace bpo = boost::program_options; namespace bpo = boost::program_options;

4
examples/readout/builder.cxx
View File

@ -13,7 +13,7 @@
namespace bpo = boost::program_options; namespace bpo = boost::program_options;
class Builder : public FairMQDevice class Builder : public fair::mq::Device
{ {
public: public:
Builder() = default; Builder() = default;
@ -24,7 +24,7 @@ class Builder : public FairMQDevice
OnData("rb", &Builder::HandleData); OnData("rb", &Builder::HandleData);
} }
bool HandleData(FairMQMessagePtr& msg, int /*index*/) bool HandleData(fair::mq::MessagePtr& msg, int /*index*/)
{ {
if (Send(msg, fOutputChannelName) < 0) { if (Send(msg, fOutputChannelName) < 0) {
return false; return false;

4
examples/readout/processor.cxx
View File

@ -18,9 +18,9 @@ struct Processor : fair::mq::Device
OnData("bp", &Processor::HandleData); OnData("bp", &Processor::HandleData);
} }
bool HandleData(FairMQMessagePtr& msg, int /*index*/) bool HandleData(fair::mq::MessagePtr& msg, int /*index*/)
{ {
FairMQMessagePtr msg2(NewMessageFor("ps", 0, msg->GetSize())); fair::mq::MessagePtr msg2(NewMessageFor("ps", 0, msg->GetSize()));
if (Send(msg2, "ps") < 0) { if (Send(msg2, "ps") < 0) {
return false; return false;
} }

6
examples/readout/readout.cxx
View File

@ -22,7 +22,7 @@ struct Readout : fair::mq::Device
fMsgSize = fConfig->GetProperty<int>("msg-size"); fMsgSize = fConfig->GetProperty<int>("msg-size");
fMaxIterations = fConfig->GetProperty<uint64_t>("max-iterations"); fMaxIterations = fConfig->GetProperty<uint64_t>("max-iterations");
fRegion = FairMQUnmanagedRegionPtr(NewUnmanagedRegionFor("rb", fRegion = fair::mq::UnmanagedRegionPtr(NewUnmanagedRegionFor("rb",
0, 0,
10000000, 10000000,
[this](const std::vector<fair::mq::RegionBlock>& blocks) { // callback to be called when message buffers no longer needed by transport [this](const std::vector<fair::mq::RegionBlock>& blocks) { // callback to be called when message buffers no longer needed by transport
@ -36,7 +36,7 @@ struct Readout : fair::mq::Device
bool ConditionalRun() override bool ConditionalRun() override
{ {
FairMQMessagePtr msg(NewMessageFor("rb", // channel fair::mq::MessagePtr msg(NewMessageFor("rb", // channel
0, // sub-channel 0, // sub-channel
fRegion, // region fRegion, // region
fRegion->GetData(), // ptr within region fRegion->GetData(), // ptr within region
@ -71,7 +71,7 @@ struct Readout : fair::mq::Device
int fMsgSize = 10000; int fMsgSize = 10000;
uint64_t fMaxIterations = 0; uint64_t fMaxIterations = 0;
uint64_t fNumIterations = 0; uint64_t fNumIterations = 0;
FairMQUnmanagedRegionPtr fRegion = nullptr; fair::mq::UnmanagedRegionPtr fRegion = nullptr;
std::atomic<uint64_t> fNumUnackedMsgs = 0; std::atomic<uint64_t> fNumUnackedMsgs = 0;
}; };

2
examples/readout/sender.cxx
View File

@ -21,7 +21,7 @@ struct Sender : fair::mq::Device
OnData(fInputChannelName, &Sender::HandleData); OnData(fInputChannelName, &Sender::HandleData);
} }
bool HandleData(FairMQMessagePtr& msg, int /*index*/) bool HandleData(fair::mq::MessagePtr& msg, int /*index*/)
{ {
if (Send(msg, "sr") < 0) { if (Send(msg, "sr") < 0) {
return false; return false;

8
examples/region/sampler.cxx
View File

@ -23,7 +23,7 @@ struct Sampler : fair::mq::Device
fLinger = fConfig->GetProperty<uint32_t>("region-linger"); fLinger = fConfig->GetProperty<uint32_t>("region-linger");
fMaxIterations = fConfig->GetProperty<uint64_t>("max-iterations"); fMaxIterations = fConfig->GetProperty<uint64_t>("max-iterations");
GetChannel("data", 0).Transport()->SubscribeToRegionEvents([](FairMQRegionInfo info) { GetChannel("data", 0).Transport()->SubscribeToRegionEvents([](fair::mq::RegionInfo info) {
LOG(info) << "Region event: " << info.event << ": " LOG(info) << "Region event: " << info.event << ": "
<< (info.managed ? "managed" : "unmanaged") << (info.managed ? "managed" : "unmanaged")
<< ", id: " << info.id << ", id: " << info.id
@ -36,7 +36,7 @@ struct Sampler : fair::mq::Device
regionCfg.linger = fLinger; // delay in ms before region destruction to collect outstanding events regionCfg.linger = fLinger; // delay in ms before region destruction to collect outstanding events
regionCfg.lock = true; // mlock region after creation regionCfg.lock = true; // mlock region after creation
regionCfg.zero = true; // zero region content after creation regionCfg.zero = true; // zero region content after creation
fRegion = FairMQUnmanagedRegionPtr(NewUnmanagedRegionFor("data", // region is created using the transport of this channel... fRegion = fair::mq::UnmanagedRegionPtr(NewUnmanagedRegionFor("data", // region is created using the transport of this channel...
0, // ... and this sub-channel 0, // ... and this sub-channel
10000000, // region size 10000000, // region size
[this](const std::vector<fair::mq::RegionBlock>& blocks) { // callback to be called when message buffers no longer needed by transport [this](const std::vector<fair::mq::RegionBlock>& blocks) { // callback to be called when message buffers no longer needed by transport
@ -50,7 +50,7 @@ struct Sampler : fair::mq::Device
bool ConditionalRun() override bool ConditionalRun() override
{ {
FairMQMessagePtr msg(NewMessageFor("data", // channel fair::mq::MessagePtr msg(NewMessageFor("data", // channel
0, // sub-channel 0, // sub-channel
fRegion, // region fRegion, // region
fRegion->GetData(), // ptr within region fRegion->GetData(), // ptr within region
@ -93,7 +93,7 @@ struct Sampler : fair::mq::Device
uint32_t fLinger = 100; uint32_t fLinger = 100;
uint64_t fMaxIterations = 0; uint64_t fMaxIterations = 0;
uint64_t fNumIterations = 0; uint64_t fNumIterations = 0;
FairMQUnmanagedRegionPtr fRegion = nullptr; fair::mq::UnmanagedRegionPtr fRegion = nullptr;
std::mutex fMtx; std::mutex fMtx;
uint64_t fNumUnackedMsgs = 0; uint64_t fNumUnackedMsgs = 0;
}; };

4
examples/req-rep/client.cxx
View File

@ -32,11 +32,11 @@ struct Client : fair::mq::Device
// its size, // its size,
// custom deletion function (called when transfer is done), // custom deletion function (called when transfer is done),
// and pointer to the object managing the data buffer // and pointer to the object managing the data buffer
FairMQMessagePtr req(NewMessage(const_cast<char*>(text->c_str()), // data fair::mq::MessagePtr req(NewMessage(const_cast<char*>(text->c_str()), // data
text->length(), // size text->length(), // size
[](void* /*data*/, void* object) { delete static_cast<std::string*>(object); }, // deletion callback [](void* /*data*/, void* object) { delete static_cast<std::string*>(object); }, // deletion callback
text)); // object that manages the data text)); // object that manages the data
FairMQMessagePtr rep(NewMessage()); fair::mq::MessagePtr rep(NewMessage());
LOG(info) << "Sending \"" << fText << "\" to server."; LOG(info) << "Sending \"" << fText << "\" to server.";

4
examples/req-rep/server.cxx
View File

@ -26,7 +26,7 @@ struct Server : fair::mq::Device
fMaxIterations = fConfig->GetProperty<uint64_t>("max-iterations"); fMaxIterations = fConfig->GetProperty<uint64_t>("max-iterations");
} }
bool HandleData(FairMQMessagePtr& req, int) bool HandleData(fair::mq::MessagePtr& req, int)
{ {
LOG(info) << "Received request from client: \"" << std::string(static_cast<char*>(req->GetData()), req->GetSize()) << "\""; LOG(info) << "Received request from client: \"" << std::string(static_cast<char*>(req->GetData()), req->GetSize()) << "\"";
@ -34,7 +34,7 @@ struct Server : fair::mq::Device
LOG(info) << "Sending reply to client."; LOG(info) << "Sending reply to client.";
FairMQMessagePtr rep(NewMessage(const_cast<char*>(text->c_str()), // data fair::mq::MessagePtr rep(NewMessage(const_cast<char*>(text->c_str()), // data
text->length(), // size text->length(), // size
[](void* /*data*/, void* object) { delete static_cast<std::string*>(object); }, // deletion callback [](void* /*data*/, void* object) { delete static_cast<std::string*>(object); }, // deletion callback
text)); // object that manages the data text)); // object that manages the data

50
fairmq/JSONParser.cxx
View File

@ -30,13 +30,13 @@
using namespace std; using namespace std;
using namespace fair::mq; using namespace fair::mq;
using namespace fair::mq::tools; using namespace tools;
using namespace boost::property_tree; using namespace boost::property_tree;
namespace fair::mq namespace fair::mq
{ {
fair::mq::Properties PtreeParser(const ptree& pt, const string& id) Properties PtreeParser(const ptree& pt, const string& id)
{ {
if (id.empty()) { if (id.empty()) {
throw ParserError("no device ID provided. Provide with `--id` cmd option"); throw ParserError("no device ID provided. Provide with `--id` cmd option");
@ -47,7 +47,7 @@ fair::mq::Properties PtreeParser(const ptree& pt, const string& id)
return helper::DeviceParser(pt.get_child("fairMQOptions"), id); return helper::DeviceParser(pt.get_child("fairMQOptions"), id);
} }
fair::mq::Properties JSONParser(const string& filename, const string& deviceId) Properties JSONParser(const string& filename, const string& deviceId)
{ {
ptree pt; ptree pt;
LOG(debug) << "Parsing JSON from " << filename << " ..."; LOG(debug) << "Parsing JSON from " << filename << " ...";
@ -58,9 +58,9 @@ fair::mq::Properties JSONParser(const string& filename, const string& deviceId)
namespace helper namespace helper
{ {
fair::mq::Properties DeviceParser(const ptree& fairMQOptions, const string& deviceId) Properties DeviceParser(const ptree& fairMQOptions, const string& deviceId)
{ {
fair::mq::Properties properties; Properties properties;
for (const auto& node : fairMQOptions) { for (const auto& node : fairMQOptions) {
if (node.first == "devices") { if (node.first == "devices") {
@ -82,27 +82,27 @@ fair::mq::Properties DeviceParser(const ptree& fairMQOptions, const string& devi
return properties; return properties;
} }
void ChannelParser(const ptree& tree, fair::mq::Properties& properties) void ChannelParser(const ptree& tree, Properties& properties)
{ {
for (const auto& node : tree) { for (const auto& node : tree) {
if (node.first == "channels") { if (node.first == "channels") {
for (const auto& cn : node.second) { for (const auto& cn : node.second) {
fair::mq::Properties commonProperties; Properties commonProperties;
commonProperties.emplace("type", cn.second.get<string>("type", FairMQChannel::DefaultType)); commonProperties.emplace("type", cn.second.get<string>("type", Channel::DefaultType));
commonProperties.emplace("method", cn.second.get<string>("method", FairMQChannel::DefaultMethod)); commonProperties.emplace("method", cn.second.get<string>("method", Channel::DefaultMethod));
commonProperties.emplace("address", cn.second.get<string>("address", FairMQChannel::DefaultAddress)); commonProperties.emplace("address", cn.second.get<string>("address", Channel::DefaultAddress));
commonProperties.emplace("transport", cn.second.get<string>("transport", FairMQChannel::DefaultTransportName)); commonProperties.emplace("transport", cn.second.get<string>("transport", Channel::DefaultTransportName));
commonProperties.emplace("sndBufSize", cn.second.get<int>("sndBufSize", FairMQChannel::DefaultSndBufSize)); commonProperties.emplace("sndBufSize", cn.second.get<int>("sndBufSize", Channel::DefaultSndBufSize));
commonProperties.emplace("rcvBufSize", cn.second.get<int>("rcvBufSize", FairMQChannel::DefaultRcvBufSize)); commonProperties.emplace("rcvBufSize", cn.second.get<int>("rcvBufSize", Channel::DefaultRcvBufSize));
commonProperties.emplace("sndKernelSize", cn.second.get<int>("sndKernelSize", FairMQChannel::DefaultSndKernelSize)); commonProperties.emplace("sndKernelSize", cn.second.get<int>("sndKernelSize", Channel::DefaultSndKernelSize));
commonProperties.emplace("rcvKernelSize", cn.second.get<int>("rcvKernelSize", FairMQChannel::DefaultRcvKernelSize)); commonProperties.emplace("rcvKernelSize", cn.second.get<int>("rcvKernelSize", Channel::DefaultRcvKernelSize));
commonProperties.emplace("sndTimeoutMs", cn.second.get<int>("sndTimeoutMs", FairMQChannel::DefaultSndTimeoutMs)); commonProperties.emplace("sndTimeoutMs", cn.second.get<int>("sndTimeoutMs", Channel::DefaultSndTimeoutMs));
commonProperties.emplace("rcvTimeoutMs", cn.second.get<int>("rcvTimeoutMs", FairMQChannel::DefaultRcvTimeoutMs)); commonProperties.emplace("rcvTimeoutMs", cn.second.get<int>("rcvTimeoutMs", Channel::DefaultRcvTimeoutMs));
commonProperties.emplace("linger", cn.second.get<int>("linger", FairMQChannel::DefaultLinger)); commonProperties.emplace("linger", cn.second.get<int>("linger", Channel::DefaultLinger));
commonProperties.emplace("rateLogging", cn.second.get<int>("rateLogging", FairMQChannel::DefaultRateLogging)); commonProperties.emplace("rateLogging", cn.second.get<int>("rateLogging", Channel::DefaultRateLogging));
commonProperties.emplace("portRangeMin", cn.second.get<int>("portRangeMin", FairMQChannel::DefaultPortRangeMin)); commonProperties.emplace("portRangeMin", cn.second.get<int>("portRangeMin", Channel::DefaultPortRangeMin));
commonProperties.emplace("portRangeMax", cn.second.get<int>("portRangeMax", FairMQChannel::DefaultPortRangeMax)); commonProperties.emplace("portRangeMax", cn.second.get<int>("portRangeMax", Channel::DefaultPortRangeMax));
commonProperties.emplace("autoBind", cn.second.get<bool>("autoBind", FairMQChannel::DefaultAutoBind)); commonProperties.emplace("autoBind", cn.second.get<bool>("autoBind", Channel::DefaultAutoBind));
string name = cn.second.get<string>("name"); string name = cn.second.get<string>("name");
int numSockets = cn.second.get<int>("numSockets", 0); int numSockets = cn.second.get<int>("numSockets", 0);
@ -128,7 +128,7 @@ void ChannelParser(const ptree& tree, fair::mq::Properties& properties)
} }
} }
void SubChannelParser(const ptree& channelTree, fair::mq::Properties& properties, const string& channelName, const fair::mq::Properties& commonProperties) void SubChannelParser(const ptree& channelTree, Properties& properties, const string& channelName, const Properties& commonProperties)
{ {
// for each socket in channel // for each socket in channel
int i = 0; int i = 0;
@ -137,7 +137,7 @@ void SubChannelParser(const ptree& channelTree, fair::mq::Properties& properties
if (node.first == "sockets") { if (node.first == "sockets") {
for (const auto& sn : node.second) { for (const auto& sn : node.second) {
// a sub-channel inherits relevant properties from the common channel ... // a sub-channel inherits relevant properties from the common channel ...
fair::mq::Properties newProperties(commonProperties); Properties newProperties(commonProperties);
// ... and adds/overwrites its own properties // ... and adds/overwrites its own properties
newProperties["type"] = sn.second.get<string>("type", boost::any_cast<string>(commonProperties.at("type"))); newProperties["type"] = sn.second.get<string>("type", boost::any_cast<string>(commonProperties.at("type")));
@ -177,7 +177,7 @@ void SubChannelParser(const ptree& channelTree, fair::mq::Properties& properties
LOG(trace) << "\tNo sockets specified,"; LOG(trace) << "\tNo sockets specified,";
LOG(trace) << "\tapplying common settings to the channel:"; LOG(trace) << "\tapplying common settings to the channel:";
fair::mq::Properties newProperties(commonProperties); Properties newProperties(commonProperties);
for (auto& p : newProperties) { for (auto& p : newProperties) {
LOG(trace) << "\t" << setw(13) << left << p.first << " : " << p.second; LOG(trace) << "\t" << setw(13) << left << p.first << " : " << p.second;

8
fairmq/MemoryResourceTools.h
View File

@ -12,7 +12,7 @@
/// ///
/// @author Mikolaj Krzewicki, mkrzewic@cern.ch /// @author Mikolaj Krzewicki, mkrzewic@cern.ch
#include <fairmq/FairMQTransportFactory.h> #include <fairmq/TransportFactory.h>
#include <fairmq/MemoryResources.h> #include <fairmq/MemoryResources.h>
namespace fair::mq namespace fair::mq
@ -28,13 +28,13 @@ template<typename ContainerT>
// pmr::polymorphic_allocator<typename // pmr::polymorphic_allocator<typename
// ContainerT::value_type>, // ContainerT::value_type>,
// typename ContainerT::allocator_type>::value == true, // typename ContainerT::allocator_type>::value == true,
// FairMQMessagePtr>::type // MessagePtr>::type
FairMQMessagePtr getMessage(ContainerT &&container_, FairMQMemoryResource *targetResource = nullptr) MessagePtr getMessage(ContainerT &&container_, MemoryResource *targetResource = nullptr)
{ {
auto container = std::move(container_); auto container = std::move(container_);
auto alloc = container.get_allocator(); auto alloc = container.get_allocator();
auto resource = dynamic_cast<FairMQMemoryResource *>(alloc.resource()); auto resource = dynamic_cast<MemoryResource *>(alloc.resource());
if (!resource && !targetResource) { if (!resource && !targetResource) {
throw std::runtime_error("Neither the container or target resource specified"); throw std::runtime_error("Neither the container or target resource specified");
} }

2
fairmq/Plugin.cxx
View File

@ -7,7 +7,7 @@
********************************************************************************/ ********************************************************************************/
#include <fairmq/Plugin.h> #include <fairmq/Plugin.h>
#include <FairMQLogger.h> #include <fairlogger/Logger.h>
#include <utility> #include <utility>
using namespace std; using namespace std;

2
fairmq/ProgOptions.cxx
View File

@ -351,7 +351,7 @@ void ProgOptions::DeleteProperty(const string& key)
vm.erase(key); vm.erase(key);
} }
void ProgOptions::AddChannel(const string& name, const FairMQChannel& channel) void ProgOptions::AddChannel(const string& name, const Channel& channel)
{ {
lock_guard<mutex> lock(fMtx); lock_guard<mutex> lock(fMtx);
unordered_map<string, int> existingChannels = GetChannelInfoImpl(); unordered_map<string, int> existingChannels = GetChannelInfoImpl();

6
fairmq/devices/BenchmarkSampler.h
View File

@ -44,7 +44,7 @@ class BenchmarkSampler : public Device
void Run() override void Run() override
{ {
// store the channel reference to avoid traversing the map on every loop iteration // store the channel reference to avoid traversing the map on every loop iteration
FairMQChannel& dataOutChannel = GetChannel(fOutChannelName, 0); Channel& dataOutChannel = GetChannel(fOutChannelName, 0);
LOG(info) << "Starting the benchmark with message size of " << fMsgSize << " and " << fMaxIterations << " iterations."; LOG(info) << "Starting the benchmark with message size of " << fMsgSize << " and " << fMaxIterations << " iterations.";
auto tStart = std::chrono::high_resolution_clock::now(); auto tStart = std::chrono::high_resolution_clock::now();
@ -53,7 +53,7 @@ class BenchmarkSampler : public Device
while (!NewStatePending()) { while (!NewStatePending()) {
if (fMultipart) { if (fMultipart) {
FairMQParts parts; Parts parts;
for (size_t i = 0; i < fNumParts; ++i) { for (size_t i = 0; i < fNumParts; ++i) {
parts.AddPart(dataOutChannel.NewMessage(fMsgSize, fair::mq::Alignment{fMsgAlignment})); parts.AddPart(dataOutChannel.NewMessage(fMsgSize, fair::mq::Alignment{fMsgAlignment}));
@ -71,7 +71,7 @@ class BenchmarkSampler : public Device
++fNumIterations; ++fNumIterations;
} }
} else { } else {
FairMQMessagePtr msg(dataOutChannel.NewMessage(fMsgSize, fair::mq::Alignment{fMsgAlignment})); MessagePtr msg(dataOutChannel.NewMessage(fMsgSize, fair::mq::Alignment{fMsgAlignment}));
if (fMemSet) { if (fMemSet) {
std::memset(msg->GetData(), 0, msg->GetSize()); std::memset(msg->GetData(), 0, msg->GetSize());
} }

10
fairmq/devices/Merger.h
View File

@ -9,7 +9,7 @@
#ifndef FAIR_MQ_MERGER_H #ifndef FAIR_MQ_MERGER_H
#define FAIR_MQ_MERGER_H #define FAIR_MQ_MERGER_H
#include <FairMQPoller.h> #include <fairmq/Poller.h>
#include <fairmq/Device.h> #include <fairmq/Device.h>
#include <fairlogger/Logger.h> #include <fairlogger/Logger.h>
@ -45,13 +45,13 @@ class Merger : public Device
{ {
int numInputs = GetNumSubChannels(fInChannelName); int numInputs = GetNumSubChannels(fInChannelName);
std::vector<FairMQChannel*> chans; std::vector<Channel*> chans;
for (auto& chan : fChannels.at(fInChannelName)) { for (auto& chan : fChannels.at(fInChannelName)) {
chans.push_back(&chan); chans.push_back(&chan);
} }
FairMQPollerPtr poller(NewPoller(chans)); PollerPtr poller(NewPoller(chans));
if (fMultipart) { if (fMultipart) {
while (!NewStatePending()) { while (!NewStatePending()) {
@ -61,7 +61,7 @@ class Merger : public Device
for (int i = 0; i < numInputs; ++i) { for (int i = 0; i < numInputs; ++i) {
// Check if the channel has data ready to be received. // Check if the channel has data ready to be received.
if (poller->CheckInput(i)) { if (poller->CheckInput(i)) {
FairMQParts payload; Parts payload;
if (Receive(payload, fInChannelName, i) >= 0) { if (Receive(payload, fInChannelName, i) >= 0) {
if (Send(payload, fOutChannelName) < 0) { if (Send(payload, fOutChannelName) < 0) {
@ -83,7 +83,7 @@ class Merger : public Device
for (int i = 0; i < numInputs; ++i) { for (int i = 0; i < numInputs; ++i) {
// Check if the channel has data ready to be received. // Check if the channel has data ready to be received.
if (poller->CheckInput(i)) { if (poller->CheckInput(i)) {
FairMQMessagePtr payload(fTransportFactory->CreateMessage()); MessagePtr payload(fTransportFactory->CreateMessage());
if (Receive(payload, fInChannelName, i) >= 0) { if (Receive(payload, fInChannelName, i) >= 0) {
if (Send(payload, fOutChannelName) < 0) { if (Send(payload, fOutChannelName) < 0) {

16
fairmq/devices/Multiplier.h
View File

@ -40,11 +40,11 @@ class Multiplier : public Device
} }
bool HandleSingleData(std::unique_ptr<FairMQMessage>& payload, int) bool HandleSingleData(std::unique_ptr<Message>& payload, int)
{ {
for (unsigned int i = 0; i < fOutChannelNames.size() - 1; ++i) { // all except last channel for (unsigned int i = 0; i < fOutChannelNames.size() - 1; ++i) { // all except last channel
for (unsigned int j = 0; j < GetNumSubChannels(fOutChannelNames.at(i)); ++j) { // all subChannels in a channel for (unsigned int j = 0; j < GetNumSubChannels(fOutChannelNames.at(i)); ++j) { // all subChannels in a channel
FairMQMessagePtr msgCopy(fTransportFactory->CreateMessage()); MessagePtr msgCopy(fTransportFactory->CreateMessage());
msgCopy->Copy(*payload); msgCopy->Copy(*payload);
Send(msgCopy, fOutChannelNames.at(i), j); Send(msgCopy, fOutChannelNames.at(i), j);
@ -54,7 +54,7 @@ class Multiplier : public Device
unsigned int lastChannelSize = GetNumSubChannels(fOutChannelNames.back()); unsigned int lastChannelSize = GetNumSubChannels(fOutChannelNames.back());
for (unsigned int i = 0; i < lastChannelSize - 1; ++i) { // iterate over all except last subChannels of the last channel for (unsigned int i = 0; i < lastChannelSize - 1; ++i) { // iterate over all except last subChannels of the last channel
FairMQMessagePtr msgCopy(fTransportFactory->CreateMessage()); MessagePtr msgCopy(fTransportFactory->CreateMessage());
msgCopy->Copy(*payload); msgCopy->Copy(*payload);
Send(msgCopy, fOutChannelNames.back(), i); Send(msgCopy, fOutChannelNames.back(), i);
@ -65,14 +65,14 @@ class Multiplier : public Device
return true; return true;
} }
bool HandleMultipartData(FairMQParts& payload, int) bool HandleMultipartData(Parts& payload, int)
{ {
for (unsigned int i = 0; i < fOutChannelNames.size() - 1; ++i) { // all except last channel for (unsigned int i = 0; i < fOutChannelNames.size() - 1; ++i) { // all except last channel
for (unsigned int j = 0; j < GetNumSubChannels(fOutChannelNames.at(i)); ++j) { // all subChannels in a channel for (unsigned int j = 0; j < GetNumSubChannels(fOutChannelNames.at(i)); ++j) { // all subChannels in a channel
FairMQParts parts; Parts parts;
for (int k = 0; k < payload.Size(); ++k) { for (int k = 0; k < payload.Size(); ++k) {
FairMQMessagePtr msgCopy(fTransportFactory->CreateMessage()); MessagePtr msgCopy(fTransportFactory->CreateMessage());
msgCopy->Copy(payload.AtRef(k)); msgCopy->Copy(payload.AtRef(k));
parts.AddPart(std::move(msgCopy)); parts.AddPart(std::move(msgCopy));
} }
@ -84,10 +84,10 @@ class Multiplier : public Device
unsigned int lastChannelSize = GetNumSubChannels(fOutChannelNames.back()); unsigned int lastChannelSize = GetNumSubChannels(fOutChannelNames.back());
for (unsigned int i = 0; i < lastChannelSize - 1; ++i) { // iterate over all except last subChannels of the last channel for (unsigned int i = 0; i < lastChannelSize - 1; ++i) { // iterate over all except last subChannels of the last channel
FairMQParts parts; Parts parts;
for (int k = 0; k < payload.Size(); ++k) { for (int k = 0; k < payload.Size(); ++k) {
FairMQMessagePtr msgCopy(fTransportFactory->CreateMessage()); MessagePtr msgCopy(fTransportFactory->CreateMessage());
msgCopy->Copy(payload.AtRef(k)); msgCopy->Copy(payload.AtRef(k));
parts.AddPart(std::move(msgCopy)); parts.AddPart(std::move(msgCopy));
} }

4
fairmq/devices/Proxy.h
View File

@ -34,7 +34,7 @@ class Proxy : public Device
{ {
if (fMultipart) { if (fMultipart) {
while (!NewStatePending()) { while (!NewStatePending()) {
FairMQParts payload; Parts payload;
if (Receive(payload, fInChannelName) >= 0) { if (Receive(payload, fInChannelName) >= 0) {
if (Send(payload, fOutChannelName) < 0) { if (Send(payload, fOutChannelName) < 0) {
LOG(debug) << "Transfer interrupted"; LOG(debug) << "Transfer interrupted";
@ -47,7 +47,7 @@ class Proxy : public Device
} }
} else { } else {
while (!NewStatePending()) { while (!NewStatePending()) {
FairMQMessagePtr payload(fTransportFactory->CreateMessage()); MessagePtr payload(fTransportFactory->CreateMessage());
if (Receive(payload, fInChannelName) >= 0) { if (Receive(payload, fInChannelName) >= 0) {
if (Send(payload, fOutChannelName) < 0) { if (Send(payload, fOutChannelName) < 0) {
LOG(debug) << "Transfer interrupted"; LOG(debug) << "Transfer interrupted";

7
fairmq/devices/Sink.h
View File

@ -9,7 +9,6 @@
#ifndef FAIR_MQ_SINK_H #ifndef FAIR_MQ_SINK_H
#define FAIR_MQ_SINK_H #define FAIR_MQ_SINK_H
#include <FairMQPoller.h>
#include <fairmq/Device.h> #include <fairmq/Device.h>
#include <fairmq/tools/Strings.h> #include <fairmq/tools/Strings.h>
@ -48,7 +47,7 @@ class Sink : public Device
void Run() override void Run() override
{ {
// store the channel reference to avoid traversing the map on every loop iteration // store the channel reference to avoid traversing the map on every loop iteration
FairMQChannel& dataInChannel = GetChannel(fInChannelName, 0); Channel& dataInChannel = GetChannel(fInChannelName, 0);
LOG(info) << "Starting sink and expecting to receive " << fMaxIterations << " messages."; LOG(info) << "Starting sink and expecting to receive " << fMaxIterations << " messages.";
auto tStart = std::chrono::high_resolution_clock::now(); auto tStart = std::chrono::high_resolution_clock::now();
@ -70,7 +69,7 @@ class Sink : public Device
while (!NewStatePending()) { while (!NewStatePending()) {
if (fMultipart) { if (fMultipart) {
FairMQParts parts; Parts parts;
if (dataInChannel.Receive(parts) < 0) { if (dataInChannel.Receive(parts) < 0) {
continue; continue;
} }
@ -80,7 +79,7 @@ class Sink : public Device
} }
} }
} else { } else {
FairMQMessagePtr msg(dataInChannel.NewMessage()); MessagePtr msg(dataInChannel.NewMessage());
if (dataInChannel.Receive(msg) < 0) { if (dataInChannel.Receive(msg) < 0) {
continue; continue;
} }

4
fairmq/devices/Splitter.h
View File

@ -34,9 +34,9 @@ class Splitter : public Device
fDirection = 0; fDirection = 0;
if (fMultipart) { if (fMultipart) {
OnData(fInChannelName, &Splitter::HandleData<FairMQParts>); OnData(fInChannelName, &Splitter::HandleData<Parts>);
} else { } else {
OnData(fInChannelName, &Splitter::HandleData<FairMQMessagePtr>); OnData(fInChannelName, &Splitter::HandleData<MessagePtr>);
} }
} }

2
fairmq/ofi/Socket.cxx
View File

@ -10,7 +10,7 @@
#include <fairmq/ofi/Socket.h> #include <fairmq/ofi/Socket.h>
#include <fairmq/ofi/TransportFactory.h> #include <fairmq/ofi/TransportFactory.h>
#include <fairmq/tools/Strings.h> #include <fairmq/tools/Strings.h>
#include <FairMQLogger.h> #include <fairlogger/Logger.h>
#include <asiofi.hpp> #include <asiofi.hpp>
#include <asio/buffer.hpp> #include <asio/buffer.hpp>

2
fairmq/plugins/PMIx/PMIxCommands.h
View File

@ -11,7 +11,7 @@
#include "PMIx.hpp" #include "PMIx.hpp"
#include <FairMQLogger.h> #include <fairlogger/Logger.h>
#include <fairmq/tools/Semaphore.h> #include <fairmq/tools/Semaphore.h>
#include <memory> // make_unique #include <memory> // make_unique
#include <string> #include <string>

2
fairmq/plugins/PMIx/PMIxPlugin.h
View File

@ -14,7 +14,7 @@
#include <fairmq/Plugin.h> #include <fairmq/Plugin.h>
#include <fairmq/Version.h> #include <fairmq/Version.h>
#include <FairMQLogger.h> #include <fairlogger/Logger.h>
#include <string> #include <string>
#include <sstream> #include <sstream>

4
fairmq/runFairMQDevice.h
View File

@ -11,7 +11,7 @@
#include <memory> #include <memory>
#include <string> #include <string>
using FairMQDevicePtr = FairMQDevice*; using FairMQDevicePtr = fair::mq::Device*;
// to be implemented by the user to return a child class of FairMQDevice // to be implemented by the user to return a child class of FairMQDevice
FairMQDevicePtr getDevice(const fair::mq::ProgOptions& config); FairMQDevicePtr getDevice(const fair::mq::ProgOptions& config);
@ -45,7 +45,7 @@ int main(int argc, char* argv[])
// }); // });
runner.AddHook<InstantiateDevice>([](DeviceRunner& r){ runner.AddHook<InstantiateDevice>([](DeviceRunner& r){
r.fDevice = std::unique_ptr<FairMQDevice>{getDevice(r.fConfig)}; r.fDevice = std::unique_ptr<fair::mq::Device>{getDevice(r.fConfig)};
}); });
return runner.Run(); return runner.Run();

16
fairmq/shmem/Message.h
View File

@ -35,7 +35,7 @@ class Message final : public fair::mq::Message
friend class Socket; friend class Socket;
public: public:
Message(Manager& manager, FairMQTransportFactory* factory = nullptr) Message(Manager& manager, fair::mq::TransportFactory* factory = nullptr)
: fair::mq::Message(factory) : fair::mq::Message(factory)
, fManager(manager) , fManager(manager)
, fQueued(false) , fQueued(false)
@ -46,7 +46,7 @@ class Message final : public fair::mq::Message
fManager.IncrementMsgCounter(); fManager.IncrementMsgCounter();
} }
Message(Manager& manager, Alignment alignment, FairMQTransportFactory* factory = nullptr) Message(Manager& manager, Alignment alignment, fair::mq::TransportFactory* factory = nullptr)
: fair::mq::Message(factory) : fair::mq::Message(factory)
, fManager(manager) , fManager(manager)
, fQueued(false) , fQueued(false)
@ -58,7 +58,7 @@ class Message final : public fair::mq::Message
fManager.IncrementMsgCounter(); fManager.IncrementMsgCounter();
} }
Message(Manager& manager, const size_t size, FairMQTransportFactory* factory = nullptr) Message(Manager& manager, const size_t size, fair::mq::TransportFactory* factory = nullptr)
: fair::mq::Message(factory) : fair::mq::Message(factory)
, fManager(manager) , fManager(manager)
, fQueued(false) , fQueued(false)
@ -70,7 +70,7 @@ class Message final : public fair::mq::Message
fManager.IncrementMsgCounter(); fManager.IncrementMsgCounter();
} }
Message(Manager& manager, const size_t size, Alignment alignment, FairMQTransportFactory* factory = nullptr) Message(Manager& manager, const size_t size, Alignment alignment, fair::mq::TransportFactory* factory = nullptr)
: fair::mq::Message(factory) : fair::mq::Message(factory)
, fManager(manager) , fManager(manager)
, fQueued(false) , fQueued(false)
@ -83,7 +83,7 @@ class Message final : public fair::mq::Message
fManager.IncrementMsgCounter(); fManager.IncrementMsgCounter();
} }
Message(Manager& manager, void* data, const size_t size, fairmq_free_fn* ffn, void* hint = nullptr, FairMQTransportFactory* factory = nullptr) Message(Manager& manager, void* data, const size_t size, fair::mq::FreeFn* ffn, void* hint = nullptr, fair::mq::TransportFactory* factory = nullptr)
: fair::mq::Message(factory) : fair::mq::Message(factory)
, fManager(manager) , fManager(manager)
, fQueued(false) , fQueued(false)
@ -102,7 +102,7 @@ class Message final : public fair::mq::Message
fManager.IncrementMsgCounter(); fManager.IncrementMsgCounter();
} }
Message(Manager& manager, UnmanagedRegionPtr& region, void* data, const size_t size, void* hint = 0, FairMQTransportFactory* factory = nullptr) Message(Manager& manager, UnmanagedRegionPtr& region, void* data, const size_t size, void* hint = 0, fair::mq::TransportFactory* factory = nullptr)
: fair::mq::Message(factory) : fair::mq::Message(factory)
, fManager(manager) , fManager(manager)
, fQueued(false) , fQueued(false)
@ -125,7 +125,7 @@ class Message final : public fair::mq::Message
fManager.IncrementMsgCounter(); fManager.IncrementMsgCounter();
} }
Message(Manager& manager, MetaHeader& hdr, FairMQTransportFactory* factory = nullptr) Message(Manager& manager, MetaHeader& hdr, fair::mq::TransportFactory* factory = nullptr)
: fair::mq::Message(factory) : fair::mq::Message(factory)
, fManager(manager) , fManager(manager)
, fQueued(false) , fQueued(false)
@ -169,7 +169,7 @@ class Message final : public fair::mq::Message
InitializeChunk(size, fAlignment); InitializeChunk(size, fAlignment);
} }
void Rebuild(void* data, size_t size, fairmq_free_fn* ffn, void* hint = nullptr) override void Rebuild(void* data, size_t size, fair::mq::FreeFn* ffn, void* hint = nullptr) override
{ {
CloseMessage(); CloseMessage();
fQueued = false; fQueued = false;

2
fairmq/shmem/Socket.h
View File

@ -51,7 +51,7 @@ struct ZMsg
class Socket final : public fair::mq::Socket class Socket final : public fair::mq::Socket
{ {
public: public:
Socket(Manager& manager, const std::string& type, const std::string& name, const std::string& id, void* context, FairMQTransportFactory* fac = nullptr) Socket(Manager& manager, const std::string& type, const std::string& name, const std::string& id, void* context, fair::mq::TransportFactory* fac = nullptr)
: fair::mq::Socket(fac) : fair::mq::Socket(fac)
, fManager(manager) , fManager(manager)
, fId(id + "." + name + "." + type) , fId(id + "." + name + "." + type)

8
fairmq/shmem/TransportFactory.h
View File

@ -113,7 +113,7 @@ class TransportFactory final : public fair::mq::TransportFactory
return std::make_unique<Message>(*fManager, size, alignment, this); return std::make_unique<Message>(*fManager, size, alignment, this);
} }
MessagePtr CreateMessage(void* data, size_t size, fairmq_free_fn* ffn, void* hint = nullptr) override MessagePtr CreateMessage(void* data, size_t size, fair::mq::FreeFn* ffn, void* hint = nullptr) override
{ {
return std::make_unique<Message>(*fManager, data, size, ffn, hint, this); return std::make_unique<Message>(*fManager, data, size, ffn, hint, this);
} }
@ -128,17 +128,17 @@ class TransportFactory final : public fair::mq::TransportFactory
return std::make_unique<Socket>(*fManager, type, name, GetId(), fZmqCtx, this); return std::make_unique<Socket>(*fManager, type, name, GetId(), fZmqCtx, this);
} }
PollerPtr CreatePoller(const std::vector<FairMQChannel>& channels) const override PollerPtr CreatePoller(const std::vector<Channel>& channels) const override
{ {
return std::make_unique<Poller>(channels); return std::make_unique<Poller>(channels);
} }
PollerPtr CreatePoller(const std::vector<FairMQChannel*>& channels) const override PollerPtr CreatePoller(const std::vector<Channel*>& channels) const override
{ {
return std::make_unique<Poller>(channels); return std::make_unique<Poller>(channels);
} }
PollerPtr CreatePoller(const std::unordered_map<std::string, std::vector<FairMQChannel>>& channelsMap, const std::vector<std::string>& channelList) const override PollerPtr CreatePoller(const std::unordered_map<std::string, std::vector<Channel>>& channelsMap, const std::vector<std::string>& channelList) const override
{ {
return std::make_unique<Poller>(channelsMap, channelList); return std::make_unique<Poller>(channelsMap, channelList);
} }

2
fairmq/shmem/UnmanagedRegionImpl.h
View File

@ -34,7 +34,7 @@ class UnmanagedRegionImpl final : public fair::mq::UnmanagedRegion
RegionCallback callback, RegionCallback callback,
RegionBulkCallback bulkCallback, RegionBulkCallback bulkCallback,
fair::mq::RegionConfig cfg, fair::mq::RegionConfig cfg,
FairMQTransportFactory* factory) fair::mq::TransportFactory* factory)
: fair::mq::UnmanagedRegion(factory) : fair::mq::UnmanagedRegion(factory)
, fManager(manager) , fManager(manager)
, fRegion(nullptr) , fRegion(nullptr)

5
fairmq/zeromq/Context.h
View File

@ -10,8 +10,9 @@
#define FAIR_MQ_ZMQ_CONTEXT_H_ #define FAIR_MQ_ZMQ_CONTEXT_H_
#include <fairmq/tools/Strings.h> #include <fairmq/tools/Strings.h>
#include <FairMQLogger.h> #include <fairmq/UnmanagedRegion.h>
#include <FairMQUnmanagedRegion.h>
#include <fairlogger/Logger.h>
#include <zmq.h> #include <zmq.h>

21
fairmq/zeromq/Message.h
View File

@ -10,9 +10,10 @@
#define FAIR_MQ_ZMQ_MESSAGE_H #define FAIR_MQ_ZMQ_MESSAGE_H
#include <fairmq/zeromq/UnmanagedRegion.h> #include <fairmq/zeromq/UnmanagedRegion.h>
#include <FairMQLogger.h> #include <fairmq/Message.h>
#include <FairMQMessage.h> #include <fairmq/UnmanagedRegion.h>
#include <FairMQUnmanagedRegion.h>
#include <fairlogger/Logger.h>
#include <zmq.h> #include <zmq.h>
@ -38,7 +39,7 @@ class Message final : public fair::mq::Message
Message& operator=(const Message&) = delete; Message& operator=(const Message&) = delete;
Message& operator=(Message&&) = delete; Message& operator=(Message&&) = delete;
Message(FairMQTransportFactory* factory = nullptr) Message(fair::mq::TransportFactory* factory = nullptr)
: fair::mq::Message(factory) : fair::mq::Message(factory)
, fMsg(std::make_unique<zmq_msg_t>()) , fMsg(std::make_unique<zmq_msg_t>())
{ {
@ -47,7 +48,7 @@ class Message final : public fair::mq::Message
} }
} }
Message(Alignment alignment, FairMQTransportFactory* factory = nullptr) Message(Alignment alignment, fair::mq::TransportFactory* factory = nullptr)
: fair::mq::Message(factory) : fair::mq::Message(factory)
, fAlignment(alignment.alignment) , fAlignment(alignment.alignment)
, fMsg(std::make_unique<zmq_msg_t>()) , fMsg(std::make_unique<zmq_msg_t>())
@ -57,7 +58,7 @@ class Message final : public fair::mq::Message
} }
} }
Message(const size_t size, FairMQTransportFactory* factory = nullptr) Message(const size_t size, fair::mq::TransportFactory* factory = nullptr)
: fair::mq::Message(factory) : fair::mq::Message(factory)
, fMsg(std::make_unique<zmq_msg_t>()) , fMsg(std::make_unique<zmq_msg_t>())
{ {
@ -80,7 +81,7 @@ class Message final : public fair::mq::Message
return {static_cast<void*>(fullBufferPtr), static_cast<void*>(alignedPartPtr)}; return {static_cast<void*>(fullBufferPtr), static_cast<void*>(alignedPartPtr)};
} }
Message(const size_t size, Alignment alignment, FairMQTransportFactory* factory = nullptr) Message(const size_t size, Alignment alignment, fair::mq::TransportFactory* factory = nullptr)
: fair::mq::Message(factory) : fair::mq::Message(factory)
, fAlignment(alignment.alignment) , fAlignment(alignment.alignment)
, fMsg(std::make_unique<zmq_msg_t>()) , fMsg(std::make_unique<zmq_msg_t>())
@ -97,7 +98,7 @@ class Message final : public fair::mq::Message
} }
} }
Message(void* data, const size_t size, fairmq_free_fn* ffn, void* hint = nullptr, FairMQTransportFactory* factory = nullptr) Message(void* data, const size_t size, fair::mq::FreeFn* ffn, void* hint = nullptr, fair::mq::TransportFactory* factory = nullptr)
: fair::mq::Message(factory) : fair::mq::Message(factory)
, fMsg(std::make_unique<zmq_msg_t>()) , fMsg(std::make_unique<zmq_msg_t>())
{ {
@ -106,7 +107,7 @@ class Message final : public fair::mq::Message
} }
} }
Message(UnmanagedRegionPtr& region, void* data, const size_t size, void* hint = 0, FairMQTransportFactory* factory = nullptr) Message(UnmanagedRegionPtr& region, void* data, const size_t size, void* hint = 0, fair::mq::TransportFactory* factory = nullptr)
: fair::mq::Message(factory) : fair::mq::Message(factory)
, fMsg(std::make_unique<zmq_msg_t>()) , fMsg(std::make_unique<zmq_msg_t>())
{ {
@ -184,7 +185,7 @@ class Message final : public fair::mq::Message
} }
} }
void Rebuild(void* data, size_t size, fairmq_free_fn* ffn, void* hint = nullptr) override void Rebuild(void* data, size_t size, fair::mq::FreeFn* ffn, void* hint = nullptr) override
{ {
CloseMessage(); CloseMessage();
fMsg = std::make_unique<zmq_msg_t>(); fMsg = std::make_unique<zmq_msg_t>();

11
fairmq/zeromq/Socket.h
View File

@ -9,14 +9,15 @@
#ifndef FAIR_MQ_ZMQ_SOCKET_H #ifndef FAIR_MQ_ZMQ_SOCKET_H
#define FAIR_MQ_ZMQ_SOCKET_H #define FAIR_MQ_ZMQ_SOCKET_H
#include <FairMQLogger.h> #include <fairmq/Message.h>
#include <FairMQMessage.h> #include <fairmq/Socket.h>
#include <FairMQSocket.h>
#include <fairmq/tools/Strings.h> #include <fairmq/tools/Strings.h>
#include <fairmq/zeromq/Common.h> #include <fairmq/zeromq/Common.h>
#include <fairmq/zeromq/Context.h> #include <fairmq/zeromq/Context.h>
#include <fairmq/zeromq/Message.h> #include <fairmq/zeromq/Message.h>
#include <fairlogger/Logger.h>
#include <zmq.h> #include <zmq.h>
#include <atomic> #include <atomic>
@ -30,7 +31,7 @@ namespace fair::mq::zmq
class Socket final : public fair::mq::Socket class Socket final : public fair::mq::Socket
{ {
public: public:
Socket(Context& ctx, const std::string& type, const std::string& name, const std::string& id, FairMQTransportFactory* factory = nullptr) Socket(Context& ctx, const std::string& type, const std::string& name, const std::string& id, fair::mq::TransportFactory* factory = nullptr)
: fair::mq::Socket(factory) : fair::mq::Socket(factory)
, fCtx(ctx) , fCtx(ctx)
, fId(id + "." + name + "." + type) , fId(id + "." + name + "." + type)
@ -219,7 +220,7 @@ class Socket final : public fair::mq::Socket
bool repeat = false; bool repeat = false;
do { do {
FairMQMessagePtr part = std::make_unique<Message>(GetTransport()); fair::mq::MessagePtr part = std::make_unique<Message>(GetTransport());
int nbytes = zmq_msg_recv(static_cast<Message*>(part.get())->GetMessage(), fSocket, flags); int nbytes = zmq_msg_recv(static_cast<Message*>(part.get())->GetMessage(), fSocket, flags);
if (nbytes >= 0) { if (nbytes >= 0) {

14
fairmq/zeromq/TransportFactory.h
View File

@ -14,7 +14,7 @@
#include <fairmq/zeromq/Socket.h> #include <fairmq/zeromq/Socket.h>
#include <fairmq/zeromq/Poller.h> #include <fairmq/zeromq/Poller.h>
#include <fairmq/zeromq/UnmanagedRegion.h> #include <fairmq/zeromq/UnmanagedRegion.h>
#include <FairMQTransportFactory.h> #include <fairmq/TransportFactory.h>
#include <fairmq/ProgOptions.h> #include <fairmq/ProgOptions.h>
#include <memory> // unique_ptr, make_unique #include <memory> // unique_ptr, make_unique
@ -24,11 +24,11 @@
namespace fair::mq::zmq namespace fair::mq::zmq
{ {
class TransportFactory final : public FairMQTransportFactory class TransportFactory final : public fair::mq::TransportFactory
{ {
public: public:
TransportFactory(const std::string& id = "", const ProgOptions* config = nullptr) TransportFactory(const std::string& id = "", const ProgOptions* config = nullptr)
: FairMQTransportFactory(id) : fair::mq::TransportFactory(id)
, fCtx(nullptr) , fCtx(nullptr)
{ {
int major = 0, minor = 0, patch = 0; int major = 0, minor = 0, patch = 0;
@ -68,7 +68,7 @@ class TransportFactory final : public FairMQTransportFactory
return std::make_unique<Message>(size, alignment, this); return std::make_unique<Message>(size, alignment, this);
} }
MessagePtr CreateMessage(void* data, size_t size, fairmq_free_fn* ffn, void* hint = nullptr) override MessagePtr CreateMessage(void* data, size_t size, fair::mq::FreeFn* ffn, void* hint = nullptr) override
{ {
return std::make_unique<Message>(data, size, ffn, hint, this); return std::make_unique<Message>(data, size, ffn, hint, this);
} }
@ -83,17 +83,17 @@ class TransportFactory final : public FairMQTransportFactory
return std::make_unique<Socket>(*fCtx, type, name, GetId(), this); return std::make_unique<Socket>(*fCtx, type, name, GetId(), this);
} }
PollerPtr CreatePoller(const std::vector<FairMQChannel>& channels) const override PollerPtr CreatePoller(const std::vector<Channel>& channels) const override
{ {
return std::make_unique<Poller>(channels); return std::make_unique<Poller>(channels);
} }
PollerPtr CreatePoller(const std::vector<FairMQChannel*>& channels) const override PollerPtr CreatePoller(const std::vector<Channel*>& channels) const override
{ {
return std::make_unique<Poller>(channels); return std::make_unique<Poller>(channels);
} }
PollerPtr CreatePoller(const std::unordered_map<std::string, std::vector<FairMQChannel>>& channelsMap, const std::vector<std::string>& channelList) const override PollerPtr CreatePoller(const std::unordered_map<std::string, std::vector<Channel>>& channelsMap, const std::vector<std::string>& channelList) const override
{ {
return std::make_unique<Poller>(channelsMap, channelList); return std::make_unique<Poller>(channelsMap, channelList);
} }

7
fairmq/zeromq/UnmanagedRegion.h
View File

@ -10,8 +10,9 @@
#define FAIR_MQ_ZMQ_UNMANAGEDREGION_H #define FAIR_MQ_ZMQ_UNMANAGEDREGION_H
#include <fairmq/zeromq/Context.h> #include <fairmq/zeromq/Context.h>
#include <FairMQUnmanagedRegion.h> #include <fairmq/UnmanagedRegion.h>
#include <FairMQLogger.h>
#include <fairlogger/Logger.h>
#include <cstddef> // size_t #include <cstddef> // size_t
#include <string> #include <string>
@ -33,7 +34,7 @@ class UnmanagedRegion final : public fair::mq::UnmanagedRegion
int64_t userFlags, int64_t userFlags,
RegionCallback callback, RegionCallback callback,
RegionBulkCallback bulkCallback, RegionBulkCallback bulkCallback,
FairMQTransportFactory* factory, fair::mq::TransportFactory* factory,
fair::mq::RegionConfig cfg) fair::mq::RegionConfig cfg)
: fair::mq::UnmanagedRegion(factory) : fair::mq::UnmanagedRegion(factory)
, fCtx(ctx) , fCtx(ctx)

4
test/device/TestReceiver.h
View File

@ -9,8 +9,8 @@
#ifndef FAIR_MQ_TEST_TESTRECEIVER_H #ifndef FAIR_MQ_TEST_TESTRECEIVER_H
#define FAIR_MQ_TEST_TESTRECEIVER_H #define FAIR_MQ_TEST_TESTRECEIVER_H
#include <FairMQDevice.h> #include <fairmq/Device.h>
#include <FairMQLogger.h> #include <fairlogger/Logger.h>
#include <string> #include <string>
#include <thread> #include <thread>

4
test/device/TestSender.h
View File

@ -9,8 +9,8 @@
#ifndef FAIR_MQ_TEST_TESTSENDER_H #ifndef FAIR_MQ_TEST_TESTSENDER_H
#define FAIR_MQ_TEST_TESTSENDER_H #define FAIR_MQ_TEST_TESTSENDER_H
#include <FairMQDevice.h> #include <fairmq/Device.h>
#include <FairMQLogger.h> #include <fairlogger/Logger.h>
#include <string> #include <string>
#include <thread> #include <thread>

103
test/device/_config.cxx
View File

@ -6,7 +6,7 @@
* copied verbatim in the file "LICENSE" * * copied verbatim in the file "LICENSE" *
********************************************************************************/ ********************************************************************************/
#include <FairMQDevice.h> #include <fairmq/Device.h>
#include <fairmq/DeviceRunner.h> #include <fairmq/DeviceRunner.h>
#include <fairmq/ProgOptions.h> #include <fairmq/ProgOptions.h>
@ -19,52 +19,53 @@ namespace _config
{ {
using namespace std; using namespace std;
using namespace fair::mq;
void control(FairMQDevice& device) void control(Device& device)
{ {
device.ChangeState(fair::mq::Transition::InitDevice); device.ChangeState(Transition::InitDevice);
device.WaitForState(fair::mq::State::InitializingDevice); device.WaitForState(State::InitializingDevice);
device.ChangeState(fair::mq::Transition::CompleteInit); device.ChangeState(Transition::CompleteInit);
device.WaitForState(fair::mq::State::Initialized); device.WaitForState(State::Initialized);
device.ChangeState(fair::mq::Transition::Bind); device.ChangeState(Transition::Bind);
device.WaitForState(fair::mq::State::Bound); device.WaitForState(State::Bound);
device.ChangeState(fair::mq::Transition::Connect); device.ChangeState(Transition::Connect);
device.WaitForState(fair::mq::State::DeviceReady); device.WaitForState(State::DeviceReady);
device.ChangeState(fair::mq::Transition::InitTask); device.ChangeState(Transition::InitTask);
device.WaitForState(fair::mq::State::Ready); device.WaitForState(State::Ready);
device.ChangeState(fair::mq::Transition::Run); device.ChangeState(Transition::Run);
device.WaitForState(fair::mq::State::Ready); device.WaitForState(State::Ready);
device.ChangeState(fair::mq::Transition::ResetTask); device.ChangeState(Transition::ResetTask);
device.WaitForState(fair::mq::State::DeviceReady); device.WaitForState(State::DeviceReady);
device.ChangeState(fair::mq::Transition::ResetDevice); device.ChangeState(Transition::ResetDevice);
device.WaitForState(fair::mq::State::Idle); device.WaitForState(State::Idle);
device.ChangeState(fair::mq::Transition::End); device.ChangeState(Transition::End);
} }
class TestDevice : public FairMQDevice class TestDevice : public Device
{ {
public: public:
TestDevice(const string& transport) TestDevice(const string& transport)
{ {
fDeviceThread = thread(&FairMQDevice::RunStateMachine, this); fDeviceThread = thread(&Device::RunStateMachine, this);
SetTransport(transport); SetTransport(transport);
ChangeState(fair::mq::Transition::InitDevice); ChangeState(Transition::InitDevice);
WaitForState(fair::mq::State::InitializingDevice); WaitForState(State::InitializingDevice);
ChangeState(fair::mq::Transition::CompleteInit); ChangeState(Transition::CompleteInit);
WaitForState(fair::mq::State::Initialized); WaitForState(State::Initialized);
ChangeState(fair::mq::Transition::Bind); ChangeState(Transition::Bind);
WaitForState(fair::mq::State::Bound); WaitForState(State::Bound);
ChangeState(fair::mq::Transition::Connect); ChangeState(Transition::Connect);
WaitForState(fair::mq::State::DeviceReady); WaitForState(State::DeviceReady);
ChangeState(fair::mq::Transition::InitTask); ChangeState(Transition::InitTask);
WaitForState(fair::mq::State::Ready); WaitForState(State::Ready);
ChangeState(fair::mq::Transition::Run); ChangeState(Transition::Run);
} }
TestDevice(const TestDevice&) = delete; TestDevice(const TestDevice&) = delete;
@ -74,15 +75,15 @@ class TestDevice : public FairMQDevice
~TestDevice() override ~TestDevice() override
{ {
WaitForState(fair::mq::State::Running); WaitForState(State::Running);
ChangeState(fair::mq::Transition::Stop); ChangeState(Transition::Stop);
WaitForState(fair::mq::State::Ready); WaitForState(State::Ready);
ChangeState(fair::mq::Transition::ResetTask); ChangeState(Transition::ResetTask);
WaitForState(fair::mq::State::DeviceReady); WaitForState(State::DeviceReady);
ChangeState(fair::mq::Transition::ResetDevice); ChangeState(Transition::ResetDevice);
WaitForState(fair::mq::State::Idle); WaitForState(State::Idle);
ChangeState(fair::mq::Transition::End); ChangeState(Transition::End);
if (fDeviceThread.joinable()) { if (fDeviceThread.joinable()) {
fDeviceThread.join(); fDeviceThread.join();
@ -100,7 +101,7 @@ class Config : public ::testing::Test
string TestDeviceSetConfig(const string& transport) string TestDeviceSetConfig(const string& transport)
{ {
fair::mq::ProgOptions config; ProgOptions config;
vector<string> emptyArgs = {"dummy", "--id", "test", "--color", "false"}; vector<string> emptyArgs = {"dummy", "--id", "test", "--color", "false"};
@ -108,10 +109,10 @@ class Config : public ::testing::Test
config.SetProperty("transport", transport); config.SetProperty("transport", transport);
FairMQDevice device; Device device;
device.SetConfig(config); device.SetConfig(config);
FairMQChannel channel; Channel channel;
channel.UpdateType("pub"); channel.UpdateType("pub");
channel.UpdateMethod("connect"); channel.UpdateMethod("connect");
channel.UpdateAddress("tcp://localhost:5558"); channel.UpdateAddress("tcp://localhost:5558");
@ -130,14 +131,14 @@ class Config : public ::testing::Test
string TestDeviceSetConfigWithPlugins(const string& transport) string TestDeviceSetConfigWithPlugins(const string& transport)
{ {
fair::mq::ProgOptions config; ProgOptions config;
vector<string> emptyArgs = {"dummy", "--id", "test", "--color", "false"}; vector<string> emptyArgs = {"dummy", "--id", "test", "--color", "false"};
config.SetProperty("transport", transport); config.SetProperty("transport", transport);
FairMQDevice device; Device device;
fair::mq::PluginManager mgr; PluginManager mgr;
mgr.LoadPlugin("s:config"); mgr.LoadPlugin("s:config");
mgr.ForEachPluginProgOptions([&](boost::program_options::options_description options) { mgr.ForEachPluginProgOptions([&](boost::program_options::options_description options) {
config.AddToCmdLineOptions(options); config.AddToCmdLineOptions(options);
@ -146,10 +147,10 @@ class Config : public ::testing::Test
mgr.InstantiatePlugins(); mgr.InstantiatePlugins();
config.ParseAll(emptyArgs, true); config.ParseAll(emptyArgs, true);
fair::mq::DeviceRunner::HandleGeneralOptions(config); DeviceRunner::HandleGeneralOptions(config);
device.SetConfig(config); device.SetConfig(config);
FairMQChannel channel; Channel channel;
channel.UpdateType("pub"); channel.UpdateType("pub");
channel.UpdateMethod("connect"); channel.UpdateMethod("connect");
channel.UpdateAddress("tcp://localhost:5558"); channel.UpdateAddress("tcp://localhost:5558");
@ -177,16 +178,16 @@ class Config : public ::testing::Test
string TestDeviceSetTransport(const string& transport) string TestDeviceSetTransport(const string& transport)
{ {
FairMQDevice device; Device device;
device.SetTransport(transport); device.SetTransport(transport);
FairMQChannel channel; Channel channel;
channel.UpdateType("pub"); channel.UpdateType("pub");
channel.UpdateMethod("connect"); channel.UpdateMethod("connect");
channel.UpdateAddress("tcp://localhost:5558"); channel.UpdateAddress("tcp://localhost:5558");
device.AddChannel("data", std::move(channel)); device.AddChannel("data", std::move(channel));
thread t(&FairMQDevice::RunStateMachine, &device); thread t(&Device::RunStateMachine, &device);
control(device); control(device);

21
test/device/_error_state.cxx
View File

@ -12,7 +12,7 @@
#include <boost/process.hpp> #include <boost/process.hpp>
#include <fairmq/tools/Process.h> #include <fairmq/tools/Process.h>
#include <fairmq/tools/Unique.h> #include <fairmq/tools/Unique.h>
#include <FairMQDevice.h> #include <fairmq/Device.h>
#include <string> #include <string>
#include <thread> #include <thread>
@ -22,24 +22,25 @@ namespace
{ {
using namespace std; using namespace std;
using namespace fair::mq;
using namespace fair::mq::test; using namespace fair::mq::test;
using namespace fair::mq::tools; using namespace fair::mq::tools;
class BadDevice : public FairMQDevice class BadDevice : public Device
{ {
public: public:
BadDevice() BadDevice()
{ {
fDeviceThread = thread([&](){ fDeviceThread = thread([&](){
EXPECT_THROW(RunStateMachine(), fair::mq::MessageError); EXPECT_THROW(RunStateMachine(), MessageError);
}); });
SetTransport("shmem"); SetTransport("shmem");
ChangeState(fair::mq::Transition::InitDevice); ChangeState(Transition::InitDevice);
WaitForState(fair::mq::State::InitializingDevice); WaitForState(State::InitializingDevice);
ChangeState(fair::mq::Transition::CompleteInit); ChangeState(Transition::CompleteInit);
WaitForState(fair::mq::State::Initialized); WaitForState(State::Initialized);
parts.AddPart(NewMessage()); parts.AddPart(NewMessage());
} }
@ -51,7 +52,7 @@ class BadDevice : public FairMQDevice
~BadDevice() override ~BadDevice() override
{ {
ChangeState(fair::mq::Transition::ResetDevice); ChangeState(Transition::ResetDevice);
if (fDeviceThread.joinable()) { if (fDeviceThread.joinable()) {
fDeviceThread.join(); fDeviceThread.join();
@ -60,12 +61,12 @@ class BadDevice : public FairMQDevice
private: private:
thread fDeviceThread; thread fDeviceThread;
FairMQParts parts; Parts parts;
}; };
void RunErrorStateIn(const string& state, const string& control, const string& input = "") void RunErrorStateIn(const string& state, const string& control, const string& input = "")
{ {
size_t session{fair::mq::tools::UuidHash()}; size_t session{tools::UuidHash()};
execute_result result{"", 100}; execute_result result{"", 100};
thread device_thread([&]() { thread device_thread([&]() {

6
test/device/_multiple_devices.cxx
View File

@ -21,7 +21,7 @@ namespace
using namespace std; using namespace std;
using namespace fair::mq; using namespace fair::mq;
void control(FairMQDevice& device) void control(Device& device)
{ {
thread t([&] { thread t([&] {
device.ChangeState(Transition::InitDevice); device.ChangeState(Transition::InitDevice);
@ -62,7 +62,7 @@ class MultipleDevices : public ::testing::Test {
test::Sender sender("data"); test::Sender sender("data");
sender.SetTransport("zeromq"); sender.SetTransport("zeromq");
FairMQChannel channel("push", "connect", "ipc://multiple-devices-test"); Channel channel("push", "connect", "ipc://multiple-devices-test");
channel.UpdateRateLogging(0); channel.UpdateRateLogging(0);
sender.AddChannel("data", std::move(channel)); sender.AddChannel("data", std::move(channel));
@ -75,7 +75,7 @@ class MultipleDevices : public ::testing::Test {
test::Receiver receiver("data"); test::Receiver receiver("data");
receiver.SetTransport("zeromq"); receiver.SetTransport("zeromq");
FairMQChannel channel("pull", "bind", "ipc://multiple-devices-test"); Channel channel("pull", "bind", "ipc://multiple-devices-test");
channel.UpdateRateLogging(0); channel.UpdateRateLogging(0);
receiver.AddChannel("data", std::move(channel)); receiver.AddChannel("data", std::move(channel));

8
test/device/_transitions.cxx
View File

@ -6,8 +6,8 @@
* copied verbatim in the file "LICENSE" * * copied verbatim in the file "LICENSE" *
********************************************************************************/ ********************************************************************************/
#include <FairMQDevice.h> #include <fairmq/Device.h>
#include <FairMQLogger.h> #include <fairlogger/Logger.h>
#include <gtest/gtest.h> #include <gtest/gtest.h>
@ -20,7 +20,7 @@ namespace
using namespace std; using namespace std;
using namespace fair::mq; using namespace fair::mq;
class SlowDevice : public FairMQDevice class SlowDevice : public Device
{ {
public: public:
SlowDevice() = default; SlowDevice() = default;
@ -34,7 +34,7 @@ class SlowDevice : public FairMQDevice
void transitionTo(const vector<State>& states, int numExpectedStates) void transitionTo(const vector<State>& states, int numExpectedStates)
{ {
FairMQDevice device; Device device;
thread t([&] { thread t([&] {
for (const auto& s : states) { for (const auto& s : states) {

2
test/device/_version.cxx
View File

@ -6,7 +6,7 @@
* copied verbatim in the file "LICENSE" * * copied verbatim in the file "LICENSE" *
********************************************************************************/ ********************************************************************************/
#include <FairMQDevice.h> #include <fairmq/Device.h>
#include <fairmq/tools/Version.h> #include <fairmq/tools/Version.h>

4
test/helper/devices/TestErrorState.h
View File

@ -9,8 +9,8 @@
#ifndef FAIR_MQ_TEST_ERROR_STATE_H #ifndef FAIR_MQ_TEST_ERROR_STATE_H
#define FAIR_MQ_TEST_ERROR_STATE_H #define FAIR_MQ_TEST_ERROR_STATE_H
#include <FairMQDevice.h> #include <fairmq/Device.h>
#include <FairMQLogger.h> #include <fairlogger/Logger.h>
#include <iostream> #include <iostream>

4
test/helper/devices/TestExceptions.h
View File

@ -9,8 +9,8 @@
#ifndef FAIR_MQ_TEST_EXCEPTIONS_H #ifndef FAIR_MQ_TEST_EXCEPTIONS_H
#define FAIR_MQ_TEST_EXCEPTIONS_H #define FAIR_MQ_TEST_EXCEPTIONS_H
#include <FairMQDevice.h> #include <fairmq/Device.h>
#include <FairMQLogger.h> #include <fairlogger/Logger.h>
#include <iostream> #include <iostream>
#include <stdexcept> #include <stdexcept>

2
test/helper/devices/TestPairLeft.h
View File

@ -9,7 +9,7 @@
#ifndef FAIR_MQ_TEST_PAIRLEFT_H #ifndef FAIR_MQ_TEST_PAIRLEFT_H
#define FAIR_MQ_TEST_PAIRLEFT_H #define FAIR_MQ_TEST_PAIRLEFT_H
#include <FairMQDevice.h> #include <fairmq/Device.h>
#include <cstddef> #include <cstddef>
#include <thread> #include <thread>

2
test/helper/devices/TestPairRight.h
View File

@ -9,7 +9,7 @@
#ifndef FAIR_MQ_TEST_PAIRRIGHT_H #ifndef FAIR_MQ_TEST_PAIRRIGHT_H
#define FAIR_MQ_TEST_PAIRRIGHT_H #define FAIR_MQ_TEST_PAIRRIGHT_H
#include <FairMQDevice.h> #include <fairmq/Device.h>
#include <cstddef> #include <cstddef>
#include <string> #include <string>
#include <thread> #include <thread>

17
test/helper/devices/TestPollIn.h
View File

@ -9,17 +9,20 @@
#ifndef FAIR_MQ_TEST_POLLIN_H #ifndef FAIR_MQ_TEST_POLLIN_H
#define FAIR_MQ_TEST_POLLIN_H #define FAIR_MQ_TEST_POLLIN_H
#include <FairMQDevice.h> #include <fairmq/Device.h>
#include <FairMQLogger.h>
#include <fairmq/ProgOptions.h> #include <fairmq/ProgOptions.h>
#include <fairlogger/Logger.h>
#include <thread> #include <thread>
namespace fair::mq::test namespace fair::mq::test
{ {
using namespace std; using namespace std;
using namespace fair::mq;
class PollIn : public FairMQDevice class PollIn : public Device
{ {
public: public:
PollIn() PollIn()
@ -35,12 +38,12 @@ class PollIn : public FairMQDevice
auto Run() -> void override auto Run() -> void override
{ {
vector<FairMQChannel*> chans; vector<Channel*> chans;
chans.push_back(&GetChannel("data1", 0)); chans.push_back(&GetChannel("data1", 0));
chans.push_back(&GetChannel("data2", 0)); chans.push_back(&GetChannel("data2", 0));
FairMQPollerPtr poller = nullptr; PollerPtr poller = nullptr;
if (fPollType == 0) if (fPollType == 0)
{ {
@ -59,8 +62,8 @@ class PollIn : public FairMQDevice
bool arrived2 = false; bool arrived2 = false;
bool bothArrived = false; bool bothArrived = false;
FairMQMessagePtr msg1(NewMessage()); MessagePtr msg1(NewMessage());
FairMQMessagePtr msg2(NewMessage()); MessagePtr msg2(NewMessage());
while (!bothArrived) while (!bothArrived)
{ {

2
test/helper/devices/TestPollOut.h
View File

@ -9,7 +9,7 @@
#ifndef FAIR_MQ_TEST_POLLOUT_H #ifndef FAIR_MQ_TEST_POLLOUT_H
#define FAIR_MQ_TEST_POLLOUT_H #define FAIR_MQ_TEST_POLLOUT_H
#include <FairMQDevice.h> #include <fairmq/Device.h>
#include <thread> #include <thread>
namespace fair::mq::test namespace fair::mq::test

4
test/helper/devices/TestPub.h
View File

@ -9,8 +9,8 @@
#ifndef FAIR_MQ_TEST_PUB_H #ifndef FAIR_MQ_TEST_PUB_H
#define FAIR_MQ_TEST_PUB_H #define FAIR_MQ_TEST_PUB_H
#include <FairMQDevice.h> #include <fairmq/Device.h>
#include <FairMQLogger.h> #include <fairlogger/Logger.h>
#include <chrono> #include <chrono>
#include <thread> #include <thread>

4
test/helper/devices/TestPull.h
View File

@ -9,8 +9,8 @@
#ifndef FAIR_MQ_TEST_PULL_H #ifndef FAIR_MQ_TEST_PULL_H
#define FAIR_MQ_TEST_PULL_H #define FAIR_MQ_TEST_PULL_H
#include <FairMQDevice.h> #include <fairmq/Device.h>
#include <FairMQLogger.h> #include <fairlogger/Logger.h>
#include <thread> #include <thread>
namespace fair::mq::test namespace fair::mq::test

2
test/helper/devices/TestPush.h
View File

@ -9,7 +9,7 @@
#ifndef FAIR_MQ_TEST_PUSH_H #ifndef FAIR_MQ_TEST_PUSH_H
#define FAIR_MQ_TEST_PUSH_H #define FAIR_MQ_TEST_PUSH_H
#include <FairMQDevice.h> #include <fairmq/Device.h>
#include <thread> #include <thread>
namespace fair::mq::test namespace fair::mq::test

4
test/helper/devices/TestRep.h
View File

@ -9,8 +9,8 @@
#ifndef FAIR_MQ_TEST_REP_H #ifndef FAIR_MQ_TEST_REP_H
#define FAIR_MQ_TEST_REP_H #define FAIR_MQ_TEST_REP_H
#include <FairMQDevice.h> #include <fairmq/Device.h>
#include <FairMQLogger.h> #include <fairlogger/Logger.h>
#include <thread> #include <thread>
namespace fair::mq::test namespace fair::mq::test

4
test/helper/devices/TestReq.h
View File

@ -9,8 +9,8 @@
#ifndef FAIR_MQ_TEST_REQ_H #ifndef FAIR_MQ_TEST_REQ_H
#define FAIR_MQ_TEST_REQ_H #define FAIR_MQ_TEST_REQ_H
#include <FairMQDevice.h> #include <fairmq/Device.h>
#include <FairMQLogger.h> #include <fairlogger/Logger.h>
#include <thread> #include <thread>
namespace fair::mq::test namespace fair::mq::test

4
test/helper/devices/TestSignals.h
View File

@ -9,8 +9,8 @@
#ifndef FAIR_MQ_TEST_SIGNALS_H #ifndef FAIR_MQ_TEST_SIGNALS_H
#define FAIR_MQ_TEST_SIGNALS_H #define FAIR_MQ_TEST_SIGNALS_H
#include <FairMQDevice.h> #include <fairmq/Device.h>
#include <FairMQLogger.h> #include <fairlogger/Logger.h>
#include <iostream> #include <iostream>
#include <csignal> #include <csignal>

4
test/helper/devices/TestSub.h
View File

@ -9,8 +9,8 @@
#ifndef FAIR_MQ_TEST_SUB_H #ifndef FAIR_MQ_TEST_SUB_H
#define FAIR_MQ_TEST_SUB_H #define FAIR_MQ_TEST_SUB_H
#include <FairMQDevice.h> #include <fairmq/Device.h>
#include <FairMQLogger.h> #include <fairlogger/Logger.h>
#include <chrono> #include <chrono>
#include <thread> #include <thread>

21
test/helper/devices/TestTransferTimeout.h
View File

@ -9,13 +9,16 @@
#ifndef FAIR_MQ_TEST_TRANSFERTIMEOUT_H #ifndef FAIR_MQ_TEST_TRANSFERTIMEOUT_H
#define FAIR_MQ_TEST_TRANSFERTIMEOUT_H #define FAIR_MQ_TEST_TRANSFERTIMEOUT_H
#include <FairMQDevice.h> #include <fairmq/Device.h>
#include <FairMQLogger.h>
#include <fairlogger/Logger.h>
namespace fair::mq::test namespace fair::mq::test
{ {
class TransferTimeout : public FairMQDevice using namespace fair::mq;
class TransferTimeout : public Device
{ {
protected: protected:
auto Run() -> void override auto Run() -> void override
@ -38,8 +41,8 @@ class TransferTimeout : public FairMQDevice
bool send2PartsCancelingAfter0ms = false; bool send2PartsCancelingAfter0ms = false;
bool receive2PartsCancelingAfter0ms = false; bool receive2PartsCancelingAfter0ms = false;
FairMQMessagePtr msg1(NewMessage()); MessagePtr msg1(NewMessage());
FairMQMessagePtr msg2(NewMessage()); MessagePtr msg2(NewMessage());
if (Send(msg1, "data-out", 0, 200) == static_cast<int>(TransferCode::timeout)) { if (Send(msg1, "data-out", 0, 200) == static_cast<int>(TransferCode::timeout)) {
LOG(info) << "send msg canceled (200ms)"; LOG(info) << "send msg canceled (200ms)";
@ -69,9 +72,9 @@ class TransferTimeout : public FairMQDevice
LOG(error) << "receive msg did not cancel (0ms)"; LOG(error) << "receive msg did not cancel (0ms)";
} }
FairMQParts parts1; Parts parts1;
parts1.AddPart(NewMessage(10)); parts1.AddPart(NewMessage(10));
FairMQParts parts2; Parts parts2;
if (Send(parts1, "data-out", 0, 200) == static_cast<int>(TransferCode::timeout)) { if (Send(parts1, "data-out", 0, 200) == static_cast<int>(TransferCode::timeout)) {
LOG(info) << "send 1 part canceled (200ms)"; LOG(info) << "send 1 part canceled (200ms)";
@ -101,10 +104,10 @@ class TransferTimeout : public FairMQDevice
LOG(error) << "receive 1 part did not cancel (0ms)"; LOG(error) << "receive 1 part did not cancel (0ms)";
} }
FairMQParts parts3; Parts parts3;
parts3.AddPart(NewMessage(10)); parts3.AddPart(NewMessage(10));
parts3.AddPart(NewMessage(10)); parts3.AddPart(NewMessage(10));
FairMQParts parts4; Parts parts4;
if (Send(parts3, "data-out", 0, 200) == static_cast<int>(TransferCode::timeout)) { if (Send(parts3, "data-out", 0, 200) == static_cast<int>(TransferCode::timeout)) {
LOG(info) << "send 2 parts canceled (200ms)"; LOG(info) << "send 2 parts canceled (200ms)";

4
test/helper/devices/TestWaitFor.h
View File

@ -9,8 +9,8 @@
#ifndef FAIR_MQ_TEST_WAITFOR_H #ifndef FAIR_MQ_TEST_WAITFOR_H
#define FAIR_MQ_TEST_WAITFOR_H #define FAIR_MQ_TEST_WAITFOR_H
#include <FairMQDevice.h> #include <fairmq/Device.h>
#include <FairMQLogger.h> #include <fairlogger/Logger.h>
#include <iostream> #include <iostream>

16
test/memory_resources/_memory_resources.cxx
View File

@ -6,7 +6,7 @@
* copied verbatim in the file "LICENSE" * * copied verbatim in the file "LICENSE" *
********************************************************************************/ ********************************************************************************/
#include <fairmq/FairMQTransportFactory.h> #include <fairmq/TransportFactory.h>
#include <fairmq/MemoryResourceTools.h> #include <fairmq/MemoryResourceTools.h>
#include <fairmq/ProgOptions.h> #include <fairmq/ProgOptions.h>
#include <fairmq/tools/Unique.h> #include <fairmq/tools/Unique.h>
@ -24,7 +24,7 @@ namespace
using namespace std; using namespace std;
using namespace fair::mq; using namespace fair::mq;
using FactoryType = shared_ptr<FairMQTransportFactory>; using FactoryType = shared_ptr<TransportFactory>;
struct TestData struct TestData
{ {
@ -81,8 +81,8 @@ TEST(MemoryResources, transportAllocatorMap)
// config.SetProperty<string>("session", to_string(session)); // config.SetProperty<string>("session", to_string(session));
config.SetProperty<string>("session", "default"); config.SetProperty<string>("session", "default");
FactoryType factoryZMQ = FairMQTransportFactory::CreateTransportFactory("zeromq", fair::mq::tools::Uuid(), &config); FactoryType factoryZMQ = TransportFactory::CreateTransportFactory("zeromq", fair::mq::tools::Uuid(), &config);
FactoryType factorySHM = FairMQTransportFactory::CreateTransportFactory("shmem", fair::mq::tools::Uuid(), &config); FactoryType factorySHM = TransportFactory::CreateTransportFactory("shmem", fair::mq::tools::Uuid(), &config);
auto allocZMQ = factoryZMQ->GetMemoryResource(); auto allocZMQ = factoryZMQ->GetMemoryResource();
auto allocSHM = factorySHM->GetMemoryResource(); auto allocSHM = factorySHM->GetMemoryResource();
@ -103,7 +103,7 @@ TEST(MemoryResources, allocator)
ProgOptions config; ProgOptions config;
config.SetProperty<string>("session", to_string(session)); config.SetProperty<string>("session", to_string(session));
FactoryType factoryZMQ = FairMQTransportFactory::CreateTransportFactory("zeromq", fair::mq::tools::Uuid(), &config); FactoryType factoryZMQ = TransportFactory::CreateTransportFactory("zeromq", fair::mq::tools::Uuid(), &config);
auto allocZMQ = factoryZMQ->GetMemoryResource(); auto allocZMQ = factoryZMQ->GetMemoryResource();
@ -131,12 +131,12 @@ TEST(MemoryResources, getMessage)
ProgOptions config; ProgOptions config;
config.SetProperty<string>("session", to_string(session)); config.SetProperty<string>("session", to_string(session));
FactoryType factoryZMQ = FairMQTransportFactory::CreateTransportFactory("zeromq", fair::mq::tools::Uuid(), &config); FactoryType factoryZMQ = TransportFactory::CreateTransportFactory("zeromq", fair::mq::tools::Uuid(), &config);
FactoryType factorySHM = FairMQTransportFactory::CreateTransportFactory("shmem", fair::mq::tools::Uuid(), &config); FactoryType factorySHM = TransportFactory::CreateTransportFactory("shmem", fair::mq::tools::Uuid(), &config);
auto allocZMQ = factoryZMQ->GetMemoryResource(); auto allocZMQ = factoryZMQ->GetMemoryResource();
FairMQMessagePtr message{nullptr}; MessagePtr message{nullptr};
int* messageArray{nullptr}; int* messageArray{nullptr};

7
test/message/_message.cxx
View File

@ -6,7 +6,6 @@
* copied verbatim in the file "LICENSE" * * copied verbatim in the file "LICENSE" *
********************************************************************************/ ********************************************************************************/
#include <fairlogger/Logger.h>
#include <fairmq/Channel.h> #include <fairmq/Channel.h>
#include <fairmq/ProgOptions.h> #include <fairmq/ProgOptions.h>
#include <fairmq/tools/Semaphore.h> #include <fairmq/tools/Semaphore.h>
@ -15,6 +14,8 @@
#include <fairmq/TransportFactory.h> #include <fairmq/TransportFactory.h>
#include <fairmq/shmem/Message.h> #include <fairmq/shmem/Message.h>
#include <fairlogger/Logger.h>
#include <gtest/gtest.h> #include <gtest/gtest.h>
#include <array> #include <array>
@ -282,8 +283,8 @@ auto ZeroCopyFromUnmanaged(string const& address) -> void
}); });
{ {
FairMQChannel push("Push", "push", factory1); Channel push("Push", "push", factory1);
FairMQChannel pull("Pull", "pull", factory2); Channel pull("Pull", "pull", factory2);
push.Bind(address); push.Bind(address);
pull.Connect(address); pull.Connect(address);

12
test/parts/_iterator_interface.cxx
View File

@ -7,8 +7,8 @@
********************************************************************************/ ********************************************************************************/
#include <gtest/gtest.h> #include <gtest/gtest.h>
#include <FairMQParts.h> #include <fairmq/Parts.h>
#include <FairMQTransportFactory.h> #include <fairmq/TransportFactory.h>
#include <string> #include <string>
#include <sstream> #include <sstream>
#include <algorithm> #include <algorithm>
@ -21,11 +21,11 @@ using namespace std;
class RandomAccessIterator : public ::testing::Test { class RandomAccessIterator : public ::testing::Test {
protected: protected:
FairMQParts mParts; fair::mq::Parts mParts;
shared_ptr<FairMQTransportFactory> mFactory; shared_ptr<fair::mq::TransportFactory> mFactory;
RandomAccessIterator() RandomAccessIterator()
: mFactory(FairMQTransportFactory::CreateTransportFactory("zeromq")) : mFactory(fair::mq::TransportFactory::CreateTransportFactory("zeromq"))
{ {
mParts.AddPart(mFactory->NewSimpleMessage("1")); mParts.AddPart(mFactory->NewSimpleMessage("1"));
mParts.AddPart(mFactory->NewSimpleMessage("2")); mParts.AddPart(mFactory->NewSimpleMessage("2"));
@ -62,7 +62,7 @@ TEST_F(RandomAccessIterator, RangeForLoopMutation)
TEST_F(RandomAccessIterator, ForEachConst) TEST_F(RandomAccessIterator, ForEachConst)
{ {
stringstream out; stringstream out;
for_each(mParts.cbegin(), mParts.cend(), [&out](const FairMQMessagePtr& part) { for_each(mParts.cbegin(), mParts.cend(), [&out](const fair::mq::MessagePtr& part) {
out << string{static_cast<char*>(part->GetData()), part->GetSize()}; out << string{static_cast<char*>(part->GetData()), part->GetSize()};
}); });

12
test/plugin_services/Fixture.h
View File

@ -9,17 +9,19 @@
#ifndef FAIR_MQ_TEST_FIXTURE #ifndef FAIR_MQ_TEST_FIXTURE
#define FAIR_MQ_TEST_FIXTURE #define FAIR_MQ_TEST_FIXTURE
#include <gtest/gtest.h>
#include <fairmq/PluginServices.h> #include <fairmq/PluginServices.h>
#include <FairMQDevice.h> #include <fairmq/Device.h>
#include <fairmq/ProgOptions.h> #include <fairmq/ProgOptions.h>
#include <gtest/gtest.h>
#include <memory> #include <memory>
#include <thread> #include <thread>
namespace fair::mq::test namespace fair::mq::test
{ {
inline auto control(FairMQDevice& device) -> void inline auto control(fair::mq::Device& device) -> void
{ {
device.ChangeState(fair::mq::Transition::InitDevice); device.ChangeState(fair::mq::Transition::InitDevice);
device.WaitForState(fair::mq::State::InitializingDevice); device.WaitForState(fair::mq::State::InitializingDevice);
@ -42,7 +44,7 @@ struct PluginServices : ::testing::Test {
, mServices(mConfig, mDevice) , mServices(mConfig, mDevice)
, fRunStateMachineThread() , fRunStateMachineThread()
{ {
fRunStateMachineThread = std::thread(&FairMQDevice::RunStateMachine, &mDevice); fRunStateMachineThread = std::thread(&fair::mq::Device::RunStateMachine, &mDevice);
mDevice.SetTransport("zeromq"); mDevice.SetTransport("zeromq");
} }
@ -55,7 +57,7 @@ struct PluginServices : ::testing::Test {
} }
fair::mq::ProgOptions mConfig; fair::mq::ProgOptions mConfig;
FairMQDevice mDevice; fair::mq::Device mDevice;
fair::mq::PluginServices mServices; fair::mq::PluginServices mServices;
std::thread fRunStateMachineThread; std::thread fRunStateMachineThread;
}; };

40
test/plugins/_plugin.cxx
View File

@ -10,7 +10,7 @@
#include <fairmq/Plugin.h> #include <fairmq/Plugin.h>
#include <fairmq/PluginServices.h> #include <fairmq/PluginServices.h>
#include <fairmq/tools/Version.h> #include <fairmq/tools/Version.h>
#include <FairMQDevice.h> #include <fairmq/Device.h>
#include <fairmq/ProgOptions.h> #include <fairmq/ProgOptions.h>
#include <memory> #include <memory>
#include <sstream> #include <sstream>
@ -24,28 +24,28 @@ namespace _plugin
using namespace std; using namespace std;
using namespace fair::mq; using namespace fair::mq;
auto control(FairMQDevice& device) -> void auto control(Device& device) -> void
{ {
device.SetTransport("zeromq"); device.SetTransport("zeromq");
device.ChangeState(fair::mq::Transition::InitDevice); device.ChangeState(Transition::InitDevice);
device.WaitForState(fair::mq::State::InitializingDevice); device.WaitForState(State::InitializingDevice);
device.ChangeState(fair::mq::Transition::CompleteInit); device.ChangeState(Transition::CompleteInit);
device.WaitForState(fair::mq::State::Initialized); device.WaitForState(State::Initialized);
device.ChangeState(fair::mq::Transition::Bind); device.ChangeState(Transition::Bind);
device.WaitForState(fair::mq::State::Bound); device.WaitForState(State::Bound);
device.ChangeState(fair::mq::Transition::Connect); device.ChangeState(Transition::Connect);
device.WaitForState(fair::mq::State::DeviceReady); device.WaitForState(State::DeviceReady);
device.ChangeState(fair::mq::Transition::ResetDevice); device.ChangeState(Transition::ResetDevice);
device.WaitForState(fair::mq::State::Idle); device.WaitForState(State::Idle);
device.ChangeState(fair::mq::Transition::End); device.ChangeState(Transition::End);
} }
TEST(Plugin, Operators) TEST(Plugin, Operators)
{ {
fair::mq::ProgOptions config; ProgOptions config;
FairMQDevice device; Device device;
PluginServices services{config, device}; PluginServices services{config, device};
Plugin p1{"dds", {1, 0, 0}, "Foo Bar <foo.bar@test.net>", "https://git.test.net/dds.git", &services}; Plugin p1{"dds", {1, 0, 0}, "Foo Bar <foo.bar@test.net>", "https://git.test.net/dds.git", &services};
Plugin p2{"dds", {1, 0, 0}, "Foo Bar <foo.bar@test.net>", "https://git.test.net/dds.git", &services}; Plugin p2{"dds", {1, 0, 0}, "Foo Bar <foo.bar@test.net>", "https://git.test.net/dds.git", &services};
@ -61,8 +61,8 @@ TEST(Plugin, Operators)
TEST(Plugin, OstreamOperators) TEST(Plugin, OstreamOperators)
{ {
fair::mq::ProgOptions config; ProgOptions config;
FairMQDevice device; Device device;
PluginServices services{config, device}; PluginServices services{config, device};
Plugin p1{"dds", {1, 0, 0}, "Foo Bar <foo.bar@test.net>", "https://git.test.net/dds.git", &services}; Plugin p1{"dds", {1, 0, 0}, "Foo Bar <foo.bar@test.net>", "https://git.test.net/dds.git", &services};
stringstream ss; stringstream ss;
@ -77,9 +77,9 @@ TEST(Plugin, OstreamOperators)
TEST(PluginVersion, Operators) TEST(PluginVersion, Operators)
{ {
struct fair::mq::tools::Version v1{1, 0, 0}; struct tools::Version v1{1, 0, 0};
struct fair::mq::tools::Version v2{1, 0, 0}; struct tools::Version v2{1, 0, 0};
struct fair::mq::tools::Version v3{1, 2, 0}; struct tools::Version v3{1, 2, 0};
EXPECT_EQ(v1, v2); EXPECT_EQ(v1, v2);
EXPECT_NE(v1, v3); EXPECT_NE(v1, v3);
EXPECT_GT(v3, v2); EXPECT_GT(v3, v2);

36
test/plugins/_plugin_manager.cxx
View File

@ -9,9 +9,9 @@
#include <gtest/gtest.h> #include <gtest/gtest.h>
#include <fairmq/PluginManager.h> #include <fairmq/PluginManager.h>
#include <fairmq/PluginServices.h> #include <fairmq/PluginServices.h>
#include <FairMQDevice.h> #include <fairmq/Device.h>
#include <fairmq/ProgOptions.h> #include <fairmq/ProgOptions.h>
#include <FairMQLogger.h> #include <fairlogger/Logger.h>
#include <fstream> #include <fstream>
#include <memory> #include <memory>
#include <vector> #include <vector>
@ -25,28 +25,28 @@ using namespace boost::filesystem;
using namespace boost::program_options; using namespace boost::program_options;
using namespace std; using namespace std;
auto control(FairMQDevice& device) -> void auto control(Device& device) -> void
{ {
device.SetTransport("zeromq"); device.SetTransport("zeromq");
device.ChangeState(fair::mq::Transition::InitDevice); device.ChangeState(Transition::InitDevice);
device.WaitForState(fair::mq::State::InitializingDevice); device.WaitForState(State::InitializingDevice);
device.ChangeState(fair::mq::Transition::CompleteInit); device.ChangeState(Transition::CompleteInit);
device.WaitForState(fair::mq::State::Initialized); device.WaitForState(State::Initialized);
device.ChangeState(fair::mq::Transition::Bind); device.ChangeState(Transition::Bind);
device.WaitForState(fair::mq::State::Bound); device.WaitForState(State::Bound);
device.ChangeState(fair::mq::Transition::Connect); device.ChangeState(Transition::Connect);
device.WaitForState(fair::mq::State::DeviceReady); device.WaitForState(State::DeviceReady);
device.ChangeState(fair::mq::Transition::ResetDevice); device.ChangeState(Transition::ResetDevice);
device.WaitForState(fair::mq::State::Idle); device.WaitForState(State::Idle);
device.ChangeState(fair::mq::Transition::End); device.ChangeState(Transition::End);
} }
TEST(PluginManager, LoadPluginDynamic) TEST(PluginManager, LoadPluginDynamic)
{ {
fair::mq::ProgOptions config; ProgOptions config;
FairMQDevice device; Device device;
PluginManager mgr; PluginManager mgr;
mgr.EmplacePluginServices(config, device); mgr.EmplacePluginServices(config, device);
@ -77,13 +77,13 @@ TEST(PluginManager, LoadPluginDynamic)
TEST(PluginManager, LoadPluginStatic) TEST(PluginManager, LoadPluginStatic)
{ {
FairMQDevice device; Device device;
PluginManager mgr; PluginManager mgr;
device.SetTransport("zeromq"); device.SetTransport("zeromq");
ASSERT_NO_THROW(mgr.LoadPlugin("s:control")); ASSERT_NO_THROW(mgr.LoadPlugin("s:control"));
fair::mq::ProgOptions config; ProgOptions config;
config.SetProperty<string>("control", "static"); config.SetProperty<string>("control", "static");
config.SetProperty("catch-signals", 0); config.SetProperty("catch-signals", 0);
mgr.EmplacePluginServices(config, device); mgr.EmplacePluginServices(config, device);

80
test/region/_region.cxx
View File

@ -6,13 +6,14 @@
* copied verbatim in the file "LICENSE" * * copied verbatim in the file "LICENSE" *
********************************************************************************/ ********************************************************************************/
#include <FairMQLogger.h> #include <fairmq/TransportFactory.h>
#include <FairMQTransportFactory.h>
#include <fairmq/ProgOptions.h> #include <fairmq/ProgOptions.h>
#include <fairmq/tools/Unique.h> #include <fairmq/tools/Unique.h>
#include <fairmq/tools/Semaphore.h> #include <fairmq/tools/Semaphore.h>
#include <fairmq/tools/Strings.h> #include <fairmq/tools/Strings.h>
#include <fairlogger/Logger.h>
#include <gtest/gtest.h> #include <gtest/gtest.h>
#include <memory> // make_unique #include <memory> // make_unique
@ -22,31 +23,32 @@ namespace
{ {
using namespace std; using namespace std;
using namespace fair::mq;
void RegionsCache(const string& transport, const string& address) void RegionsCache(const string& transport, const string& address)
{ {
size_t session1 = fair::mq::tools::UuidHash(); size_t session1 = tools::UuidHash();
size_t session2 = fair::mq::tools::UuidHash(); size_t session2 = tools::UuidHash();
fair::mq::ProgOptions config1; ProgOptions config1;
fair::mq::ProgOptions config2; ProgOptions config2;
config1.SetProperty<string>("session", to_string(session1)); config1.SetProperty<string>("session", to_string(session1));
config2.SetProperty<string>("session", to_string(session2)); config2.SetProperty<string>("session", to_string(session2));
auto factory1 = FairMQTransportFactory::CreateTransportFactory(transport, fair::mq::tools::Uuid(), &config1); auto factory1 = TransportFactory::CreateTransportFactory(transport, tools::Uuid(), &config1);
auto factory2 = FairMQTransportFactory::CreateTransportFactory(transport, fair::mq::tools::Uuid(), &config2); auto factory2 = TransportFactory::CreateTransportFactory(transport, tools::Uuid(), &config2);
auto region1 = factory1->CreateUnmanagedRegion(1000000, [](void*, size_t, void*) {}); auto region1 = factory1->CreateUnmanagedRegion(1000000, [](void*, size_t, void*) {});
auto region2 = factory2->CreateUnmanagedRegion(1000000, [](void*, size_t, void*) {}); auto region2 = factory2->CreateUnmanagedRegion(1000000, [](void*, size_t, void*) {});
void* r1ptr = region1->GetData(); void* r1ptr = region1->GetData();
void* r2ptr = region2->GetData(); void* r2ptr = region2->GetData();
FairMQChannel push1("Push1", "push", factory1); Channel push1("Push1", "push", factory1);
FairMQChannel pull1("Pull1", "pull", factory1); Channel pull1("Pull1", "pull", factory1);
push1.Bind(address + to_string(1)); push1.Bind(address + to_string(1));
pull1.Connect(address + to_string(1)); pull1.Connect(address + to_string(1));
FairMQChannel push2("Push2", "push", factory2); Channel push2("Push2", "push", factory2);
FairMQChannel pull2("Pull2", "pull", factory2); Channel pull2("Pull2", "pull", factory2);
push2.Bind(address + to_string(2)); push2.Bind(address + to_string(2));
pull2.Connect(address + to_string(2)); pull2.Connect(address + to_string(2));
@ -56,27 +58,27 @@ void RegionsCache(const string& transport, const string& address)
static_cast<char*>(r2ptr)[0] = 99; // c static_cast<char*>(r2ptr)[0] = 99; // c
static_cast<char*>(static_cast<char*>(r2ptr) + 100)[0] = 100; // d static_cast<char*>(static_cast<char*>(r2ptr) + 100)[0] = 100; // d
FairMQMessagePtr m1(push1.NewMessage(region1, r1ptr, 100, nullptr)); MessagePtr m1(push1.NewMessage(region1, r1ptr, 100, nullptr));
FairMQMessagePtr m2(push1.NewMessage(region1, static_cast<char*>(r1ptr) + 100, 100, nullptr)); MessagePtr m2(push1.NewMessage(region1, static_cast<char*>(r1ptr) + 100, 100, nullptr));
push1.Send(m1); push1.Send(m1);
push1.Send(m2); push1.Send(m2);
FairMQMessagePtr m3(push2.NewMessage(region2, r2ptr, 100, nullptr)); MessagePtr m3(push2.NewMessage(region2, r2ptr, 100, nullptr));
FairMQMessagePtr m4(push2.NewMessage(region2, static_cast<char*>(r2ptr) + 100, 100, nullptr)); MessagePtr m4(push2.NewMessage(region2, static_cast<char*>(r2ptr) + 100, 100, nullptr));
push2.Send(m3); push2.Send(m3);
push2.Send(m4); push2.Send(m4);
} }
{ {
FairMQMessagePtr m1(pull1.NewMessage()); MessagePtr m1(pull1.NewMessage());
FairMQMessagePtr m2(pull1.NewMessage()); MessagePtr m2(pull1.NewMessage());
ASSERT_EQ(pull1.Receive(m1), 100); ASSERT_EQ(pull1.Receive(m1), 100);
ASSERT_EQ(pull1.Receive(m2), 100); ASSERT_EQ(pull1.Receive(m2), 100);
ASSERT_EQ(static_cast<char*>(m1->GetData())[0], 'a'); ASSERT_EQ(static_cast<char*>(m1->GetData())[0], 'a');
ASSERT_EQ(static_cast<char*>(m2->GetData())[0], 'b'); ASSERT_EQ(static_cast<char*>(m2->GetData())[0], 'b');
FairMQMessagePtr m3(pull2.NewMessage()); MessagePtr m3(pull2.NewMessage());
FairMQMessagePtr m4(pull2.NewMessage()); MessagePtr m4(pull2.NewMessage());
ASSERT_EQ(pull2.Receive(m3), 100); ASSERT_EQ(pull2.Receive(m3), 100);
ASSERT_EQ(pull2.Receive(m4), 100); ASSERT_EQ(pull2.Receive(m4), 100);
ASSERT_EQ(static_cast<char*>(m3->GetData())[0], 'c'); ASSERT_EQ(static_cast<char*>(m3->GetData())[0], 'c');
@ -86,17 +88,17 @@ void RegionsCache(const string& transport, const string& address)
void RegionEventSubscriptions(const string& transport) void RegionEventSubscriptions(const string& transport)
{ {
size_t session{fair::mq::tools::UuidHash()}; size_t session{tools::UuidHash()};
fair::mq::ProgOptions config; ProgOptions config;
config.SetProperty<string>("session", to_string(session)); config.SetProperty<string>("session", to_string(session));
auto factory = FairMQTransportFactory::CreateTransportFactory(transport, fair::mq::tools::Uuid(), &config); auto factory = TransportFactory::CreateTransportFactory(transport, tools::Uuid(), &config);
constexpr int size1 = 1000000; constexpr int size1 = 1000000;
constexpr int size2 = 5000000; constexpr int size2 = 5000000;
constexpr int64_t userFlags = 12345; constexpr int64_t userFlags = 12345;
fair::mq::tools::Semaphore blocker; tools::Semaphore blocker;
{ {
auto region1 = factory->CreateUnmanagedRegion(size1, [](void*, size_t, void*) {}); auto region1 = factory->CreateUnmanagedRegion(size1, [](void*, size_t, void*) {});
@ -110,14 +112,14 @@ void RegionEventSubscriptions(const string& transport)
ASSERT_EQ(region2->GetSize(), size2); ASSERT_EQ(region2->GetSize(), size2);
ASSERT_EQ(factory->SubscribedToRegionEvents(), false); ASSERT_EQ(factory->SubscribedToRegionEvents(), false);
factory->SubscribeToRegionEvents([&, id1, id2, ptr1, ptr2](FairMQRegionInfo info) { factory->SubscribeToRegionEvents([&, id1, id2, ptr1, ptr2](RegionInfo info) {
LOG(info) << ">>> " << info.event << ": " LOG(info) << ">>> " << info.event << ": "
<< (info.managed ? "managed" : "unmanaged") << (info.managed ? "managed" : "unmanaged")
<< ", id: " << info.id << ", id: " << info.id
<< ", ptr: " << info.ptr << ", ptr: " << info.ptr
<< ", size: " << info.size << ", size: " << info.size
<< ", flags: " << info.flags; << ", flags: " << info.flags;
if (info.event == FairMQRegionEvent::created) { if (info.event == RegionEvent::created) {
if (info.id == id1) { if (info.id == id1) {
ASSERT_EQ(info.size, size1); ASSERT_EQ(info.size, size1);
ASSERT_EQ(info.ptr, ptr1); ASSERT_EQ(info.ptr, ptr1);
@ -128,7 +130,7 @@ void RegionEventSubscriptions(const string& transport)
ASSERT_EQ(info.flags, userFlags); ASSERT_EQ(info.flags, userFlags);
blocker.Signal(); blocker.Signal();
} }
} else if (info.event == FairMQRegionEvent::destroyed) { } else if (info.event == RegionEvent::destroyed) {
if (info.id == id1) { if (info.id == id1) {
blocker.Signal(); blocker.Signal();
} else if (info.id == id2) { } else if (info.id == id2) {
@ -157,22 +159,22 @@ void RegionEventSubscriptions(const string& transport)
void RegionCallbacks(const string& transport, const string& _address) void RegionCallbacks(const string& transport, const string& _address)
{ {
size_t session(fair::mq::tools::UuidHash()); size_t session(tools::UuidHash());
std::string address(fair::mq::tools::ToString(_address, "_", transport)); std::string address(tools::ToString(_address, "_", transport));
fair::mq::ProgOptions config; ProgOptions config;
config.SetProperty<string>("session", to_string(session)); config.SetProperty<string>("session", to_string(session));
auto factory = FairMQTransportFactory::CreateTransportFactory(transport, fair::mq::tools::Uuid(), &config); auto factory = TransportFactory::CreateTransportFactory(transport, tools::Uuid(), &config);
unique_ptr<int> intPtr1 = make_unique<int>(42); unique_ptr<int> intPtr1 = make_unique<int>(42);
unique_ptr<int> intPtr2 = make_unique<int>(43); unique_ptr<int> intPtr2 = make_unique<int>(43);
fair::mq::tools::Semaphore blocker; tools::Semaphore blocker;
FairMQChannel push("Push", "push", factory); Channel push("Push", "push", factory);
push.Bind(address); push.Bind(address);
FairMQChannel pull("Pull", "pull", factory); Channel pull("Pull", "pull", factory);
pull.Connect(address); pull.Connect(address);
void* ptr1 = nullptr; void* ptr1 = nullptr;
@ -189,7 +191,7 @@ void RegionCallbacks(const string& transport, const string& _address)
}); });
ptr1 = region1->GetData(); ptr1 = region1->GetData();
auto region2 = factory->CreateUnmanagedRegion(3000000, [&](const std::vector<fair::mq::RegionBlock>& blocks) { auto region2 = factory->CreateUnmanagedRegion(3000000, [&](const std::vector<RegionBlock>& blocks) {
ASSERT_EQ(blocks.size(), 1); ASSERT_EQ(blocks.size(), 1);
ASSERT_EQ(blocks.at(0).ptr, ptr2); ASSERT_EQ(blocks.at(0).ptr, ptr2);
ASSERT_EQ(blocks.at(0).size, size2); ASSERT_EQ(blocks.at(0).size, size2);
@ -200,15 +202,15 @@ void RegionCallbacks(const string& transport, const string& _address)
ptr2 = region2->GetData(); ptr2 = region2->GetData();
{ {
FairMQMessagePtr msg1out(push.NewMessage(region1, ptr1, size1, intPtr1.get())); MessagePtr msg1out(push.NewMessage(region1, ptr1, size1, intPtr1.get()));
FairMQMessagePtr msg2out(push.NewMessage(region2, ptr2, size2, intPtr2.get())); MessagePtr msg2out(push.NewMessage(region2, ptr2, size2, intPtr2.get()));
ASSERT_EQ(push.Send(msg1out), size1); ASSERT_EQ(push.Send(msg1out), size1);
ASSERT_EQ(push.Send(msg2out), size2); ASSERT_EQ(push.Send(msg2out), size2);
} }
{ {
FairMQMessagePtr msg1in(pull.NewMessage()); MessagePtr msg1in(pull.NewMessage());
FairMQMessagePtr msg2in(pull.NewMessage()); MessagePtr msg2in(pull.NewMessage());
ASSERT_EQ(pull.Receive(msg1in), size1); ASSERT_EQ(pull.Receive(msg1in), size1);
ASSERT_EQ(pull.Receive(msg2in), size2); ASSERT_EQ(pull.Receive(msg2in), size2);
} }

2
test/state_machine/_state_machine.cxx
View File

@ -8,7 +8,7 @@
#include <gtest/gtest.h> #include <gtest/gtest.h>
#include <fairmq/StateMachine.h> #include <fairmq/StateMachine.h>
#include <FairMQLogger.h> #include <fairlogger/Logger.h>
#include <string> #include <string>
#include <thread> #include <thread>

8
test/transport/_options.cxx
View File

@ -39,7 +39,7 @@ void CheckOldOptionInterface(Channel& channel, const string& option)
void RunOptionsTest(const string& transport) void RunOptionsTest(const string& transport)
{ {
fair::mq::ProgOptions config; ProgOptions config;
config.SetProperty<string>("session", tools::Uuid()); config.SetProperty<string>("session", tools::Uuid());
auto factory = TransportFactory::CreateTransportFactory(transport, tools::Uuid(), &config); auto factory = TransportFactory::CreateTransportFactory(transport, tools::Uuid(), &config);
Channel channel("Push", "push", factory); Channel channel("Push", "push", factory);
@ -85,15 +85,15 @@ void ZeroingAndMlock(const string& transport)
void ZeroingAndMlockOnCreation(const string& transport) void ZeroingAndMlockOnCreation(const string& transport)
{ {
size_t session{fair::mq::tools::UuidHash()}; size_t session{tools::UuidHash()};
fair::mq::ProgOptions config; ProgOptions config;
config.SetProperty<string>("session", to_string(session)); config.SetProperty<string>("session", to_string(session));
config.SetProperty<size_t>("shm-segment-size", 16384); // NOLINT config.SetProperty<size_t>("shm-segment-size", 16384); // NOLINT
config.SetProperty<bool>("shm-mlock-segment-on-creation", true); config.SetProperty<bool>("shm-mlock-segment-on-creation", true);
config.SetProperty<bool>("shm-zero-segment-on-creation", true); config.SetProperty<bool>("shm-zero-segment-on-creation", true);
auto factory = FairMQTransportFactory::CreateTransportFactory(transport, fair::mq::tools::Uuid(), &config); auto factory = TransportFactory::CreateTransportFactory(transport, tools::Uuid(), &config);
constexpr size_t size{10000}; constexpr size_t size{10000};
auto outMsg(factory->CreateMessage(size)); auto outMsg(factory->CreateMessage(size));

4
test/transport/_shmem.cxx
View File

@ -9,7 +9,7 @@
#include <fairmq/ProgOptions.h> #include <fairmq/ProgOptions.h>
#include <fairmq/shmem/Monitor.h> #include <fairmq/shmem/Monitor.h>
#include <fairmq/tools/Unique.h> #include <fairmq/tools/Unique.h>
#include <FairMQTransportFactory.h> #include <fairmq/TransportFactory.h>
#include <gtest/gtest.h> #include <gtest/gtest.h>
@ -29,7 +29,7 @@ void GetFreeMemory()
ASSERT_THROW(shmem::Monitor::GetFreeMemory(shmem::SessionId{sessionId}, 0), shmem::Monitor::MonitorError); ASSERT_THROW(shmem::Monitor::GetFreeMemory(shmem::SessionId{sessionId}, 0), shmem::Monitor::MonitorError);
auto factory = FairMQTransportFactory::CreateTransportFactory("shmem", tools::Uuid(), &config); auto factory = TransportFactory::CreateTransportFactory("shmem", tools::Uuid(), &config);
ASSERT_NO_THROW(shmem::Monitor::GetFreeMemory(shmem::SessionId{sessionId}, 0)); ASSERT_NO_THROW(shmem::Monitor::GetFreeMemory(shmem::SessionId{sessionId}, 0));
ASSERT_THROW(shmem::Monitor::GetFreeMemory(shmem::SessionId{sessionId}, 1), shmem::Monitor::MonitorError); ASSERT_THROW(shmem::Monitor::GetFreeMemory(shmem::SessionId{sessionId}, 1), shmem::Monitor::MonitorError);

24
test/transport/_transfer_timeout.cxx
View File

@ -7,13 +7,16 @@
********************************************************************************/ ********************************************************************************/
#include "runner.h" #include "runner.h"
#include <FairMQChannel.h>
#include <FairMQLogger.h> #include <fairmq/Channel.h>
#include <FairMQTransportFactory.h> #include <fairmq/TransportFactory.h>
#include <fairmq/ProgOptions.h> #include <fairmq/ProgOptions.h>
#include <fairmq/tools/Process.h> #include <fairmq/tools/Process.h>
#include <fairmq/tools/Strings.h> #include <fairmq/tools/Strings.h>
#include <fairmq/tools/Unique.h> #include <fairmq/tools/Unique.h>
#include <fairlogger/Logger.h>
#include <gtest/gtest.h> #include <gtest/gtest.h>
#include <chrono> #include <chrono>
@ -24,10 +27,11 @@ namespace
{ {
using namespace std; using namespace std;
using namespace fair::mq;
using namespace fair::mq::test; using namespace fair::mq::test;
using namespace fair::mq::tools; using namespace fair::mq::tools;
void delayedInterruptor(FairMQTransportFactory& transport) void delayedInterruptor(TransportFactory& transport)
{ {
std::this_thread::sleep_for(std::chrono::milliseconds(100)); std::this_thread::sleep_for(std::chrono::milliseconds(100));
transport.Interrupt(); transport.Interrupt();
@ -35,7 +39,7 @@ void delayedInterruptor(FairMQTransportFactory& transport)
auto RunTransferTimeout(string transport) -> void auto RunTransferTimeout(string transport) -> void
{ {
size_t session{fair::mq::tools::UuidHash()}; size_t session{UuidHash()};
stringstream cmd; stringstream cmd;
cmd << runTestDevice << " --id transfer_timeout_" << transport << " --control static --transport " << transport cmd << runTestDevice << " --id transfer_timeout_" << transport << " --control static --transport " << transport
<< " --session " << session << " --color false --mq-config \"" << mqConfig << "\""; << " --session " << session << " --color false --mq-config \"" << mqConfig << "\"";
@ -48,18 +52,18 @@ auto RunTransferTimeout(string transport) -> void
void InterruptTransfer(const string& transport, const string& _address) void InterruptTransfer(const string& transport, const string& _address)
{ {
size_t session{fair::mq::tools::UuidHash()}; size_t session{UuidHash()};
std::string address(fair::mq::tools::ToString(_address, "_", transport)); std::string address(ToString(_address, "_", transport));
fair::mq::ProgOptions config; fair::mq::ProgOptions config;
config.SetProperty<string>("session", to_string(session)); config.SetProperty<string>("session", to_string(session));
auto factory = FairMQTransportFactory::CreateTransportFactory(transport, fair::mq::tools::Uuid(), &config); auto factory = TransportFactory::CreateTransportFactory(transport, Uuid(), &config);
FairMQChannel pull{"Pull", "pull", factory}; Channel pull{"Pull", "pull", factory};
pull.Bind(address); pull.Bind(address);
FairMQMessagePtr msg(pull.NewMessage()); MessagePtr msg(pull.NewMessage());
auto t = thread(delayedInterruptor, ref(*factory)); auto t = thread(delayedInterruptor, ref(*factory));