diff --git a/include/opendht/dht_proxy_client.h b/include/opendht/dht_proxy_client.h index cb0880e4..71d30027 100644 --- a/include/opendht/dht_proxy_client.h +++ b/include/opendht/dht_proxy_client.h @@ -281,7 +281,7 @@ private: struct ListenState; void sendListen(const std::shared_ptr& request, const ValueCallback&, const Value::Filter& filter, const Sp& state); - void sendSubscribe(const std::shared_ptr& request, const Sp&, const Sp& state); + void sendSubscribe(const std::shared_ptr& request, const Sp& state); void doPut(const InfoHash&, Sp, DoneCallback, time_point created, bool permanent); @@ -373,7 +373,7 @@ private: const std::function loopSignal_; #if OPENDHT_PUSH_NOTIFICATIONS - void fillBodyToGetToken(std::shared_ptr request, unsigned token = 0); + void fillBody(std::shared_ptr request); void getPushRequest(Json::Value&) const; #endif // OPENDHT_PUSH_NOTIFICATIONS diff --git a/src/dht_proxy_client.cpp b/src/dht_proxy_client.cpp index 4cd33173..49640949 100644 --- a/src/dht_proxy_client.cpp +++ b/src/dht_proxy_client.cpp @@ -53,7 +53,6 @@ struct DhtProxyClient::Listener std::thread thread; unsigned callbackId; Sp state; - Sp pushNotifToken; Sp refreshJob; Listener(ValueCache&& c, Sp&& j, const Sp& r, Value::Filter&& f) : cache(std::move(c)), @@ -751,31 +750,19 @@ DhtProxyClient::sendListen(const std::shared_ptr& req, const V } void -DhtProxyClient::sendSubscribe(const std::shared_ptr& req, const Sp& token, const Sp& state) { +DhtProxyClient::sendSubscribe(const std::shared_ptr& req, const Sp& state) { #if OPENDHT_PUSH_NOTIFICATIONS req->set_method("SUBSCRIBE"); try { - fillBodyToGetToken(req); - restbed::Http::async(req, [this,state, token](const std::shared_ptr&, + fillBody(req); + restbed::Http::async(req, [this,state](const std::shared_ptr&, const std::shared_ptr& reply) { auto code = reply->get_status_code(); if (code == 200) { try { restbed::Http::fetch("\n", reply); - std::string body; - reply->get_body(body); - - std::string err; - Json::Value json; - Json::CharReaderBuilder rbuilder; - auto reader = std::unique_ptr(rbuilder.newCharReader()); - if (reader->parse(body.data(), body.data() + body.size(), &json, &err)) { - if (not json.isMember("token")) { - state->ok = false; - return; - } - *token = unpackId(json, "token"); - } + auto code = reply->get_status_code(); + state->ok = code == 200; } catch (const std::exception& e) { if (not state->cancel) { DHT_LOG.e("sendSubscribe: error: %s", e.what()); @@ -860,9 +847,7 @@ DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter fi return true; }; - auto pushNotifToken = std::make_shared(0); auto vcb = l->second.cb; - l->second.pushNotifToken = pushNotifToken; l->second.req = req; if (not deviceKey_.empty()) { @@ -879,8 +864,8 @@ DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter fi } } }); - l->second.thread = std::thread([this, req,pushNotifToken, state](){ - sendSubscribe(req,pushNotifToken,state); + l->second.thread = std::thread([this, req, state](){ + sendSubscribe(req, state); }); } else { l->second.thread = std::thread([this, req, vcb, filter, state]{ @@ -920,7 +905,6 @@ DhtProxyClient::doCancelListen(const InfoHash& key, size_t ltoken) Json::Value body; body["key"] = deviceKey_; body["client_id"] = pushClientId_; - body["token"] = std::to_string(*listener.pushNotifToken); Json::StreamWriterBuilder wbuilder; wbuilder["commentStyle"] = "None"; wbuilder["indentation"] = ""; @@ -1054,10 +1038,12 @@ DhtProxyClient::pushNotificationReceived(const std::mapsecond); + for (auto& search: searches_) { for (auto& list : search.second.listeners) { - if (*list.second.pushNotifToken != token or list.second.state->cancel) + if (search.first != key || list.second.state->cancel) continue; DHT_LOG.d(search.first, "[search %s] handling push notification", search.first.to_c_str()); auto cb = list.second.cb; @@ -1087,17 +1073,15 @@ DhtProxyClient::resubscribe(const InfoHash& key, Listener& listener) auto req = std::make_shared(uri); req->set_method("SUBSCRIBE"); - auto pushNotifToken = std::make_shared(0); auto state = listener.state; if (listener.thread.joinable()) listener.thread.join(); state->cancel = false; 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]() { - sendSubscribe(req, pushNotifToken, state); + listener.thread = std::thread([this, req, state]() { + sendSubscribe(req, state); }); #endif } @@ -1117,17 +1101,14 @@ DhtProxyClient::getPushRequest(Json::Value& body) const } void -DhtProxyClient::fillBodyToGetToken(std::shared_ptr req, unsigned token) +DhtProxyClient::fillBody(std::shared_ptr req) { // Fill body with // { // "key":"device_key", - // "token": xxx // } Json::Value body; getPushRequest(body); - if (token > 0) - body["token"] = token; Json::StreamWriterBuilder wbuilder; wbuilder["commentStyle"] = "None"; wbuilder["indentation"] = ""; diff --git a/src/dht_proxy_server.cpp b/src/dht_proxy_server.cpp index e84e92ea..0f499b05 100644 --- a/src/dht_proxy_server.cpp +++ b/src/dht_proxy_server.cpp @@ -370,7 +370,6 @@ DhtProxyServer::listen(const Sp& session) #if OPENDHT_PUSH_NOTIFICATIONS struct DhtProxyServer::Listener { - std::shared_ptr token; std::string clientId; std::future internalToken; Sp expireJob; @@ -409,12 +408,11 @@ DhtProxyServer::subscribe(const std::shared_ptr& session) s->close(restbed::BAD_REQUEST, "{\"err\":\"No token\"}"); return; } - auto tokenFromReq = unpackId(root, "token"); auto platform = root["platform"].asString(); auto isAndroid = platform == "android"; auto clientId = root.isMember("client_id") ? root["client_id"].asString() : std::string(); - std::cout << "Subscribe " << infoHash << " token:" << tokenFromReq << " client:" << clientId << std::endl; + std::cout << "Subscribe " << infoHash << " client:" << clientId << std::endl; { std::lock(schedulerLock_, lockListener_); @@ -428,10 +426,9 @@ DhtProxyServer::subscribe(const std::shared_ptr& session) auto listeners = pushListener->second.listeners.emplace(infoHash, std::vector{}).first; for (auto& listener: listeners->second) { if (listener.clientId == clientId) { - *listener.token = tokenFromReq; scheduler_.edit(listener.expireJob, timeout); scheduler_.edit(listener.expireNotifyJob, timeout - proxy::OP_MARGIN); - s->close(restbed::OK, "{\"token\": " + std::to_string(tokenFromReq) + "}\n"); + s->close(restbed::OK, "{}\n"); schedulerCv_.notify_one(); return; } @@ -444,15 +441,12 @@ DhtProxyServer::subscribe(const std::shared_ptr& session) pushListener->second.isAndroid = isAndroid; // The listener is not found, so add it. - auto token = std::make_shared(tokenFromReq); - listener.token = token; listener.internalToken = dht_->listen(infoHash, - [this, token, infoHash, pushToken, isAndroid, clientId](std::vector> /*value*/) { + [this, 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); sendPushNotification(pushToken, json, isAndroid); return true; } @@ -463,18 +457,17 @@ DhtProxyServer::subscribe(const std::shared_ptr& session) } ); listener.expireNotifyJob = scheduler_.add(timeout - proxy::OP_MARGIN, - [this, token, infoHash, pushToken, isAndroid, clientId] { + [this, infoHash, pushToken, isAndroid, clientId] { std::cout << "Listener: sending refresh " << infoHash << std::endl; Json::Value json; json["timeout"] = infoHash.toString(); json["to"] = clientId; - json["token"] = std::to_string(*token); sendPushNotification(pushToken, json, isAndroid); } ); } schedulerCv_.notify_one(); - s->close(restbed::OK, "{\"token\": " + std::to_string(tokenFromReq) + "}\n"); + s->close(restbed::OK, "{}\n"); } catch (...) { s->close(restbed::INTERNAL_SERVER_ERROR, "{\"err\":\"Internal server error\"}"); }