wip

include/opendht/dhtrunner.h

@@ -407,7 +407,7 @@ public:
     /**
      * Insert a push notification to process for OpenDHT
      */
-    void pushNotificationReceived(const std::map<std::string, std::string>& data) const;
+    void pushNotificationReceived(const std::map<std::string, std::string>& data);
 
     /* Proxy server mothods */
     void forwardAllMessages(bool forward);

include/opendht/proxy.h

@@ -22,8 +22,8 @@
 namespace dht {
 namespace proxy {
 
-constexpr const std::chrono::seconds OP_TIMEOUT {1 * 60 * 60}; // onw hour
-constexpr const std::chrono::seconds OP_MARGIN {5 * 60}; // 5 minutes
+constexpr const std::chrono::minutes OP_TIMEOUT {1 * 10}; // one hour
+constexpr const std::chrono::minutes OP_MARGIN {5}; // 5 minutes
 constexpr const char* const HTTP_PROTO {"http://"};
 using ListenToken = uint64_t;
 

src/dht_proxy_client.cpp

@@ -91,6 +91,7 @@ DhtProxyClient::startProxy()
     if (serverHost_.empty()) return;
     DHT_LOG.w("Staring proxy client to %s", serverHost_.c_str());
     nextProxyConfirmation = scheduler.add(scheduler.time(), std::bind(&DhtProxyClient::confirmProxy, this));
+    loopSignal_();
 }
 
 DhtProxyClient::~DhtProxyClient()
@@ -324,15 +325,18 @@ DhtProxyClient::put(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_po
         auto nextRefresh = scheduler.time() + proxy::OP_TIMEOUT;
         search->second.puts.erase(id);
         search->second.puts.emplace(id, PermanentPut {val, scheduler.add(nextRefresh, [this, key, id]{
+            DHT_LOG.w(key, "[search %s]: PermanentPut refresh callback vid %llu", key.to_c_str(), id);
+            std::lock_guard<std::mutex> lock(searchLock_);
             auto s = searches_.find(key);
-            if (s == searches_.end())
-                return;
-            auto p = s->second.puts.find(id);
-            if (p == s->second.puts.end())
-                return;
-            const auto& now = scheduler.time();
-            doPut(key, p->second.value, {}, now, true);
-            scheduler.edit(p->second.refreshJob, now + proxy::OP_TIMEOUT);
+            if (s != searches_.end()) {
+                auto p = s->second.puts.find(id);
+                if (p != s->second.puts.end()) {
+                    doPut(key, p->second.value, [this](bool ok, const std::vector<Sp<Node>>&) {
+                        DHT_LOG.w("PermanentPut refresh %s", (ok ? "ok" : "failed"));
+                    }, time_point::max(), true);
+                    scheduler.edit(p->second.refreshJob, scheduler.time() + proxy::OP_TIMEOUT);
+                }
+            }
         })});
     }
     doPut(key, val, std::move(cb), created, permanent);
@@ -648,6 +652,7 @@ DhtProxyClient::cancelListen(const InfoHash& key, size_t gtoken) {
         });
     }
     scheduler.edit(it->second.opExpirationJob, ops.getExpiration());
+    loopSignal_();
     return canceled;
 }
 
@@ -711,6 +716,7 @@ DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter fi
         }
         auto next = l->second.cache.onValues(expired ? new_values_empty : values, std::vector<Value::Id>{}, expired_ids, types, scheduler.time());
         scheduler.edit(l->second.cacheExpirationJob, next);
+        loopSignal_();
         return true;
     };
     auto pushNotifToken = std::make_shared<proxy::ListenToken>(0);
@@ -955,6 +961,10 @@ DhtProxyClient::pushNotificationReceived(const std::map<std::string, std::string
 {
 #if OPENDHT_PUSH_NOTIFICATIONS
     scheduler.syncTime();
+    DHT_LOG.d("Got push notification");
+    for (const auto& n : notification) {
+        DHT_LOG.d("   %s -> %s", n.first.c_str(), n.second.c_str());
+    }
     try {
         std::lock_guard<std::mutex> lock(searchLock_);
         auto timeout = notification.find("timeout");
@@ -965,8 +975,10 @@ 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);
+                DHT_LOG.d("refresh PermanentPut: %s %llx %p %s", key.to_c_str(), vid, put.refreshJob.get(), put.refreshJob ? (put.refreshJob->do_ ? "set":"not set") : "(null job)");
                 scheduler.edit(put.refreshJob, scheduler.time());
+                loopSignal_();
             } else {
                 // Refresh listen
                 for (auto& list : search.listeners)
@@ -1042,7 +1054,7 @@ DhtProxyClient::resubscribe(const InfoHash& key, Listener& listener)
             }
         }, settings).get();
         auto& s = *state;
-        if (s.ok and not s.cancel)
+        if (not s.ok and not s.cancel)
             opFailed();
     });
 #endif

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;
@@ -599,35 +599,41 @@ DhtProxyServer::put(const std::shared_ptr<restbed::Session>& session)
                                     clientId = pVal["client_id"].asString();
                                     platform = pVal["platform"].asString();
                                 }
+                                std::cout << "permanent: pushToken " << pushToken << " clientId " << clientId << " platform " << platform << std::endl;
                                 bool isAndroid = platform == "android";
                                 std::unique_lock<std::mutex> lock(schedulerLock_);
                                 scheduler_.syncTime();
-                                auto sPuts = puts_.emplace(infoHash, SearchPuts{}).first;
                                 auto timeout = scheduler_.time() + proxy::OP_TIMEOUT;
                                 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;
+                                    std::cout << "New ppush entry" << 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()) {
+                                        std::cout << "Adding refresh job" << std::endl;
+                                        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 << " " << std::to_string(vid) << " " << clientId << 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);
+                                    std::cout << "Refreshing put jobs" << std::endl;
+                                    scheduler_.edit(pput.expireJob, timeout);
+                                    if (pput.expireNotifyJob)
+                                        scheduler_.edit(pput.expireNotifyJob, timeout - proxy::OP_MARGIN);
                                 }
                                 lock.unlock();
                                 schedulerCv_.notify_one();
@@ -652,7 +658,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

@@ -966,11 +966,17 @@ DhtRunner::setPushNotificationToken(const std::string& token) {
 }
 
 void
-DhtRunner::pushNotificationReceived(const std::map<std::string, std::string>& data) const
+DhtRunner::pushNotificationReceived(const std::map<std::string, std::string>& data)
 {
 #if OPENDHT_PROXY_CLIENT && OPENDHT_PUSH_NOTIFICATIONS
-    if (dht_via_proxy_)
-        dht_via_proxy_->pushNotificationReceived(data);
+    {
+        std::lock_guard<std::mutex> lck(storage_mtx);
+        pending_ops_prio.emplace([=](SecureDht& dht) {
+            if (dht_via_proxy_)
+                dht_via_proxy_->pushNotificationReceived(data);
+        });
+    }
+    cv.notify_all();
 #endif
 }