proxy server: keep only one push listener per token.key.clientId

include/opendht/dht_proxy_client.h

@@ -29,6 +29,7 @@
 #include "def.h"
 #include "dht_interface.h"
 #include "scheduler.h"
+#include "proxy.h"
 
 namespace restbed {
     class Request;
@@ -365,6 +366,7 @@ private:
 
 #if OPENDHT_PUSH_NOTIFICATIONS
     void fillBodyToGetToken(std::shared_ptr<restbed::Request> request, unsigned token = 0);
+    void getPushRequest(Json::Value&) const;
 #endif // OPENDHT_PUSH_NOTIFICATIONS
 
     bool isDestroying_ {false};

include/opendht/dht_proxy_server.h

@@ -1,6 +1,7 @@
 /*
  *  Copyright (C) 2017-2018 Savoir-faire Linux Inc.
  *  Author: Sébastien Blin <sebastien.blin@savoirfairelinux.com>
+ *          Adrien Béraud <adrien.beraud@savoirfairelinux.com>
  *
  *  This program is free software; you can redistribute it and/or modify
  *  it under the terms of the GNU General Public License as published by
@@ -25,6 +26,7 @@
 #include "infohash.h"
 #include "scheduler.h"
 #include "value.h"
+#include "proxy.h"
 
 #include <thread>
 #include <memory>
@@ -179,6 +181,12 @@ private:
      */
     void handleOptionsMethod(const std::shared_ptr<restbed::Session>& session) const;
 
+    /**
+     * Remove finished listeners
+     * @param testSession if we remove the listener only if the session is closed
+     */
+    void removeClosedListeners(bool testSession = true);
+
 #if OPENDHT_PUSH_NOTIFICATIONS
     /**
      * Subscribe to push notifications for an iOS or Android device.
@@ -204,6 +212,10 @@ private:
      * @param json, the content to send
      */
     void sendPushNotification(const std::string& key, const Json::Value& json, bool isAndroid) const;
+
+    void cancelPushListen(const std::string& pushToken, const InfoHash& key, proxy::ListenToken token);
+
+
 #endif //OPENDHT_PUSH_NOTIFICATIONS
 
     using clock = std::chrono::steady_clock;
@@ -234,25 +246,18 @@ private:
     struct SearchPuts;
     std::map<InfoHash, SearchPuts> puts_;
 
+    mutable std::atomic<size_t> requestNum_ {0};
+    mutable std::atomic<time_point> lastStatsReset_ {time_point::min()};
+
+    const std::string pushServer_;
+
 #if OPENDHT_PUSH_NOTIFICATIONS
     struct Listener;
     struct PushListener;
     std::mutex lockPushListeners_;
     std::map<std::string, PushListener> pushListeners_;
-    uint64_t tokenPushNotif_ {0};
-
-    void cancelPushListen(const std::string& pushToken, const InfoHash& key, unsigned token);
+    proxy::ListenToken tokenPushNotif_ {0};
 #endif //OPENDHT_PUSH_NOTIFICATIONS
-    const std::string pushServer_;
-
-    mutable std::atomic<size_t> requestNum_ {0};
-    mutable std::atomic<time_point> lastStatsReset_ {time_point::min()};
-
-    /**
-     * Remove finished listeners
-     * @param testSession if we remove the listener only if the session is closed
-     */
-    void removeClosedListeners(bool testSession = true);
 };
 
 }

include/opendht/proxy.h

