diff --git a/include/opendht/dht_proxy_client.h b/include/opendht/dht_proxy_client.h index 0d59c754..5843f45a 100644 --- a/include/opendht/dht_proxy_client.h +++ b/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 request, unsigned token = 0); + void getPushRequest(Json::Value&) const; #endif // OPENDHT_PUSH_NOTIFICATIONS bool isDestroying_ {false}; diff --git a/include/opendht/dht_proxy_server.h b/include/opendht/dht_proxy_server.h index 9371b0cc..a83b07ed 100644 --- a/include/opendht/dht_proxy_server.h +++ b/include/opendht/dht_proxy_server.h @@ -1,6 +1,7 @@ /* * Copyright (C) 2017-2018 Savoir-faire Linux Inc. * Author: Sébastien Blin + * Adrien Béraud * * 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 #include @@ -179,6 +181,12 @@ private: */ void handleOptionsMethod(const std::shared_ptr& 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 puts_; + mutable std::atomic requestNum_ {0}; + mutable std::atomic lastStatsReset_ {time_point::min()}; + + const std::string pushServer_; + #if OPENDHT_PUSH_NOTIFICATIONS struct Listener; struct PushListener; std::mutex lockPushListeners_; std::map 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 requestNum_ {0}; - mutable std::atomic 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); }; } diff --git a/include/opendht/proxy.h b/include/opendht/proxy.h new file mode 100644 index 00000000..103e131c --- /dev/null +++ b/include/opendht/proxy.h @@ -0,0 +1,31 @@ +/* + * Copyright (C) 2017-2018 Savoir-faire Linux Inc. + * Author: Adrien Béraud + * + * 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 . + */ +#pragma once + +#include + +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; + +} +} diff --git a/src/dht_proxy_client.cpp b/src/dht_proxy_client.cpp index 0f436e32..6217c6b3 100644 --- a/src/dht_proxy_client.cpp +++ b/src/dht_proxy_client.cpp @@ -31,9 +31,6 @@ #include #include -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; - Sp pushNotifToken; // NOTE: unused if not using push notifications + Sp pushNotifToken; Sp refreshJob; Listener(ValueCache&& c, Sp&& j, const Sp& 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(uri); Value::Filter filter = w.empty() ? f : f.chain(w.getFilter()); @@ -323,7 +320,7 @@ DhtProxyClient::put(const InfoHash& key, Sp val, DoneCallback cb, time_po std::lock_guard 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 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 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(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(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(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(0); + auto pushNotifToken = std::make_shared(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(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(uri); req->set_method("LISTEN"); listener.req = req; @@ -923,29 +931,38 @@ void DhtProxyClient::pushNotificationReceived(const std::map& notification) { #if OPENDHT_PUSH_NOTIFICATIONS + scheduler.syncTime(); try { std::lock_guard 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>& 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(uri); req->set_method("SUBSCRIBE"); - auto pushNotifToken = std::make_shared(0); + auto pushNotifToken = std::make_shared(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 req, unsigned token) { @@ -1017,16 +1047,9 @@ DhtProxyClient::fillBodyToGetToken(std::shared_ptr 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"] = ""; diff --git a/src/dht_proxy_server.cpp b/src/dht_proxy_server.cpp index 9bf1a9db..bbec69ca 100644 --- a/src/dht_proxy_server.cpp +++ b/src/dht_proxy_server.cpp @@ -1,6 +1,7 @@ /* * Copyright (C) 2017-2018 Savoir-faire Linux Inc. * Author: Sébastien Blin + * Adrien Béraud * * 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 #include -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 expireJob; + Sp expireNotifyJob; }; struct DhtProxyServer::SearchPuts { std::map puts; @@ -304,7 +303,8 @@ DhtProxyServer::listen(const std::shared_ptr& session) #if OPENDHT_PUSH_NOTIFICATIONS struct DhtProxyServer::Listener { - unsigned token; + std::shared_ptr token; + std::string clientId; std::future internalToken; Sp expireJob; }; @@ -346,56 +346,58 @@ DhtProxyServer::subscribe(const std::shared_ptr& session) auto isAndroid = platform == "android"; auto clientId = root.isMember("client_id") ? root["client_id"].asString() : std::string(); - uint64_t token = 0; { std::lock_guard 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{}).first; for (auto& listener: listeners->second) { - if (listener.token == tokenFromReq) { + if (listener.clientId == clientId) { + *listener.token = tokenFromReq; { std::lock_guard 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(tokenFromReq); listener.token = token; listener.internalToken = dht_->listen(infoHash, - [this, pushToken, token, isAndroid, clientId](std::vector> /*value*/) { + [this, token, infoHash, pushToken, isAndroid, clientId](std::vector> /*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 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& 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 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(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& session) if (reader->parse(char_data, char_data + strJson.size(), &root, &err)) { // Build the Value from json auto value = std::make_shared(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 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();