wip

include/opendht/default_types.h

@@ -114,8 +114,6 @@ public:
     ImMessage() {}
     ImMessage(dht::Value::Id id, std::string&& m, long d = 0)
         : id(id), msg(std::move(m)), date(d) {}
-    ImMessage(dht::Value::Id id, std::string&& dt, std::string&& m, long d = 0)
-        : id(id), msg(std::move(m)), datatype(std::move(dt)), date(d) {}
 
     virtual void unpackValue(const Value& v) override {
         to = v.recipient;
@@ -125,11 +123,10 @@ public:
     dht::InfoHash to;
     dht::Value::Id id {0};
     std::string msg;
-    std::string datatype;
     long date {0};
     ImStatus status {ImStatus::NONE};
 
-    MSGPACK_DEFINE_MAP(id, msg, date, status, datatype)
+    MSGPACK_DEFINE_MAP(id, msg, date, status)
 };
 
 class OPENDHT_PUBLIC TrustRequest : public EncryptedValue<TrustRequest>

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

@@ -322,16 +322,19 @@ 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 - proxy::OP_MARGIN;
+        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()) {
                 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 - proxy::OP_MARGIN);
+                    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);
                 }
             }
         })});
@@ -716,22 +719,6 @@ DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter fi
         loopSignal_();
         return true;
     };
-
-    if (not deviceKey_.empty())
-        // Relaunch push listeners even if a timeout is not received (if the proxy crash for any reason)
-        l->second.refreshJob = scheduler.add(scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN, [this, key, token, state] {
-            if (state->cancel)
-                return;
-            std::lock_guard<std::mutex> lock(searchLock_);
-            auto s = searches_.find(key);
-            if (s != searches_.end()) {
-                auto l = s->second.listeners.find(token);
-                if (l != s->second.listeners.end()) {
-                    resubscribe(key, l->second);
-                }
-            }
-        });
-
     auto pushNotifToken = std::make_shared<proxy::ListenToken>(0);
     auto vcb = l->second.cb;
     l->second.pushNotifToken = pushNotifToken;
@@ -796,15 +783,9 @@ DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter fi
                             }
                         }
                     } catch (std::runtime_error&) {
-                        // NOTE: Http::close() can occurs here. Ignore this if it's a cancelListen
-                        // else, the proxy server is down so we must call opFailed
-                        if (!state->cancel) {
-                            DHT_LOG.w("Listen closed by the proxy server");
-                            state->ok = false;
-                        }
+                        // NOTE: Http::close() can occurs here. Ignore this.
                     }
                 } else {
-                    DHT_LOG.w("Listen operation did not return a code 200.");
                     state->ok = false;
                 }
             }, settings).get();
@@ -980,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");
@@ -991,6 +976,7 @@ DhtProxyClient::pushNotificationReceived(const std::map<std::string, std::string
                 // Refresh put
                 auto vid = std::stoull(vidIt->second);
                 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 {
@@ -1025,7 +1011,6 @@ DhtProxyClient::resubscribe(const InfoHash& key, Listener& listener)
 {
 #if OPENDHT_PUSH_NOTIFICATIONS
     if (deviceKey_.empty()) return;
-    scheduler.syncTime();
     DHT_LOG.d(key, "[search %s] resubscribe push listener", key.to_c_str());
     // Subscribe
     restbed::Uri uri(proxy::HTTP_PROTO + serverHost_ + "/" + key.toString());
@@ -1040,7 +1025,6 @@ DhtProxyClient::resubscribe(const InfoHash& key, Listener& listener)
     state->ok = true;
     listener.req = req;
     listener.pushNotifToken = pushNotifToken;
-    scheduler.edit(listener.refreshJob, scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN);
     listener.thread = std::thread([this, req, pushNotifToken, state]() {
         fillBodyToGetToken(req);
         auto settings = std::make_shared<restbed::Settings>();

src/dht_proxy_server.cpp

@@ -599,6 +599,7 @@ 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();
@@ -608,16 +609,18 @@ DhtProxyServer::put(const std::shared_ptr<restbed::Session>& session)
                                 auto r = sPuts->second.puts.emplace(vid, PermanentPut{});
                                 auto& pput = r.first->second;
                                 if (r.second) {
+                                    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()) {
+                                        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 << " " << vid << 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;
@@ -627,6 +630,7 @@ DhtProxyServer::put(const std::shared_ptr<restbed::Session>& session)
                                     }
 #endif
                                 } else {
+                                    std::cout << "Refreshing put jobs" << std::endl;
                                     scheduler_.edit(pput.expireJob, timeout);
                                     if (pput.expireNotifyJob)
                                         scheduler_.edit(pput.expireNotifyJob, timeout - proxy::OP_MARGIN);

src/dhtrunner.cpp

@@ -599,24 +599,20 @@ 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);
+    std::lock_guard<std::mutex> lck(storage_mtx);
+    pending_ops.emplace([=](SecureDht& dht) {
 #if OPENDHT_PROXY_CLIENT
-        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);
-        });
+        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
-        pending_ops.emplace([=](SecureDht& dht) {
-            dht.cancelListen(h, token);
-        });
+        dht.cancelListen(h, token);
 #endif // OPENDHT_PROXY_CLIENT
-    }
+    });
     cv.notify_all();
 }
 
@@ -625,21 +621,20 @@ 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
-        pending_ops.emplace([=](SecureDht&) {
-            auto it = listeners_.find(ftoken.get());
+            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
-        pending_ops.emplace([=](SecureDht& dht) {
-            dht.cancelListen(h, ftoken.get());
-        });
+            dht.cancelListen(h, token);
 #endif // OPENDHT_PROXY_CLIENT
+        });
     }
     cv.notify_all();
 }
@@ -976,7 +971,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&) {
+        pending_ops_prio.emplace([=](SecureDht& dht) {
             if (dht_via_proxy_)
                 dht_via_proxy_->pushNotificationReceived(data);
         });