bug fixed ps_sub thread join

zhangmeng

2023-03-01 0f53806de20f07b0bc15877c7a9b293758acbfa7

 src/nng_wrap.cpp

@@ -205,7 +205,7 @@
        PRNTVITAG("client_socket faild\n");
        return -1;
    }
    // set_socket_timeout(sock, timeout_req_rep);
    set_socket_timeout(sock, 300);
    sub->socket_ = sock;
    sub->t_ = get_thread([](const auto sub){
        while (!sub->t_quit_.load()) {
@@ -246,7 +246,7 @@
                        }
                    }
                }
                // this_thread::sleep_for(chrono::milliseconds{6});
                this_thread::sleep_for(chrono::milliseconds{6});
                // printf("======>> subscribe nn_recv failed %s\n", nn_strerror(nn_errno()));
            }
        }
@@ -590,20 +590,22 @@

    if (!w) return -1;

    *src = malloc(sizeof(uint64_t));
    *(uint64_t*)(*src) = key;
    *src = malloc(sizeof(key));
    memcpy(*src, &key, sizeof(key));

    return w->mode;
}

int send_reply(const void* src, const void* msg, const int msg_len, void* arg/*=NULL*/){
int send_reply(void* src, const void* msg, const int msg_len, void* arg/*=NULL*/){
    _rr* rep = (_rr*)arg;
    if (!rep) rep = singleton<_rr>();

    struct work* w{};
    {
        auto key = *(static_cast<uint64_t*>(const_cast<void*>(src)));
        free(const_cast<void*>(src));
        uint64_t key;
        memcpy(&key, src, sizeof(key));
        // auto key = *(static_cast<uint64_t*>(const_cast<void*>(src)));
        free(src);

        lock_guard<mutex> l{rep->mtx_msg_};
        auto iter = rep->works_.find(key);