bug fixed query_proc

zhangmeng

2023-01-29 e4cb33f25012c534f0582d485c4f491f9dd3cd25

 src/nng_wrap.cpp

@@ -210,7 +210,8 @@
    sub->t_ = get_thread([](const auto sub){
        while (!sub->t_quit_.load()) {
            char* m{};
            int m_len = nn_recv(sub->socket_, &m, NN_MSG, NN_DONTWAIT);
            // int m_len = nn_recv(sub->socket_, &m, NN_MSG, NN_DONTWAIT);
            int m_len = nn_recv(sub->socket_, &m, NN_MSG, 0);
            if (m_len > 0){
                string tmp_msg{m, (size_t)m_len};
                nn_freemsg(m);
@@ -245,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()));
            }
        }
@@ -589,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);