| | |
| | | #include "center.h" |
| | | #include "defs.h" |
| | | #include "pubsub.h" |
| | | #include "socket.h" |
| | | #include "topic_node.h" |
| | | #include "util.h" |
| | | #include <atomic> |
| | | #include <boost/uuid/uuid_generators.hpp> |
| | |
| | | const uint64_t nmsg = 100 * 2; |
| | | const int timeout = 1000; |
| | | auto Sub = [&](int id, const std::vector<std::string> &topics) { |
| | | SocketSubscribe client(shm); |
| | | bool r = client.Subscribe(sub_proc_id, topics, timeout); |
| | | DemoNode client("client_" + std::to_string(id), shm); |
| | | |
| | | bool r = client.Subscribe(topics, timeout); |
| | | if (!r) { |
| | | printf("client subscribe failed.\n"); |
| | | } |
| | | std::mutex mutex; |
| | | std::condition_variable cv; |
| | | |
| | |
| | | } |
| | | // printf("sub %2d recv: %s/%s\n", id, pub.topic().c_str(), pub.data().c_str()); |
| | | }; |
| | | client.StartRecv(OnTopicData, 1); |
| | | client.SubscribeStartWorker(OnTopicData, 1); |
| | | |
| | | std::unique_lock<std::mutex> lk(mutex); |
| | | cv.wait(lk); |
| | | }; |
| | | |
| | | auto Pub = [&](const std::string &topic) { |
| | | SocketPublish provider(shm); |
| | | DemoNode provider("server_" + topic, shm); |
| | | |
| | | for (unsigned i = 0; i < nmsg; ++i) { |
| | | std::string data = topic + std::to_string(i) + std::string(1000, '-'); |
| | | |
| | | bool r = provider.Publish(pub_proc_id, topic, data.data(), data.size(), timeout); |
| | | bool r = provider.Publish(topic, data.data(), data.size(), timeout); |
| | | if (!r) { |
| | | printf("pub ret: %s\n", r ? "ok" : "fail"); |
| | | } |
| | |
| | | std::atomic<bool> run(true); |
| | | |
| | | auto Client = [&](const std::string &topic, const int nreq) { |
| | | TopicNode client(shm); |
| | | MsgRegister reg; |
| | | reg.mutable_proc()->set_proc_id(client_proc_id + topic); |
| | | MsgCommonReply reply_body; |
| | | |
| | | if (!client.Register(reg, reply_body, 1000)) { |
| | | printf("client register failed\n"); |
| | | return; |
| | | } |
| | | DemoNode client(client_proc_id + topic, shm); |
| | | |
| | | std::atomic<int> count(0); |
| | | std::string reply; |
| | |
| | | do { |
| | | std::this_thread::yield(); |
| | | } while (count.load() < nreq); |
| | | client.ClientStopWorker(); |
| | | client.StopAll(); |
| | | printf("request %s %d done ", topic.c_str(), count.load()); |
| | | }; |
| | | |
| | | std::atomic_uint64_t server_msg_count(0); |
| | | auto Server = [&](const std::string &name, const std::vector<std::string> &topics) { |
| | | TopicNode server(shm); |
| | | MsgRegister reg; |
| | | reg.mutable_proc()->set_proc_id(server_proc_id); |
| | | reg.mutable_proc()->set_name(name); |
| | | MsgCommonReply reply_body; |
| | | |
| | | if (!server.Register(reg, reply_body, 100)) { |
| | | printf("server register failed\n"); |
| | | return; |
| | | } |
| | | DemoNode server(name, shm); |
| | | |
| | | auto onData = [&](const std::string &topic, const std::string &data, std::string &reply) { |
| | | ++server_msg_count; |
| | |
| | | for (auto &topic : topics) { |
| | | rpc.add_topics(topic); |
| | | } |
| | | MsgCommonReply reply_body; |
| | | if (!server.RegisterRPC(rpc, reply_body, 100)) { |
| | | printf("server register topic failed\n"); |
| | | return; |
| | |
| | | clients.Launch(Client, t, 1000 * 1); |
| | | } |
| | | clients.WaitAll(); |
| | | printf("clients done, server replyed: %d\n", server_msg_count.load()); |
| | | printf("clients done, server replyed: %ld\n", server_msg_count.load()); |
| | | run = false; |
| | | servers.WaitAll(); |
| | | } |