bug fixed ps_sub thread join

zhangmeng

2023-03-01 777333ff834744ac5665fa9abe5ec6373d25cda8

 main.cpp

@@ -4,13 +4,14 @@
#include <string>
#include <thread>
#include <memory>
using namespace std;

#include "cbhomeclient.h"
#include "message.h"

// #include "3dparty/bus_nng/bn_api.h"
// #include "3rdparty/bus_nng/bn_api.h"
#include "bhome_msg_api.pb.h"

using namespace std;
using namespace bhome_msg;

template <class F> void ignoref(F&& f){}
@@ -41,7 +42,7 @@

            int ret = bus_client_publish(handle, i.data(), i.size(), msg.data(), msg.size());
            printf("======>> bus_client_pubmsg [%s] ret %d\n", msg.c_str(), ret);
            this_thread::sleep_for(chrono::seconds{2});
            this_thread::sleep_for(chrono::seconds{1});
        }
    }
}
@@ -61,6 +62,12 @@
    while (true) {
        auto msg = bus_client_get_submsg(handle);
        printf("SUB msg topic [%s] data [%s]\n", msg->topic, msg->msg);
        free_submsg(msg);

        size_t count = 0;
        auto p = bus_client_query_procs(handle, &count);
        printf("bus_client_query_procs size %lu\n", count);
        free_query_procs(p, count);
    }

    bus_client_free(handle);
@@ -76,10 +83,10 @@
    creg_free(reg);

    size_t count = 0;
    string base_msg("test_request==");
    string base_msg("test_request==request message -> msg-");
    this_thread::sleep_for(chrono::seconds(3));
    while (true) {
        auto msg = base_msg + "request message -> msg-"+to_string(count++);
        auto msg = base_msg + to_string(count++);
        auto reqmsg = make_req_msg(topic, topicl, msg.data(), msg.size());
        crepmsg* repmsg = NULL;
        if (bus_client_request(handle, reqmsg, &repmsg)){
@@ -113,12 +120,12 @@
        void* src = NULL;
        auto msg = bus_client_get_reqmsg(handle, &src);
        auto repmsg = make_reply_msg(0, NULL, 0, "recv request", 12);
        bus_client_reply_msg(handle, src, repmsg);
        bus_client_reply(handle, src, repmsg);
        free_reply_msg(repmsg);
        printf("REPREQ msg [%s] \n", msg->msg);

        free_reqmsg(msg);
        this_thread::sleep_for(chrono::seconds{2});
        // this_thread::sleep_for(chrono::seconds{2});
    }
}

@@ -134,7 +141,8 @@
    printf("start RR\n");
    const char* rrtopic = "cbhomeclient_req_rep";
    thread([&]{ req(rrtopic); }).detach();

    reply(rrtopic);

    return 0;
}
}