dhtproxy: allow token refresh, set clientID (#6)

* add method to set the push token independently from other configuration,
  since the push token can be updated during runtime.
* allow to provide a "clientID" to be sent in push notifications,
  used to demultiplex push notifications to multiple OpenDHT nodes running
  on the same host. For instance Ring uses the accountID as the clientID
  to know which account received the push notification.
* use {"platform":"android"/"ios"} to communicate the platform instead of {"isAndroid":true/false}.

include/opendht/dht.h

@@ -71,11 +71,6 @@ public:
     Dht(int s, int s6, Config config);
     virtual ~Dht();
 
-
-#if OPENDHT_PROXY_CLIENT
-    void startProxy(const std::string&, const std::string&) {};
-#endif
-
     /**
      * Get the ID of the node.
      */

include/opendht/dht_interface.h

@@ -29,7 +29,10 @@ public:
     virtual ~DhtInterface() = default;
 
 #if OPENDHT_PROXY_CLIENT
-    virtual void startProxy(const std::string& host, const std::string& deviceKey = "") = 0;
+    //virtual void startProxy() {};
+#if OPENDHT_PUSH_NOTIFICATIONS
+    virtual void setPushNotificationToken(const std::string& token) {};
+#endif
 #endif
 
     // [[deprecated]]

include/opendht/dht_proxy_client.h

@@ -46,13 +46,13 @@ public:
      * and an ID for the node.
      * @param serverHost the proxy address
      */
-    explicit DhtProxyClient(const std::string& serverHost);
-    /**
-     * Start the connection with a server.
-     * @param serverHost the server address
-     * @param deviceKey if we use push notifications
-     */
-    void startProxy(const std::string& serverHost, const std::string& deviceKey = "");
+    explicit DhtProxyClient(const std::string& serverHost, const std::string& pushClientId = "");
+
+#if OPENDHT_PUSH_NOTIFICATIONS
+    virtual void setPushNotificationToken(const std::string& token) {
+        deviceKey_ = token;
+    }
+#endif
 
     virtual ~DhtProxyClient();
 
@@ -248,6 +248,12 @@ public:
 
 private:
     const ValueType NO_VALUE;
+
+    /**
+     * Start the connection with a server.
+     */
+    void startProxy();
+
     /**
      * Get informations from the proxy node
      * @return the JSON returned by the proxy
@@ -266,6 +272,7 @@ private:
      */
     void cancelAllOperations();
     std::string serverHost_;
+    std::string pushClientId_;
     NodeStatus statusIpv4_ {NodeStatus::Disconnected};
     NodeStatus statusIpv6_ {NodeStatus::Disconnected};
 

include/opendht/dht_proxy_server.h

@@ -210,13 +210,14 @@ private:
 
 #if OPENDHT_PUSH_NOTIFICATIONS
     struct PushListener {
-        std::string key;
+        std::string pushToken;
         InfoHash hash;
         unsigned token;
         std::future<size_t> internalToken;
         std::chrono::steady_clock::time_point deadline;
         bool started {false};
         unsigned callbackId {0};
+        std::string clientId {};
         bool isAndroid {true};
     };
     mutable std::mutex lockPushListeners_;

include/opendht/dhtrunner.h

@@ -304,6 +304,7 @@ public:
         bool threaded;
 #if OPENDHT_PROXY_CLIENT
         std::string proxy_server;
+        std::string push_node_id;
 #endif //OPENDHT_PROXY_CLIENT
     };
 
@@ -313,11 +314,7 @@ public:
      * @param threaded: If false, ::loop() must be called periodically. Otherwise a thread is launched.
      * @param cb: Optional callback to receive general state information.
      */
