dhtrunner: fix warnings

src/dht_proxy_client.cpp

@@ -322,15 +322,16 @@ DhtProxyClient::put(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_po
         std::lock_guard<std::mutex> lock(searchLock_);
         auto id = val->id;
         auto search = searches_.emplace(key, ProxySearch{}).first;
-        auto nextRefresh = scheduler.time() + proxy::OP_TIMEOUT;
+        auto nextRefresh = scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN;
         search->second.puts.erase(id);
         search->second.puts.emplace(id, PermanentPut {val, scheduler.add(nextRefresh, [this, key, id]{
+            std::lock_guard<std::mutex> lock(searchLock_);
             auto s = searches_.find(key);
             if (s != searches_.end()) {
                 auto p = s->second.puts.find(id);
                 if (p != s->second.puts.end()) {
                     doPut(key, p->second.value, {}, time_point::max(), true);
-                    scheduler.edit(p->second.refreshJob, scheduler.time() + proxy::OP_TIMEOUT);
+                    scheduler.edit(p->second.refreshJob, scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN);
                 }
             }
         })});
@@ -967,7 +968,7 @@ DhtProxyClient::pushNotificationReceived(const std::map<std::string, std::string
             if (vidIt != notification.end()) {
                 // Refresh put
                 auto vid = std::stoull(vidIt->second);
-                auto put = search.puts.at(vid);
+                auto& put = search.puts.at(vid);
                 scheduler.edit(put.refreshJob, scheduler.time());
                 loopSignal_();
             } else {

src/dht_proxy_server.cpp

@@ -588,8 +588,8 @@ DhtProxyServer::put(const std::shared_ptr<restbed::Session>& session)
                         if (reader->parse(char_data, char_data + b.size(), &root, &err)) {
                             // Build the Value from json
                             auto value = std::make_shared<Value>(root);
-                            auto permanent = root.isMember("permanent");
-                            std::cout << "Got put " << infoHash << " " << value << " " << (permanent ? "permanent" : "") << std::endl;
+                            bool permanent = root.isMember("permanent");
+                            std::cout << "Got put " << infoHash << " " << *value << " " << (permanent ? "permanent" : "") << std::endl;
 
                             if (permanent) {
                                 std::string pushToken, clientId, platform;
@@ -606,28 +606,30 @@ DhtProxyServer::put(const std::shared_ptr<restbed::Session>& session)
                                 auto vid = value->id;
                                 auto sPuts = puts_.emplace(infoHash, SearchPuts{}).first;
                                 auto r = sPuts->second.puts.emplace(vid, PermanentPut{});
+                                auto& pput = r.first->second;
                                 if (r.second) {
-                                    r.first->second.expireJob = scheduler_.add(timeout, [this, infoHash, vid]{
-                                        std::cout << "Permanent put expired: " << infoHash << std::endl;
+                                    pput.expireJob = scheduler_.add(timeout, [this, infoHash, vid]{
+                                        std::cout << "Permanent put expired: " << infoHash << " " << vid << std::endl;
                                         cancelPut(infoHash, vid);
                                     });
 #if OPENDHT_PUSH_NOTIFICATIONS
-                                    if (not pushToken.empty())
-                                        r.first->second.expireNotifyJob = scheduler_.add(timeout - proxy::OP_MARGIN,
+                                    if (not pushToken.empty()) {
+                                        pput.expireNotifyJob = scheduler_.add(timeout - proxy::OP_MARGIN,
                                             [this, infoHash, vid, pushToken, clientId, isAndroid]
                                         {
-                                            std::cout << "Permanent put refresh: " << infoHash << std::endl;
+                                            std::cout << "Permanent put refresh: " << infoHash << " " << vid << std::endl;
                                             Json::Value json;
                                             json["timeout"] = infoHash.toString();
                                             json["to"] = clientId;
                                             json["vid"] = std::to_string(vid);
                                             sendPushNotification(pushToken, json, isAndroid);
                                         });
+                                    }
 #endif
                                 } else {
-                                    scheduler_.edit(r.first->second.expireJob, timeout);
-                                    if (r.first->second.expireNotifyJob)
-                                        scheduler_.edit(r.first->second.expireNotifyJob, timeout - proxy::OP_MARGIN);
+                                    scheduler_.edit(pput.expireJob, timeout);
+                                    if (pput.expireNotifyJob)
+                                        scheduler_.edit(pput.expireNotifyJob, timeout - proxy::OP_MARGIN);
                                 }
                                 lock.unlock();
                                 schedulerCv_.notify_one();
@@ -652,7 +654,8 @@ DhtProxyServer::put(const std::shared_ptr<restbed::Session>& session)
                 } else {
                     s->close(restbed::SERVICE_UNAVAILABLE, "{\"err\":\"Incorrect DhtRunner\"}");
                 }
-            } catch (...) {
+            } catch (const std::exception& e) {
+                std::cout << "Error performing put: " << e.what() << std::endl;
                 s->close(restbed::INTERNAL_SERVER_ERROR, "{\"err\":\"Internal server error\"}");
             }
         }

src/dhtrunner.cpp

@@ -599,20 +599,24 @@ DhtRunner::listen(const std::string& key, GetCallback vcb, Value::Filter f, Wher
 void
 DhtRunner::cancelListen(InfoHash h, size_t token)
 {
-    std::lock_guard<std::mutex> lck(storage_mtx);
-    pending_ops.emplace([=](SecureDht& dht) {
+    {
+        std::lock_guard<std::mutex> lck(storage_mtx);
 #if OPENDHT_PROXY_CLIENT
-        auto it = listeners_.find(token);
-        if (it == listeners_.end()) return;
-        if (it->second.tokenClassicDht)
-            dht_->cancelListen(h, it->second.tokenClassicDht);
-        if (it->second.tokenProxyDht and dht_via_proxy_)
-            dht_via_proxy_->cancelListen(h, it->second.tokenProxyDht);
-        listeners_.erase(it);
+        pending_ops.emplace([=](SecureDht&) {
+            auto it = listeners_.find(token);
+            if (it == listeners_.end()) return;
+            if (it->second.tokenClassicDht)
+                dht_->cancelListen(h, it->second.tokenClassicDht);
+            if (it->second.tokenProxyDht and dht_via_proxy_)
+                dht_via_proxy_->cancelListen(h, it->second.tokenProxyDht);
+            listeners_.erase(it);
+        });
 #else
-        dht.cancelListen(h, token);
+        pending_ops.emplace([=](SecureDht& dht) {
+            dht.cancelListen(h, token);
+        });
 #endif // OPENDHT_PROXY_CLIENT
-    });
+    }
     cv.notify_all();
 }
 
@@ -621,20 +625,21 @@ DhtRunner::cancelListen(InfoHash h, std::shared_future<size_t> ftoken)
 {
     {
         std::lock_guard<std::mutex> lck(storage_mtx);
-        pending_ops.emplace([=](SecureDht& dht) {
-            auto token = ftoken.get();
 #if OPENDHT_PROXY_CLIENT
-            auto it = listeners_.find(token);
+        pending_ops.emplace([=](SecureDht&) {
+            auto it = listeners_.find(ftoken.get());
             if (it == listeners_.end()) return;
             if (it->second.tokenClassicDht)
                 dht_->cancelListen(h, it->second.tokenClassicDht);
             if (it->second.tokenProxyDht and dht_via_proxy_)
                 dht_via_proxy_->cancelListen(h, it->second.tokenProxyDht);
             listeners_.erase(it);
-#else
-            dht.cancelListen(h, token);
-#endif // OPENDHT_PROXY_CLIENT
         });
+#else
+        pending_ops.emplace([=](SecureDht& dht) {
+            dht.cancelListen(h, ftoken.get());
+        });
+#endif // OPENDHT_PROXY_CLIENT
     }
     cv.notify_all();
 }
@@ -971,7 +976,7 @@ DhtRunner::pushNotificationReceived(const std::map<std::string, std::string>& da
 #if OPENDHT_PROXY_CLIENT && OPENDHT_PUSH_NOTIFICATIONS
     {
         std::lock_guard<std::mutex> lck(storage_mtx);
-        pending_ops_prio.emplace([=](SecureDht& dht) {
+        pending_ops_prio.emplace([=](SecureDht&) {
             if (dht_via_proxy_)
                 dht_via_proxy_->pushNotificationReceived(data);
         });