multiplexed_socket: avoid incorrect buffer in setOnRecv

setOnRecv must lock socketsMutex. Generally it,s the case because onAccept
will be ready right away and still locked, however if onChannelReady_ retrigger
the callback, setOnRecv will be used without socketsMutex locked.

Retrigger in a thread and always lock socketsMutex to be sure.

Change-Id: If44d240e113585a5ce3320b8f17feaaaa8be8515

src/jamidht/multiplexed_socket.cpp

@@ -724,14 +724,18 @@ MultiplexedSocket::waitForData(const uint16_t& channel,
 void
 MultiplexedSocket::setOnRecv(const uint16_t& channel, GenericSocket<uint8_t>::RecvCb&& cb)
 {
+    // Re run on ioPool, socketsMtx can be locked here (via onAccept), so retrigger
+    // to avoid double lock
+    dht::ThreadPool::io().run([w = weak(), channel, cb = std::move(cb)]() {
+        if (auto shared = w.lock()) {
+            std::lock_guard<std::mutex> lkSockets(shared->pimpl_->socketsMutex);
             std::deque<uint8_t> recv;
             {
-        // NOTE: here socketsMtx is already locked via onAccept
-        std::lock_guard<std::mutex> lk(pimpl_->channelCbsMtx_);
-        pimpl_->channelCbs_[channel] = cb;
+                std::lock_guard<std::mutex> lk(shared->pimpl_->channelCbsMtx_);
+                shared->pimpl_->channelCbs_[channel] = cb;
 
-        auto dataIt = pimpl_->channelDatas_.find(channel);
-        if (dataIt != pimpl_->channelDatas_.end() && dataIt->second) {
+                auto dataIt = shared->pimpl_->channelDatas_.find(channel);
+                if (dataIt != shared->pimpl_->channelDatas_.end() && dataIt->second) {
                     std::lock_guard<std::mutex> lk(dataIt->second->mutex);
                     recv = std::move(dataIt->second->buf);
                 }
@@ -740,6 +744,8 @@ MultiplexedSocket::setOnRecv(const uint16_t& channel, GenericSocket<uint8_t>::Re
                 cb(&recv[0], recv.size());
             }
         }
+    });
+}
 
 void
 MultiplexedSocket::shutdown()