@@ -0,0 +1,31 @@
+/*
+ *  Copyright (C) 2017-2018 Savoir-faire Linux Inc.
+ *  Author: Adrien Béraud <adrien.beraud@savoirfairelinux.com>
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+#pragma once
+
+#include <chrono>
+
+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 char* const HTTP_PROTO {"http://"};
+using ListenToken = uint64_t;
+
+}
+}

src/dht_proxy_client.cpp

@@ -31,9 +31,6 @@
 #include <chrono>
 #include <vector>
 
-constexpr const char* const HTTP_PROTO {"http://"};
-constexpr const std::chrono::seconds OP_TIMEOUT {1 * 60 * 60 - 60}; // one hour minus margin
-
 namespace dht {
 
 struct DhtProxyClient::Listener
@@ -50,7 +47,7 @@ struct DhtProxyClient::Listener
     std::thread thread;
     unsigned callbackId;
     Sp<State> state;
-    Sp<unsigned> pushNotifToken; // NOTE: unused if not using push notifications
+    Sp<proxy::ListenToken> pushNotifToken;
     Sp<Scheduler::Job> refreshJob;
     Listener(ValueCache&& c, Sp<Scheduler::Job>&& j, const Sp<restbed::Request>& r, Value::Filter&& f)
         : cache(std::move(c)),
@@ -235,7 +232,7 @@ void
 DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Value::Filter&& f, Where&& w)
 {
     DHT_LOG.d(key, "[search %s]: get", key.to_c_str());
-    restbed::Uri uri(HTTP_PROTO + serverHost_ + "/" + key.toString());
+    restbed::Uri uri(proxy::HTTP_PROTO + serverHost_ + "/" + key.toString());
     auto req = std::make_shared<restbed::Request>(uri);
     Value::Filter filter = w.empty() ? f : f.chain(w.getFilter());
 
@@ -323,7 +320,7 @@ 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() + OP_TIMEOUT;
+        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]{
             auto s = searches_.find(key);
@@ -334,7 +331,7 @@ DhtProxyClient::put(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_po
                 return;
             const auto& now = scheduler.time();
             doPut(key, p->second.value, {}, now, true);
-            scheduler.edit(p->second.refreshJob, now + OP_TIMEOUT);
+            scheduler.edit(p->second.refreshJob, now + proxy::OP_TIMEOUT);
         })});
     }
     doPut(key, val, std::move(cb), created, permanent);
@@ -344,13 +341,24 @@ void
 DhtProxyClient::doPut(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_point created, bool permanent)
 {
     DHT_LOG.d(key, "[search %s] performing put of %s", key.to_c_str(), val->toString().c_str());
-    restbed::Uri uri(HTTP_PROTO + serverHost_ + "/" + key.toString());
+    restbed::Uri uri(proxy::HTTP_PROTO + serverHost_ + "/" + key.toString());
     auto req = std::make_shared<restbed::Request>(uri);
     req->set_method("POST");
 
     auto json = val->toJson();
-    if (permanent)
-        json["permanent"] = true;
+    if (permanent) {
+        if (deviceKey_.empty()) {
+            json["permanent"] = true;
+        } else {
+#if OPENDHT_PUSH_NOTIFICATIONS
+            Json::Value refresh;
+            getPushRequest(refresh);
+            json["permanent"] = refresh;
+#else
+            json["permanent"] = true;
+#endif
+        }
+    }
     Json::StreamWriterBuilder wbuilder;
     wbuilder["commentStyle"] = "None";
     wbuilder["indentation"] = "";
@@ -495,7 +503,7 @@ DhtProxyClient::getProxyInfos()
                 // See https://github.com/Corvusoft/restbed/issues/290.
                 server = serverHost;
             }
-            restbed::Uri uri(HTTP_PROTO + server + "/");
+            restbed::Uri uri(proxy::HTTP_PROTO + server + "/");
             auto req = std::make_shared<restbed::Request>(uri);
             reqs.emplace_back(restbed::Http::async(req,
             [this, resolved_proxy, successCount](
@@ -627,7 +635,7 @@ size_t
 DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter filter/*, Where where*/)
 {
     scheduler.syncTime();
-    restbed::Uri uri(HTTP_PROTO + serverHost_ + "/" + key.toString());
+    restbed::Uri uri(proxy::HTTP_PROTO + serverHost_ + "/" + key.toString());
     auto req = std::make_shared<restbed::Request>(uri);
     req->set_method(deviceKey_.empty() ? "LISTEN" : "SUBSCRIBE");
 
@@ -685,7 +693,7 @@ DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter fi
         scheduler.edit(l->second.cacheExpirationJob, next);
         return true;
     };
-    auto pushNotifToken = std::make_shared<unsigned>(0);
+    auto pushNotifToken = std::make_shared<proxy::ListenToken>(0);
     auto vcb = l->second.cb;
     l->second.pushNotifToken = pushNotifToken;
     l->second.thread = std::thread([this,req,filter,vcb,pushNotifToken,state]()
@@ -786,7 +794,7 @@ DhtProxyClient::doCancelListen(const InfoHash& key, size_t ltoken)
             listener.thread.join();
         }
         // UNSUBSCRIBE
-        restbed::Uri uri(HTTP_PROTO + serverHost_ + "/" + key.toString());
+        restbed::Uri uri(proxy::HTTP_PROTO + serverHost_ + "/" + key.toString());
         auto req = std::make_shared<restbed::Request>(uri);
         req->set_method("UNSUBSCRIBE");
         // fill request body
@@ -857,7 +865,7 @@ DhtProxyClient::restartListeners()
         // Redo listen
         auto filter = listener.filter;
         auto cb = listener.cb;
