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

Add empty msg check for transport compatibility checker

Browse Source
This commit is contained in:
Alexey Rybalchenko 2021-06-24 10:50:37 +02:00 committed by Dennis Klein
parent a8bdb91165
commit 4dbb5535c3
4 changed files with 46 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

42
fairmq/FairMQChannel.h
View File

@ -385,22 +385,7 @@ class FairMQChannel
void CheckSendCompatibility(FairMQMessagePtr& msg) void CheckSendCompatibility(FairMQMessagePtr& msg)
{ {
if (fTransportType != msg->GetType()) { if (fTransportType != msg->GetType()) {
FairMQMessagePtr msgWrapper(NewMessage( if (msg->GetSize() > 0) {
msg->GetData(),
msg->GetSize(),
[](void* /*data*/, void* _msg) { delete static_cast<FairMQMessage*>(_msg); },
msg.get()
));
msg.release();
msg = move(msgWrapper);
}
}
void CheckSendCompatibility(std::vector<FairMQMessagePtr>& msgVec)
{
for (auto& msg : msgVec) {
if (fTransportType != msg->GetType()) {
FairMQMessagePtr msgWrapper(NewMessage( FairMQMessagePtr msgWrapper(NewMessage(
msg->GetData(), msg->GetData(),
msg->GetSize(), msg->GetSize(),
@ -409,6 +394,30 @@ class FairMQChannel
)); ));
msg.release(); msg.release();
msg = move(msgWrapper); msg = move(msgWrapper);
} else {
FairMQMessagePtr newMsg(NewMessage());
msg = move(newMsg);
}
}
}
void CheckSendCompatibility(std::vector<FairMQMessagePtr>& msgVec)
{
for (auto& msg : msgVec) {
if (fTransportType != msg->GetType()) {
if (msg->GetSize() > 0) {
FairMQMessagePtr msgWrapper(NewMessage(
msg->GetData(),
msg->GetSize(),
[](void* /*data*/, void* _msg) { delete static_cast<FairMQMessage*>(_msg); },
msg.get()
));
msg.release();
msg = move(msgWrapper);
} else {
FairMQMessagePtr newMsg(NewMessage());
msg = move(newMsg);
}
} }
} }
} }
@ -425,7 +434,6 @@ class FairMQChannel
{ {
for (auto& msg : msgVec) { for (auto& msg : msgVec) {
if (fTransportType != msg->GetType()) { if (fTransportType != msg->GetType()) {
FairMQMessagePtr newMsg(NewMessage()); FairMQMessagePtr newMsg(NewMessage());
msg = move(newMsg); msg = move(newMsg);
} }

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

@ -27,15 +27,13 @@ class Rep : public FairMQDevice
auto Run() -> void override auto Run() -> void override
{ {
auto request1 = FairMQMessagePtr{NewMessage()}; auto request1 = FairMQMessagePtr{NewMessage()};
if (Receive(request1, "data") >= 0) if (Receive(request1, "data") >= 0) {
{
LOG(info) << "Received request 1"; LOG(info) << "Received request 1";
auto reply = FairMQMessagePtr{NewMessage()}; auto reply = FairMQMessagePtr{NewMessage()};
Send(reply, "data"); Send(reply, "data");
} }
auto request2 = FairMQMessagePtr{NewMessage()}; auto request2 = FairMQMessagePtr{NewMessage()};
if (Receive(request2, "data") >= 0) if (Receive(request2, "data") >= 0) {
{
LOG(info) << "Received request 2"; LOG(info) << "Received request 2";
auto reply = FairMQMessagePtr{NewMessage()}; auto reply = FairMQMessagePtr{NewMessage()};
Send(reply, "data"); Send(reply, "data");

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

@ -30,8 +30,7 @@ class Req : public FairMQDevice
Send(request, "data"); Send(request, "data");
auto reply = FairMQMessagePtr{NewMessage()}; auto reply = FairMQMessagePtr{NewMessage()};
if (Receive(reply, "data") >= 0) if (Receive(reply, "data") >= 0) {
{
LOG(info) << "received reply"; LOG(info) << "received reply";
} }
}; };

27
test/protocols/_req_rep.cxx
View File

@ -24,27 +24,36 @@ auto RunReqRep(string transport) -> void
{ {
size_t session{fair::mq::tools::UuidHash()}; size_t session{fair::mq::tools::UuidHash()};
auto rep = execute_result{ "", 0 }; auto rep = execute_result{"", 0};
thread rep_thread([&]() { thread rep_thread([&]() {
stringstream cmd; stringstream cmd;
cmd << runTestDevice << " --id rep_" << transport << " --control static " cmd << runTestDevice << " --id rep_" << transport
<< "--session " << session << " --color false --mq-config \"" << mqConfig << "\""; << " --control static"
<< " --session " << session
<< " --color false"
<< " --mq-config \"" << mqConfig << "\"";
rep = execute(cmd.str(), "[REP]"); rep = execute(cmd.str(), "[REP]");
}); });
auto req1 = execute_result{ "", 0 }; auto req1 = execute_result{"", 0};
thread req1_thread([&]() { thread req1_thread([&]() {
stringstream cmd; stringstream cmd;
cmd << runTestDevice << " --id req_1" << transport << " --control static " cmd << runTestDevice << " --id req_1" << transport
<< "--session " << session << " --color false --mq-config \"" << mqConfig << "\""; << " --control static"
<< " --session " << session
<< " --color false"
<< " --mq-config \"" << mqConfig << "\"";
req1 = execute(cmd.str(), "[REQ1]"); req1 = execute(cmd.str(), "[REQ1]");
}); });
auto req2 = execute_result{ "", 0 }; auto req2 = execute_result{"", 0};
thread req2_thread([&]() { thread req2_thread([&]() {
stringstream cmd; stringstream cmd;
cmd << runTestDevice << " --id req_2" << transport << " --control static " cmd << runTestDevice << " --id req_2" << transport
<< "--session " << session << " --color false --mq-config \"" << mqConfig << "\""; << " --control static"
<< " --session " << session
<< " --color false"
<< " --mq-config \"" << mqConfig << "\"";
req2 = execute(cmd.str(), "[REQ2]"); req2 = execute(cmd.str(), "[REQ2]");
}); });