-    void run(in_port_t port, const crypto::Identity identity, bool threaded = false, NetId network = 0
-#if OPENDHT_PROXY_CLIENT
-    , const std::string& proxy_server = "127.0.0.1:8000"
-#endif //OPENDHT_PROXY_CLIENT
-) {
+    void run(in_port_t port, const crypto::Identity identity, bool threaded = false, NetId network = 0) {
         run(port, {
             /*.dht_config = */{
                 /*.node_config = */{
@@ -330,7 +327,8 @@ public:
             },
             /*.threaded = */threaded,
 #if OPENDHT_PROXY_CLIENT
-            /*.proxy_server = */proxy_server
+            /*.proxy_server = */"",
+            /*.push_node_id = */""
 #endif //OPENDHT_PROXY_CLIENT
         });
     }
@@ -378,21 +376,29 @@ public:
     void join();
 
 #if OPENDHT_PROXY_CLIENT
-    void setProxyServer(const std::string& url = "127.0.0.1:8000") {
+    void setProxyServer(const std::string& url = "127.0.0.1:8000", const std::string& pushNodeId = "") {
         config_.proxy_server = url;
+        config_.push_node_id = pushNodeId;
     }
+
     /**
      * Start or stop the proxy
      * @param proxify if we want to use the proxy
      * @param deviceKey non empty to enable push notifications
      */
-    void enableProxy(bool proxify, const std::string& deviceKey = "");
+    void enableProxy(bool proxify);
+
 #endif // OPENDHT_PROXY_CLIENT
 #if OPENDHT_PROXY_SERVER
     void forwardAllMessages(bool forward);
 #endif // OPENDHT_PROXY_SERVER
 
 #if OPENDHT_PUSH_NOTIFICATIONS
+    /**
+     * Updates the push notification device token
+     */
+    void setPushNotificationToken(const std::string& token);
+
     /**
      * Insert a push notification to process for OpenDHT
      */
@@ -442,6 +448,10 @@ private:
      */
     std::unique_ptr<SecureDht> dht_via_proxy_;
     Config config_;
+
+#if OPENDHT_PUSH_NOTIFICATIONS
+    std::string pushToken_;
+#endif
 #endif // OPENDHT_PROXY_CLIENT
     /**
      * Store current listeners and translates global tokens for each client.

include/opendht/securedht.h

@@ -294,10 +294,12 @@ public:
     }
 
 #if OPENDHT_PROXY_CLIENT
-    void startProxy(const std::string& host, const std::string& deviceKey = "") {
-        dht_->startProxy(host, deviceKey);
+#if OPENDHT_PUSH_NOTIFICATIONS
+    void setPushNotificationToken(const std::string& token = "") {
+        dht_->setPushNotificationToken(token);
     }
 #endif
+#endif
 
 #if OPENDHT_PROXY_SERVER
     void forwardAllMessages(bool forward) {

src/dht_proxy_client.cpp

@@ -31,12 +31,12 @@ constexpr const char* const HTTP_PROTO {"http://"};
 
 namespace dht {
 
-DhtProxyClient::DhtProxyClient(const std::string& serverHost)
-: serverHost_(serverHost), lockCurrentProxyInfos_(new std::mutex()),
+DhtProxyClient::DhtProxyClient(const std::string& serverHost, const std::string& pushClientId)
+: serverHost_(serverHost), pushClientId_(pushClientId), lockCurrentProxyInfos_(new std::mutex()),
   scheduler(DHT_LOG), currentProxyInfos_(new Json::Value())
 {
     if (!serverHost_.empty())
-        startProxy(serverHost_);
+        startProxy();
 }
 
 void
@@ -56,11 +56,9 @@ DhtProxyClient::confirmProxy()
 }
 
 void
-DhtProxyClient::startProxy(const std::string& serverHost, const std::string& deviceKey)
+DhtProxyClient::startProxy()
 {
-    serverHost_ = serverHost;
     if (serverHost_.empty()) return;
-    deviceKey_ = deviceKey;
     auto confirm_proxy_time = scheduler.time() + std::chrono::seconds(5);
     nextProxyConfirmation = scheduler.add(confirm_proxy_time, std::bind(&DhtProxyClient::confirmProxy, this));
     auto confirm_connectivity = scheduler.time() + std::chrono::seconds(5);
@@ -753,16 +751,17 @@ DhtProxyClient::fillBodyToGetToken(std::shared_ptr<restbed::Request> req)
     // }
     Json::Value body;
     body["key"] = deviceKey_;
+    body["client_id"] = pushClientId_;
     {
         std::lock_guard<std::mutex> lock(lockCallback_);
         callbackId_ += 1;
         body["callback_id"] = callbackId_;
     }
 #ifdef __ANDROID__
-    body["isAndroid"] = true;
+    body["platform"] = "android";
 #endif
 #ifdef __APPLE__
-    body["isAndroid"] = false;
+    body["platform"] = "apple";
 #endif
     Json::StreamWriterBuilder wbuilder;
     wbuilder["commentStyle"] = "None";

src/dht_proxy_server.cpp

@@ -284,17 +284,19 @@ DhtProxyServer::subscribe(const std::shared_ptr<restbed::Session>& session) cons
                     s->close(restbed::BAD_REQUEST, "{\"err\":\"Incorrect JSON\"}");
                     return;
                 }
-                auto userKey = root["key"].asString();
-                if (userKey.empty()) return;
+                auto pushToken = root["key"].asString();
+                if (pushToken.empty()) return;
                 auto callbackId = root.isMember("callback_id") ? root["callback_id"].asLargestUInt() : 0;
-                auto isAndroid = root.isMember("isAndroid") ? root["isAndroid"].asBool() : true;
+                auto platform = root["platform"].asString();
+                auto isAndroid = platform == "android";
+                auto clientId = root.isMember("client_id") ? root["client_id"].asString() : std::string();
 
                 auto token = 0;
                 {
                     std::lock_guard<std::mutex> lock(lockListener_);
                     // Check if listener is already present and refresh timeout if launched
                     for(auto& listener: pushListeners_) {
-                        if (listener.key == userKey && listener.hash == infoHash
+                        if (listener.pushToken == pushToken && listener.hash == infoHash
                         && listener.callbackId == callbackId) {
                             if (listener.started)
                                 listener.deadline = std::chrono::steady_clock::now()
@@ -307,11 +309,12 @@ DhtProxyServer::subscribe(const std::shared_ptr<restbed::Session>& session) cons
                     ++tokenPushNotif_;
                     token = tokenPushNotif_;
                     PushListener listener;
-                    listener.key = userKey;
+                    listener.pushToken = pushToken;
                     listener.hash = std::move(infoHash);
                     listener.token = token;
                     listener.started = false;
                     listener.callbackId = callbackId;
+                    listener.clientId = clientId;
                     listener.isAndroid = isAndroid;
                     pushListeners_.emplace_back(std::move(listener));
                 }
@@ -348,8 +351,8 @@ DhtProxyServer::unsubscribe(const std::shared_ptr<restbed::Session>& session) co
                     s->close(restbed::BAD_REQUEST, "{\"err\":\"Incorrect JSON\"}");
                     return;
                 }
-                auto userKey = root["key"].asString();
-                if (userKey.empty()) return;
+                auto pushToken = root["key"].asString();
+                if (pushToken.empty()) return;
                 auto token = root["token"].asLargestUInt();
                 if (token == 0) return;
                 auto callbackId = root.isMember("callback_id") ? root["callback_id"].asLargestUInt() : 0;
@@ -358,7 +361,7 @@ DhtProxyServer::unsubscribe(const std::shared_ptr<restbed::Session>& session) co
                 // Check if listener is already present and refresh timeout if launched
                 auto listener = pushListeners_.begin();
                 while (listener != pushListeners_.end()) {
-                    if (listener->key == userKey && listener->token == token
+                    if (listener->pushToken == pushToken && listener->token == token
                     && listener->hash == infoHash && listener->callbackId == callbackId) {
                         if (dht_ && listener->started)
                             dht_->cancelListen(listener->hash, std::move(listener->internalToken.get()));
@@ -415,23 +418,24 @@ DhtProxyServer::handlePushListeners()
     while (pushListener != pushListeners_.end()) {
         if (dht_ && !pushListener->started) {
             // Try to start unstarted listeners
-            auto key = pushListener->key;
+            auto key = pushListener->pushToken;
             auto token = pushListener->token;
             auto callbackId = pushListener->callbackId;
             auto isAndroid = pushListener->isAndroid;
-            auto internalToken = std::move(dht_->listen(pushListener->hash,
-                [this, key, callbackId, token, isAndroid](std::shared_ptr<Value> /*value*/) {
+            auto clientId = pushListener->clientId;
+            pushListener->internalToken = dht_->listen(pushListener->hash,
+                [this, key, callbackId, token, isAndroid, clientId](std::shared_ptr<Value> /*value*/) {
                     // Build message content.
                     Json::Value json;
                     if (callbackId > 0) {
                         json["callback_id"] = callbackId;
                     }
+                    json["to"] = clientId;
                     json["token"] = token;
                     sendPushNotification(key, json, isAndroid);
                     return true;
                 }
-            ));
-            pushListener->internalToken = std::move(internalToken);
+            );
             pushListener->deadline = std::chrono::steady_clock::now() + std::chrono::seconds(TIMEOUT);
             pushListener->started = true;
             pushListener++;
@@ -444,8 +448,9 @@ DhtProxyServer::handlePushListeners()
             if (pushListener->callbackId > 0) {
                 json["callback_id"] = pushListener->callbackId;
             }
+            json["to"] = pushListener->clientId;
             json["token"] = pushListener->token;
-            sendPushNotification(pushListener->key, json, pushListener->isAndroid);
+            sendPushNotification(pushListener->pushToken, json, pushListener->isAndroid);
             pushListener = pushListeners_.erase(pushListener);
         } else {
             pushListener++;

src/dhtrunner.cpp

@@ -89,8 +89,7 @@ DhtRunner::run(const SockAddr& local4, const SockAddr& local6, DhtRunner::Config
         rcv_thread.join();
     running = true;
 #if OPENDHT_PROXY_CLIENT
-    config_.dht_config = config.dht_config;
-    config_.threaded = config.threaded;
+    config_ = config;
 #endif //OPENDHT_PROXY_CLIENT
     doRun(local4, local6, config.dht_config);
     if (not config.threaded)
@@ -436,7 +435,7 @@ DhtRunner::doRun(const SockAddr& sin4, const SockAddr& sin6, SecureDht::Config c
 #if OPENDHT_PROXY_CLIENT
     if (!dht_via_proxy_) {
         auto dht_via_proxy = std::unique_ptr<DhtInterface>(
-            new DhtProxyClient(config_.proxy_server)
+            new DhtProxyClient(config_.proxy_server, config_.push_node_id)
         );
         dht_via_proxy_ = std::unique_ptr<SecureDht>(new SecureDht(std::move(dht_via_proxy), config_.dht_config));
     }
@@ -861,16 +860,20 @@ DhtRunner::activeDht() const
 
 #if OPENDHT_PROXY_CLIENT
 void
-DhtRunner::enableProxy(bool proxify, const std::string& deviceKey) {
-    if (!dht_via_proxy_) {
-        auto dht_via_proxy = std::unique_ptr<DhtInterface>(
-            new DhtProxyClient(config_.proxy_server)
-        );
-        dht_via_proxy_ = std::unique_ptr<SecureDht>(new SecureDht(std::move(dht_via_proxy), config_.dht_config));
+DhtRunner::enableProxy(bool proxify)
+{
+    if (dht_via_proxy_) {
+        dht_via_proxy_->shutdown({});
     }
     if (proxify) {
         // Init the proxy client
-        dht_via_proxy_->startProxy(config_.proxy_server, deviceKey);
+        auto dht_via_proxy = std::unique_ptr<DhtInterface>(
+            new DhtProxyClient(config_.proxy_server, config_.push_node_id)
+        );
+        dht_via_proxy_ = std::unique_ptr<SecureDht>(new SecureDht(std::move(dht_via_proxy), config_.dht_config));
+        if (not pushToken_.empty())
+            dht_via_proxy_->setPushNotificationToken(pushToken_);
+        //dht_via_proxy_->startProxy();
         // add current listeners
         for (auto& listener: listeners_) {
             auto tokenProxy = dht_via_proxy_->listen(listener->hash, listener->gcb, std::move(listener->f), std::move(listener->w));
@@ -881,9 +884,6 @@ DhtRunner::enableProxy(bool proxify, const std::string& deviceKey) {
     } else {
         use_proxy = proxify;
         loop_(); // Restart the classic DHT.
-        // We doesn't need to maintain the connection with the proxy.
-        // Delete it
-        dht_via_proxy_->shutdown({});
         // update all proxyToken for all proxyListener
         auto it = listeners_.begin();
         for (; it != listeners_.end(); ++it) {
@@ -899,6 +899,17 @@ DhtRunner::enableProxy(bool proxify, const std::string& deviceKey) {
         }
     }
 }
+
+/**
+ * Updates the push notification device token
+ */
+void
+DhtRunner::setPushNotificationToken(const std::string& token) {
+    pushToken_ = token;
+    if (dht_via_proxy_)
+        dht_via_proxy_->setPushNotificationToken(token);
+}
+
 #endif // OPENDHT_PROXY_CLIENT
 
 #if OPENDHT_PROXY_SERVER

tools/dhtnode.cpp

@@ -220,8 +220,10 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params
 #if OPENDHT_PROXY_CLIENT
         else if (op == "stt") {
             iss >> idstr >> deviceKey;
+            if (not deviceKey.empty())
+                dht->setPushNotificationToken(deviceKey);
             dht->setProxyServer(idstr);
-            dht->enableProxy(true, deviceKey);
+            dht->enableProxy(true);
             continue;
         } else if (op == "stp") {
             dht->enableProxy(false);
@@ -483,6 +485,13 @@ main(int argc, char **argv)
         }
 
         dht->run(params.port, crt, true, params.network);
+#if OPENDHT_PROXY_CLIENT
+        if (!params.proxyclient.empty()) {
+            dht->setPushNotificationToken(params.devicekey);
+            dht->setProxyServer(params.proxyclient, "dhtnode");
+            dht->enableProxy(true);
+        }
+#endif //OPENDHT_PROXY_CLIENT
 
         if (params.log) {
             if (params.syslog or (params.daemonize and params.logfile.empty()))
@@ -508,12 +517,6 @@ main(int argc, char **argv)
                 ));
             }
 #endif //OPENDHT_PROXY_SERVER
-        #if OPENDHT_PROXY_CLIENT
-            if (!params.proxyclient.empty()) {
-                dht->setProxyServer(params.proxyclient);
-                dht->enableProxy(true, params.devicekey);
-            }
-        #endif //OPENDHT_PROXY_CLIENT
 
         if (params.daemonize or params.service)
             while (runner.wait());