-        restbed::Uri uri(HTTP_PROTO + serverHost_ + "/" + search.first.toString());
+        restbed::Uri uri(proxy::HTTP_PROTO + serverHost_ + "/" + search.first.toString());
         auto req = std::make_shared<restbed::Request>(uri);
         req->set_method("LISTEN");
         listener.req = req;
@@ -923,29 +931,38 @@ void
 DhtProxyClient::pushNotificationReceived(const std::map<std::string, std::string>& notification)
 {
 #if OPENDHT_PUSH_NOTIFICATIONS
+    scheduler.syncTime();
     try {
         std::lock_guard<std::mutex> lock(searchLock_);
-        bool timeout = notification.find("timeout") == notification.cend();
-        uint64_t token = std::stoull(notification.at("token"));
-        for (auto& search: searches_) {
-            for (auto& list : search.second.listeners) {
-                auto& listener = list.second;
-                if (*listener.pushNotifToken!= token)
-                    continue;
-                DHT_LOG.d(search.first, "[search %s] handling push notification", search.first.to_c_str());
-                if (timeout) {
-                    // Wake up daemon and get values
-                    auto cb = listener.cb;
-                    auto filter = listener.filter;
+        auto timeout = notification.find("timeout");
+        if (timeout != notification.cend()) {
+            InfoHash key(timeout->second);
+            auto& search = searches_.at(key);
+            auto vidIt = notification.find("vid");
+            if (vidIt != notification.end()) {
+                // Refresh put
+                auto vid = std::stoull(vidIt->second);
+                auto put = search.puts.at(vid);
+                scheduler.edit(put.refreshJob, scheduler.time() + proxy::OP_TIMEOUT);
+            } else {
+                // Refresh listen
+                for (auto& list : search.listeners)
+                    resubscribe(key, list.second);
+            }
+        } else {
+            auto token = std::stoull(notification.at("token"));
+            for (auto& search: searches_) {
+                for (auto& list : search.second.listeners) {
+                    if (*list.second.pushNotifToken != token)
+                        continue;
+                    DHT_LOG.d(search.first, "[search %s] handling push notification", search.first.to_c_str());
+                    auto cb = list.second.cb;
+                    auto filter = list.second.filter;
                     get(search.first, [cb](const std::vector<Sp<Value>>& vals){
                         cb(vals, false);
                         return true;
                     }, DoneCallbackSimple{}, std::move(filter));
-                } else {
-                    // A timeout has occured, we need to relaunch the listener
-                    resubscribe(search.first, listener);
                 }
-                break;
             }
         }
     } catch (const std::exception& e) {
@@ -961,11 +978,11 @@ DhtProxyClient::resubscribe(const InfoHash& key, Listener& listener)
     if (deviceKey_.empty()) return;
     DHT_LOG.d(key, "[search %s] resubscribe push listener", key.to_c_str());
     // Subscribe
-    restbed::Uri uri(HTTP_PROTO + serverHost_ + "/" + key.toString());
+    restbed::Uri uri(proxy::HTTP_PROTO + serverHost_ + "/" + key.toString());
     auto req = std::make_shared<restbed::Request>(uri);
     req->set_method("SUBSCRIBE");
 
-    auto pushNotifToken = std::make_shared<unsigned>(0);
+    auto pushNotifToken = std::make_shared<proxy::ListenToken>(0);
 
     if (listener.thread.joinable())
         listener.thread.join();
@@ -1008,6 +1025,19 @@ DhtProxyClient::resubscribe(const InfoHash& key, Listener& listener)
 }
 
 #if OPENDHT_PUSH_NOTIFICATIONS
+void
+DhtProxyClient::getPushRequest(Json::Value& body) const
+{
+    body["key"] = deviceKey_;
+    body["client_id"] = pushClientId_;
+#ifdef __ANDROID__
+    body["platform"] = "android";
+#endif
+#ifdef __APPLE__
+    body["platform"] = "apple";
+#endif
+}
+
 void
 DhtProxyClient::fillBodyToGetToken(std::shared_ptr<restbed::Request> req, unsigned token)
 {
@@ -1017,16 +1047,9 @@ DhtProxyClient::fillBodyToGetToken(std::shared_ptr<restbed::Request> req, unsign
     //   "token": xxx
     // }
     Json::Value body;
-    body["key"] = deviceKey_;
-    body["client_id"] = pushClientId_;
+    getPushRequest(body);
     if (token > 0)
         body["token"] = token;
-#ifdef __ANDROID__
-    body["platform"] = "android";
-#endif
-#ifdef __APPLE__
-    body["platform"] = "apple";
-#endif
     Json::StreamWriterBuilder wbuilder;
     wbuilder["commentStyle"] = "None";
     wbuilder["indentation"] = "";

src/dht_proxy_server.cpp

@@ -1,6 +1,7 @@
 /*
  *  Copyright (C) 2017-2018 Savoir-faire Linux Inc.
  *  Author: Sébastien Blin <sebastien.blin@savoirfairelinux.com>
+ *          Adrien Béraud <adrien.beraud@savoirfairelinux.com>
  *
  *  This program is free software; you can redistribute it and/or modify
  *  it under the terms of the GNU General Public License as published by
@@ -29,18 +30,16 @@
 #include <functional>
 #include <limits>
 
-constexpr const std::chrono::seconds OP_TIMEOUT {1 * 60 * 60}; // one hour
-#if OPENDHT_PUSH_NOTIFICATIONS
-constexpr const char* const HTTP_PROTO {"http://"}; // TODO, https for prod
-#endif //OPENDHT_PUSH_NOTIFICATIONS
-
 using namespace std::placeholders;
 
 namespace dht {
 
 struct DhtProxyServer::PermanentPut {
     time_point expiration;
+    std::string pushToken;
+    std::string clientId;
     Sp<Scheduler::Job> expireJob;
+    Sp<Scheduler::Job> expireNotifyJob;
 };
 struct DhtProxyServer::SearchPuts {
     std::map<dht::Value::Id, PermanentPut> puts;
@@ -304,7 +303,8 @@ DhtProxyServer::listen(const std::shared_ptr<restbed::Session>& session)
 #if OPENDHT_PUSH_NOTIFICATIONS
 
 struct DhtProxyServer::Listener {
-    unsigned token;
+    std::shared_ptr<proxy::ListenToken> token;
+    std::string clientId;
     std::future<size_t> internalToken;
     Sp<Scheduler::Job> expireJob;
 };
@@ -346,56 +346,58 @@ DhtProxyServer::subscribe(const std::shared_ptr<restbed::Session>& session)
                 auto isAndroid = platform == "android";
                 auto clientId = root.isMember("client_id") ? root["client_id"].asString() : std::string();
 
-                uint64_t token = 0;
                 {
                     std::lock_guard<std::mutex> lock(lockListener_);
                     // Check if listener is already present and refresh timeout if launched
-
+                    // One push listener per pushToken.infoHash.clientId
                     auto pushListener = pushListeners_.emplace(pushToken, PushListener{}).first;
                     auto listeners = pushListener->second.listeners.emplace(infoHash, std::vector<Listener>{}).first;
                     for (auto& listener: listeners->second) {
-                        if (listener.token == tokenFromReq) {
+                        if (listener.clientId == clientId) {
+                            *listener.token = tokenFromReq;
                             {
                                 std::lock_guard<std::mutex> l(schedulerLock_);
-                                scheduler_.edit(listener.expireJob, scheduler_.time() + OP_TIMEOUT);
+                                scheduler_.edit(listener.expireJob, scheduler_.time() + proxy::OP_TIMEOUT);
                             }
-                            s->close(restbed::OK, "{\"token\": " + std::to_string(listener.token) + "}\n");
+                            s->close(restbed::OK, "{\"token\": " + std::to_string(tokenFromReq) + "}\n");
                             schedulerCv_.notify_one();
                             return;
                         }
                     }
+                    listeners->second.emplace_back(Listener{});
+                    auto& listener = listeners->second.back();
+
+                    // New listener
                     pushListener->second.isAndroid = isAndroid;
 
                     // The listener is not found, so add it.
-                    ++tokenPushNotif_;
-                    token = tokenPushNotif_;
-                    Listener listener;
+                    auto token = std::make_shared<proxy::ListenToken>(tokenFromReq);
                     listener.token = token;
                     listener.internalToken = dht_->listen(infoHash,
-                        [this, pushToken, token, isAndroid, clientId](std::vector<std::shared_ptr<Value>> /*value*/) {
+                        [this, token, infoHash, pushToken, isAndroid, clientId](std::vector<std::shared_ptr<Value>> /*value*/) {
                             // Build message content.
                             Json::Value json;
+                            json["key"] = infoHash.toString();
                             json["to"] = clientId;
-                            json["token"] = std::to_string(token);
+                            json["token"] = std::to_string(*token);
                             sendPushNotification(pushToken, json, isAndroid);
                             return true;
                         }
                     );
                     std::lock_guard<std::mutex> l(schedulerLock_);
-                    listener.expireJob = scheduler_.add(scheduler_.time() + OP_TIMEOUT,
-                        [this, pushToken, infoHash, token, clientId, isAndroid] {
-                            cancelPushListen(pushToken, infoHash, token);
+                    listener.expireJob = scheduler_.add(scheduler_.time() + proxy::OP_TIMEOUT,
+                        [this, token, infoHash, pushToken, isAndroid, clientId] {
+                            cancelPushListen(pushToken, infoHash, *token);
                             // Send a push notification to inform the client that this listen has timeout
                             Json::Value json;
                             json["timeout"] = infoHash.toString();
                             json["to"] = clientId;
-                            json["token"] = std::to_string(token);
+                            json["token"] = std::to_string(*token);
                             sendPushNotification(pushToken, json, isAndroid);
                         }
                     );
-                    listeners->second.emplace_back(std::move(listener));
                 }
-                s->close(restbed::OK, "{\"token\": " + std::to_string(token) + "}\n");
+                s->close(restbed::OK, "{\"token\": " + std::to_string(tokenFromReq) + "}\n");
             } catch (...) {
                 // do nothing
             }
@@ -443,7 +445,7 @@ DhtProxyServer::unsubscribe(const std::shared_ptr<restbed::Session>& session)
 }
 
 void
-DhtProxyServer::cancelPushListen(const std::string& pushToken, const dht::InfoHash& key, unsigned token)
+DhtProxyServer::cancelPushListen(const std::string& pushToken, const dht::InfoHash& key, proxy::ListenToken token)
 {
     std::lock_guard<std::mutex> lock(lockListener_);
     auto pushListener = pushListeners_.find(pushToken);
@@ -453,7 +455,7 @@ DhtProxyServer::cancelPushListen(const std::string& pushToken, const dht::InfoHa
     if (listeners == pushListener->second.listeners.end())
         return;
     for (auto listener = listeners->second.begin(); listener != listeners->second.end();) {
-        if (listener->token == token) {
+        if (*listener->token == token) {
             if (dht_)
                 dht_->cancelListen(key, std::move(listener->internalToken));
             listener = listeners->second.erase(listener);
@@ -472,7 +474,7 @@ DhtProxyServer::cancelPushListen(const std::string& pushToken, const dht::InfoHa
 void
 DhtProxyServer::sendPushNotification(const std::string& token, const Json::Value& json, bool isAndroid) const
 {
-    restbed::Uri uri(HTTP_PROTO + pushServer_ + "/api/push");
+    restbed::Uri uri(proxy::HTTP_PROTO + pushServer_ + "/api/push");
     auto req = std::make_shared<restbed::Request>(uri);
     req->set_method("POST");
 
@@ -522,6 +524,8 @@ DhtProxyServer::cancelPut(const InfoHash& key, Value::Id vid)
         return;
     if (dht_)
         dht_->cancelPut(key, vid);
+    if (put->second.expireNotifyJob)
+        put->second.expireNotifyJob->cancel();
     sPutsMap.erase(put);
     if (sPutsMap.empty())
         puts_.erase(sPuts);
@@ -557,21 +561,43 @@ DhtProxyServer::put(const std::shared_ptr<restbed::Session>& session)
                     if (reader->parse(char_data, char_data + strJson.size(), &root, &err)) {
                         // Build the Value from json
                         auto value = std::make_shared<Value>(root);
-                        auto permanent = root.isMember("permanent") ? root["permanent"].asBool() : false;
+                        auto permanent = root.isMember("permanent");
 
                         if (permanent) {
+                            std::string pushToken, clientId, platform;
+                            auto& pVal = root["permanent"];
+                            if (pVal.isObject()) {
+                                pushToken = pVal["key"].asString();
+                                clientId = pVal["client_id"].asString();
+                                platform = pVal["platform"].asString();
+                            }
+                            bool isAndroid = platform == "android";
                             std::unique_lock<std::mutex> lock(schedulerLock_);
                             scheduler_.syncTime();
                             auto sPuts = puts_.emplace(infoHash, SearchPuts{}).first;
-                            auto timeout = scheduler_.time() + OP_TIMEOUT;
+                            auto timeout = scheduler_.time() + proxy::OP_TIMEOUT;
                             auto vid = value->id;
                             auto r = sPuts->second.puts.emplace(vid, PermanentPut{});
                             if (r.second) {
                                 r.first->second.expireJob = scheduler_.add(timeout, [this, infoHash, vid]{
                                     cancelPut(infoHash, vid);
                                 });
+#if OPENDHT_PUSH_NOTIFICATIONS
+                                if (not pushToken.empty())
+                                    r.first->second.expireNotifyJob = scheduler_.add(timeout - proxy::OP_MARGIN,
+                                        [this, infoHash, vid, pushToken, clientId, isAndroid]
+                                    {
+                                        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);
                             }
                             lock.unlock();
                             schedulerCv_.notify_one();