florian/mesytec-mnode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

tweak ping pong

Browse source
This commit is contained in:
Florian Lüke 2024-11-21 23:00:13 +01:00
parent 78cab52b91
commit 9e95897283
1 changed files with 58 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

70
src/mnode_proto_test1.cc
View file

@ -1,14 +1,25 @@
#include <mesytec-mvlc/mesytec-mvlc.h>
#include "proto/service.pb.h" #include "proto/service.pb.h"
#include "mesy_nng.h" #include "mesy_nng.h"
using namespace mesytec; using namespace mesytec;
using namespace std::literals;
void requester(nng_socket socket, unsigned id) void requester(nng_socket socket, unsigned id)
{ {
mvlc::util::Stopwatch sw;
std::uint32_t seq = 0; std::uint32_t seq = 0;
mnode::Ping ping;
mnode::Pong pong;
while (true) while (true)
{ {
mnode::Ping ping; if (sw.get_elapsed() >= 1s)
{
spdlog::info("Requester {} sent {} requests", id, seq);
break;
}
ping.set_peer_id(id); ping.set_peer_id(id);
ping.set_sequence_number(seq++); ping.set_sequence_number(seq++);
auto reqMsg = nng::make_message(ping.SerializeAsString()); auto reqMsg = nng::make_message(ping.SerializeAsString());
@ -19,7 +30,7 @@ void requester(nng_socket socket, unsigned id)
return; return;
continue; continue;
} }
spdlog::info("Sent ping request: {}", ping.ShortDebugString()); spdlog::debug("Sent ping request: {}", ping.ShortDebugString());
auto [repMsg, rc] = nng::receive_message(socket); auto [repMsg, rc] = nng::receive_message(socket);
if (rc) if (rc)
{ {
@ -28,16 +39,25 @@ void requester(nng_socket socket, unsigned id)
return; return;
continue; continue;
} }
mnode::Pong pong;
pong.ParseFromArray(nng_msg_body(repMsg.get()), nng_msg_len(repMsg.get())); pong.ParseFromArray(nng_msg_body(repMsg.get()), nng_msg_len(repMsg.get()));
spdlog::info("Received pong response: {}", pong.ShortDebugString()); spdlog::debug("Received pong response: {}", pong.ShortDebugString());
} }
} }
void responder(nng_socket socket, unsigned id) void responder(nng_socket socket, unsigned id)
{ {
mvlc::util::Stopwatch sw;
mnode::Ping ping;
mnode::Pong pong;
while (true) while (true)
{ {
if (sw.get_elapsed() >= 1s)
{
spdlog::info("Responder {} answered {} requests", id, ping.sequence_number());
break;
}
auto [reqMsg, rc] = nng::receive_message(socket); auto [reqMsg, rc] = nng::receive_message(socket);
if (rc) if (rc)
{ {
@ -46,11 +66,9 @@ void responder(nng_socket socket, unsigned id)
return; return;
continue; continue;
} }
mnode::Ping ping;
ping.ParseFromArray(nng_msg_body(reqMsg.get()), nng_msg_len(reqMsg.get())); ping.ParseFromArray(nng_msg_body(reqMsg.get()), nng_msg_len(reqMsg.get()));
spdlog::info("Received ping request: {}", ping.ShortDebugString()); spdlog::debug("Received ping request: {}", ping.ShortDebugString());
mnode::Pong pong;
pong.set_peer_id(id); pong.set_peer_id(id);
pong.set_sequence_number(ping.sequence_number()); pong.set_sequence_number(ping.sequence_number());
auto repMsg = nng::make_message(pong.SerializeAsString()); auto repMsg = nng::make_message(pong.SerializeAsString());
@ -61,22 +79,50 @@ void responder(nng_socket socket, unsigned id)
return; return;
continue; continue;
} }
spdlog::info("Sent pong response: {}", pong.ShortDebugString()); spdlog::debug("Sent pong response: {}", pong.ShortDebugString());
} }
} }
void client_aio(nng_socket socket, unsigned id)
{
}
void server_aio(nng_socket socket, unsigned id)
{
}
int main() int main()
{ {
spdlog::set_level(spdlog::level::info); spdlog::set_level(spdlog::level::info);
unsigned peerId = 0;
std::vector<std::thread> threads;
#if 1
{
auto reqSocket = nng::make_req_socket(); auto reqSocket = nng::make_req_socket();
auto repSocket = nng::make_rep_socket(); auto repSocket = nng::make_rep_socket();
if (int rc = nng::marry_listen_dial(reqSocket, repSocket, "inproc://test")) if (int rc = nng::marry_listen_dial(reqSocket, repSocket, fmt::format("inproc://test{}", peerId).c_str()))
return rc; return rc;
std::vector<std::thread> threads; threads.emplace_back(std::thread(requester, reqSocket, peerId++));
threads.emplace_back(std::thread(requester, reqSocket, 0)); threads.emplace_back(std::thread(responder, repSocket, peerId++));
threads.emplace_back(std::thread(responder, repSocket, 1)); }
#endif
#if 1
{
auto reqSocket = nng::make_req_socket();
auto repSocket = nng::make_rep_socket();
if (int rc = nng::marry_listen_dial(reqSocket, repSocket, fmt::format("inproc://test{}", peerId).c_str()))
return rc;
threads.emplace_back(std::thread(client_aio, reqSocket, peerId++));
threads.emplace_back(std::thread(server_aio, repSocket, peerId++));
}
#endif
for (auto &t: threads) for (auto &t: threads)
if (t.joinable()) t.join(); if (t.joinable()) t.join();